Group sends keep the outbox promise, with a real completion rule
"Undelivered - will send" on a group bubble was a lie: group text and group stickers recorded PENDING when members were unreachable and nothing queued them. The 1:1 outbox could not be stretched over it - a group send fails per member - so the entry grew the thread it restates, and the design questions got answers instead of a shrug: WHO is queued: exactly the members nothing reached. Group text reads them out of the failure map, where the no-device guard files them as "user:<id>"; the sticker loop sees per-member results directly. A member whose devices took the message is NOT queued - resending to them would duplicate it. WHERE it restates: the group thread, carried on the entry. Restating threadForUser(member) would touch a 1:1 thread that may not even exist. WHEN the bubble flips: only when the LAST queued member is reached. Flipping on the first success would mark the message delivered while other members are still owed it - "will send" would quietly stop being true for exactly the people it was still true for. Until the set drains, PENDING remains the honest label, because the queue genuinely will keep sending. The test walks the whole rule: two members away, one returns (bubble still PENDING, one entry left), the second returns (bubble DELIVERED, group thread redrawn, the prose body not replaced by the wire form). ⚠️ iOS sends to groups too (sendGroupText/sendGroupImage) and still has no group outbox - it stays on the roadmap with this as the reference implementation. Engine, client and app suites green; seven outbox tests. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Paul Hitt
co-authored by
Claude Opus 5
Claude Opus 5
parent
09a4fb90fc
commit
59fa88c2ad
@@ -257,13 +257,20 @@ class Session(
|
||||
Conversations.Sent {
|
||||
var delivered = 0
|
||||
var failed = 0
|
||||
val at = System.currentTimeMillis()
|
||||
val wire = io.flamenet.e2e.Content.sticker(pack, stickerId, groupId = group.id)
|
||||
for (member in group.memberIds) {
|
||||
if (member == userId) continue
|
||||
val one = conversations.sendSticker(member, pack, stickerId, groupId = group.id)
|
||||
delivered += one.delivered.size
|
||||
failed += one.failed.size
|
||||
// The same rule as group text: queue exactly the members nothing
|
||||
// reached, with the group thread to restate.
|
||||
if (!one.reachedSomeone) {
|
||||
enqueue(member, wire, at, thread = Vault.threadForGroup(group.id))
|
||||
}
|
||||
}
|
||||
record(Vault.threadForGroup(group.id), "", mine = true, author = userId,
|
||||
record(Vault.threadForGroup(group.id), "", mine = true, author = userId, at = at,
|
||||
sticker = Sticker(pack, stickerId),
|
||||
delivery = if (delivered > 0) Delivery.DELIVERED else Delivery.PENDING)
|
||||
return Conversations.Sent(
|
||||
@@ -343,10 +350,15 @@ class Session(
|
||||
* which transmits its argument as the plaintext verbatim — so both kinds
|
||||
* take the same path out.
|
||||
*/
|
||||
private fun enqueue(peerUserId: Long, text: String, at: Long) {
|
||||
private fun enqueue(peerUserId: Long, text: String, at: Long,
|
||||
thread: String? = null) {
|
||||
val pending = vault.outbox()
|
||||
pending.put(JSONObject()
|
||||
.put("to", peerUserId).put("text", text).put("at", at))
|
||||
.put("to", peerUserId).put("text", text).put("at", at)
|
||||
// ⚠️ Present only for group entries: the bubble to restate lives in
|
||||
// the GROUP thread, not in a 1:1 thread with this member — which
|
||||
// may not even exist.
|
||||
.also { if (thread != null) it.put("thread", thread) })
|
||||
vault.saveOutbox(pending)
|
||||
}
|
||||
|
||||
@@ -366,23 +378,50 @@ class Session(
|
||||
|
||||
val touched = mutableSetOf<String>()
|
||||
val stillWaiting = org.json.JSONArray()
|
||||
// Group entries that delivered this pass, keyed "thread|at". The bubble
|
||||
// only flips when the LAST member of its set is reached — see below.
|
||||
val groupDelivered = mutableSetOf<Pair<String, Long>>()
|
||||
|
||||
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 groupThread = item.optString("thread").takeIf { it.isNotBlank() }
|
||||
|
||||
val sent = runCatching { conversations.send(to, text) }.getOrNull()
|
||||
if (sent?.reachedSomeone == true) {
|
||||
restate(threadId, at, Delivery.DELIVERED)
|
||||
touched += threadId
|
||||
if (groupThread == null) {
|
||||
restate(Vault.threadForUser(to), at, Delivery.DELIVERED)
|
||||
touched += Vault.threadForUser(to)
|
||||
} else {
|
||||
groupDelivered += groupThread to at
|
||||
}
|
||||
} else {
|
||||
stillWaiting.put(item)
|
||||
}
|
||||
}
|
||||
vault.saveOutbox(stillWaiting)
|
||||
|
||||
// 🔴 **A group bubble flips only when nobody is left waiting on it.**
|
||||
// One message fans out to many members and each gets its own entry;
|
||||
// restating on the first success would mark it delivered while other
|
||||
// members are still owed it, and "will send" would quietly stop being
|
||||
// true for them. Until the set drains, PENDING remains the honest
|
||||
// label — the queue genuinely will keep sending.
|
||||
for ((thread, at) in groupDelivered) {
|
||||
var remaining = false
|
||||
for (i in 0 until stillWaiting.length()) {
|
||||
val item = stillWaiting.optJSONObject(i) ?: continue
|
||||
if (item.optString("thread") == thread && item.optLong("at") == at) {
|
||||
remaining = true; break
|
||||
}
|
||||
}
|
||||
if (!remaining) {
|
||||
restate(thread, at, Delivery.DELIVERED)
|
||||
touched += thread
|
||||
}
|
||||
}
|
||||
return touched
|
||||
}
|
||||
|
||||
@@ -446,8 +485,20 @@ 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,
|
||||
val at = System.currentTimeMillis()
|
||||
record(Vault.threadForGroup(group.id), text, mine = true, author = userId, at = at,
|
||||
delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING)
|
||||
// 🔴 **The group half of the outbox promise.** Members with no
|
||||
// registered device are filed in the failure map as "user:<id>" by the
|
||||
// no-device guard; each gets an entry carrying the SAME group body and
|
||||
// the group thread to restate. A member whose devices took it is not
|
||||
// queued — resending to them would duplicate the message.
|
||||
val body = io.flamenet.e2e.Content.groupText(text, group.id)
|
||||
sent.failed.keys.filter { it.startsWith("user:") }
|
||||
.mapNotNull { it.removePrefix("user:").toLongOrNull() }
|
||||
.forEach { member ->
|
||||
enqueue(member, body, at, thread = Vault.threadForGroup(group.id))
|
||||
}
|
||||
return sent
|
||||
}
|
||||
|
||||
|
||||
@@ -222,6 +222,64 @@ class OutboxTest {
|
||||
"restating the delivery must not turn the sticker into text")
|
||||
}
|
||||
|
||||
/**
|
||||
* 🔴 The group half of the promise, with its completion rule made
|
||||
* observable: the bubble flips only when the LAST queued member is
|
||||
* reached. Restating on the first success would mark the message delivered
|
||||
* while other members are still owed it — "will send" would quietly stop
|
||||
* being true for exactly the people it was still true for.
|
||||
*/
|
||||
@Test
|
||||
fun `a group message waits for every queued member before the bubble flips`() {
|
||||
val session = session()
|
||||
val group = io.flamenet.client.RelayClient.Group(
|
||||
id = "g1", name = "test", owner = 1, memberIds = listOf(101L, 102L))
|
||||
|
||||
// Both members signed out: two device lookups, both empty.
|
||||
server.enqueue(json("""{"devices":[]}"""))
|
||||
server.enqueue(json("""{"devices":[]}"""))
|
||||
session.sendGroup(group, "meeting moved to three")
|
||||
|
||||
val thread = Vault.threadForGroup("g1")
|
||||
assertEquals(Session.Delivery.PENDING, session.history(thread)[0].delivery)
|
||||
assertEquals(2, session.vault.outbox().length(),
|
||||
"one entry per member nothing reached")
|
||||
|
||||
// Member 101 registers; 102 is still away.
|
||||
val m1 = DeviceIdentity.create()
|
||||
val opk1 = m1.generateOneTimePreKeys(1).first()
|
||||
server.enqueue(json(org.json.JSONObject().put("devices",
|
||||
org.json.JSONArray().put(m1.registrationWire().put("opk",
|
||||
org.json.JSONObject().put("id", opk1.id).put("pub",
|
||||
android.util.Base64.encodeToString(opk1.publicKey,
|
||||
android.util.Base64.NO_WRAP))))).toString()))
|
||||
server.enqueue(json("""{"ok":true}"""))
|
||||
server.enqueue(json("""{"devices":[]}""")) // 102 still gone
|
||||
session.flushOutbox()
|
||||
|
||||
assertEquals(1, session.vault.outbox().length(),
|
||||
"the reached member leaves the queue; the other stays")
|
||||
assertEquals(Session.Delivery.PENDING, session.history(thread)[0].delivery,
|
||||
"one member reached is not the group delivered")
|
||||
|
||||
// Member 102 comes back.
|
||||
val m2 = DeviceIdentity.create()
|
||||
val opk2 = m2.generateOneTimePreKeys(1).first()
|
||||
server.enqueue(json(org.json.JSONObject().put("devices",
|
||||
org.json.JSONArray().put(m2.registrationWire().put("opk",
|
||||
org.json.JSONObject().put("id", opk2.id).put("pub",
|
||||
android.util.Base64.encodeToString(opk2.publicKey,
|
||||
android.util.Base64.NO_WRAP))))).toString()))
|
||||
server.enqueue(json("""{"ok":true}"""))
|
||||
val touched = session.flushOutbox()
|
||||
|
||||
assertEquals(0, session.vault.outbox().length())
|
||||
assertTrue(thread in touched, "the GROUP thread is what gets redrawn")
|
||||
assertEquals(Session.Delivery.DELIVERED, session.history(thread)[0].delivery)
|
||||
assertEquals("meeting moved to three", session.history(thread)[0].text,
|
||||
"the group body must not be replaced by the wire form")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a message stamped once is the one that gets restated`() {
|
||||
val session = session()
|
||||
|
||||
Reference in New Issue
Block a user