From 9107b81c999864b3a8737fde40974dfac65ac32f Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sun, 6 Sep 2026 18:56:33 +0200 Subject: [PATCH] fix: hand the notary the whole queue, not the one row standing at its head A message sent into a marmot group sometimes sticks on the unsealed icon and never leaves. It is not that message that is broken. Something ahead of it in the queue cannot be sent, and because the notary was handed one row at a time, that row was the queue. **The queue.** MarmotInnerEventDao.observeUnprocessedMarmotInnerEvents selected every unsent row for a key and returned `Flow`, so Room handed back the first one and dropped the rest: SELECT * FROM MarmotInnerEvent WHERE publicKey = :publicKey AND marmotGroupEventId IS NULL ORDER BY createdAt ASC The only exit from that queue is a successful send. MarmotOutboundDao.encryptAndSendMarmotInnerEvent stamps the row with the group event it became, inside the same @Transaction that writes the event, the NostrEvent and the BroadcastNostrEventRequests. Nothing else clears `marmotGroupEventId`, there is no attempt count, no failure status and no sweep. A row that cannot be sent therefore does not move, and it is selected again, and again, at the head of every later emission. Note what the filter is: the sender's key, not the room. One room whose MLS state is gone silences every group on the device. **The stopper.** DatabaseMarmotRepository.encryptAndSendMarmotInnerEvent was two nested `?.let`: database.chatRoomDao().findChatRoomById(...)?.let { localChatRoom -> localChatRoom.chatRoom.toMlsGroup()?.let { mlsGroup -> ... } } A row for a room this device holds no MLS state for -- the shape a room restored from an inbound gift wrap has -- fell out of both and returned. No send, no throw, no log, and no mark on the row. So the queue manufactured its own permanent head: a message that could never succeed, reported as though nothing had happened, sitting in front of everything else forever. MarmotOutboundDao.inviteMemberToChatRoom already throws MarmotMissingChatGroupException on exactly this condition, with a comment saying the point is to "say so instead of silently doing nothing and letting the caller report success". The send path disagreed with the invite path about the same missing group. **distinctUntilChanged.** Both notary collectors sat behind it, which is the wrong question to ask a work queue. A queue re-emits because its table changed; that is the signal to look again, not a duplicate to discard. Comparing an emission against the previous one asks "is this new work?" when the question is "is there work left?". The two queues then failed by different mechanics, which is worth writing down because it explains why the symptom looks like a retry loop in one place and a dead collector in the other: - MarmotInnerEvent.equals compares `logger`, an @Ignore'd `Logger.withTag(TAG)` initialised per instance. Kermit's `withTag` returns `Logger(this.config, tag)` -- a fresh object -- and neither Logger nor BaseLogger overrides equals, so two reads of one row are never equal. distinctUntilChanged suppressed nothing here, and the notary spent the session retrying the stopper and never looking past it. Correct behaviour by accident, resting on a field that is not part of the row. - GiftWrapPayload.equals is an honest value comparison with no logger in it. The refused payload's re-emission compared equal and was dropped, so after one refusal nothing on that queue was collected again for the life of the session, whatever was queued afterwards. **Why it reads as unsealed.** ChatMessageListViewModel picks its status icon off three relations, in order: a broadcast receipt, a broadcast request, a NostrEvent. A queued marmot message has none of them until the notary turns it into a kind:445, so it falls to the last branch -- KeyOff, "Unsealed message status". The icon is accurate. The message is exactly as unsealed as it looks, and will stay that way. **The gift wrap queue has it too, and a Welcome rides it.** MIP-02 addresses kind:444 to a joiner who holds no group state and cannot read a kind:445, so MarmotOutboundDao.deliveryWelcome queues one as a GiftWrapPayload deliberately -- marmot traffic on the NIP-17 path. That queue had the same single-row shape and no ORDER BY at all, so which row was "the head" was whatever SQLite returned first. Two known refusals leave a payload there with `giftWrapSealId` still null: sealGiftWrapPayload refuses outright to seal a non-Welcome payload belonging to an MLS room (65e4a3a, and the comment there already named the blockage this causes), and a Welcome whose joiner published no key package matches no participant and produces no wraps at all. **The fix.** Both queries return the backlog instead of its head, ordered `createdAt ASC, id ASC` -- the same ordering BroadcastNostrEventRequestDao settled on, and for the same reason: createdAt is persisted at second resolution, a burst of sends shares one, and an order that is only ever "some row with this timestamp" lets two passes disagree about what comes next. NotaryViewModel walks the list and keeps guardNotarization per row, so a failure costs only itself. It walks it serially and in order on purpose: each send ratchets its room's MLS state forward and writes it back, and encryptAndSendMarmotInnerEvent re-reads that state per row, so two sends for one room in parallel would encrypt from the same generation and the group could read only one of them. The two `?.let`s become two throws, which the per-row guard logs. A failed row writes nothing -- the DAO is one transaction -- so it stays queued and is tried again on the next pass. That is wanted: a room whose state has not caught up yet deserves the retry, and a room that never will is at least no longer standing in front of anybody. The retry is bounded by the fact that it is Room's invalidation driving it: a pass in which every remaining row fails writes nothing, invalidates nothing and emits nothing further. No schema change. The queue's shape was in the query and the collector, not in the table. **Tests.** MarmotOutboundQueueJvmTest, eight of them, Room-backed against a real MlsGroup -- the DAO seam MarmotOutboundDaoJvmTest opened, which 0211764 could not use and said so. Two rooms stand side by side, one holding real MLS state and one holding none, and the unsendable row is queued first on purpose because under the old queue it was the only row the notary ever saw. The one that matters is "a message that cannot be sent no longer holds up the ones behind it": the stopper fails, exactly once, and the message behind it in another room still comes out with a group event and one pending BroadcastNostrEventRequest per relay -- pending because, per 0211764, that is the only status the broadcaster looks at and the only thing that actually puts a kind:445 on a relay. The rest pin the supporting facts: the backlog arrives whole and oldest first, rows sharing a second come back in the same order twice, a room with no MLS state and a room that does not exist are each refused rather than ignored, a refused send writes neither a group event nor a broadcast request, and two messages for one room each ratchet the group forward and both leave the queue. **Not covered, deliberately.** The notary's third queue, unsigned Nostr events, still has this shape, and UnsignedNostrEvent.equals is a value comparison, so it is the frozen-collector variant rather than the retrying one. NostrDao.kt's comment on commitPublishedNostrEvent records that it has already bitten once -- an indexing throw rolled back `signedAt` and "every later event (including the MLS key package, which is enqueued last) would never be signed at all" -- fixed point-wise by moving indexing out of the transaction, leaving the queue shape untouched. It carries account traffic rather than group messages and NavigationViewModel gates the user on it, so it is its own change. Nothing here surfaces *why* a message is stuck. A row that can never be sent is still never sent; it is only no longer contagious. Telling the sender that would want a persisted attempt count and a place in the UI to put it, which is also its own change. Verified: :composeApp:compileDebugKotlinAndroid succeeds; 511 tests pass, 503 before these eight. That a stuck room no longer silences a healthy one is asserted against a real group in a real database, not inferred -- but that a second participant now receives the messages that were backing up is inference from the code, since it wants two devices. Co-Authored-By: Claude Opus 5 --- .../database/dao/GiftWrapPayloadDao.kt | 23 +- .../database/dao/MarmotInnerEventDao.kt | 26 +- .../repository/DatabaseChatRepository.kt | 8 +- .../repository/DatabaseMarmotRepository.kt | 38 ++- .../compose/repository/ChatRepository.kt | 11 +- .../compose/repository/MarmotRepository.kt | 11 +- .../compose/ui/view/model/NotaryViewModel.kt | 44 ++- .../dao/MarmotOutboundQueueJvmTest.kt | 301 ++++++++++++++++++ 8 files changed, 435 insertions(+), 27 deletions(-) create mode 100644 composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/MarmotOutboundQueueJvmTest.kt diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt index 1ff37322..bc35c1fe 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt @@ -13,8 +13,27 @@ interface GiftWrapPayloadDao { @Query("SELECT * FROM GiftWrapPayload") suspend fun getAllGiftWrapUnsigned(): List - @Query("SELECT * FROM GiftWrapPayload WHERE publicKey = :publicKey AND giftWrapSealId IS NULL") - fun observeUnsealedGiftWrapPayloads(publicKey: String): Flow + /** + * Everything this key has queued for sealing and not yet sealed, oldest first. + * + * A backlog, not a head, for the reason spelled out on + * `MarmotInnerEventDao.observeUnprocessedMarmotInnerEvents`: nothing takes a row off + * this queue but a successful seal, so one payload that cannot be sealed used to be + * the whole queue. That is not hypothetical here -- `sealGiftWrapPayload` refuses a + * payload belonging to an MLS room outright, and a Welcome whose joiner published no + * key package produces no wraps at all. Either one returns leaving `giftWrapSealId` + * null, and every payload behind it waited on a row that would never move. + * + * `id` breaks the tie because `createdAt` is stored to the second; there was no + * ordering here at all before, so which row was "the head" was whatever SQLite + * happened to return first. + */ + @Query( + "SELECT * FROM GiftWrapPayload " + + "WHERE publicKey = :publicKey AND giftWrapSealId IS NULL " + + "ORDER BY createdAt ASC, id ASC" + ) + fun observeUnsealedGiftWrapPayloads(publicKey: String): Flow> /** * Every payload of the given kinds this device holds for a room, oldest first. diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt index 37b43f32..35d1682d 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt @@ -9,8 +9,30 @@ import kotlinx.coroutines.flow.Flow @Dao interface MarmotInnerEventDao { - @Query("SELECT * FROM MarmotInnerEvent WHERE publicKey = :publicKey AND marmotGroupEventId IS NULL ORDER BY createdAt ASC") - fun observeUnprocessedMarmotInnerEvents(publicKey: String): Flow + /** + * Everything this key has queued for the group and not yet sent, oldest first. + * + * A backlog, not a head. The only exit from this queue is a successful send: the + * row is stamped with the group event it became, in the same transaction that + * writes that event. So a row that cannot be sent does not leave, and while the + * notary was handed one row at a time that row *was* the queue -- a message that + * will never encrypt (a room whose MLS state is gone, a direct message to a key + * that is no longer a member) sat at the head and every message queued behind it + * waited on it forever. Not just in its own room, either: the filter is the + * sender's key, so one wedged room silenced every group this device belongs to. + * Nothing reports it; the messages simply stay unsealed, which is what they are. + * + * Handing over the whole backlog lets the caller fail one row without losing the + * rest. It must still work through them in order and one at a time: each send + * ratchets the room's MLS state forward and saves it, and the next send in that + * room is encrypted against what the last one left behind. + * + * `id` breaks the tie because `createdAt` is stored to the second and a burst of + * sends shares one. Ordering that is only ever "some row with this timestamp" + * would let two passes disagree about what comes next. + */ + @Query("SELECT * FROM MarmotInnerEvent WHERE publicKey = :publicKey AND marmotGroupEventId IS NULL ORDER BY createdAt ASC, id ASC") + fun observeUnprocessedMarmotInnerEvents(publicKey: String): Flow> @Upsert suspend fun upsert(marmotInnerEvent: MarmotInnerEvent) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt index 80b04c00..5152230c 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt @@ -328,7 +328,7 @@ class DatabaseChatRepository( return null } - override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow { + override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow> { return database.giftWrapPayloadDao().observeUnsealedGiftWrapPayloads(publicKey) } @@ -349,8 +349,10 @@ class DatabaseChatRepository( // is the only way to reach them, and `MarmotOutboundDao.deliveryWelcome` queues one // here deliberately. Every Welcome carries its room's nostrGroupId, so a refusal // keyed on `mlsGroupState` alone catches all of them and no invite is ever - // delivered -- and because the unsealed queue is a single-row flow, the refused - // Welcome sits at its head and blocks every payload behind it too. + // delivered. The refusal below no longer strands the rest of the queue behind it + // -- the notary is handed the whole backlog and refuses one payload at a time -- + // but a refused Welcome is still an invite that never arrives, so the exception + // for it stays. if (giftWrapPayload.kind != WelcomeEvent.KIND) { val chatRoom = database.chatRoomDao().findChatRoomById(giftWrapPayload.chatRoomId) if (chatRoom?.chatRoom?.mlsGroupState != null) { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt index 8a644e57..e4cf72a6 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt @@ -4,6 +4,7 @@ import press.mantra.compose.database.MantraDatabase import press.mantra.compose.database.model.MarmotInnerEvent import press.mantra.compose.database.model.MarmotKeyPackageBundle import press.mantra.compose.database.model.UnsignedNostrEvent +import press.mantra.compose.exceptions.MarmotMissingChatGroupException import press.mantra.compose.extensions.toHex import press.mantra.compose.nostr.Relays import press.mantra.compose.repository.MarmotRepository @@ -120,7 +121,7 @@ class DatabaseMarmotRepository( return database.marmotKeyPackageBundleDao().getAllMarmotKeyPackageBundles(publicKey) } - override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow { + override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow> { return database.marmotInnerEventDao().observeUnprocessedMarmotInnerEvents(publicKey) } @@ -128,23 +129,36 @@ class DatabaseMarmotRepository( return database.marmotKeyPackageBundleDao().observeActiveMarmotKeyPackageBundle(publicKey) } + /** + * Both lookups used to be `?.let`, so a row for a room this device holds no MLS state + * for returned quietly, having done nothing. Nothing marks such a row, so it stayed + * queued and was handed back on the next pass, forever -- and while the queue was a + * single row, forever meant every message queued behind it too. Saying so is the same + * call `MarmotOutboundDao.inviteMemberToChatRoom` makes, for the same reason: there is + * no group to encrypt to, and reporting success is a lie the sender acts on. + */ override suspend fun encryptAndSendMarmotInnerEvent( marmotInnerEvent: MarmotInnerEvent, nostrSignerSync: NostrSignerSync ) { - database.chatRoomDao().findChatRoomById(marmotInnerEvent.chatRoomId)?.let { localChatRoom -> - localChatRoom.chatRoom.toMlsGroup()?.let { mlsGroup -> - logger.d("mlsGroup: $mlsGroup") + val localChatRoom = database.chatRoomDao().findChatRoomById(marmotInnerEvent.chatRoomId) + ?: throw MarmotMissingChatGroupException( + "Cannot send ${marmotInnerEvent.id}: no chat room ${marmotInnerEvent.chatRoomId}" + ) - database.marmotOutboundDao().encryptAndSendMarmotInnerEvent( - localChatRoom = localChatRoom, - mlsGroup = mlsGroup, - marmotInnerEvent = marmotInnerEvent, - nostrSignerSync = nostrSignerSync - ) - } + val mlsGroup = localChatRoom.chatRoom.toMlsGroup() + ?: throw MarmotMissingChatGroupException( + "Cannot send ${marmotInnerEvent.id}: ${marmotInnerEvent.chatRoomId} holds no MLS group state" + ) - } + logger.d("mlsGroup: $mlsGroup") + + database.marmotOutboundDao().encryptAndSendMarmotInnerEvent( + localChatRoom = localChatRoom, + mlsGroup = mlsGroup, + marmotInnerEvent = marmotInnerEvent, + nostrSignerSync = nostrSignerSync + ) } private fun generateKeyPackage( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt index 57592dd4..dcec7bec 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt @@ -108,7 +108,14 @@ interface ChatRepository { fun updateChatRoomSubject(publicKey: String, subject: String): ChatRoom? - suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow + /** + * The whole unsealed backlog for [publicKey], oldest first. + * + * See `GiftWrapPayloadDao.observeUnsealedGiftWrapPayloads`: a payload that cannot be + * sealed used to be the entire queue. Callers work through it and a refusal costs + * only the payload it refuses. + */ + suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow> suspend fun sealGiftWrapPayload(giftWrapPayload: GiftWrapPayload, nostrSignerSync: NostrSignerSync) @@ -241,7 +248,7 @@ interface ChatRepository { return null } - override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow { + override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow> { TODO("Not yet implemented") } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt index 44deb99f..c9875665 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt @@ -19,7 +19,14 @@ interface MarmotRepository { suspend fun getMarmotKeyPackageBundles(publicKey: HexKey): List - suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow + /** + * The whole outbound backlog for [publicKey], oldest first. + * + * See `MarmotInnerEventDao.observeUnprocessedMarmotInnerEvents` for why this is a + * list: a row that cannot be sent used to be the entire queue. Callers work + * through it in order, and a row that fails costs only itself. + */ + suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow> suspend fun observeActiveMarmotKeyPackageBundle(publicKey: HexKey): Flow @@ -46,7 +53,7 @@ interface MarmotRepository { return emptyList() } - override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow { + override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow> { TODO("Not yet implemented") } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt index d87b080d..6875c636 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt @@ -151,6 +151,16 @@ class NotaryViewModel( } + /** + * Seal everything queued for sealing, not just the oldest thing. + * + * Same shape as [observeUnprocessedMarmotInnerEvents] and a harder stop. A payload + * `sealGiftWrapPayload` declines to seal keeps `giftWrapSealId` null, so it came back + * as the head of the next emission -- and `GiftWrapPayload.equals` is a plain value + * comparison, so `distinctUntilChanged` read that as nothing having changed and + * dropped it. One refusal and nothing on this queue was collected again for the rest + * of the session, whatever was queued afterwards. + */ private suspend fun observeUnsealedGiftWrapPayloads( keyPair: KeyPair ) { @@ -160,8 +170,8 @@ class NotaryViewModel( logger.i { "observeUnsealedGiftWrapPayloads: ${keyPair.pubKey.toHexKey()}" } chatRepository.observeUnsealedGiftWrapPayloads( publicKey = keyPair.pubKey.toHexKey() - ).distinctUntilChanged().collect { giftWrapPayloadOrNull -> - giftWrapPayloadOrNull?.let { giftWrapPayload -> + ).collect { giftWrapPayloads -> + giftWrapPayloads.forEach { giftWrapPayload -> guardNotarization("gift wrap payload ${giftWrapPayload.id}") { logger.d("seal and deliver giftWrapPayload: $giftWrapPayload") @@ -174,6 +184,28 @@ class NotaryViewModel( } } + /** + * Encrypt and send everything a marmot room has queued, not just the oldest thing. + * + * The queue used to arrive one row at a time and the row only left it by being sent, + * so a message that could never be encrypted was a stopper: it came back at the head + * of every later emission and everything behind it -- in every room this key belongs + * to -- waited on a send that was never going to happen. Those messages show in the + * chat as unsealed, and stay that way. + * + * `distinctUntilChanged` went with it. A queue re-emits because its table changed, + * which is the signal to look again; comparing that emission against the last one + * asks "is this new work?" when the question is "is there work left?". Here it + * happened to suppress nothing -- `MarmotInnerEvent.equals` compares an `@Ignore`d + * per-instance Logger, so no two reads of a row are ever equal -- which is not a + * property to leave a queue standing on. Next door, where the payload's equals is an + * honest value comparison, it dropped every emission after the first refusal. + * + * Serially, in the order the query gives them: each send ratchets its room's MLS + * state forward and writes it back, and `encryptAndSendMarmotInnerEvent` reads that + * state fresh per row. Two sends for one room in parallel would encrypt from the same + * generation and one of them would be undecryptable. + */ private suspend fun observeUnprocessedMarmotInnerEvents( keyPair: KeyPair ) { @@ -183,8 +215,12 @@ class NotaryViewModel( logger.i { "observeUnprocessedMarmotInnerEvents: ${keyPair.pubKey.toHexKey()}" } marmotRepository.observeUnprocessedMarmotInnerEvents( publicKey = keyPair.pubKey.toHexKey() - ).distinctUntilChanged().collect { marmotInnerEventOrNull -> - marmotInnerEventOrNull?.let { marmotInnerEvent -> + ).collect { marmotInnerEvents -> + marmotInnerEvents.forEach { marmotInnerEvent -> + // Per row, so one that cannot be encrypted costs only itself. It stays + // queued and is tried again on the next pass -- a room whose state has + // not caught up yet is worth retrying, and one that never will is at + // least no longer standing in front of anybody. guardNotarization("marmot inner event ${marmotInnerEvent.id}") { logger.d("encrypt and broadcast: $marmotInnerEvent") diff --git a/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/MarmotOutboundQueueJvmTest.kt b/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/MarmotOutboundQueueJvmTest.kt new file mode 100644 index 00000000..56dab9c0 --- /dev/null +++ b/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/MarmotOutboundQueueJvmTest.kt @@ -0,0 +1,301 @@ +package press.mantra.compose.database.dao + +import androidx.room3.Room +import com.vitorpamplona.quartz.marmot.mip01Groups.MarmotGroupData +import com.vitorpamplona.quartz.marmot.mls.group.MlsGroup +import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray +import com.vitorpamplona.quartz.nip01Core.core.toHexKey +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nipC7Chats.ChatEvent +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.runBlocking +import press.mantra.compose.database.MantraDatabase +import press.mantra.compose.database.builder.getRoomDatabase +import press.mantra.compose.database.model.ChatRoom +import press.mantra.compose.database.model.MarmotInnerEvent +import press.mantra.compose.database.model.NostrEvent +import press.mantra.compose.database.model.Profile +import press.mantra.compose.database.repository.DatabaseMarmotRepository +import press.mantra.compose.exceptions.MarmotMissingChatGroupException +import press.mantra.compose.extensions.toHex +import press.mantra.compose.nostr.Relays +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue +import kotlin.time.Instant + +/** + * The outbound queue the notary drains, and what a message it cannot send does to the rest. + * + * Nothing takes a row off this queue but a successful send: `encryptAndSendMarmotInnerEvent` + * stamps the row with the group event it became, in the transaction that writes that event. + * So a row that can never be sent never leaves, and while the queue handed back one row at a + * time, that row was the queue. Every message behind it -- in every room the key belongs to, + * because the filter is the sender rather than the room -- waited on a send that was not + * coming, and showed in the chat as unsealed for as long as the user cared to look. + * + * These stand the two rooms side by side: one holding real MLS state, one holding none, which + * is the shape a room restored from an inbound gift wrap has. The unsendable row is queued + * *first* on purpose. Under the old queue it was the only row the notary ever saw. + */ +class MarmotOutboundQueueJvmTest { + + private val db: MantraDatabase = getRoomDatabase( + Room.inMemoryDatabaseBuilder() + ) + + @AfterTest + fun closeDb() = db.close() + + private val keyPair = KeyPair() + private val user = keyPair.pubKey.toHexKey() + private val signer = NostrSignerSync(keyPair) + + private val repository = DatabaseMarmotRepository( + database = db, + scope = CoroutineScope(Dispatchers.Default), + ) + + private val statelessRoomId = "a".repeat(64) + private val mlsRoomId = "b".repeat(64) + + /** Participant/ChatRoom rows point at a Profile, which points at the event it came from. */ + private suspend fun seedProfile() { + val nostrEventId = "c".repeat(64) + db.nostrEventDao().upsert( + NostrEvent( + id = nostrEventId, + pubKey = user, + kind = 0, + tags = emptyArray(), + content = "{}", + sig = "0".repeat(128), + ) + ) + db.profileDao().upsert(Profile(publicKey = user, userName = "user", nostrEventId = nostrEventId)) + } + + /** A room with `mlsGroupState = null`: there is no group here to encrypt anything to. */ + private suspend fun seedStatelessRoom() { + db.chatRoomDao().upsert( + ChatRoom( + id = statelessRoomId, + userPublicKey = user, + subject = "a room restored without its group", + description = null, + mlsGroupState = null, + ) + ) + } + + /** A room holding real MLS state, as a room this device created does. */ + private suspend fun seedMlsRoom() { + val mlsGroup = MlsGroup.create( + identity = user.hexToByteArray(), + initialExtensions = listOf( + MarmotGroupData.bootstrap( + nostrGroupId = mlsRoomId, + creatorPubKey = user, + outboxRelays = Relays.DefaultDMRelayList.map { it.url }, + ).toExtension() + ), + ) + db.chatRoomDao().upsert( + ChatRoom( + id = mlsRoomId, + userPublicKey = user, + subject = "a room this device created", + description = null, + mlsGroupState = mlsGroup.saveState().encodeTls().toHex(), + ) + ) + } + + private suspend fun queue(id: String, chatRoomId: String, text: String, atEpochSeconds: Long): MarmotInnerEvent { + val row = MarmotInnerEvent( + id = id, + publicKey = user, + chatRoomId = chatRoomId, + kind = ChatEvent.KIND, + tags = emptyArray(), + content = text, + createdAt = Instant.fromEpochSeconds(atEpochSeconds), + ) + db.marmotInnerEventDao().upsert(row) + return row + } + + private suspend fun backlog() = db.marmotInnerEventDao() + .observeUnprocessedMarmotInnerEvents(user) + .first() + + /** + * What the notary does with an emission, reduced to the part under test: every row, in + * order, each one's failure its own. Returns what failed, so a test can assert that a row + * was refused rather than quietly skipped. + */ + private suspend fun drain(): List = backlog().mapNotNull { row -> + runCatching { repository.encryptAndSendMarmotInnerEvent(row, signer) }.exceptionOrNull() + } + + @Test + fun `the queue hands back the whole backlog, oldest first`() = runBlocking { + seedProfile() + seedMlsRoom() + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "second", atEpochSeconds = 200) + queue(id = "1".repeat(64), chatRoomId = mlsRoomId, text = "first", atEpochSeconds = 100) + + assertEquals( + listOf("first", "second"), + backlog().map { it.content }, + "the notary is handed the backlog to work through, not just its head", + ) + } + + /** + * `createdAt` is stored to the second, so a burst of sends shares one. Without the `id` + * tiebreak the order within that second is SQLite's business, and two passes over the same + * backlog need not agree on it. + */ + @Test + fun `rows sharing a second come back in a stable order`() = runBlocking { + seedProfile() + seedMlsRoom() + queue(id = "f".repeat(64), chatRoomId = mlsRoomId, text = "f", atEpochSeconds = 100) + queue(id = "0".repeat(64), chatRoomId = mlsRoomId, text = "0", atEpochSeconds = 100) + queue(id = "7".repeat(64), chatRoomId = mlsRoomId, text = "7", atEpochSeconds = 100) + + assertEquals(listOf("0", "7", "f"), backlog().map { it.content }) + assertEquals(listOf("0", "7", "f"), backlog().map { it.content }, "the order changed between passes") + } + + /** + * Both lookups were `?.let`, so this returned having done nothing at all: no send, no + * error, and no mark on the row -- which is what made it a permanent head of the queue. + */ + @Test + fun `a message for a room with no mls state is refused rather than ignored`() = runBlocking { + seedProfile() + seedStatelessRoom() + val row = queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "nowhere to send this", atEpochSeconds = 100) + + assertFailsWith { + repository.encryptAndSendMarmotInnerEvent(row, signer) + } + } + + @Test + fun `a message for a room this device does not have is refused rather than ignored`() = runBlocking { + seedProfile() + val row = queue(id = "1".repeat(64), chatRoomId = "d".repeat(64), text = "no such room", atEpochSeconds = 100) + + assertFailsWith { + repository.encryptAndSendMarmotInnerEvent(row, signer) + } + } + + /** + * The one that matters. The unsendable row is the oldest, so it is what a queue that + * yields only its head would offer forever; the message behind it is in a different room + * and has everything it needs to go out. + */ + @Test + fun `a message that cannot be sent no longer holds up the ones behind it`() = runBlocking { + seedProfile() + seedStatelessRoom() + seedMlsRoom() + queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "the stopper", atEpochSeconds = 100) + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "sent anyway", atEpochSeconds = 200) + + val failures = drain() + + assertEquals(1, failures.size, "exactly the stopper should have failed") + assertTrue(failures.single() is MarmotMissingChatGroupException) + + val sent = assertNotNull(db.marmotInnerEventDao().getByChatRoomAndKinds(mlsRoomId, listOf(ChatEvent.KIND)).singleOrNull()) + assertNotNull(sent.marmotGroupEventId, "the message behind the stopper was never sent") + assertEquals( + listOf("the stopper"), + backlog().map { it.content }, + "the stopper stays queued and nothing else does", + ) + } + + /** + * Sent means handed to the broadcaster. `BroadcastNostrEventRequest` rows in "pending" are + * the only thing that puts a kind:445 on a relay, so a row that got a group event but no + * request is still a message nobody receives -- see 0211764. + */ + @Test + fun `a message past the stopper is queued for the relays`() = runBlocking { + seedProfile() + seedStatelessRoom() + seedMlsRoom() + queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "the stopper", atEpochSeconds = 100) + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "sent anyway", atEpochSeconds = 200) + + drain() + + val groupEventId = assertNotNull( + db.marmotInnerEventDao().getByChatRoomAndKinds(mlsRoomId, listOf(ChatEvent.KIND)).single().marmotGroupEventId + ) + val requests = db.broadcastNostrEventRequestDao().getAllBroadcastNostrEventRequests() + .filter { it.nostrEventId == groupEventId } + + assertEquals( + Relays.DefaultDMRelayList.size, + requests.size, + "one broadcast request per relay is what actually sends it", + ) + assertTrue(requests.all { it.status == "pending" }, "pending is the only status the broadcaster looks at") + } + + /** + * Two messages queued for the same room go out in order and each under its own generation: + * the send ratchets the room's MLS state forward and writes it back, and the next one is + * encrypted against what it left behind. A drain that read the room once and reused it + * would send both from the same generation, and the group could only read one of them. + */ + @Test + fun `two messages for one room each ratchet the group forward`() = runBlocking { + seedProfile() + seedMlsRoom() + val stateBefore = assertNotNull(db.chatRoomDao().findChatRoomById(mlsRoomId)).chatRoom.mlsGroupState + queue(id = "1".repeat(64), chatRoomId = mlsRoomId, text = "first", atEpochSeconds = 100) + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "second", atEpochSeconds = 200) + + assertEquals(emptyList(), drain()) + + assertEquals(emptyList(), backlog(), "both messages should have left the queue") + val sent = db.marmotInnerEventDao().getByChatRoomAndKinds(mlsRoomId, listOf(ChatEvent.KIND)) + assertEquals(2, sent.mapNotNull { it.marmotGroupEventId }.distinct().size, "both were sent as the same event") + assertTrue( + assertNotNull(db.chatRoomDao().findChatRoomById(mlsRoomId)).chatRoom.mlsGroupState != stateBefore, + "the ratchet never reached the database", + ) + } + + /** + * A refused send must leave nothing behind. `encryptAndSendMarmotInnerEvent` is one + * transaction, so a failure that had written a group event or a broadcast request would + * have the room believing it sent something the members will never see. + */ + @Test + fun `a refused send writes nothing`() = runBlocking { + seedProfile() + seedStatelessRoom() + val row = queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "the stopper", atEpochSeconds = 100) + + runCatching { repository.encryptAndSendMarmotInnerEvent(row, signer) } + + assertNull(backlog().single().marmotGroupEventId, "the refused row was marked as sent") + assertEquals(emptyList(), db.broadcastNostrEventRequestDao().getAllBroadcastNostrEventRequests()) + } +}