feat: read a room's group events again when they arrived out of order
Relays impose no ordering, so a kind:445 can turn up before the group can read it: an application message encrypted under an epoch whose commit has not landed, or a commit for an epoch ahead of the local one. Both are stored and then dropped -- MarmotInboundManager refuses an out-of-epoch commit precisely so it does not half-mutate the group -- and nothing goes back for them once the missing event fills the gap. The message is on disk, readable, and never read. A "Reindex Events" button at the bottom of the group's detail screen is that second look. Only events with nothing to show for them are replayed: no chat line at all, or one of the two placeholder types. A room where nothing went wrong is left exactly as it was, which is what makes the button safe to press on a hunch. Passes repeat while a pass recovers something, because created_at order is not epoch order and a commit recovered by one pass is what lets the next read the messages that were waiting on it. **Replaying was not safe as it stood.** Every row the path writes is keyed on an event id and upserts in place -- MarmotGroupEvent, MarmotInnerEvent, and the nip30303 entities -- with one exception. ChatMessage's primary key is autogenerated, so writing a freshly built line always inserts, and a re-read would have left the room showing each recovered message twice, once as "Undecryptable Message" and once as itself. ChatMessage.reconcileMarmotLine matches on the group event id instead, so a re-read is an update, and refuses to let a placeholder overwrite a line that says something. That last rule is what protects the line this device wrote on the way out for a message it sent: our own kind:445 cannot be read back, since the sender ratchet has consumed the generation, and without the rule a replay would have replaced our words with "Undecryptable Message". The MLS group itself was already safe to replay against, which is worth saying because it is the part that looks dangerous: a commit behind the current epoch is rejected as a duplicate before it touches the group, one ahead is refused, and a consumed ratchet generation throws before mutating anything. The exception was quartz's EpochCommitTracker, which does not dedupe and only empties when a commit applies -- so replaying a held commit just grew the list and left it pending forever. forgetPendingCommits drops the room's entries first, and the sweep feeds the events back in the order CommitOrdering picks a winner in, so a contested epoch resolves the same way it would have on every other device. **What is testable, and what is not.** The DAO is not: testDebugUnitTest is plain JVM and Room's in-memory builder wants an Android Context. So the two pieces carrying decisions are lifted out where they can be run without one -- MarmotReindexSweep for the stopping rule, and reconcileMarmotLine for which of two lines wins -- and the DAO is left as query, sweep, write. The filter tests pin why the query's `tags LIKE` is a prefilter and not a test: an event belonging to another room can mention this one in a q tag, and its own h tag is what rejects it. **Not recovered by any of this.** A message whose key is gone -- one the ratchet has already advanced past, or one from an epoch predating this device's join. And events that never reached disk at all: storeNostrEvent is a single transaction, so a kind:445 arriving before its room exists rolls back its own insert along with the failed indexing, and only a re-sync brings it back. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
@@ -0,0 +1,161 @@
|
||||
package press.mantra.compose.database.model
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertSame
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* What happens to a room's chat line when its group event is read a second time.
|
||||
*
|
||||
* [ChatMessage.id] is autogenerated, so writing a freshly built line always
|
||||
* inserts. That is correct exactly once. Reading the same kind:445 again -- which
|
||||
* is what a reindex does -- would otherwise leave the room showing the recovered
|
||||
* message twice, once as the placeholder that was written when it could not be
|
||||
* read and once as itself. And the other way round matters just as much: a group
|
||||
* event this device sent already has its line, and a replay that cannot read our
|
||||
* own ratchet-consumed message must not replace it with "Undecryptable Message".
|
||||
*
|
||||
* Both directions are decided by [ChatMessage.reconcileMarmotLine], which is why
|
||||
* they are asserted rather than left to the shape of the calling code.
|
||||
*/
|
||||
class MarmotChatLineReconciliationTest {
|
||||
|
||||
private val groupEventId = "a".repeat(64)
|
||||
private val sender = "b".repeat(64)
|
||||
private val room = "c".repeat(64)
|
||||
|
||||
private fun line(
|
||||
id: Long = 0,
|
||||
messageType: String,
|
||||
content: String,
|
||||
viewedAt: Instant? = null,
|
||||
savedAt: Instant = Instant.fromEpochSeconds(1_000),
|
||||
) = ChatMessage(
|
||||
id = id,
|
||||
senderPublicKey = sender,
|
||||
isUserMessage = false,
|
||||
giftWrapPayloadId = null,
|
||||
marmotGroupEventId = groupEventId,
|
||||
marmotInnerEventId = null,
|
||||
chatRoomId = room,
|
||||
content = content,
|
||||
messageType = messageType,
|
||||
savedAt = savedAt,
|
||||
viewedAt = viewedAt,
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `a group event read for the first time is written as it comes`() {
|
||||
val fresh = line(messageType = "message", content = "hello")
|
||||
|
||||
assertSame(fresh, ChatMessage.reconcileMarmotLine(fresh = fresh, existing = null))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a recovered message replaces the placeholder rather than joining it`() {
|
||||
val placeholder = line(
|
||||
id = 42,
|
||||
messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER,
|
||||
content = "Undecryptable Message",
|
||||
)
|
||||
val recovered = line(messageType = "message", content = "hello")
|
||||
|
||||
val row = ChatMessage.reconcileMarmotLine(fresh = recovered, existing = placeholder)
|
||||
|
||||
assertEquals(42, row?.id, "reusing the row id is what makes this an update, not a second line")
|
||||
assertEquals("hello", row?.content)
|
||||
assertEquals("message", row?.messageType)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a placeholder never overwrites a line that says something`() {
|
||||
val sent = line(id = 7, messageType = "message", content = "hello")
|
||||
val placeholder = line(
|
||||
messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER,
|
||||
content = "Undecryptable Message",
|
||||
)
|
||||
|
||||
assertNull(
|
||||
ChatMessage.reconcileMarmotLine(fresh = placeholder, existing = sent),
|
||||
"a replay that cannot read our own message must leave its line alone",
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a commit still waiting may replace the placeholder it already wrote`() {
|
||||
val pending = line(
|
||||
id = 3,
|
||||
messageType = ChatMessage.TYPE_PENDING_COMMIT,
|
||||
content = "Pending Commit in epoch 4",
|
||||
)
|
||||
val stillPending = line(
|
||||
messageType = ChatMessage.TYPE_PENDING_COMMIT,
|
||||
content = "Pending Commit in epoch 5",
|
||||
)
|
||||
|
||||
val row = ChatMessage.reconcileMarmotLine(fresh = stillPending, existing = pending)
|
||||
|
||||
assertEquals(3, row?.id, "one placeholder replacing another is still one line")
|
||||
assertEquals("Pending Commit in epoch 5", row?.content)
|
||||
}
|
||||
|
||||
/**
|
||||
* When a line was first seen, and when it was first stored, are facts about
|
||||
* the reader rather than about the event. Reading the event again is not the
|
||||
* message arriving again, so neither may be reset.
|
||||
*/
|
||||
@Test
|
||||
fun `replacing a line keeps when it was first saved and seen`() {
|
||||
val seenAt = Instant.fromEpochSeconds(2_000)
|
||||
val storedAt = Instant.fromEpochSeconds(1_500)
|
||||
|
||||
val placeholder = line(
|
||||
id = 9,
|
||||
messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER,
|
||||
content = "Undecryptable Message",
|
||||
viewedAt = seenAt,
|
||||
savedAt = storedAt,
|
||||
)
|
||||
val recovered = line(
|
||||
messageType = "message",
|
||||
content = "hello",
|
||||
viewedAt = null,
|
||||
savedAt = Instant.fromEpochSeconds(9_999),
|
||||
)
|
||||
|
||||
val row = ChatMessage.reconcileMarmotLine(fresh = recovered, existing = placeholder)
|
||||
|
||||
assertEquals(seenAt, row?.viewedAt, "a re-read must not mark a read message unread")
|
||||
assertEquals(storedAt, row?.savedAt)
|
||||
}
|
||||
|
||||
/**
|
||||
* The lines a replay is allowed to replace are exactly the ones that stand in
|
||||
* for a message still to come. Adding a type to the transcript without adding
|
||||
* it here silently makes those events unrecoverable; adding one here that
|
||||
* reports something real makes them overwritable.
|
||||
*/
|
||||
@Test
|
||||
fun `only the placeholder types count as unresolved`() {
|
||||
assertEquals(
|
||||
setOf(ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER, ChatMessage.TYPE_PENDING_COMMIT),
|
||||
ChatMessage.UNRESOLVED_MARMOT_TYPES,
|
||||
)
|
||||
|
||||
listOf("message", "processedCommit", "proposalStaged", "artifact", "dialect")
|
||||
.forEach { messageType ->
|
||||
assertNull(
|
||||
ChatMessage.reconcileMarmotLine(
|
||||
fresh = line(
|
||||
messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER,
|
||||
content = "Undecryptable Message",
|
||||
),
|
||||
existing = line(id = 1, messageType = messageType, content = "something"),
|
||||
),
|
||||
"a $messageType line reports something and must survive a replay",
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package press.mantra.compose.database.model
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNotEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* Which room a stored kind:445 belongs to.
|
||||
*
|
||||
* `NostrEventDao.getMarmotGroupNostrEventsByChatRoomId` finds a room's group
|
||||
* events with `tags LIKE '%' || :chatRoomId || '%'`, because the tags are one
|
||||
* serialised column and there is nothing better to match on. That is a prefilter
|
||||
* and not a test: a group id can appear in a tag that is not the `h` tag, and an
|
||||
* event matched that way belongs to a different room entirely. Feeding one of
|
||||
* those to the replay would read another group's event against this group's MLS
|
||||
* state.
|
||||
*
|
||||
* So the reindex confirms `groupId()` in Kotlin after the query. These pin the
|
||||
* cases that confirmation is there for.
|
||||
*/
|
||||
class MarmotGroupEventRoomFilterTest {
|
||||
|
||||
private val room = "a".repeat(64)
|
||||
private val otherRoom = "d".repeat(64)
|
||||
private val ephemeralSender = "e".repeat(64)
|
||||
|
||||
private fun storedGroupEvent(
|
||||
tags: Array<Array<String>>,
|
||||
kind: Int = 445,
|
||||
) = NostrEvent(
|
||||
id = "1".repeat(64),
|
||||
pubKey = ephemeralSender,
|
||||
kind = kind,
|
||||
tags = tags,
|
||||
content = "bm9uY2UrY2lwaGVydGV4dA==",
|
||||
sig = "f".repeat(128),
|
||||
createdAt = Instant.fromEpochSeconds(1_700_000_000),
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `the h tag is what names the room`() {
|
||||
val stored = storedGroupEvent(arrayOf(arrayOf("h", room)))
|
||||
|
||||
assertEquals(room, stored.toGroupEvent(userPublicKey = ephemeralSender)?.groupId())
|
||||
}
|
||||
|
||||
/**
|
||||
* The case the `LIKE` cannot tell apart: this event is another room's, and
|
||||
* only mentions ours in a tag that says nothing about routing.
|
||||
*/
|
||||
@Test
|
||||
fun `an event mentioning the room outside its h tag belongs to the other room`() {
|
||||
val stored = storedGroupEvent(
|
||||
arrayOf(
|
||||
arrayOf("h", otherRoom),
|
||||
arrayOf("q", room),
|
||||
)
|
||||
)
|
||||
|
||||
val groupId = stored.toGroupEvent(userPublicKey = ephemeralSender)?.groupId()
|
||||
|
||||
assertEquals(otherRoom, groupId)
|
||||
assertNotEquals(room, groupId, "the LIKE would match this event; groupId() is what rejects it")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an event with no h tag names no room`() {
|
||||
val stored = storedGroupEvent(arrayOf(arrayOf("q", room)))
|
||||
|
||||
assertNull(stored.toGroupEvent(userPublicKey = ephemeralSender)?.groupId())
|
||||
}
|
||||
|
||||
/**
|
||||
* Only kind:445 is a group event. The query pins the kind in SQL, and this is
|
||||
* the other half of that: nothing else can be read as one by mistake.
|
||||
*/
|
||||
@Test
|
||||
fun `an event of another kind is not a group event`() {
|
||||
val stored = storedGroupEvent(tags = arrayOf(arrayOf("h", room)), kind = 9)
|
||||
|
||||
assertNull(stored.toGroupEvent(userPublicKey = ephemeralSender))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
package press.mantra.compose.managers
|
||||
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* The stopping rule a reindex sweep runs on.
|
||||
*
|
||||
* A replay is worth doing at all because the events depend on each other: the
|
||||
* commit nobody could apply is what the room's later messages were waiting on.
|
||||
* That is also what makes one ordered pass insufficient, and what makes "keep
|
||||
* going" dangerous — a room full of events whose keys are gone would be swept
|
||||
* forever. Both halves of that are pinned here.
|
||||
*
|
||||
* The sweep never sees a database. What counts as recovered is whatever the
|
||||
* `isUnresolved` probe says afterwards, so these fakes are the real contract and
|
||||
* not a stand-in for one.
|
||||
*/
|
||||
class MarmotReindexSweepTest {
|
||||
|
||||
/**
|
||||
* A backlog where each event is unlocked by the one before it, which is the
|
||||
* shape a chain of out-of-order commits arrives in.
|
||||
*
|
||||
* [unlockedBy] is the event that has to be replayed successfully first;
|
||||
* null means it can be read straight away.
|
||||
*/
|
||||
private class Backlog(unlocks: Map<String, String?>) {
|
||||
private val unlockedBy = unlocks
|
||||
private val resolved = mutableSetOf<String>()
|
||||
|
||||
/** Every event replayed, in order, across every pass. */
|
||||
val replayed = mutableListOf<String>()
|
||||
|
||||
suspend fun replay(id: String) {
|
||||
replayed.add(id)
|
||||
val blocker = unlockedBy.getValue(id)
|
||||
if (blocker == null || blocker in resolved) {
|
||||
resolved.add(id)
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun isUnresolved(id: String): Boolean = id !in resolved
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a chain of events unlocking each other is read in as many passes as it takes`() = runTest {
|
||||
// c waits on b, b waits on a. One pass in this order recovers only a.
|
||||
val backlog = Backlog(mapOf("c" to "b", "b" to "a", "a" to null))
|
||||
|
||||
val report = MarmotReindexSweep.run(
|
||||
stored = 10,
|
||||
unresolved = listOf("c", "b", "a"),
|
||||
replay = backlog::replay,
|
||||
isUnresolved = backlog::isUnresolved,
|
||||
)
|
||||
|
||||
assertEquals(3, report.recovered, "every event in the chain should have been recovered")
|
||||
assertEquals(0, report.failed)
|
||||
assertEquals(3, report.unresolved, "unresolved is what the sweep started with")
|
||||
assertEquals(10, report.stored, "stored is reported as handed in")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an event that can never be read stops the sweep after one pass`() = runTest {
|
||||
// Nothing unlocks these: the keys are gone, replaying achieves nothing.
|
||||
val backlog = Backlog(mapOf("a" to "never", "b" to "never"))
|
||||
|
||||
val report = MarmotReindexSweep.run(
|
||||
stored = 2,
|
||||
unresolved = listOf("a", "b"),
|
||||
replay = backlog::replay,
|
||||
isUnresolved = backlog::isUnresolved,
|
||||
)
|
||||
|
||||
assertEquals(listOf("a", "b"), backlog.replayed, "a pass that recovers nothing must not repeat")
|
||||
assertEquals(0, report.recovered)
|
||||
assertEquals(2, report.failed)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `events that can be read are still recovered alongside ones that cannot`() = runTest {
|
||||
val backlog = Backlog(mapOf("stuck" to "never", "b" to "a", "a" to null))
|
||||
|
||||
val report = MarmotReindexSweep.run(
|
||||
stored = 3,
|
||||
unresolved = listOf("stuck", "b", "a"),
|
||||
replay = backlog::replay,
|
||||
isUnresolved = backlog::isUnresolved,
|
||||
)
|
||||
|
||||
assertEquals(2, report.recovered)
|
||||
assertEquals(1, report.failed, "the unreadable one is reported, not retried forever")
|
||||
assertEquals(report.unresolved, report.recovered + report.failed, "every event is accounted for")
|
||||
}
|
||||
|
||||
/**
|
||||
* The pass cap only bites on the shape where each pass recovers exactly one
|
||||
* event, which is the only way the sweep can go quadratic. Six events in a
|
||||
* chain replayed worst-first need six passes and get five.
|
||||
*/
|
||||
@Test
|
||||
fun `the pass cap bounds a chain longer than it`() = runTest {
|
||||
val chain = listOf("f", "e", "d", "c", "b", "a")
|
||||
val backlog = Backlog(
|
||||
mapOf("f" to "e", "e" to "d", "d" to "c", "c" to "b", "b" to "a", "a" to null)
|
||||
)
|
||||
|
||||
val report = MarmotReindexSweep.run(
|
||||
stored = chain.size,
|
||||
unresolved = chain,
|
||||
replay = backlog::replay,
|
||||
isUnresolved = backlog::isUnresolved,
|
||||
)
|
||||
|
||||
assertEquals(MarmotReindexSweep.DEFAULT_MAX_PASSES, report.recovered)
|
||||
assertEquals(1, report.failed, "what the cap cut short is reported as still unread")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a replay that throws is survived by the rest of the sweep`() = runTest {
|
||||
val resolved = mutableSetOf<String>()
|
||||
|
||||
val report = MarmotReindexSweep.run(
|
||||
stored = 3,
|
||||
unresolved = listOf("throws", "a", "b"),
|
||||
replay = { id ->
|
||||
if (id == "throws") error("no key for this epoch")
|
||||
resolved.add(id)
|
||||
},
|
||||
isUnresolved = { id -> id !in resolved },
|
||||
)
|
||||
|
||||
assertEquals(2, report.recovered, "a throw must not abandon the events behind it")
|
||||
assertEquals(1, report.failed)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `nothing unresolved does no work at all`() = runTest {
|
||||
var replays = 0
|
||||
|
||||
val report = MarmotReindexSweep.run(
|
||||
stored = 40,
|
||||
unresolved = emptyList<String>(),
|
||||
replay = { replays++ },
|
||||
isUnresolved = { true },
|
||||
)
|
||||
|
||||
assertEquals(0, replays, "a room where nothing went wrong must be left alone")
|
||||
assertTrue(report.isNoOp)
|
||||
assertEquals(40, report.stored)
|
||||
}
|
||||
|
||||
/**
|
||||
* The probe, not the replay call, is what counts a recovery: a group event can
|
||||
* be replayed without error and still leave the room with nothing to show for
|
||||
* it -- a commit rejected as a duplicate does exactly that.
|
||||
*/
|
||||
@Test
|
||||
fun `a replay that raises nothing but resolves nothing is not counted as recovered`() = runTest {
|
||||
val report = MarmotReindexSweep.run(
|
||||
stored = 1,
|
||||
unresolved = listOf("duplicateCommit"),
|
||||
replay = { /* applies cleanly, writes no line */ },
|
||||
isUnresolved = { true },
|
||||
)
|
||||
|
||||
assertEquals(0, report.recovered)
|
||||
assertEquals(1, report.failed)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user