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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
57411ea703
commit
6584d313ca
@@ -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) {
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user