From 6584d313ca398e05b14e403e613527b126e75ee1 Mon Sep 17 00:00:00 2001 From: Paul Hitt Date: Thu, 10 Sep 2026 08:18:34 -0400 Subject: [PATCH 1/2] Ask the relay to hold a poll open The relay has always been willing to hold an inbox request until something arrives, and the Android client has never asked it to. The constant naming the hold was declared in MessengerService and read nowhere, and the comment next to the 500 ms gap already described a long poll that was not happening. What that cost, measured against production: two devices sent 1,177 inbox requests in eight minutes, every one of them held open by the relay for about 30 ms and answered "nothing". Every message and every call signal waited out the gap to the next poll. Someone had already noticed the effect on calls and shortened the gap to a second while one was being set up, which made the polling denser without making the signal arrive any sooner. So `wait` reaches the wire. The signed-in mailbox is asked to hold; a backup home never is, because two held reads in a row would put a backup mailbox's latency in front of the one nearly all mail arrives at. A held request needs longer than the ordinary call timeout, so it gets a clone of the client with room for the hold. Both loops now pace on whether the relay held the request rather than on a fixed gap: a poll that took a while was itself the wait, and the next one goes straight back out. Co-Authored-By: Claude Opus 5 --- .../messenger/app/MessengerService.kt | 14 +- .../messenger/app/MessengerViewModel.kt | 26 +++- .../io/flamenet/messenger/app/Session.kt | 4 +- .../io/flamenet/client/Conversations.kt | 17 ++- .../kotlin/io/flamenet/client/RelayClient.kt | 36 +++++- .../io/flamenet/client/LongPollWaitTest.kt | 122 ++++++++++++++++++ 6 files changed, 198 insertions(+), 21 deletions(-) create mode 100644 client/src/test/kotlin/io/flamenet/client/LongPollWaitTest.kt diff --git a/app/src/main/kotlin/io/flamenet/messenger/app/MessengerService.kt b/app/src/main/kotlin/io/flamenet/messenger/app/MessengerService.kt index fa89d9d..1f1e50d 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/MessengerService.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/MessengerService.kt @@ -176,7 +176,8 @@ class MessengerService : Service() { } while (true) { - val touched = runCatching { s.receive() }.getOrDefault(emptySet()) + val started = System.currentTimeMillis() + val touched = runCatching { s.receive(WAIT_SECONDS) }.getOrDefault(emptySet()) if (touched.isNotEmpty()) notifyMessages(touched.size) // Reply reminders fire from THIS loop, after `receive()`, and the // order matters: an inbound message clears the reminder inside @@ -185,10 +186,13 @@ class MessengerService : Service() { fireDueReplyReminders(s) // An unanswered ring has to stop ringing. expireStaleRing() - // A short gap, not a poll interval: `receive()` is held open by the - // relay for WAIT_SECONDS, so this only paces a relay that answers - // immediately. - delay(500) + // A short gap, not a poll interval: `receive()` asks the relay to + // hold the request open for WAIT_SECONDS, so a poll that took a + // while was already the wait, and going straight back is what keeps + // the mailbox continuously watched. Only a relay that answered at + // once, one that has no hold to offer, gets paced. + val held = System.currentTimeMillis() - started >= 2_000 + if (!held) delay(500) } } diff --git a/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt b/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt index 92fdf56..f960452 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt @@ -3399,6 +3399,7 @@ class MessengerViewModel( runCatching { session?.conversations?.ensureDeliveryKey() } } while (true) { + val tick = System.currentTimeMillis() // An offer the background service consumed while the app // was off screen. PendingIncomingCall.take()?.let { parked -> @@ -3421,7 +3422,7 @@ class MessengerViewModel( // here. val received = withContext(Dispatchers.IO) { - val got = runCatching { current.receive() } + val got = runCatching { current.receive(POLL_WAIT_SECONDS) } // Cached plaintext for anything nothing refers // to any more, an expired message's photo, a // story past its lifetime. @@ -3536,9 +3537,12 @@ class MessengerViewModel( } } } - // A call being set up lives or dies on signalling latency, - // the 5 s cadence made every answer feel geological, and - // stacked with the same cadence on the other end. + // A call being set up lives or dies on signalling latency. + // The relay is asked to hold each poll open, so a tick that + // took a while was the wait itself and the next one goes + // straight back out; only a relay that answered at once + // needs pacing, and then a call being set up cannot afford + // the ordinary cadence. val settingUp = when (calls?.state) { CallManager.State.DIALLING, @@ -3548,7 +3552,12 @@ class MessengerViewModel( else -> false } - kotlinx.coroutines.delay(if (settingUp) CALL_SETUP_POLL_MS else POLL_INTERVAL_MS) + val held = System.currentTimeMillis() - tick >= 2_000 + if (!held) { + kotlinx.coroutines.delay( + if (settingUp) CALL_SETUP_POLL_MS else POLL_INTERVAL_MS, + ) + } } } } @@ -3570,6 +3579,13 @@ class MessengerViewModel( const val CALLS_SEEN_AT = "calls_seen_at" const val POLL_INTERVAL_MS = 5_000L const val CALL_SETUP_POLL_MS = 1_000L + + /** + * How long the relay may hold a poll open. Asking for a hold is what + * turns the cadence above into a ceiling rather than the latency every + * message and call signal actually pays. + */ + const val POLL_WAIT_SECONDS = 20 } // / The relay said this device was removed. diff --git a/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt b/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt index 014b2f2..08945e9 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt @@ -1153,8 +1153,8 @@ class Session( */ var onCallSignal: ((Long, io.flamenet.e2e.Content.CallSignal, Long) -> Unit)? = null - fun receive(): Set { - val everything = conversations.poll() + fun receive(waitSeconds: Int = 0): Set { + val everything = conversations.poll(waitSeconds = waitSeconds) // Control traffic consumed during the poll may have carried group-name // claims. drainGroupNames() diff --git a/client/src/main/kotlin/io/flamenet/client/Conversations.kt b/client/src/main/kotlin/io/flamenet/client/Conversations.kt index ba777b8..fd1527e 100644 --- a/client/src/main/kotlin/io/flamenet/client/Conversations.kt +++ b/client/src/main/kotlin/io/flamenet/client/Conversations.kt @@ -953,7 +953,10 @@ class Conversations( /** * Read the inbox, decrypt what we can, and advance the watermark. */ - fun poll(limit: Int = 100): List { + fun poll( + limit: Int = 100, + waitSeconds: Int = 0, + ): List { // S4: keep auditing the log we depend on. trust.verifyLogConsistencyIfDue() val out = mutableListOf() @@ -962,13 +965,16 @@ class Conversations( val reached = mutableListOf() var firstFailure: Throwable? = null - runCatching { readInbox(relay, home = null, limit = limit, into = out) } - .onSuccess { reached += true } + runCatching { + readInbox(relay, home = null, limit = limit, into = out, waitSeconds = waitSeconds) + }.onSuccess { reached += true } .onFailure { firstFailure = it } // After the signed-in relay and never instead of it: a backup home that // is slow or down must not delay the mailbox nearly all mail actually - // arrives at. + // arrives at. They are never asked to hold, for the same reason: two + // held reads in a row would put a backup mailbox's latency in front of + // the main one's, and this loop is what a call signal waits on. for (home in extraHomes()) { runCatching { readInbox(home.second, home.first, limit, out) } .onSuccess { reached += true } @@ -1001,9 +1007,10 @@ class Conversations( home: String?, limit: Int, into: MutableList, + waitSeconds: Int = 0, ) { val watermark = if (home == null) vault.inboxWatermark else vault.watermarkFor(home) - val incoming = client.pollInbox(identity.deviceId, watermark, limit) + val incoming = client.pollInbox(identity.deviceId, watermark, limit, waitSeconds) val out = into for (message in incoming) { diff --git a/client/src/main/kotlin/io/flamenet/client/RelayClient.kt b/client/src/main/kotlin/io/flamenet/client/RelayClient.kt index 3f5c33e..eccf9f2 100644 --- a/client/src/main/kotlin/io/flamenet/client/RelayClient.kt +++ b/client/src/main/kotlin/io/flamenet/client/RelayClient.kt @@ -29,6 +29,18 @@ class RelayClient( ) { private val base: HttpUrl = baseUrl.trimEnd('/').toHttpUrl() + /** + * The same client with room for a held request. The ordinary call timeout + * is shorter than the longest hold the relay will take, so a long poll on + * `http` would be cancelled from this side before the relay ever answered. + */ + private val heldHttp: OkHttpClient by lazy { + http + .newBuilder() + .callTimeout(MAX_WAIT_SECONDS + 20L, TimeUnit.SECONDS) + .build() + } + /** * The relay this client talks to, for anything that has to remember a fact * about a specific one, §19's "has this relay ever served a log". @@ -50,6 +62,13 @@ class RelayClient( /** The relay's own 10 MiB per-blob limit, with slack (audit M5). */ const val HARD_ATTACHMENT_CAP = 12L * 1024 * 1024 + /** + * The longest hold to ask the relay for. It answers the moment + * something arrives, so this only bounds how long an idle connection + * stays open. + */ + const val MAX_WAIT_SECONDS = 20 + /** * @param bridge when set, every request is carried inside obfs4 to * that bridge, which proxies on to the relay. @@ -270,7 +289,9 @@ class RelayClient( deviceId: String, after: Long, limit: Int = 100, + waitSeconds: Int = 0, ): List { + val held = waitSeconds.coerceIn(0, MAX_WAIT_SECONDS) val url = base .newBuilder() @@ -279,8 +300,13 @@ class RelayClient( addQueryParameter("device_id", deviceId) addQueryParameter("after", after.toString()) addQueryParameter("limit", limit.toString()) + // Ask the relay to hold the request open rather than + // answering "nothing" straight away. Without it the only + // way to notice a message is to ask again, and a call + // signal then waits out the whole gap between polls. + if (held > 0) addQueryParameter("wait", held.toString()) }.build() - val body = requestJson("GET", url, "/e2e/messages", null) + val body = requestJson("GET", url, "/e2e/messages", null, if (held > 0) heldHttp else http) val messages = body.optJSONArray("messages") ?: JSONArray() return (0 until messages.length()).map { val m = messages.getJSONObject(it) @@ -744,7 +770,8 @@ class RelayClient( url: HttpUrl, logicalPath: String, body: ByteArray?, - ): JSONObject = request(method, url, logicalPath, body?.toRequestBody(JSON)).use(::json) + client: OkHttpClient = http, + ): JSONObject = request(method, url, logicalPath, body?.toRequestBody(JSON), client).use(::json) /** * One request, with one retry on 401. @@ -754,12 +781,13 @@ class RelayClient( url: HttpUrl, logicalPath: String, body: RequestBody?, + client: OkHttpClient = http, ): Response { - val first = http.newCall(build(method, url, logicalPath, body)).execute() + val first = client.newCall(build(method, url, logicalPath, body)).execute() if (first.code != 401 || auth == null) return checked(first, logicalPath) first.close() auth?.invalidate() - return checked(http.newCall(build(method, url, logicalPath, body)).execute(), logicalPath) + return checked(client.newCall(build(method, url, logicalPath, body)).execute(), logicalPath) } private fun build( diff --git a/client/src/test/kotlin/io/flamenet/client/LongPollWaitTest.kt b/client/src/test/kotlin/io/flamenet/client/LongPollWaitTest.kt new file mode 100644 index 0000000..b5413de --- /dev/null +++ b/client/src/test/kotlin/io/flamenet/client/LongPollWaitTest.kt @@ -0,0 +1,122 @@ +package io.flamenet.client + +import io.flamenet.e2e.DeviceIdentity +import io.flamenet.e2e.Vault +import okhttp3.mockwebserver.Dispatcher +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import okhttp3.mockwebserver.RecordedRequest +import javax.crypto.KeyGenerator +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +/** + * The relay will hold a poll open, but only if it is asked to. The constant + * naming the hold once existed on its own, never reached the wire, and the + * apps paid the whole gap between polls for every message and call signal. + */ +class LongPollWaitTest { + private class MapStorage : Vault.Storage { + val records = LinkedHashMap() + + override fun read(name: String) = records[name] + + override fun write( + name: String, + bytes: ByteArray, + ) { + records[name] = bytes + } + + override fun delete(name: String) { + records.remove(name) + } + + override fun names() = records.keys.toList() + } + + private lateinit var home: MockWebServer + private lateinit var backup: MockWebServer + + /** Every `/e2e/messages` URL each mailbox was asked for. */ + private val homePaths = mutableListOf() + private val backupPaths = mutableListOf() + + @BeforeTest + fun start() { + home = MockWebServer().apply { dispatcher = recorder(homePaths); start() } + backup = MockWebServer().apply { dispatcher = recorder(backupPaths); start() } + } + + @AfterTest + fun stop() { + home.shutdown() + backup.shutdown() + } + + private fun recorder(into: MutableList) = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val path = request.path.orEmpty() + if ("/e2e/messages" in path) { + into += path + return MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody("""{"messages":[],"cursor":0}""") + } + return MockResponse().setResponseCode(404) + } + } + + private fun identity(): DeviceIdentity = DeviceIdentity.create() + + @Test + fun `a hold is asked for on the wire`() { + val client = RelayClient(home.url("/").toString(), "me", "pw") + client.pollInbox(identity().deviceId, after = 0, waitSeconds = 20) + assertEquals(1, homePaths.size) + assertTrue("wait=20" in homePaths.single(), homePaths.single()) + } + + @Test + fun `no hold is asked for when none was wanted`() { + val client = RelayClient(home.url("/").toString(), "me", "pw") + client.pollInbox(identity().deviceId, after = 0) + assertFalse("wait=" in homePaths.single(), homePaths.single()) + } + + @Test + fun `the hold is capped at what the relay will honour`() { + val client = RelayClient(home.url("/").toString(), "me", "pw") + client.pollInbox(identity().deviceId, after = 0, waitSeconds = 3_600) + assertTrue("wait=${RelayClient.MAX_WAIT_SECONDS}" in homePaths.single(), homePaths.single()) + } + + @Test + fun `a backup mailbox is never held open`() { + val mine = identity() + val vault = + Vault(MapStorage(), KeyGenerator.getInstance("AES").apply { init(256) }.generateKey()) + vault.saveExtraHomes( + listOf( + Vault.ExtraHome( + relay = backup.hostName, + username = "me", + appPassword = "pw", + baseUrl = backup.url("/").toString(), + ), + ), + ) + Conversations(mine, vault, RelayClient(home.url("/").toString(), "me", "pw")) + .poll(waitSeconds = 20) + assertTrue("wait=20" in homePaths.single(), homePaths.toString()) + // Two held reads in a row would put the backup's latency in front of + // the mailbox almost everything actually arrives at. + assertFalse("wait=" in backupPaths.single(), backupPaths.toString()) + } +} -- 2.54.0 From a70787e96076cf87e6b4ef3983c20ee7f5ddd8ed Mon Sep 17 00:00:00 2001 From: Paul Hitt Date: Thu, 10 Sep 2026 08:39:25 -0400 Subject: [PATCH 2/2] Hand CI the backup vectors it was never given MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The §16 backup port shipped with a vector suite that finds its input by walking relative paths out of the checkout. That works on a developer's machine, where flamenet-e2e sits beside this repo, and cannot work on the runner, where it does not. The suite fails rather than skips, deliberately, so it said so: nine BackupVectorTest failures on every pipeline since the port merged, on main included. Two things were missing, and both had to be fixed or the gate would have stayed red. The pin predates the vectors, so `unit-tests` cloned a flamenet-e2e that has no backup-v1.json in it. Bumped; nothing else under test/vectors moved between the two commits, so no other suite changes what it checks. And neither job exported FLAMENET_BACKUP_VECTORS, so even the job that clones HEAD, which did have the file, never pointed the suite at it. It is exported in both now, and named in the pre-flight check, which is there precisely so a missing file reads as a missing file rather than a protocol bug. Co-Authored-By: Claude Opus 5 --- .e2e-ref | 2 +- .gitlab-ci.yml | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/.e2e-ref b/.e2e-ref index b1041c0..f1b77fb 100644 --- a/.e2e-ref +++ b/.e2e-ref @@ -1 +1 @@ -9775fbfe08c823921b6493b0fd39a0e4a13c8aa2 +32d3a8824b06e53a0204feebedfee6dd9d384e7b diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 407a0d0..4a78e22 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -60,11 +60,13 @@ unit-tests: - export FLAMENET_VECTORS="$CI_PROJECT_DIR/.ci-vectors/test/vectors/protocol-v1.json" - export FLAMENET_PQ_VECTORS="$CI_PROJECT_DIR/.ci-vectors/test/vectors/pqx3dh-v2.json" - export FLAMENET_ACCOUNT_VECTORS="$CI_PROJECT_DIR/.ci-vectors/test/vectors/account-kdf-v1.json" + - export FLAMENET_BACKUP_VECTORS="$CI_PROJECT_DIR/.ci-vectors/test/vectors/backup-v1.json" # Checked before the suite runs. A path that silently does not exist puts # the tests straight back where they were, and the failure would read as a # protocol bug rather than a missing file. - | - for f in "$FLAMENET_VECTORS" "$FLAMENET_PQ_VECTORS" "$FLAMENET_ACCOUNT_VECTORS"; do + for f in "$FLAMENET_VECTORS" "$FLAMENET_PQ_VECTORS" "$FLAMENET_ACCOUNT_VECTORS" \ + "$FLAMENET_BACKUP_VECTORS"; do [ -f "$f" ] || { echo "GATE: vector file missing: $f"; exit 1; } done # Clear old results before running, or the gate counts ghosts. The @@ -155,6 +157,7 @@ vectors-head: - export FLAMENET_VECTORS="$CI_PROJECT_DIR/.ci-vectors-head/test/vectors/protocol-v1.json" - export FLAMENET_PQ_VECTORS="$CI_PROJECT_DIR/.ci-vectors-head/test/vectors/pqx3dh-v2.json" - export FLAMENET_ACCOUNT_VECTORS="$CI_PROJECT_DIR/.ci-vectors-head/test/vectors/account-kdf-v1.json" + - export FLAMENET_BACKUP_VECTORS="$CI_PROJECT_DIR/.ci-vectors-head/test/vectors/backup-v1.json" # Only the module that reads vectors. The rest of the suite has nothing # to say about whether flamenet-e2e moved, and running it here would double # the board's cost to answer a question it cannot answer. -- 2.54.0