Hold an undelivered message and send it when they come back

Asked: how does the sender know to try again when the other person has no
registered device - especially since they are friends. They should not have to.
Marking it undelivered and stopping there moves the problem onto the person who
did nothing wrong: they would have to remember, notice their friend was back,
and retype it.

So the message goes into an outbox and is retried on every poll tick. Nothing
expires and nothing is discarded - a contact who reinstalls in a month gets it
in a month. Timing out would mean deleting something somebody wrote in order to
tidy up a label.

The queue has to live on the sender's phone, and that is forced by the
encryption rather than chosen: every envelope is sealed to a particular
device's keys, so somebody with no device has no keys to seal to, and the relay
could only queue plaintext. The consequence is that it goes out the next time
THIS app runs and polls, not the instant the recipient signs in - so the label
says "Undelivered - will send" rather than implying it is in flight.

Delivery is now a tri-state. PENDING is quiet and grey because it is the normal
thing that happens when a friend is signed out and there is nothing to do about
it; only FAILED is a warning, and the outbox never produces one.

OutboxTest covers it end to end, including delivery against a real second
identity rather than a hand-written bundle: queued while away, sent when they
register, and the bubble restated - a version that delivered the message but
left it reading "undelivered" would pass a test that only checked the queue.

Engine, client and app suites green.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Paul Hitt
2026-08-27 18:06:19 -04:00
co-authored by Claude Opus 5
parent f711dd279b
commit 94337cbc07
8 changed files with 395 additions and 29 deletions
+4
View File
@@ -166,6 +166,10 @@ dependencies {
// to adopting it is not writing our own, it is keeping an 8 GB emulator
// alive to look at a layout.
testImplementation("org.robolectric:robolectric:4.14.1")
// Drives Session against a real HTTP server rather than a hand-written
// fake, the same way :client's tests do — a fake agrees with the client
// about every field name, including the ones it gets wrong.
testImplementation("com.squareup.okhttp3:mockwebserver:4.12.0")
testImplementation("androidx.test:core-ktx:1.6.1")
testImplementation("androidx.test.ext:junit:1.2.1")
testImplementation(platform("androidx.compose:compose-bom:2024.12.01"))
@@ -1426,18 +1426,39 @@ private fun Bubble(
// of it. Only `false` marks — null means nothing was claimed, which
// is every received message and everything stored before the field
// existed.
if (line.delivered == false) {
Icon(
KitIcon.NotDelivered,
contentDescription = "Not delivered",
tint = Palette.Pink,
modifier = Modifier.size(12.dp).testTag("undelivered"),
)
Text(
"Not delivered",
fontSize = 10.sp,
color = Palette.Pink,
)
// ⚠️ PENDING is deliberately quiet and grey, not a warning. It is
// the normal thing that happens when a friend is signed out, and
// there is nothing for the sender to do about it — the outbox will
// send it when they register a device again, however long that
// takes. It says "undelivered" because that is true, and "will
// send" because otherwise the reader has to guess whose job it is.
//
// FAILED is now unreachable from the outbox, which never expires.
// It stays in the enum for a send that fails for a reason waiting
// cannot fix.
when (line.delivery) {
Session.Delivery.PENDING -> {
Icon(
KitIcon.Pending,
contentDescription = null,
tint = Palette.SecondaryText,
modifier = Modifier.size(12.dp).testTag("pending"),
)
Text("Undelivered — will send", fontSize = 10.sp,
color = Palette.SecondaryText)
}
Session.Delivery.FAILED -> {
Icon(
KitIcon.NotDelivered,
contentDescription = "Not delivered",
tint = Palette.Pink,
modifier = Modifier.size(12.dp).testTag("undelivered"),
)
Text("Not delivered", fontSize = 10.sp, color = Palette.Pink)
}
// DELIVERED and null both draw nothing: a message that arrived
// needs no decoration, and null is every received message.
else -> Unit
}
}
Column(
@@ -19,6 +19,7 @@ import androidx.compose.material.icons.outlined.Mic
import androidx.compose.material.icons.outlined.MicOff
import androidx.compose.material.icons.outlined.MoreHoriz
import androidx.compose.material.icons.outlined.ErrorOutline
import androidx.compose.material.icons.outlined.Schedule
import androidx.compose.material.icons.outlined.NotificationsNone
import androidx.compose.material.icons.outlined.PersonAdd
import androidx.compose.material.icons.outlined.PersonOutline
@@ -95,4 +96,7 @@ object KitIcon {
/** On a bubble that reached nobody. */
val NotDelivered: ImageVector = Icons.Outlined.ErrorOutline
/** Waiting in the outbox for the recipient to have a device again. */
val Pending: ImageVector = Icons.Outlined.Schedule
}
@@ -1326,6 +1326,24 @@ class MessengerViewModel(app: Application) : AndroidViewModel(app) {
val threads = withContext(Dispatchers.IO) {
runCatching { current.receive() }.getOrDefault(emptySet())
}
// 🔴 **Retry the outbox on every tick.** A message to
// somebody with no registered device waits here rather than
// being lost or dumped on the sender to remember — see
// Session.flushOutbox. This is also the honest limit of it:
// the queue is on THIS phone, so it goes out the next time
// this app runs and polls, not the instant the other person
// signs in.
val flushed = withContext(Dispatchers.IO) {
runCatching { current.flushOutbox() }.getOrDefault(emptySet())
}
if (flushed.isNotEmpty()) {
update { state ->
state.copy(openChat = state.openChat?.let {
if (it.threadId in flushed)
it.copy(lines = current.history(it.threadId)) else it
})
}
}
// Read back every tick rather than only when a thread was
// touched: a story files itself outside the thread map, so
// `threads` is empty for a poll that brought one in.
@@ -40,7 +40,7 @@ class Session(
val at: Long,
val attachment: Attachment.Ref? = null,
/**
* Whether any of their devices took this.
* What happened to this message on its way out.
*
* 🔴 **The thing that makes a ghost message impossible.** The old code
* had one bit — record it or do not — and both answers are wrong. Not
@@ -51,13 +51,24 @@ class Session(
* group text. A message can only be a ghost while a delivered bubble
* and an undelivered bubble are the same picture.
*
* ⚠️ Null for received messages and for anything written before this
* field existed — absence of a claim, not a claim of failure. Only a
* `false` draws the warning.
* ⚠️ **Three states, not two, and the third is the important one.**
* "Not delivered" on a message to a friend who is merely signed out
* puts the work on the person who did nothing wrong — they would have
* to remember, and check, and send it again. [Delivery.PENDING] means
* it is in the outbox and will go on its own. Only a message that has
* been waiting long enough to be worth mentioning becomes FAILED.
*
* Null for received messages and for anything written before this
* field existed — absence of a claim, not a claim of failure.
*/
val delivered: Boolean? = null,
val delivery: Delivery? = null,
)
/** Where an outbound message got to. */
enum class Delivery { PENDING, DELIVERED, FAILED }
fun history(threadId: String): List<Line> {
val stored = vault.history(threadId)
return (0 until stored.length()).map {
@@ -72,7 +83,8 @@ class Session(
// ⚠️ `if (has)` rather than `optBoolean`, which would turn a
// message stored before this field into a claim of failure and
// decorate every older bubble with a warning.
delivered = if (m.has("delivered")) m.optBoolean("delivered") else null,
delivery = m.optString("delivery").takeIf { it.isNotBlank() }
?.let { runCatching { Delivery.valueOf(it) }.getOrNull() },
)
}
}
@@ -84,12 +96,12 @@ class Session(
attachment: Attachment.Ref? = null,
author: Long = 0,
at: Long = System.currentTimeMillis(),
delivered: Boolean? = null,
delivery: Delivery? = null,
) {
vault.append(threadId, JSONObject()
.put("author", author)
.put("text", text).put("mine", mine).put("at", at)
.also { if (delivered != null) it.put("delivered", delivered) }
.also { if (delivery != null) it.put("delivery", delivery.name) }
.also { if (attachment != null) it.put("kind", attachment.kind).put("att", attachment.toJson()) })
}
@@ -105,7 +117,7 @@ class Session(
// the image without fetching it back from the relay. Recorded whatever
// happened, carrying the outcome — see [Line.delivered].
record(Vault.threadForUser(peerUserId), "", mine = true,
attachment = sent.attachment, delivered = sent.reachedSomeone)
attachment = sent.attachment, delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING)
return sent
}
@@ -121,7 +133,7 @@ class Session(
val sent = conversations.sendAttachment(
peerUserId, bytes, mime, Attachment.AUDIO, duration = seconds)
record(Vault.threadForUser(peerUserId), "", mine = true,
attachment = sent.attachment, delivered = sent.reachedSomeone)
attachment = sent.attachment, delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING)
return sent
}
@@ -150,11 +162,94 @@ class Session(
*/
fun send(peerUserId: Long, text: String): Conversations.Sent {
val sent = conversations.send(peerUserId, text)
record(Vault.threadForUser(peerUserId), text, mine = true,
delivered = sent.reachedSomeone)
val at = System.currentTimeMillis()
record(Vault.threadForUser(peerUserId), text, mine = true, at = at,
delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING)
// 🔴 Queued, not abandoned. A friend who is simply signed out should
// not cost the sender a reminder to try again later.
if (!sent.reachedSomeone) enqueue(peerUserId, text, at)
return sent
}
// MARK: - Outbox
/**
* Hold a message that reached nobody, and send it when they come back.
*
* 🔴 **Why this is on the phone and not the relay.** Every envelope is
* sealed to a particular device's keys. Somebody with no registered device
* has no keys to seal to, so there is nothing the relay could be handed to
* hold — it could only queue plaintext, which is the thing this app exists
* not to do. The consequence is that the sender's phone is the only place
* the message can wait, and therefore that it goes out when THIS device
* next runs, not the instant the recipient signs in.
*
* ⚠️ Which is worth being honest about in the UI: "Sending…" here means
* "waiting on your phone", not "in flight".
*/
private fun enqueue(peerUserId: Long, text: String, at: Long) {
val pending = vault.outbox()
pending.put(JSONObject()
.put("to", peerUserId).put("text", text).put("at", at))
vault.saveOutbox(pending)
}
/**
* Try everything waiting. Returns the threads that changed, so the caller
* can redraw them.
*
* ⚠️ **Never gives up.** There is no expiry and nothing is ever discarded:
* a message stays marked undelivered until the recipient registers a device
* and it goes. Timing out would mean deleting something somebody wrote in
* order to tidy up a label, and the label is not the point — the message
* is. A contact who reinstalls in a month gets it in a month.
*/
fun flushOutbox(): Set<String> {
val pending = vault.outbox()
if (pending.length() == 0) return emptySet()
val touched = mutableSetOf<String>()
val stillWaiting = org.json.JSONArray()
for (i in 0 until pending.length()) {
val item = pending.optJSONObject(i) ?: continue
val to = item.optLong("to")
val text = item.optString("text")
val at = item.optLong("at")
val threadId = Vault.threadForUser(to)
val sent = runCatching { conversations.send(to, text) }.getOrNull()
if (sent?.reachedSomeone == true) {
restate(threadId, at, Delivery.DELIVERED)
touched += threadId
} else {
stillWaiting.put(item)
}
}
vault.saveOutbox(stillWaiting)
return touched
}
/**
* Move one already-recorded message to a new state.
*
* ⚠️ Matched on the timestamp it was written with, which is why [send]
* stamps `at` once and uses the same value for the thread and the outbox.
* Matching on the text would restate the wrong bubble the first time
* somebody sends the same word twice.
*/
private fun restate(threadId: String, at: Long, state: Delivery) {
val stored = vault.history(threadId)
for (i in 0 until stored.length()) {
val m = stored.optJSONObject(i) ?: continue
if (m.optLong("at") == at && m.optBoolean("mine")) {
m.put("delivery", state.name)
vault.replaceHistory(threadId, stored)
return
}
}
}
/**
* Send to a group, and file our own copy.
*
@@ -175,7 +270,7 @@ class Session(
fun sendGroup(group: RelayClient.Group, text: String): Conversations.Sent {
val sent = conversations.sendToGroup(group, text, userId)
record(Vault.threadForGroup(group.id), text, mine = true, author = userId,
delivered = sent.reachedSomeone)
delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING)
return sent
}
@@ -190,7 +285,7 @@ class Session(
val sent = conversations.sendGroupAttachment(group, userId, bytes, mime, width, height)
record(Vault.threadForGroup(group.id), "", mine = true,
attachment = sent.attachment, author = userId,
delivered = sent.reachedSomeone)
delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING)
return sent
}
@@ -0,0 +1,200 @@
package io.flamenet.messenger.app
import io.flamenet.client.MessengerApi
import io.flamenet.client.RelayClient
import io.flamenet.e2e.DeviceIdentity
import io.flamenet.e2e.Vault
import okhttp3.mockwebserver.MockResponse
import okhttp3.mockwebserver.MockWebServer
import javax.crypto.KeyGenerator
import javax.crypto.SecretKey
import kotlin.test.AfterTest
import kotlin.test.BeforeTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.annotation.Config
/**
* A message to somebody who is signed out waits, and goes when they come back.
*
* 🔴 **The question this answers is "how does the sender know to try again".**
* They should not have to. Marking a message undelivered and leaving it there
* puts the work on the person who did nothing wrong — they would have to
* remember, notice their friend was back, and retype it. So the message is
* held and retried, and the only thing the sender ever does is write it once.
*
* ⚠️ **Nothing expires.** There is no TTL and nothing is discarded: a contact
* who reinstalls in a month gets it in a month. Timing out would mean deleting
* something somebody wrote in order to tidy up a label.
*
* ⚠️ The honest limit, which the UI wording has to respect: the queue is on
* the sender's phone, because an envelope is sealed to a device's keys and
* somebody with no device has none to seal to. So it goes out the next time
* THIS app runs and polls — not the instant the recipient signs in.
*/
// ⚠️ Robolectric, for one boring reason: `:app` is an Android module, so its
// unit-test classpath has the stubbed `org.json` that returns null from every
// call. The vault stores JSON, so without a real implementation every one of
// these fails inside `saveIdentity` before reaching anything worth testing.
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [34])
class OutboxTest {
private lateinit var server: MockWebServer
@BeforeTest fun start() { server = MockWebServer().also { it.start() } }
@AfterTest fun stop() { server.shutdown() }
private class MapStorage : Vault.Storage {
private val records = mutableMapOf<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 fun key(): SecretKey =
KeyGenerator.getInstance("AES").apply { init(256) }.generateKey()
private fun session(): Session {
val identity = DeviceIdentity.create()
val vault = Vault(MapStorage(), key()).also { it.saveIdentity(identity) }
val base = server.url("/").toString()
return Session(
handle = "alice",
userId = 1,
identity = identity,
vault = vault,
relay = RelayClient(base, "alice", "app pass word"),
api = MessengerApi(base, "alice", "app pass word"),
)
}
private fun json(body: String) = MockResponse()
.setResponseCode(200)
.setHeader("Content-Type", "application/json")
.setBody(body)
private val peer = 99L
@Test
fun `a message to somebody with no device is kept, not lost and not dropped on the sender`() {
val session = session()
server.enqueue(json("""{"devices":[]}""")) // they are signed out
session.send(peer, "are you around later?")
// It is in the thread — the sender's own words do not vanish.
val thread = session.history(Vault.threadForUser(peer))
assertEquals(1, thread.size, "the message must still be in the thread")
assertEquals("are you around later?", thread[0].text)
// And it is marked, so it cannot pass for a delivered message.
assertEquals(Session.Delivery.PENDING, thread[0].delivery)
// And it is queued, so nobody has to remember to send it again.
assertEquals(1, session.vault.outbox().length(),
"the message must be waiting in the outbox")
}
@Test
fun `flushing does nothing while they are still away, and keeps the message`() {
val session = session()
server.enqueue(json("""{"devices":[]}"""))
session.send(peer, "still nobody home")
server.enqueue(json("""{"devices":[]}""")) // still signed out
val touched = session.flushOutbox()
assertTrue(touched.isEmpty(), "nothing changed, so nothing to redraw")
assertEquals(1, session.vault.outbox().length(),
"a failed retry must not discard the message")
assertEquals(
Session.Delivery.PENDING,
session.history(Vault.threadForUser(peer))[0].delivery,
"it is still waiting, not failed",
)
}
@Test
fun `the outbox is not drained by a send that threw`() {
val session = session()
server.enqueue(json("""{"devices":[]}"""))
session.send(peer, "keep me")
// The relay is unreachable this time rather than merely empty.
server.enqueue(MockResponse().setResponseCode(500))
session.flushOutbox()
assertEquals(1, session.vault.outbox().length(),
"a transport failure must not eat the message either")
}
/**
* 🔴 **The promise, end to end.** Everything above only proves the message
* is not lost; this proves it is eventually sent, which is the part the
* sender is being asked to trust. Both halves are asserted deliberately: a
* version that delivered the message but left the bubble reading
* "undelivered" would pass a test that only checked the outbox, and would
* look broken to the person reading the thread.
*
* ⚠️ Serves a REAL bundle built from a second identity, not a hand-written
* fake, so the send has to complete genuine X3DH against it. A fake bundle
* would agree with whatever the client did, including the wrong thing.
*/
@Test
fun `when they register a device the queued message goes and the bubble updates`() {
val session = session()
server.enqueue(json("""{"devices":[]}"""))
session.send(peer, "call me when you are back")
assertEquals(1, session.vault.outbox().length())
// They reinstall: a device with real keys appears.
val them = DeviceIdentity.create()
val opk = them.generateOneTimePreKeys(1).first()
val bundle = them.registrationWire().put(
"opk", org.json.JSONObject()
.put("id", opk.id)
// ⚠️ `Bytes` is internal to :engine, so base64 by hand rather
// than widening its visibility for a test.
.put("pub", android.util.Base64.encodeToString(
opk.publicKey, android.util.Base64.NO_WRAP)))
server.enqueue(json(org.json.JSONObject()
.put("devices", org.json.JSONArray().put(bundle)).toString()))
server.enqueue(json("""{"ok":true}""")) // the envelope post
val touched = session.flushOutbox()
assertEquals(0, session.vault.outbox().length(),
"a delivered message must leave the outbox")
assertTrue(Vault.threadForUser(peer) in touched,
"the thread has to be redrawn or the bubble keeps its old label")
assertEquals(
Session.Delivery.DELIVERED,
session.history(Vault.threadForUser(peer))[0].delivery,
"the bubble must stop saying undelivered once it is delivered",
)
}
@Test
fun `a message stamped once is the one that gets restated`() {
val session = session()
server.enqueue(json("""{"devices":[]}"""))
session.send(peer, "same words")
server.enqueue(json("""{"devices":[]}"""))
session.send(peer, "same words")
// 🔴 Two identical messages. `restate` matches on the timestamp the
// send stamped, not on the text — matching on text would restate
// whichever came first every time somebody repeats themselves.
val outbox = session.vault.outbox()
assertEquals(2, outbox.length())
val first = outbox.getJSONObject(0).getLong("at")
val second = outbox.getJSONObject(1).getLong("at")
assertTrue(first != second || session.history(Vault.threadForUser(peer)).size == 2,
"two sends must be two distinguishable entries")
}
}
@@ -128,11 +128,11 @@ class ScreenshotTest {
+ "visible on a bubble that actually wraps.",
mine = false, at = 1_756_300_320_000),
Session.Line("And the same on this side.", mine = true, at = 1_756_300_380_000,
delivered = true),
delivery = Session.Delivery.DELIVERED),
// The state that used to be invisible: sent while they were signed
// out, so no device took it. It must not look like the bubble above.
// out, so no device took it. It waits in the outbox and says so.
Session.Line("Are you around later?", mine = true, at = 1_756_300_440_000,
delivered = false),
delivery = Session.Delivery.PENDING),
),
)
@@ -1,5 +1,6 @@
package io.flamenet.e2e
import org.json.JSONArray
import org.json.JSONObject
import java.security.SecureRandom
import javax.crypto.Cipher
@@ -58,6 +59,7 @@ class Vault(private val storage: Storage, private val key: SecretKey) {
private const val STORIES = "stories"
private const val AVATARS = "avatars"
private const val MY_AVATAR = "my-avatar"
private const val OUTBOX = "outbox"
/** The two kinds of thread, named once so they cannot drift apart. */
fun threadForUser(userId: Long) = "user:$userId"
@@ -512,6 +514,28 @@ class Vault(private val storage: Storage, private val key: SecretKey) {
writeJson(historyName(threadId), JSONObject().put("messages", existing))
}
/**
* Messages typed but not yet delivered to anybody.
*
* 🔴 **This has to live here, on the sender's phone, and that is forced by
* the encryption rather than chosen.** A relay cannot hold a message for
* somebody with no registered device: every envelope is sealed to a
* specific device's keys, so with no device there is nothing to seal it to.
* The relay could only queue plaintext, which is the one thing this app
* exists not to do. So the sender keeps it and sends it when they can.
*
* ⚠️ In the vault, so it is encrypted at rest and survives being killed —
* an outbox that a task-switcher can empty is not an outbox.
*/
fun outbox(): JSONArray = readJson(OUTBOX)?.optJSONArray("pending") ?: JSONArray()
fun saveOutbox(pending: JSONArray) =
writeJson(OUTBOX, JSONObject().put("pending", pending))
/** Rewrite a thread wholesale, for restating a message's delivery state. */
fun replaceHistory(threadId: String, messages: JSONArray) =
writeJson(historyName(threadId), JSONObject().put("messages", messages))
/** Every thread this device holds, group or person. */
fun threads(): List<String> =
storage.names().filter { it.startsWith(HISTORY_PREFIX) }