From 925099125b90c41155da3bae36ec21cea0185598 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sun, 6 Sep 2026 01:49:57 +0200 Subject: [PATCH] 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 --- composeApp/build.gradle.kts | 4 + .../compose/database/dao/ChatMessageDao.kt | 20 + .../mantra/compose/database/dao/NostrDao.kt | 348 ++++++++++++------ .../compose/database/dao/NostrEventDao.kt | 20 + .../compose/database/model/ChatMessage.kt | 67 +++- .../model/types/MarmotReindexReport.kt | 27 ++ .../repository/DatabaseChatRepository.kt | 18 + .../compose/managers/MarmotInboundManager.kt | 25 ++ .../compose/managers/MarmotReindexSweep.kt | 95 +++++ .../compose/repository/ChatRepository.kt | 19 + .../ui/composable/ChatRoomDetailScreen.kt | 90 +++++ .../ui/view/model/ChatRoomDetailViewModel.kt | 55 +++ .../model/MarmotChatLineReconciliationTest.kt | 161 ++++++++ .../model/MarmotGroupEventRoomFilterTest.kt | 85 +++++ .../managers/MarmotReindexSweepTest.kt | 173 +++++++++ gradle/libs.versions.toml | 1 + 16 files changed, 1098 insertions(+), 110 deletions(-) create mode 100644 composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/types/MarmotReindexReport.kt create mode 100644 composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotReindexSweep.kt create mode 100644 composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotChatLineReconciliationTest.kt create mode 100644 composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotGroupEventRoomFilterTest.kt create mode 100644 composeApp/src/commonTest/kotlin/press/mantra/compose/managers/MarmotReindexSweepTest.kt diff --git a/composeApp/build.gradle.kts b/composeApp/build.gradle.kts index 71501a81..a951896c 100644 --- a/composeApp/build.gradle.kts +++ b/composeApp/build.gradle.kts @@ -121,6 +121,10 @@ kotlin { } commonTest.dependencies { implementation(libs.kotlin.test) + // runTest: the DAO, model and relay layers are all suspending, and there is no + // runBlocking in a common source set — so anything worth asserting about them + // needs a coroutine, and a scheduler, to assert it in. + implementation(libs.kotlinx.coroutinesTest) } jvmMain.dependencies { implementation(compose.desktop.currentOs) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt index daa45b01..d0657be0 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt @@ -30,6 +30,26 @@ interface ChatMessageDao { @Query("SELECT COUNT(*) FROM ChatMessage WHERE chatRoomId = :chatRoomId AND senderPublicKey = :senderPublicKey") fun countChatMessagesBySenderPublicKey(chatRoomId: HexKey, senderPublicKey: HexKey): Int + /** + * The room's group events that already have something to show for them. + * + * One query so a replay can skip the events it has no business touching without + * a lookup per event -- `marmotGroupEventId` carries no index, so each of those + * is a scan of the whole table. + * + * [unresolvedTypes] is [press.mantra.compose.database.model.ChatMessage.UNRESOLVED_MARMOT_TYPES]: + * lines that stand in for a message still to come rather than reporting one. + */ + @Query( + "SELECT marmotGroupEventId FROM ChatMessage " + + "WHERE chatRoomId = :chatRoomId AND marmotGroupEventId IS NOT NULL " + + "AND messageType NOT IN (:unresolvedTypes)" + ) + suspend fun getResolvedMarmotGroupEventIds( + chatRoomId: String, + unresolvedTypes: Collection + ): List + @Upsert suspend fun upsert(chatMessage: press.mantra.compose.database.model.ChatMessage): Long } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt index 0aaba9be..f3344e5b 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt @@ -19,6 +19,7 @@ import press.mantra.compose.database.model.Participant import press.mantra.compose.database.model.Post import press.mantra.compose.database.model.Profile import press.mantra.compose.database.model.intermdiate.LocalChatRoom +import press.mantra.compose.database.model.types.MarmotReindexReport import press.mantra.compose.database.model.types.SynchronizationFilter import press.mantra.compose.exceptions.GiftWrapImpersonationException import press.mantra.compose.exceptions.GiftWrapSealDecryptionException @@ -35,12 +36,15 @@ import press.mantra.compose.nostr.frost.FrostSigningEvents import press.mantra.compose.managers.FrostSigningManager import press.mantra.compose.managers.MlsGroupCache import press.mantra.compose.managers.MarmotInboundManager +import press.mantra.compose.managers.MarmotReindexSweep import co.touchlab.kermit.Logger import kotlinx.coroutines.CancellationException import com.vitorpamplona.quartz.marmot.GroupEventResult import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageEvent import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageRelayListEvent import com.vitorpamplona.quartz.marmot.mip02Welcome.WelcomeEvent +import com.vitorpamplona.quartz.marmot.mip03GroupMessages.CommitOrdering +import com.vitorpamplona.quartz.marmot.mip03GroupMessages.GroupEvent import com.vitorpamplona.quartz.marmot.mls.codec.TlsReader import com.vitorpamplona.quartz.marmot.mls.group.MlsGroup import com.vitorpamplona.quartz.marmot.mls.messages.KeyPackageBundle @@ -386,114 +390,11 @@ abstract class NostrDao( logger.d("groupEvent: ${groupEvent.toJson()}") groupEvent.groupId()?.let { chatRoomId -> - val localChatRoom = database.chatRoomDao().findChatRoomById(chatRoomId) - - if (localChatRoom != null) { - // Through the cache rather than rebuilt here, so the secret - // tree's skipped-generation keys survive from one message to - // the next. Two events published in the same instant arrive in - // whatever order the relay feels like, and rebuilding between - // them loses the earlier one for good -- see MlsGroupCache. - val handled = MlsGroupCache.withGroup( - chatRoomId = chatRoomId, - storedStateHex = localChatRoom.chatRoom.mlsGroupState, - build = { localChatRoom.chatRoom.toMlsGroup() }, - save = { stateHex -> - database.chatRoomDao().upsert( - localChatRoom.chatRoom.copy(mlsGroupState = stateHex) - ) - } - ) { mlsGroup -> - val memberPubkeys = mlsGroup.members().mapNotNull { (leafIndex, leafNode) -> - - val pubkey = when (val cred = leafNode.credential) { - is Credential.Basic -> cred.identity.toHexKey() - else -> null - } - - logger.d("leafIndex=$leafIndex pubKey=$pubkey") - pubkey - } - - if (memberPubkeys.contains(activeKeyPair.pubKey.toHex())) { - database.marmotGroupEventDao().upsert( - MarmotGroupEvent( - id = groupEvent.id, - userPublicKey = activeKeyPair.pubKey.toHex(), - publicKey = groupEvent.pubKey, - encryptedContent = groupEvent.encryptedContent(), - chatRoomId = chatRoomId, - createdAt = Instant.fromEpochSeconds(groupEvent.createdAt), - expiresAt = groupEvent.expiration()?.let { Instant.fromEpochSeconds(it) }, - signature = groupEvent.sig - ) - ) - MarmotInboundManager.processGroupEvent( - database = database, - activeKeyPair = activeKeyPair, - localChatRoom = localChatRoom, - mlsGroup = mlsGroup, - groupEvent = groupEvent - )?.let { groupEventResult -> - logger.d("groupEventResult: $groupEventResult") - - when(groupEventResult) { - is GroupEventResult.CommitProcessed -> { - MarmotInboundManager.processGroupMembershipChanges( - database = database, - mlsGroup = mlsGroup, - localChatRoom = localChatRoom - ) - } - else -> { - logger.i("No app logic to handle: $groupEventResult") - } - } - - ChatMessage.fromGroupEventResult( - database = database, - activeKeyPair = activeKeyPair, - groupEvent = groupEvent, - groupEventResult = groupEventResult, - )?.let { chatMessage -> - logger.d("chatMessage: $chatMessage") - database.chatMessageDao().upsert( - chatMessage - ) - } - - // A FROST signing message for this group. Driven from - // here rather than from ChatMessage because the - // manager needs the room to publish its own replies - // into, and because it writes its transcript lines - // itself. The manager is idempotent, so a redelivered - // message re-runs a step it has already taken. - if (groupEventResult is GroupEventResult.ApplicationMessage) { - Event.fromJsonOrNull(groupEventResult.innerEventJson) - ?.takeIf { FrostSigningEvents.isFrostSigningKind(it.kind) } - ?.let { innerEvent -> - FrostSigningManager.processSigningPayload( - database = database, - localChatRoom = localChatRoom, - innerEvent = innerEvent, - userPublicKey = activeKeyPair.pubKey.toHex() - ) - } - } - } - } else { - throw MarmotNotMemberOfChatGroupException("We are not a member of the chat room ${localChatRoom.chatRoom.id}") - } - } - - // Null means the room has no usable group state, which is what - // a failed toMlsGroup() meant before the cache existed. - if (handled == null) { - throw MarmotMissingNostrGroupDataExtension("Couldn't find chatRoom for $nostrEvent") - } - } else { - throw MarmotMissingChatGroupException("Couldn't find chatRoom for $nostrEvent") - } + indexMarmotGroupEvent( + groupEvent = groupEvent, + chatRoomId = chatRoomId, + activeKeyPair = activeKeyPair, + ) } } @@ -1279,6 +1180,236 @@ abstract class NostrDao( } } + /** + * Reads one kind:445 against the room's MLS group and files whatever it turns + * out to be: a message, a commit that advances the group, an entity for the + * library, or a placeholder saying it could not be read yet. + * + * Its own function so a replay can run exactly what a first delivery ran -- + * see [reindexMarmotGroupEvents]. Throws for a room this device cannot process + * the event against at all, which the caller decides what to do about: a first + * delivery lets it roll back its transaction, a replay logs it and moves on. + */ + private suspend fun indexMarmotGroupEvent( + groupEvent: GroupEvent, + chatRoomId: String, + activeKeyPair: KeyPair, + ) { + val localChatRoom = database.chatRoomDao().findChatRoomById(chatRoomId) + ?: throw MarmotMissingChatGroupException("Couldn't find chatRoom for ${groupEvent.id}") + + // Through the cache rather than rebuilt here, so the secret + // tree's skipped-generation keys survive from one message to + // the next. Two events published in the same instant arrive in + // whatever order the relay feels like, and rebuilding between + // them loses the earlier one for good -- see MlsGroupCache. + val handled = MlsGroupCache.withGroup( + chatRoomId = chatRoomId, + storedStateHex = localChatRoom.chatRoom.mlsGroupState, + build = { localChatRoom.chatRoom.toMlsGroup() }, + save = { stateHex -> + database.chatRoomDao().upsert( + localChatRoom.chatRoom.copy(mlsGroupState = stateHex) + ) + } + ) { mlsGroup -> + val memberPubkeys = mlsGroup.members().mapNotNull { (leafIndex, leafNode) -> + + val pubkey = when (val cred = leafNode.credential) { + is Credential.Basic -> cred.identity.toHexKey() + else -> null + } + + logger.d("leafIndex=$leafIndex pubKey=$pubkey") + pubkey + } + + if (memberPubkeys.contains(activeKeyPair.pubKey.toHex())) { + database.marmotGroupEventDao().upsert( + MarmotGroupEvent( + id = groupEvent.id, + userPublicKey = activeKeyPair.pubKey.toHex(), + publicKey = groupEvent.pubKey, + encryptedContent = groupEvent.encryptedContent(), + chatRoomId = chatRoomId, + createdAt = Instant.fromEpochSeconds(groupEvent.createdAt), + expiresAt = groupEvent.expiration()?.let { Instant.fromEpochSeconds(it) }, + signature = groupEvent.sig + ) + ) + MarmotInboundManager.processGroupEvent( + database = database, + activeKeyPair = activeKeyPair, + localChatRoom = localChatRoom, + mlsGroup = mlsGroup, + groupEvent = groupEvent + )?.let { groupEventResult -> + logger.d("groupEventResult: $groupEventResult") + + when(groupEventResult) { + is GroupEventResult.CommitProcessed -> { + MarmotInboundManager.processGroupMembershipChanges( + database = database, + mlsGroup = mlsGroup, + localChatRoom = localChatRoom + ) + } + else -> { + logger.i("No app logic to handle: $groupEventResult") + } + } + + ChatMessage.fromGroupEventResult( + database = database, + activeKeyPair = activeKeyPair, + groupEvent = groupEvent, + groupEventResult = groupEventResult, + )?.let { chatMessage -> + logger.d("chatMessage: $chatMessage") + persistMarmotChatMessage(chatMessage) + } + + // A FROST signing message for this group. Driven from + // here rather than from ChatMessage because the + // manager needs the room to publish its own replies + // into, and because it writes its transcript lines + // itself. The manager is idempotent, so a redelivered + // message re-runs a step it has already taken. + if (groupEventResult is GroupEventResult.ApplicationMessage) { + Event.fromJsonOrNull(groupEventResult.innerEventJson) + ?.takeIf { FrostSigningEvents.isFrostSigningKind(it.kind) } + ?.let { innerEvent -> + FrostSigningManager.processSigningPayload( + database = database, + localChatRoom = localChatRoom, + innerEvent = innerEvent, + userPublicKey = activeKeyPair.pubKey.toHex() + ) + } + } + } + } else { + throw MarmotNotMemberOfChatGroupException("We are not a member of the chat room ${localChatRoom.chatRoom.id}") + } + } + + // Null means the room has no usable group state, which is what + // a failed toMlsGroup() meant before the cache existed. + if (handled == null) { + throw MarmotMissingNostrGroupDataExtension("Couldn't find chatRoom for ${groupEvent.id}") + } + } + + /** + * Files the chat line a kind:445 produced, replacing the line that event + * already has rather than adding a second one. The rule for which of the two + * wins is [ChatMessage.reconcileMarmotLine]. + */ + private suspend fun persistMarmotChatMessage(chatMessage: ChatMessage) { + val existing = chatMessage.marmotGroupEventId?.let { + database.chatMessageDao().getChatMessagesByMarmotGroupEventId(it) + } + + val row = ChatMessage.reconcileMarmotLine(fresh = chatMessage, existing = existing) + + if (row == null) { + logger.d("Keeping ${existing?.messageType} line for ${chatMessage.marmotGroupEventId}") + return + } + + database.chatMessageDao().upsert(row) + } + + /** + * Reads a room's stored kind:445 events again, oldest first. + * + * Relays impose no ordering, so a group event can arrive before the group is + * able to read it: an application message encrypted under an epoch whose commit + * has not landed yet, or a commit for an epoch ahead of the local one. Both are + * stored and then dropped -- `MarmotInboundManager` refuses to apply an + * out-of-epoch commit precisely so it does not half-mutate the group -- and + * nothing revisits them once the missing event fills the gap. This is the + * revisit, driven from the group's detail screen. + * + * Only events with nothing to show for them are replayed: no chat line at all, + * or one of [ChatMessage.UNRESOLVED_MARMOT_TYPES]. Everything already read is + * left untouched, which is what keeps the replay from disturbing a room where + * there was nothing wrong. Within that set a replay is safe to repeat: the ids + * of every row it writes are derived from the events themselves, commits that + * are behind the current epoch are rejected as duplicates before they can touch + * the group, and the one row with a generated id is reconciled by + * [persistMarmotChatMessage]. + * + * Passes repeat while a pass recovers something -- see [MarmotReindexSweep] -- + * because created_at order is not necessarily epoch order, and a commit + * recovered by one pass can be what lets the next read the messages that were + * waiting on it. + * + * What this cannot do is recover a message whose key is gone: an application + * message the ratchet has already advanced past, or one from an epoch that + * predates this device joining. Those stay unreadable however often they are + * replayed. + */ + open suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + activeKeyPair: KeyPair, + ): MarmotReindexReport { + val userPublicKey = activeKeyPair.pubKey.toHex() + + // The query's LIKE only narrows; the h tag is what decides the room. Sorted + // by CommitOrdering's own comparator rather than left in createdAt order, so + // that where two commits compete for an epoch the first one replayed is the + // one every other member also picked. + val groupEvents = database.nostrEventDao() + .getMarmotGroupNostrEventsByChatRoomId(chatRoomId) + .mapNotNull { storedNostrEvent -> + storedNostrEvent + .toGroupEvent(userPublicKey = userPublicKey) + ?.takeIf { it.groupId() == chatRoomId } + } + .sortedWith(CommitOrdering.comparator) + + logger.i("Reindex $chatRoomId: ${groupEvents.size} stored group event(s)") + + // Held commits are what a replay is most often for, and they are only ever + // cleared by one applying -- see MarmotInboundManager.forgetPendingCommits. + MarmotInboundManager.forgetPendingCommits(chatRoomId) + + val resolved = database.chatMessageDao().getResolvedMarmotGroupEventIds( + chatRoomId = chatRoomId, + unresolvedTypes = ChatMessage.UNRESOLVED_MARMOT_TYPES, + ).toSet() + + return MarmotReindexSweep.run( + stored = groupEvents.size, + unresolved = groupEvents.filterNot { it.id in resolved }, + replay = { groupEvent -> + indexMarmotGroupEvent( + groupEvent = groupEvent, + chatRoomId = chatRoomId, + activeKeyPair = activeKeyPair, + ) + }, + isUnresolved = { groupEvent -> isUnresolvedMarmotGroupEvent(groupEvent.id) }, + ).also { + logger.i("Reindex $chatRoomId: $it") + } + } + + /** + * Whether one group event still has nothing to show for it, asked again after + * a replay to see whether that replay achieved anything. + * + * Some events can never leave this state and will be swept every time: a commit + * rejected as a duplicate writes no line, and neither does a commit this device + * sent, whose line was written against the send rather than the event. Both cost + * one refused decrypt per sweep, which is cheaper than a bookkeeping column that + * would have to be migrated in. + */ + private suspend fun isUnresolvedMarmotGroupEvent(groupEventId: String): Boolean { + val chatMessage = database.chatMessageDao().getChatMessagesByMarmotGroupEventId(groupEventId) + return chatMessage == null || chatMessage.messageType in ChatMessage.UNRESOLVED_MARMOT_TYPES + } /** * Files an arriving NIP-17 chat message so it shows up in the room's feed. @@ -1425,4 +1556,5 @@ abstract class NostrDao( } } } + } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt index 1066c814..f354e547 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt @@ -143,6 +143,26 @@ interface NostrEventDao { now: Instant = Clock.System.now(), ): List + /** + * Every kind:445 held locally that mentions [chatRoomId], oldest first. + * + * The `LIKE` is a prefilter over the serialised tags, not the test: it can match + * a group id that happens to appear in some other tag, so callers must confirm + * the event's own `h` tag before treating it as this room's. Ordered ascending + * because a replay has to apply commits in the order they were sent. + * + * Deliberately not a join on `MarmotGroupEvent`: an event whose indexing failed + * part way may have been stored without ever reaching that table, and those are + * exactly the ones worth replaying. + */ + @Transaction + @Query( + "SELECT * FROM NostrEvent WHERE kind = ${GroupEvent.KIND} " + + "AND tags LIKE '%' || :chatRoomId || '%' " + + "ORDER BY createdAt ASC" + ) + fun getMarmotGroupNostrEventsByChatRoomId(chatRoomId: HexKey): List + @Transaction @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND pubKey in (:authors) AND createdAt > :since ORDER BY createdAt ASC LIMIT :limit") fun getAuthoredNostrEvents( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt index feba4e8c..f6df2e83 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt @@ -249,6 +249,69 @@ data class ChatMessage( TYPE_DKG_FAILED, ) + /** + * A kind:445 arrived that the group could not read at the time. + * + * Both are placeholders for a message still to come: the outer layer was + * encrypted under an epoch whose exporter secret we did not hold yet, or + * the commit is one of several competing for an epoch and none has been + * applied. Neither says anything a member wrote. + */ + const val TYPE_UNDECRYPTABLE_OUTER_LAYER = "undecryptableOuterLayer" + const val TYPE_PENDING_COMMIT = "pendingCommit" + + /** + * The lines a reindex is allowed to replace. + * + * A group event that left one of these behind was never read, so replaying + * it can only improve on what is there. Every other line -- a message, an + * applied commit, anything this device sent -- is the final word on its + * group event and a replay must leave it alone. + */ + val UNRESOLVED_MARMOT_TYPES = setOf( + TYPE_UNDECRYPTABLE_OUTER_LAYER, + TYPE_PENDING_COMMIT, + ) + + /** + * The row to write for a group event that was just read, given the line that + * event already has. Null means leave what is there alone. + * + * [ChatMessage.id] is autogenerated, so writing [fresh] as it comes always + * inserts. That is right the first time a group event is read and wrong every + * time after: a replay would leave the room showing each recovered message + * twice, once as "Undecryptable Message" and once as itself. A group event has + * exactly one line and a deterministic id, so [existing] is what makes a + * re-read an update instead. + * + * A placeholder never overwrites a line that says something. A group event + * this device sent already has its line, written on the way out, and a replay + * that cannot read back our own ratchet-consumed message must not replace it + * with "Undecryptable Message". + * + * [viewedAt] and [savedAt] stay with [existing]: when a line was first seen is + * a fact about the reader, not about the event, and a re-read is not a new + * arrival. Its id stays too, so the broadcast and nostr-event relations + * pointing at that row keep pointing at it. + */ + fun reconcileMarmotLine( + fresh: ChatMessage, + existing: ChatMessage?, + ): ChatMessage? { + if (existing == null) return fresh + + val freshIsPlaceholder = fresh.messageType in UNRESOLVED_MARMOT_TYPES + val existingIsPlaceholder = existing.messageType in UNRESOLVED_MARMOT_TYPES + + if (freshIsPlaceholder && !existingIsPlaceholder) return null + + return fresh.copy( + id = existing.id, + savedAt = existing.savedAt, + viewedAt = existing.viewedAt, + ) + } + suspend fun fromGroupEventResult( database: MantraDatabase, activeKeyPair: KeyPair, @@ -326,7 +389,7 @@ data class ChatMessage( giftWrapPayloadId = null, marmotGroupEventId = groupEvent.id, marmotInnerEventId = null, - messageType = "pendingCommit", + messageType = TYPE_PENDING_COMMIT, senderPublicKey = groupEvent.pubKey, isUserMessage = activeKeyPair.pubKey.toHex() == groupEvent.pubKey, chatRoomId = groupEventResult.groupId, @@ -392,7 +455,7 @@ data class ChatMessage( giftWrapPayloadId = null, marmotGroupEventId = groupEvent.id, marmotInnerEventId = null, - messageType = "undecryptableOuterLayer", + messageType = TYPE_UNDECRYPTABLE_OUTER_LAYER, senderPublicKey = groupEvent.pubKey, isUserMessage = activeKeyPair.pubKey.toHex() == groupEvent.pubKey, chatRoomId = groupEventResult.groupId, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/types/MarmotReindexReport.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/types/MarmotReindexReport.kt new file mode 100644 index 00000000..0933e6da --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/types/MarmotReindexReport.kt @@ -0,0 +1,27 @@ +package press.mantra.compose.database.model.types + +/** + * What a replay of a room's stored kind:445 events did. + * + * Nothing here is persisted -- it exists so the screen that asked for the replay + * can say what happened rather than leaving the user to guess from the message + * list whether anything moved. + * + * @param stored every kind:445 held locally for the room. + * @param unresolved how many of those had nothing to show for them, or only a + * placeholder line -- the ones a replay was allowed to touch. + * @param recovered how many of [unresolved] came out with something to show: + * a message, an applied commit, an entity added to the group's library. + * @param failed how many threw while being replayed. Expected to be non-zero + * on a room with events from before this device joined, whose epoch secrets it + * never held and never will. + */ +data class MarmotReindexReport( + val stored: Int = 0, + val unresolved: Int = 0, + val recovered: Int = 0, + val failed: Int = 0, +) { + /** Nothing was left to try, so the replay was a no-op. */ + val isNoOp: Boolean get() = unresolved == 0 +} 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 b12e3723..663336a0 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 @@ -10,9 +10,11 @@ import press.mantra.compose.database.model.MarmotKeyPackage import press.mantra.compose.database.model.Participant import press.mantra.compose.database.model.intermdiate.LocalChatMessage import press.mantra.compose.database.model.intermdiate.LocalChatRoom +import press.mantra.compose.database.model.types.MarmotReindexReport import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.marmot.mip02Welcome.WelcomeEvent import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray import com.vitorpamplona.quartz.nip01Core.core.Kind import com.vitorpamplona.quartz.nip01Core.crypto.EventHasher import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair @@ -407,6 +409,22 @@ class DatabaseChatRepository( onCompletion.invoke() } + /** + * Only the public key is needed: reading a group event is done with the MLS + * group's own secrets, held in the room's stored state, and the identity is + * used for nothing but deciding whether this device is a member and whether a + * line is its own. So a read-only [KeyPair] is enough, and the nostr private + * key stays where it is. + */ + override suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + userPublicKey: HexKey + ): MarmotReindexReport = + database.nostrDao().reindexMarmotGroupEvents( + chatRoomId = chatRoomId, + activeKeyPair = KeyPair(pubKey = userPublicKey.hexToByteArray()), + ) + companion object { const val TAG = "DatabaseChatRepository" } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt index c7be3dba..f2f7ffa6 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt @@ -413,6 +413,31 @@ object MarmotInboundManager { } } + /** + * Forget the commits being held for [nostrGroupId], so a replay decides its + * epochs again from the events it is about to feed back in. + * + * [handleCommitEvent] holds a commit whenever a second one turns up for an + * epoch, and only [CommitOrdering.EpochCommitTracker.clearEpoch] on a + * successful apply ever empties that. Nothing applies while more than one is + * held, so the set is a dead end: replaying those commits without clearing + * first just adds them again and leaves every one of them pending, and the + * list grows with each attempt. + * + * Safe to drop because the tracker is a scratch pad, not a record -- the + * commits themselves are on disk, and a replay hands them back oldest first, + * which is the order [CommitOrdering.comparator] picks a winner in. + */ + suspend fun forgetPendingCommits(nostrGroupId: HexKey) { + commitTracker + .pendingGroupEpochs() + .filter { it.groupId == nostrGroupId } + .forEach { pending -> + logger.d("Forgetting pending commits for $nostrGroupId epoch ${pending.epoch}") + commitTracker.clearEpoch(pending.groupId, pending.epoch) + } + } + private suspend fun applyCommit( database: MantraDatabase, mlsGroup: MlsGroup, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotReindexSweep.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotReindexSweep.kt new file mode 100644 index 00000000..47071e5d --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotReindexSweep.kt @@ -0,0 +1,95 @@ +package press.mantra.compose.managers + +import co.touchlab.kermit.Logger +import kotlinx.coroutines.CancellationException +import press.mantra.compose.database.model.types.MarmotReindexReport + +/** + * Reads a backlog of group events again, repeatedly, until repeating stops helping. + * + * The whole reason a replay is worth anything is that the events depend on each + * other: a commit nobody could apply is what a room's later messages were waiting + * on, and applying it makes them readable. So one pass in order is not enough -- + * a pass that recovers something has changed what the next pass can do. + * + * Ordering is the caller's business; this only decides how many times to go round. + * A pass that recovers nothing ends it, because everything left is waiting on + * something no later pass will produce either. + * + * Kept apart from the database so the stopping rule can be read, and tested, on + * its own -- it is the part with something to get wrong. + */ +object MarmotReindexSweep { + private const val TAG = "MarmotReindexSweep" + + private val logger = Logger.withTag(TAG) + + /** + * How many times a sweep will go round. + * + * A pass that recovers nothing already ends the sweep, so this only bounds the + * pathological shape where every pass recovers exactly one event and the sweep + * turns quadratic. A chain of five events each waiting on the last is already + * far past what relay reordering produces. + */ + const val DEFAULT_MAX_PASSES = 5 + + /** + * @param stored how many group events the room holds in total, for the report. + * @param unresolved those with nothing to show for them, in the order to replay + * them. Anything already read must be left out: a replay is only ever allowed + * to touch events it cannot make worse. + * @param replay reads one event again. Its failures are expected -- a message + * whose key is gone throws every time -- so they are logged, not raised. + * @param isUnresolved asked again after [replay] to see whether it achieved + * anything. This, rather than [replay]'s return, is what counts a recovery: + * an event is recovered when the room has something to show for it, which is + * a fact about the database and not about the call that just ran. + */ + suspend fun run( + stored: Int, + unresolved: List, + maxPasses: Int = DEFAULT_MAX_PASSES, + replay: suspend (T) -> Unit, + isUnresolved: suspend (T) -> Boolean, + ): MarmotReindexReport { + var remaining = unresolved + var recovered = 0 + var pass = 0 + + while (remaining.isNotEmpty() && pass < maxPasses) { + pass++ + val stillUnresolved = mutableListOf() + + for (candidate in remaining) { + try { + replay(candidate) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + logger.w("Replay of $candidate failed", e) + } + + if (isUnresolved(candidate)) { + stillUnresolved.add(candidate) + } else { + recovered++ + } + } + + val madeProgress = stillUnresolved.size < remaining.size + remaining = stillUnresolved + + if (!madeProgress) break + } + + return MarmotReindexReport( + stored = stored, + unresolved = unresolved.size, + recovered = recovered, + failed = remaining.size, + ).also { + logger.i("Swept ${unresolved.size} unresolved event(s) in $pass pass(es): $it") + } + } +} 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 4669fae9..34174c20 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt @@ -6,6 +6,7 @@ import press.mantra.compose.database.model.MarmotKeyPackage import press.mantra.compose.database.model.Participant import press.mantra.compose.database.model.intermdiate.LocalChatMessage import press.mantra.compose.database.model.intermdiate.LocalChatRoom +import press.mantra.compose.database.model.types.MarmotReindexReport import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.Kind import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync @@ -106,6 +107,19 @@ interface ChatRepository { onCompletion: () -> Unit ) + /** + * Reads the room's stored marmot group events again, so ones the group could + * not make sense of when they arrived get another chance now that the events + * they were waiting on have landed. + * + * Only events with nothing to show for them are replayed, so a room where + * nothing went wrong is left exactly as it was. + */ + suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + userPublicKey: HexKey + ): MarmotReindexReport + companion object { val NO_OP_CHAT_REPOSITORY = object: ChatRepository { override suspend fun observeChatRoomListByPublicKey(publicKey: String): Flow> { @@ -232,6 +246,11 @@ interface ChatRepository { ) { TODO("Not yet implemented") } + + override suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + userPublicKey: HexKey + ): MarmotReindexReport = MarmotReindexReport() } } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt index ba444544..bc19ed8e 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt @@ -7,12 +7,14 @@ import androidx.compose.foundation.layout.fillMaxSize import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.height import androidx.compose.foundation.layout.padding +import androidx.compose.foundation.layout.size import androidx.compose.foundation.layout.width import androidx.compose.foundation.lazy.LazyColumn import androidx.compose.foundation.lazy.items import androidx.compose.material.icons.Icons import androidx.compose.material.icons.filled.Key import androidx.compose.material.icons.filled.AccountTree +import androidx.compose.material.icons.filled.Autorenew import androidx.compose.material.icons.filled.ChevronRight import androidx.compose.material.icons.filled.DeleteForever import androidx.compose.material.icons.filled.LibraryBooks @@ -23,6 +25,7 @@ import androidx.compose.material.icons.filled.Unsubscribe import androidx.compose.material.icons.filled.WaterfallChart import androidx.compose.material3.ButtonDefaults import androidx.compose.material3.Card +import androidx.compose.material3.CircularProgressIndicator import androidx.compose.material3.ExperimentalMaterial3Api import androidx.compose.material3.ExperimentalMaterial3ExpressiveApi import androidx.compose.material3.HorizontalDivider @@ -493,6 +496,22 @@ fun ChatRoomDetailScreen( Text("Delete Group") } } + + // Only MLS rooms have group events to read again. A NIP-17 + // room's messages are gift wraps, which carry no ordering + // for anything to go wrong with. + if (chatRoomDetailUIState.localChatRoom.chatRoom.mlsGroupState != null) { + item { + HorizontalDivider() + } + + item { + ReindexMarmotGroupEventsButton( + reindexState = chatRoomDetailViewModel.reindexState, + onReindex = chatRoomDetailViewModel::reindexMarmotGroupEvents + ) + } + } } } } @@ -539,6 +558,77 @@ fun ChatRoomDetailScreen( } } +/** + * Reads the room's stored marmot group events again. + * + * Sits at the very bottom, below leaving and deleting, because it is a repair + * rather than something anyone should need in the ordinary course of using the + * group. It says what it found afterwards -- a replay that recovered nothing + * looks identical to one that was never pressed, and the difference between + * "nothing was broken" and "nothing could be fixed" is worth telling. + */ +@Composable +private fun ReindexMarmotGroupEventsButton( + reindexState: ChatRoomDetailViewModel.ReindexState, + onReindex: () -> Unit +) { + Column( + horizontalAlignment = Alignment.CenterHorizontally, + verticalArrangement = Arrangement.spacedBy(5.dp) + ) { + TextButton( + enabled = reindexState !is ChatRoomDetailViewModel.ReindexState.Running, + onClick = onReindex + ) { + if (reindexState is ChatRoomDetailViewModel.ReindexState.Running) { + CircularProgressIndicator( + modifier = Modifier.size(18.dp), + strokeWidth = 2.dp + ) + } else { + Icon( + Icons.Default.Autorenew, + contentDescription = "Reindex group events" + ) + } + + Spacer( + modifier = Modifier.width(10.dp) + ) + + Text("Reindex Events") + } + + when (reindexState) { + is ChatRoomDetailViewModel.ReindexState.Done -> { + val report = reindexState.report + Text( + text = when { + report.isNoOp -> "Nothing to reindex · ${report.stored} event(s) all read" + report.recovered > 0 && report.failed > 0 -> + "Recovered ${report.recovered} of ${report.unresolved} · ${report.failed} still unreadable" + report.recovered > 0 -> "Recovered ${report.recovered} of ${report.unresolved} event(s)" + else -> "${report.failed} event(s) still unreadable" + }, + style = MaterialTheme.typography.bodySmall, + textAlign = TextAlign.Center + ) + } + + is ChatRoomDetailViewModel.ReindexState.Failed -> { + Text( + text = reindexState.message, + color = MaterialTheme.colorScheme.error, + style = MaterialTheme.typography.bodySmall, + textAlign = TextAlign.Center + ) + } + + else -> Unit + } + } +} + @Preview @Composable private fun ChatRoomMessagingScreenPreview() { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt index 9e1d7068..b2cff93a 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt @@ -18,9 +18,11 @@ import press.mantra.compose.ui.composable.navigation.routes.Route import press.mantra.compose.ui.view.state.ChatRoomDetailUIState import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.nip01Core.core.HexKey +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.launch +import press.mantra.compose.database.model.types.MarmotReindexReport class ChatRoomDetailViewModel( val chatRoomId: String, @@ -35,6 +37,26 @@ class ChatRoomDetailViewModel( var chatRoomDetailUIState: ChatRoomDetailUIState by mutableStateOf(initialChatRoomDetailUIState) private set + /** + * Where the room's replay has got to, for the button that started it. + * + * Held here rather than in [ChatRoomDetailUIState] because a replay does not + * change what the screen is showing -- the message list is observed elsewhere + * and updates itself -- it only changes what the button has to say for itself. + */ + var reindexState: ReindexState by mutableStateOf(ReindexState.Idle) + private set + + sealed interface ReindexState { + data object Idle : ReindexState + + data object Running : ReindexState + + data class Done(val report: MarmotReindexReport) : ReindexState + + data class Failed(val message: String) : ReindexState + } + private val logger = Logger.withTag(TAG) @@ -81,6 +103,39 @@ class ChatRoomDetailViewModel( } } + /** + * Reads the room's stored marmot group events again. + * + * Nostr relays impose no ordering, so a group event can arrive before the group + * can read it -- a message from an epoch whose commit has not landed, a commit + * for an epoch ahead of the local one. Nothing revisits those once the gap + * fills, which is what this is for. + * + * Only events with nothing to show for them are replayed, so pressing this on a + * room where nothing went wrong changes nothing. The feed is observed, so + * anything recovered appears in the message list without a refresh. + */ + fun reindexMarmotGroupEvents() { + if (reindexState is ReindexState.Running) return + + reindexState = ReindexState.Running + viewModelScope.launch(Dispatchers.IO) { + reindexState = try { + ReindexState.Done( + chatRepository.reindexMarmotGroupEvents( + chatRoomId = chatRoomId, + userPublicKey = activeUserPublicKey, + ) + ) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + logger.e("Failed to reindex $chatRoomId", e) + ReindexState.Failed(e.message ?: "Reindex failed") + } + } + } + fun softDeleteGroup( localChatRoom: LocalChatRoom, onPopBackToRoute: (Route) -> Unit diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotChatLineReconciliationTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotChatLineReconciliationTest.kt new file mode 100644 index 00000000..cd898372 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotChatLineReconciliationTest.kt @@ -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", + ) + } + } +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotGroupEventRoomFilterTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotGroupEventRoomFilterTest.kt new file mode 100644 index 00000000..509d803b --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotGroupEventRoomFilterTest.kt @@ -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>, + 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)) + } +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/MarmotReindexSweepTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/MarmotReindexSweepTest.kt new file mode 100644 index 00000000..51a0f4b0 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/MarmotReindexSweepTest.kt @@ -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) { + private val unlockedBy = unlocks + private val resolved = mutableSetOf() + + /** Every event replayed, in order, across every pass. */ + val replayed = mutableListOf() + + 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() + + 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(), + 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) + } +} diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index e51ecfde..9d846a9f 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -63,6 +63,7 @@ kermit = { module = "co.touchlab:kermit", version.ref = "kermit" } kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotlin" } kotlin-testJunit = { module = "org.jetbrains.kotlin:kotlin-test-junit", version.ref = "kotlin" } kotlinx-coroutinesSwing = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-swing", version.ref = "kotlinx-coroutines" } +kotlinx-coroutinesTest = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "kotlinx-coroutines" } kotlinx-datetime = { module = "org.jetbrains.kotlinx:kotlinx-datetime", version.ref = "kotlinx-datetime" } kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinxSerialization" } kotlinx-serialization-cbor = { module = "org.jetbrains.kotlinx:kotlinx-serialization-cbor", version.ref = "kotlinxSerialization" }