From dab946da82f59428b978df72d188b83609281eea Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:17:31 +0200 Subject: [PATCH 1/2] fix(sync): limit what a LAN connection can hold before its hello is done --- .../2026-11-07-fix-sync2-lan-limits.md | 2 + .../aedev/flow/sync/transport/WsClient.kt | 2 + .../aedev/flow/sync/transport/WsServer.kt | 15 +- .../aedev/flow/sync2/crypto/Payloads.kt | 14 +- .../github/aedev/flow/sync2/lan/LanGreeter.kt | 31 +++ .../io/github/aedev/flow/sync2/lan/LanLink.kt | 39 ++- .../io/github/aedev/flow/sync2/lan/LanSync.kt | 20 +- .../github/aedev/flow/sync2/LanLimitsTest.kt | 261 ++++++++++++++++++ 8 files changed, 362 insertions(+), 22 deletions(-) create mode 100644 .github/nightly-notes/2026-11-07-fix-sync2-lan-limits.md create mode 100644 app/src/main/java/io/github/aedev/flow/sync2/lan/LanGreeter.kt create mode 100644 app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt diff --git a/.github/nightly-notes/2026-11-07-fix-sync2-lan-limits.md b/.github/nightly-notes/2026-11-07-fix-sync2-lan-limits.md new file mode 100644 index 000000000..e5f6a67bf --- /dev/null +++ b/.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 diff --git a/app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt b/app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt index 950426184..a2defd0ad 100644 --- a/app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt +++ b/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(if (limited) QUEUED_MESSAGES else Channel.UNLIMITED) diff --git a/app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt b/app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt index 1693caeea..706144531 100644 --- a/app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt +++ b/app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt @@ -40,6 +40,8 @@ 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) { @@ -47,8 +49,10 @@ class WsServer( 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(Channel.UNLIMITED) + private val inbound = Channel(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() { diff --git a/app/src/main/java/io/github/aedev/flow/sync2/crypto/Payloads.kt b/app/src/main/java/io/github/aedev/flow/sync2/crypto/Payloads.kt index 4d134a353..cac9d9758 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/crypto/Payloads.kt +++ b/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() diff --git a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanGreeter.kt b/app/src/main/java/io/github/aedev/flow/sync2/lan/LanGreeter.kt new file mode 100644 index 000000000..b314b4fde --- /dev/null +++ b/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? = + try { + val first = connection.receiveHello() ?: throw LinkedException.Unexpected() + session()?.let { acceptLink(connection, it, first) to it } + } finally { + slots.release() + } +} diff --git a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt b/app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt index 797dfa7ae..e7505a1ac 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt +++ b/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) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt b/app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt index 939cfcffc..8f50c84cf 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt +++ b/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) { diff --git a/app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt new file mode 100644 index 000000000..bc73a045c --- /dev/null +++ b/app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt @@ -0,0 +1,261 @@ +package io.github.aedev.flow.sync2 + +import android.app.Application +import androidx.test.ext.junit.runners.AndroidJUnit4 +import io.github.aedev.flow.sync.transport.SyncConnection +import io.github.aedev.flow.sync.transport.WsServer +import io.github.aedev.flow.sync2.crypto.ProtocolError +import io.github.aedev.flow.sync2.crypto.ProtocolException +import io.github.aedev.flow.sync2.crypto.Purpose +import io.github.aedev.flow.sync2.crypto.decodePayload +import io.github.aedev.flow.sync2.crypto.encodePayload +import io.github.aedev.flow.sync2.crypto.lanHelloAad +import io.github.aedev.flow.sync2.crypto.purposeKey +import io.github.aedev.flow.sync2.crypto.seal +import io.github.aedev.flow.sync2.crypto.u32 +import io.github.aedev.flow.sync2.engine.LinkedException +import io.github.aedev.flow.sync2.engine.Session +import io.github.aedev.flow.sync2.lan.GREETING_CONNECTIONS +import io.github.aedev.flow.sync2.lan.LanGreeter +import io.github.aedev.flow.sync2.lan.MAX_LAN_FRAME +import io.github.aedev.flow.sync2.lan.acceptLink +import io.github.aedev.flow.sync2.lan.connectLink +import io.github.aedev.flow.sync2.linking.LanEndpoint +import io.github.aedev.flow.sync2.linking.dial +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import okio.ByteString.Companion.toByteString +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.annotation.Config +import java.net.InetAddress +import java.net.ServerSocket +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.concurrent.thread + +/** What a connection on the LAN gets before anyone knows it's a member (spec/live.md, Session; spec/store.md, Untrusted input). */ +@RunWith(AndroidJUnit4::class) +@Config(sdk = [34], application = Application::class) +class LanLimitsTest { + private val path = "/flow-sync/2/test" + + private class Silent : SyncConnection { + var closed = false + + override val remoteHost: String? = null + + override suspend fun send(bytes: ByteArray) = Unit + + override suspend fun receive(): ByteArray? = awaitCancellation() + + override fun close() { + closed = true + } + } + + private class OneFrame( + private val frame: ByteArray, + ) : SyncConnection { + private var read = false + override val remoteHost: String? = null + + override suspend fun send(bytes: ByteArray) = Unit + + override suspend fun receive(): ByteArray? = if (read) null else frame.also { read = true } + + override fun close() = Unit + } + + @Test + fun a_payload_stops_inflating_at_its_limit() { + val content = ByteArray(1000) + assertArrayEquals(content, decodePayload(encodePayload(content), 1000)) + val error = runCatching { decodePayload(encodePayload(content), 999) }.exceptionOrNull() as ProtocolException + assertEquals(ProtocolError.LENGTH, error.error) + } + + @Test + fun a_hello_with_more_content_than_a_hello_holds_is_refused() = + runTest { + val g = LinkedGroup(this).apply { create() } + val session = Session.open(g.deviceDb(), g.store, g.b) + val lanKey = purposeKey(g.epochKey, g.group, 0, Purpose.LAN) + val frame = u32(0) + seal(lanKey, lanHelloAad(1), encodePayload(ByteArray(5 * 1024))) + + val error = runCatching { acceptLink(OneFrame(frame), session) }.exceptionOrNull() as ProtocolException + + assertEquals(ProtocolError.LENGTH, error.error) + } + + @Test + fun a_side_that_goes_quiet_in_the_hello_ends_it() = + runTest { + val g = LinkedGroup(this).apply { create() } + val session = Session.open(g.deviceDb(), g.store, g.b) + assertTrue(runCatching { acceptLink(Silent(), session) }.exceptionOrNull() is LinkedException.TimedOut) + assertTrue(runCatching { connectLink(Silent(), session) }.exceptionOrNull() is LinkedException.TimedOut) + } + + @Test + fun connections_past_the_limit_are_closed_until_a_hello_ends() = + runTest { + val greeter = LanGreeter { null } + val waiting = List(GREETING_CONNECTIONS) { Silent() } + assertTrue(waiting.all(greeter::takes)) + val extra = Silent() + assertFalse(greeter.takes(extra)) + assertTrue(extra.closed) + + assertTrue(runCatching { greeter.greet(waiting.first()) }.exceptionOrNull() is LinkedException.TimedOut) + assertTrue("the slot is free again", greeter.takes(Silent())) + } + + @Test + fun the_roster_is_not_read_until_a_hello_arrives() = + runTest { + val opened = AtomicInteger() + val greeter = + LanGreeter { + opened.incrementAndGet() + null + } + val silent = Silent() + greeter.takes(silent) + runCatching { greeter.greet(silent) } + assertEquals(0, opened.get()) + + val speaking = OneFrame(ByteArray(8)) + greeter.takes(speaking) + assertNull(greeter.greet(speaking)) + assertEquals(1, opened.get()) + } + + private fun open( + client: OkHttpClient, + port: Int, + closed: CountDownLatch? = null, + ): WebSocket { + val opened = CountDownLatch(1) + val socket = + client.newWebSocket( + Request.Builder().url("ws://127.0.0.1:$port$path").build(), + object : WebSocketListener() { + override fun onOpen( + webSocket: WebSocket, + response: Response, + ) = opened.countDown() + + override fun onClosing( + webSocket: WebSocket, + code: Int, + reason: String, + ) { + closed?.countDown() + } + + override fun onFailure( + webSocket: WebSocket, + t: Throwable, + response: Response?, + ) { + closed?.countDown() + } + }, + ) + assertTrue(opened.await(5, TimeUnit.SECONDS)) + return socket + } + + @Test + fun a_server_for_one_session_closes_a_second_connection() { + val server = WsServer(path, MAX_LAN_FRAME) + val port = runBlocking { server.start() } + val client = OkHttpClient() + try { + val first = open(client, port) + runBlocking { server.awaitConnection() } + val closed = CountDownLatch(1) + val second = open(client, port, closed) + assertTrue("nothing would ever read it", closed.await(5, TimeUnit.SECONDS)) + first.cancel() + second.cancel() + } finally { + server.stop() + } + } + + @Test + fun a_connection_nobody_reads_stops_being_read_from_the_socket() { + val unread = Channel(1) + val server = WsServer(path, MAX_LAN_FRAME) { unread.trySend(it) } + val port = runBlocking { server.start() } + val client = OkHttpClient() + try { + val socket = open(client, port) + val frame = ByteArray(1024 * 1024).toByteString() + // Far past what the socket and the server's own queues hold. + val goal = 256 + var sent = 0 + val deadline = System.currentTimeMillis() + 20_000 + while (sent < goal && System.currentTimeMillis() < deadline) { + if (socket.queueSize() < 8L * 1024 * 1024 && socket.send(frame)) sent++ else Thread.sleep(5) + } + Thread.sleep(2_000) + assertTrue("$sent frames went into a connection nobody reads and ${socket.queueSize()} bytes wait", socket.queueSize() > 0) + socket.cancel() + } finally { + server.stop() + } + } + + @Test + fun a_dial_does_not_follow_a_redirect() { + val hits = AtomicInteger() + val target = ServerSocket(0, 50, InetAddress.getByName("127.0.0.1")) + val origin = ServerSocket(0, 50, InetAddress.getByName("127.0.0.1")) + thread(isDaemon = true) { + runCatching { + while (true) { + target.accept().use { hits.incrementAndGet() } + } + } + } + thread(isDaemon = true) { + runCatching { + while (true) { + origin.accept().use { + it.getInputStream().read(ByteArray(4096)) + val answer = "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:${target.localPort}/elsewhere\r\nContent-Length: 0\r\n\r\n" + it.getOutputStream().write(answer.toByteArray()) + } + } + } + } + try { + val endpoint = LanEndpoint("127.0.0.1", origin.localPort) + val pipe = runBlocking { withContext(Dispatchers.IO) { withTimeout(10_000) { dial(listOf(endpoint)) } } } + assertNull(pipe) + assertEquals(0, hits.get()) + } finally { + target.close() + origin.close() + } + } +} From 2d112b48f4cd36b20686943e1d5f31fb63396a2d Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:18:26 +0200 Subject: [PATCH 2/2] style(sync): wrap a long test line --- app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt index bc73a045c..bb8b8e999 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/LanLimitsTest.kt @@ -242,7 +242,8 @@ class LanLimitsTest { while (true) { origin.accept().use { it.getInputStream().read(ByteArray(4096)) - val answer = "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:${target.localPort}/elsewhere\r\nContent-Length: 0\r\n\r\n" + val location = "http://127.0.0.1:${target.localPort}/elsewhere" + val answer = "HTTP/1.1 302 Found\r\nLocation: $location\r\nContent-Length: 0\r\n\r\n" it.getOutputStream().write(answer.toByteArray()) } }