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. 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()) + } +}