fix(sync): limit what a LAN connection can hold before its hello is done

.github/nightly-notes/2026-11-07-fix-sync2-lan-limits.md

@@ -0,0 +1,2 @@
+## Fixed
+- Linked devices: connections on the local network that never finish connecting no longer tie up the app

app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt

@@ -38,6 +38,8 @@ class WsClient(
                     .writeTimeout(15, TimeUnit.SECONDS)
                     .connectTimeout(4, TimeUnit.SECONDS) // AP-isolation / firewall → fail fast
                     .pingInterval(20, TimeUnit.SECONDS)
+                    .followRedirects(false)
+                    .followSslRedirects(false)
                     .apply { if (limited) addInterceptor(MessageLimit(maxMessage)) }
                     .build()
             val incoming = Channel<ByteArray>(if (limited) QUEUED_MESSAGES else Channel.UNLIMITED)

app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt

@@ -40,15 +40,19 @@ class WsServer(
                     pingPeriodMillis = PING_PERIOD_MS
                     timeoutMillis = PING_TIMEOUT_MS
                     maxFrameSize = maxFrame
+                    // Ktor queues without a limit by default, so the socket would keep being read for nobody.
+                    channels { incoming = bounded(QUEUED_FRAMES) }
                 }
                 routing {
                     webSocket(path) {
                         val conn = KtorServerConnection(this, call.request.local.remoteAddress)
                         val handler = onConnection
                         if (handler != null) {
                             launch { handler(conn) }
-                        } else if (!connectionDeferred.isCompleted) {
-                            connectionDeferred.complete(conn)
+                        } else if (!connectionDeferred.complete(conn)) {
+                            // One session per server: nothing would ever read this connection.
+                            conn.close()
+                            return@webSocket
                         }
                         try {
                             for (frame in incoming) {
@@ -87,15 +91,18 @@ class WsServer(
     }
 }
 
+private const val QUEUED_FRAMES = 4
+
 /** Wraps a Ktor server [DefaultWebSocketSession] as a [SyncConnection]. */
 private class KtorServerConnection(
     private val session: DefaultWebSocketSession,
     override val remoteHost: String,
 ) : SyncConnection {
-    private val inbound = Channel<ByteArray>(Channel.UNLIMITED)
+    private val inbound = Channel<ByteArray>(QUEUED_FRAMES)
 
+    /** Waits while frames are queued, so the socket isn't read further until someone receives them. */
     suspend fun deliver(bytes: ByteArray) {
-        inbound.trySend(bytes)
+        inbound.send(bytes)
     }
 
     fun onClosed() {

app/src/main/java/io/github/aedev/flow/sync2/crypto/Payloads.kt

@@ -25,19 +25,26 @@ fun encodePayload(content: ByteArray): ByteArray {
     return payload.copyOf(paddedSize(payload.size))
 }
 
-fun decodePayload(payload: ByteArray): ByteArray {
+/** The content of [payload], failing as soon as it inflates past [limit] bytes. */
+fun decodePayload(
+    payload: ByteArray,
+    limit: Int = MAX_INFLATED,
+): ByteArray {
     if (payload.size < 4) reject(ProtocolError.PAYLOAD)
     val length = ByteBuffer.wrap(payload, 0, 4).int
     if (length < 0 || length > payload.size - 4) reject(ProtocolError.PAYLOAD)
     if ((4 + length until payload.size).any { payload[it] != 0.toByte() }) reject(ProtocolError.PAYLOAD)
     return try {
-        inflate(payload.inputStream(4, length))
+        inflate(payload.inputStream(4, length), limit)
     } catch (_: IOException) {
         reject(ProtocolError.PAYLOAD)
     }
 }
 
-private fun inflate(body: InputStream): ByteArray {
+private fun inflate(
+    body: InputStream,
+    limit: Int,
+): ByteArray {
     val out = ByteArrayOutputStream()
     val buffer = ByteArray(8192)
     GZIPInputStream(body).use { input ->
@@ -46,6 +53,7 @@ private fun inflate(body: InputStream): ByteArray {
             if (read < 0) break
             out.write(buffer, 0, read)
             if (out.size() > MAX_INFLATED) reject(ProtocolError.PAYLOAD)
+            if (out.size() > limit) reject(ProtocolError.LENGTH)
         }
     }
     return out.toByteArray()

app/src/main/java/io/github/aedev/flow/sync2/lan/LanGreeter.kt

@@ -0,0 +1,31 @@
+package io.github.aedev.flow.sync2.lan
+
+import io.github.aedev.flow.sync.transport.SyncConnection
+import io.github.aedev.flow.sync2.engine.LinkedException
+import io.github.aedev.flow.sync2.engine.Session
+import kotlinx.coroutines.sync.Semaphore
+
+/** Connections that haven't finished their hello. Far more than a group has members. */
+internal const val GREETING_CONNECTIONS = 16
+
+/**
+ * Takes connections that dialed this device through their hello, a limited number at once. Nothing is
+ * read from the roster until a hello arrives, so a connection that says nothing costs next to nothing.
+ */
+internal class LanGreeter(
+    private val session: suspend () -> Session?,
+) {
+    private val slots = Semaphore(GREETING_CONNECTIONS)
+
+    /** Whether [connection] has a slot, for [greet] to use. Without one it is closed. */
+    fun takes(connection: SyncConnection): Boolean = slots.tryAcquire().also { if (!it) connection.close() }
+
+    /** The link once the hello is done and the session it opened on, or null when this device isn't linked. */
+    suspend fun greet(connection: SyncConnection): Pair<LanLink, Session>? =
+        try {
+            val first = connection.receiveHello() ?: throw LinkedException.Unexpected()
+            session()?.let { acceptLink(connection, it, first) to it }
+        } finally {
+            slots.release()
+        }
+}

app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt

@@ -26,8 +26,10 @@ import io.github.aedev.flow.sync2.engine.Session
 import io.github.aedev.flow.sync2.roster.Role
 import io.github.aedev.flow.sync2.roster.b64
 import io.github.aedev.flow.sync2.store.MAX_OBJECT_SIZE
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withTimeout
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.JsonPrimitive
@@ -48,6 +50,12 @@ const val MAX_LAN_CONTENT = MAX_OBJECT_SIZE + 64 * 1024
 /** The largest frame a device accepts: that content after gzip, padding, box and `seq`. */
 const val MAX_LAN_FRAME = MAX_OBJECT_SIZE + 256 * 1024L
 
+/** The most a hello's content may hold: a device id, a nonce, a signature and a port. */
+private const val MAX_HELLO_CONTENT = 4 * 1024
+
+/** How long a side waits for each hello message. */
+private const val HELLO_MS = 10_000L
+
 /**
  * A LAN session between two members (spec/live.md, Session), once the hello has checked both: each
  * frame is `u64(seq) ‖ Box(K, "FLOW-SYNC/2 lan" ‖ u64(seq), payload)`, counted strictly in each direction.
@@ -76,8 +84,7 @@ class LanLink internal constructor(
     suspend fun receive(): ByteArray? {
         val frame = connection.receive() ?: return null
         if (frame.size < 8 || ByteBuffer.wrap(frame, 0, 8).long != received) reject(ProtocolError.REPLAY)
-        val content = decodePayload(open(receiveKey, lanFrameAad(received), frame.copyOfRange(8, frame.size)))
-        if (content.size > MAX_LAN_CONTENT) reject(ProtocolError.LENGTH)
+        val content = decodePayload(open(receiveKey, lanFrameAad(received), frame.copyOfRange(8, frame.size)), MAX_LAN_CONTENT)
         received++
         return content
     }
@@ -98,7 +105,7 @@ private fun opened(
     message: Int,
 ): JsonObject {
     if (frame == null || frame.size < 4) throw LinkedException.Unexpected()
-    val content = decodePayload(open(lanKey, lanHelloAad(message), frame.copyOfRange(4, frame.size)))
+    val content = decodePayload(open(lanKey, lanHelloAad(message), frame.copyOfRange(4, frame.size)), MAX_HELLO_CONTENT)
     return runCatching { Json.parseToJsonElement(content.toString(Charsets.UTF_8)).jsonObject }.getOrNull()
         ?: throw LinkedException.Unexpected()
 }
@@ -110,6 +117,14 @@ private fun JsonObject.nonce(): ByteArray = b64(field("n"), NONCE_SIZE) ?: throw
 
 private fun JsonObject.signature(): ByteArray = b64(field("sig"), SIGNATURE_SIZE) ?: reject(ProtocolError.SIGNATURE)
 
+/** The next hello message, or null once the other side closed. A timeout ends only this connection, unlike the caller's own cancellation. */
+internal suspend fun SyncConnection.receiveHello(): ByteArray? =
+    try {
+        withTimeout(HELLO_MS) { receive() }
+    } catch (_: TimeoutCancellationException) {
+        throw LinkedException.TimedOut()
+    }
+
 /** The signing key of a current device other than this one, or a rejection. */
 private fun Session.memberKey(did: String): ByteArray =
     state.members[did]?.takeIf { it.role == Role.DEVICE && did != me }?.sig ?: reject(ProtocolError.SENDER)
@@ -142,7 +157,7 @@ suspend fun connectLink(
         ),
     )
 
-    val second = opened(connection.receive() ?: throw LinkedException.LanUnreachable(), lanKey, 2)
+    val second = opened(connection.receiveHello() ?: throw LinkedException.LanUnreachable(), lanKey, 2)
     val serverDid = second.field("did")
     val serverKey = session.memberKey(serverDid)
     val server = SyncBytes.fromHex(serverDid)
@@ -182,16 +197,20 @@ suspend fun dialLink(
     return null
 }
 
-/** The accepting side: answers under the epoch the client named, if this device holds its key. */
+/**
+ * The accepting side: answers the hello in [first] under the epoch the client named, if this device holds its key.
+ * [first] is the connection's first frame, read here unless the caller already has it.
+ */
 suspend fun acceptLink(
     connection: SyncConnection,
     session: Session,
+    first: ByteArray? = null,
 ): LanLink {
-    val first = connection.receive() ?: throw LinkedException.Unexpected()
-    if (first.size < 4) throw LinkedException.Unexpected()
-    val epoch = ByteBuffer.wrap(first, 0, 4).int
+    val opening = first ?: connection.receiveHello() ?: throw LinkedException.Unexpected()
+    if (opening.size < 4) throw LinkedException.Unexpected()
+    val epoch = ByteBuffer.wrap(opening, 0, 4).int
     val lanKey = purposeKey(session.epochKey(epoch), session.group, epoch, Purpose.LAN)
-    val hello = opened(first, lanKey, 1)
+    val hello = opened(opening, lanKey, 1)
     val clientDid = hello.field("did")
     val clientKey = session.memberKey(clientDid)
     val client = SyncBytes.fromHex(clientDid)
@@ -209,7 +228,7 @@ suspend fun acceptLink(
         )
     connection.send(hello(epoch, lanKey, 2, second))
 
-    val third = opened(connection.receive(), lanKey, 3)
+    val third = opened(connection.receiveHello(), lanKey, 3)
     val transcript = lanTranscript(LanRole.CLIENT, clientNonce, serverNonce, client, server)
     if (!verify(clientKey, transcript, third.signature())) reject(ProtocolError.SIGNATURE)
     val keys = lanSessionKeys(lanKey, clientNonce, serverNonce, client, server)

app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt

@@ -212,7 +212,11 @@ class LanSync
                 val session = open()
                 val lanKey = purposeKey(session.currentKey(), session.group, session.state.epoch, Purpose.LAN)
                 // Outside the socket's scope: closing a link cancels that scope, and the session still has to be noted.
-                val server = WsServer(SESSION_PATH, MAX_LAN_FRAME) { connection -> launch { accept(connection) } }
+                val greeter = LanGreeter { session() }
+                val server =
+                    WsServer(SESSION_PATH, MAX_LAN_FRAME) { connection ->
+                        if (greeter.takes(connection)) launch { accept(connection, greeter) }
+                    }
                 val discovery = LanDiscovery(context)
                 val name = "Flow-" + SyncBytes.toHex(SyncCrypto.randomBytes(2))
                 names += name
@@ -301,8 +305,11 @@ class LanSync
             }
 
         /** A member that dialed this one is remembered where its hello says it listens. */
-        private suspend fun accept(connection: SyncConnection) {
-            val link = meet(connection) ?: return
+        private suspend fun accept(
+            connection: SyncConnection,
+            greeter: LanGreeter,
+        ) {
+            val link = meet(connection, greeter) ?: return
             val host = connection.remoteHost
             val port = link.peerPort
             if (host != null && port != null) remember(link.peer, LocalNetworkAddress.urlHost(host), port)
@@ -318,9 +325,12 @@ class LanSync
         }
 
         /** One session with the member that dialed this device over [connection]. Returns the closed link when it got that far. */
-        private suspend fun meet(connection: SyncConnection): LanLink? =
+        private suspend fun meet(
+            connection: SyncConnection,
+            greeter: LanGreeter,
+        ): LanLink? =
             try {
-                session()?.let { session -> talk(acceptLink(connection, session), session, "LAN session accepted") }
+                greeter.greet(connection)?.let { (link, session) -> talk(link, session, "LAN session accepted") }
             } catch (e: CancellationException) {
                 throw e
             } catch (e: Exception) {