bound pairing relay inbox with frame/inbox ceilings (PB-26-205, v2.14.98)

@@ -1 +1 @@1-2.14.971+2.14.98

@@ -619,6 +619,74 @@619619         .pingInterval(RELAY_PING_INTERVAL_SECONDS, TimeUnit.SECONDS)620620         .build()621621 622+    /**623+     * PB-26-205 pairing/adoption dial: same relay rooms as [connect], but the624+     * inbound path is bounded by the shared pairing ceilings625+     * ([MAX_PAIRING_FRAME_BYTES] per frame, [MAX_PAIRING_INBOX_BYTES] /626+     * [MAX_PAIRING_INBOX_MESSAGES] unconsumed) instead of the sync ceilings.627+     * Any violation closes with status 1009 and fails readers with628+     * [PairingResourceLimitException]. Pairing call sites must use this; the629+     * ongoing-sync paths keep [connect].630+     */631+    suspend fun connectPairing(url: String): RelayBinarySocket = withContext(Dispatchers.IO) {632+        val opened = CompletableDeferred<WebSocket>()633+        lateinit var inbox: RelayPairingInbox634+        var socketRef: WebSocket? = null635+        inbox = RelayPairingInbox { socketRef?.close(1009, "pairing frame too large") }636+        val listener = object : WebSocketListener() {637+            override fun onOpen(webSocket: WebSocket, response: Response) {638+                socketRef = webSocket639+                opened.complete(webSocket)640+            }641+642+            override fun onMessage(webSocket: WebSocket, bytes: ByteString) {643+                // The size check runs before toByteArray() allocates/copies.644+                inbox.onFrame(bytes.size) { bytes.toByteArray() }645+            }646+647+            override fun onMessage(webSocket: WebSocket, text: String) {648+                inbox.reject()649+            }650+651+            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {652+                val detail = reason.takeIf { it.isNotBlank() }653+                    ?.let(::relayLogDetail)654+                    ?: "no reason"655+                inbox.close(IllegalStateException("relay_closed_$code: $detail"))656+            }657+658+            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {659+                inbox.close(t)660+                opened.completeExceptionally(t)661+            }662+        }663+        val webSocket = client.newWebSocket(Request.Builder().url(url).build(), listener)664+        val connected = try {665+            withTimeout(RELAY_CONNECT_TIMEOUT_MS) { opened.await() }666+        } catch (error: Throwable) {667+            webSocket.cancel()668+            throw error669+        }670+        object : RelayBinarySocket {671+            override fun send(frame: ByteArray) {672+                if (frame.size > MAX_PAIRING_FRAME_BYTES) throw PairingResourceLimitException()673+                check(connected.send(frame.toByteString())) { "relay_send_failed" }674+            }675+676+            override suspend fun receive(): ByteArray = inbox.receive()677+678+            override fun close(code: Int, reason: String) {679+                inbox.close(IllegalStateException("relay_closed_$code: $reason"))680+                connected.close(code, reason)681+                connected.cancel()682+            }683+684+            override fun closeAfterQueuedFrames(code: Int, reason: String) {685+                connected.close(code, reason)686+            }687+        }688+    }689+622690     suspend fun connect(url: String): RelayBinarySocket = withContext(Dispatchers.IO) {623691         val opened = CompletableDeferred<WebSocket>()624692         val incoming = Channel<ByteArray>(RELAY_INBOX_CAPACITY)

@@ -1,9 +1,11 @@11 package com.peachpasswords.sync22 33 import kotlinx.coroutines.Dispatchers4+import kotlinx.coroutines.channels.Channel45 import kotlinx.coroutines.withContext56 import java.io.DataInputStream67 import java.io.EOFException8+import java.io.IOException79 import java.io.InputStream810 import java.io.OutputStream911 @@ -35,6 +37,90 @@3537  * gigabytes before the Noise layer has authenticated anything.3638  */3739 const val MAX_PAIRING_FRAME_BYTES: Int = 1 shl 20 // 1 MiB40+41+/**42+ * PB-26-205: relay pairing/adoption inbox backstop, mirrored from the43+ * browser (`pairing-dial.ts`). The 1 MiB frame ceiling clears the 768 KiB44+ * adoption frame ceiling with headroom; the 12 MiB / 256-message inbox45+ * clears the 10 MiB aggregate adoption budget plus handshake/control46+ * overhead. Both clients agree on accept/reject at these boundaries.47+ */48+const val MAX_PAIRING_INBOX_BYTES: Int = 12 * 1024 * 102449+const val MAX_PAIRING_INBOX_MESSAGES: Int = 25650+51+/** Terminal pairing error: an IOException so it lands in the ERRORED outcome like a transport failure. */52+class PairingResourceLimitException(53+    message: String = "pairing_resource_limit",54+) : IOException(message)55+56+/**57+ * Bounded relay inbox for one pairing/adoption session. The size check runs58+ * before [read] performs the allocate/copy, so an oversized frame never59+ * materializes. On any violation the channel closes with60+ * [PairingResourceLimitException] — every parked and future reader gets that61+ * one terminal error — [onLimitExceeded] fires once (the owner closes the62+ * socket with status 1009), and later frames are ignored.63+ */64+internal class RelayPairingInbox(private val onLimitExceeded: () -> Unit) {65+    private val channel = Channel<ByteArray>(Channel.UNLIMITED)66+    private var queuedBytes = 067+    private var queuedMessages = 068+    private var terminal: Throwable? = null69+70+    @Synchronized71+    fun onFrame(size: Int, read: () -> ByteArray): Boolean {72+        if (terminal != null) return false73+        if (74+            size > MAX_PAIRING_FRAME_BYTES ||75+            queuedMessages + 1 > MAX_PAIRING_INBOX_MESSAGES ||76+            queuedBytes + size > MAX_PAIRING_INBOX_BYTES77+        ) {78+            tripLimit()79+            return false80+        }81+        val frame = read()82+        queuedBytes += frame.size83+        queuedMessages += 184+        channel.trySend(frame)85+        return true86+    }87+88+    /** Trips the limit for a non-frame violation (e.g. a text frame). */89+    @Synchronized90+    fun reject(): Boolean {91+        if (terminal != null) return false92+        tripLimit()93+        return false94+    }95+96+    /** Benign socket close/failure: fails parked readers without a limit close. */97+    @Synchronized98+    fun close(cause: Throwable) {99+        if (terminal != null) return100+        terminal = cause101+        channel.close(cause)102+    }103+104+    private fun tripLimit() {105+        terminal = PairingResourceLimitException()106+        // A closed channel still delivers buffered elements first, so drop107+        // the queued refs for parity with the browser's cleared inbox.108+        while (channel.tryReceive().isSuccess) { /* drop */ }109+        queuedBytes = 0110+        queuedMessages = 0111+        channel.close(terminal)112+        onLimitExceeded()113+    }114+115+    suspend fun receive(): ByteArray {116+        val frame = channel.receive()117+        synchronized(this) {118+            queuedMessages -= 1119+            queuedBytes -= frame.size120+        }121+        return frame122+    }123+}38124 39125 /** Blocking write of a single length-prefixed frame. */40126 fun writeFrame(out: OutputStream, payload: ByteArray) {

@@ -63,7 +63,7 @@6363 6464         job = scope.launch(Dispatchers.IO) {6565             try {66-                val connected = OkHttpRelaySocket.connect(66+                val connected = OkHttpRelaySocket.connectPairing(6767                     Rendezvous.buildUrl(relayUrl, token),6868                 )6969                 socket = connected

@@ -33,7 +33,7 @@3333         require(invitation.vaultStatus == VaultStatus.POPULATED) {3434             "receiver adopt() expects a populated invitation; got ${invitation.vaultStatus}"3535         }36-        val socket = OkHttpRelaySocket.connect(invitation.roomUrl())36+        val socket = OkHttpRelaySocket.connectPairing(invitation.roomUrl())3737         return adoptFromRelaySocket(3838             socket = socket,3939             invitation = invitation,@@ -61,7 +61,7 @@6161         require(invitation.vaultStatus == VaultStatus.EMPTY) {6262             "grantFromInvitation() expects an empty invitation; got ${invitation.vaultStatus}"6363         }64-        val socket = OkHttpRelaySocket.connect(invitation.roomUrl())64+        val socket = OkHttpRelaySocket.connectPairing(invitation.roomUrl())6565         return grantFromRelaySocket(6666             socket = socket,6767             invitation = invitation,