extant2000/flamenet-messenger-android
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'fix/android-long-poll' into 'main'

Browse Source 
Ask the relay to hold a poll open

See merge request paulhitt/flamenet-messenger-android!109
This commit was merged in pull request #136.
This commit is contained in:
Paul Hitt
2026-09-10 12:44:59 +00:00
parent 57411ea703 a70787e960
commit 44c7b512d6
8 changed files with 203 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.e2e-ref
+1 -1
View File
@@ -1 +1 @@
9775fbfe08c823921b6493b0fd39a0e4a13c8aa2
32d3a8824b06e53a0204feebedfee6dd9d384e7b
.gitlab-ci.yml
+4 -1
View File
@@ -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.
app/src/main/kotlin/io/flamenet/messenger/app/MessengerService.kt
+9 -5
View File
@@ -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)
}
}
app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt
+21 -5
View File
@@ -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.
app/src/main/kotlin/io/flamenet/messenger/app/Session.kt
+2 -2
View File
@@ -1153,8 +1153,8 @@ class Session(
*/
var onCallSignal: ((Long, io.flamenet.e2e.Content.CallSignal, Long) -> Unit)? = null
fun receive(): Set<String> {
val everything = conversations.poll()
fun receive(waitSeconds: Int = 0): Set<String> {
val everything = conversations.poll(waitSeconds = waitSeconds)
// Control traffic consumed during the poll may have carried group-name
// claims.
drainGroupNames()
client/src/main/kotlin/io/flamenet/client/Conversations.kt
+12 -5
View File
@@ -953,7 +953,10 @@ class Conversations(
/**
* Read the inbox, decrypt what we can, and advance the watermark.
*/
fun poll(limit: Int = 100): List<Received> {
fun poll(
limit: Int = 100,
waitSeconds: Int = 0,
): List<Received> {
// S4: keep auditing the log we depend on.
trust.verifyLogConsistencyIfDue()
val out = mutableListOf<Received>()
@@ -962,13 +965,16 @@ class Conversations(
val reached = mutableListOf<Boolean>()
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<Received>,
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) {
client/src/main/kotlin/io/flamenet/client/RelayClient.kt
+32 -4
View File
@@ -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<Incoming> {
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(
client/src/test/kotlin/io/flamenet/client/LongPollWaitTest.kt
+122
View File
@@ -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<String, ByteArray>()
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<String>()
private val backupPaths = mutableListOf<String>()
@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<String>) =
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())
}
}