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 7aba6e26..7eaebda8 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 @@ -10,6 +10,7 @@ import press.mantra.compose.database.model.traits.SoftDeletableEntity import press.mantra.compose.database.model.traits.TimestampedEntity import press.mantra.compose.database.model.traits.UserViewableEntity import press.mantra.compose.exceptions.MarmotUnprocessableInnerEventException +import press.mantra.compose.managers.ArchiveManager import press.mantra.compose.managers.GroupKeyStateManager import press.mantra.compose.extensions.toHex import com.vitorpamplona.quartz.marmot.GroupEventResult @@ -25,6 +26,7 @@ import press.mantra.compose.nostr.nip30303.ArtifactVersionEvent import press.mantra.compose.nostr.nip30303.ChapterEvent import press.mantra.compose.nostr.nip30303.ChunkEvent import press.mantra.compose.nostr.nip30303.DialectEvent +import press.mantra.compose.nostr.archive.ArchiveEvent import press.mantra.compose.nostr.nip30303.SubmissionEvent import press.mantra.compose.nostr.nip30303.TranslationArtifactVersionContributorListEvent import press.mantra.compose.nostr.nip30303.TranslationArtifactVersionEvent @@ -613,6 +615,28 @@ data class ChatMessage( ) } + // An archive is neither a document nor a submission: it is a + // bundle of documents addressed to one member who is missing + // them. Intercepted here rather than in applyInnerEvent for + // the same reason the gift wrap above is -- deciding whether + // to act needs the active key, which applyInnerEvent has no + // business knowing. + if (event.kind == ArchiveEvent.KIND) { + ArchiveManager.receive( + database = database, + chatRoomId = groupEventResult.groupId, + userPublicKey = activeKeyPair.pubKey.toHex(), + page = event, + ) + + // No chat line, and not for want of one worth writing. + // One line per archive is right; one per page is not, and + // the pages of an archive are not distinguishable from + // each other here. That is Phase 7's, and until then a + // silent catch-up beats a transcript full of envelopes. + return null + } + // A submission whose payload will not parse, or which carries // another submission, is kept but not applied: there is nothing // here we can turn into a row. diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/ArchiveManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/ArchiveManager.kt index 5c8171ae..6045760d 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/ArchiveManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/ArchiveManager.kt @@ -7,6 +7,8 @@ import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate import com.vitorpamplona.quartz.utils.RandomInstance import com.vitorpamplona.quartz.utils.TimeUtils import press.mantra.compose.database.MantraDatabase +import press.mantra.compose.database.model.ChatMessage +import press.mantra.compose.database.model.MarmotInnerEvent import press.mantra.compose.extensions.toHex import press.mantra.compose.nostr.archive.ArchiveEvent import press.mantra.compose.nostr.frost.GroupKeyStateEvent @@ -98,6 +100,226 @@ object ArchiveManager { } } + // ---- Receiving ------------------------------------------------------ + + /** + * What one pass over a room's stored pages did. + * + * [skipped] and [failed] are kept apart because only one of them is worth + * trying again. A payload the allowlist or the signature refused will be + * refused identically forever; a payload that threw was probably missing a + * row it depends on, and the next page may bring it. + */ + data class Outcome( + val applied: Int = 0, + val skipped: Int = 0, + val failed: Int = 0, + ) { + operator fun plus(other: Outcome) = Outcome( + applied = applied + other.applied, + skipped = skipped + other.skipped, + failed = failed + other.failed, + ) + } + + /** + * Take delivery of an archive page for [userPublicKey]. + * + * The page is already on disk by the time this runs -- the inbound path + * stores every inner event it decrypts before dispatching on kind -- so this + * does not apply the arriving page as such. It sweeps every page the room + * has, which covers the new one and any that arrived before the rows they + * depend on. + * + * **A device that is not the named recipient does nothing.** The page is an + * ordinary group message and it can read it; it has no reason to. It already + * holds the work, and re-applying would rewrite every one of its rows to + * point at an archive page rather than at the event that introduced it. That + * is also what keeps the sweep bounded: only the member being caught up ever + * builds the list. + */ + suspend fun receive( + database: MantraDatabase, + chatRoomId: String, + userPublicKey: HexKey, + page: Event, + ): Outcome { + val recipient = ArchiveEvent( + id = page.id, + pubKey = page.pubKey, + createdAt = page.createdAt, + tags = page.tags, + content = page.content, + sig = page.sig, + ).recipient() + + if (!recipient.equals(userPublicKey, ignoreCase = true)) { + logger.d("Archive page ${page.id.take(8)} is for ${recipient?.take(8)}; not applying") + return Outcome() + } + + return sweep(database, chatRoomId, userPublicKey) + } + + /** + * Apply every archive page this room holds for [userPublicKey], repeatedly, + * until a pass stops making progress. + * + * Pages arrive over relays in no order, so page 3 can land before page 2 and + * its chunks have no chapter to hang off yet. Those payloads throw a foreign + * key violation and would be lost -- unless something runs them again. + * + * **No table is needed for that**, because the inbound path already stores + * every inner event it decrypts. This is the same shape + * `FrostSigningManager.replayStoredMessages` has, and for the same reason: + * nothing was actually lost, it just had nowhere to go at the time. + * + * **Progress is measured by failures falling, not by rows written.** Every + * write here is an upsert keyed on the event id, so "applied something" is + * true on every pass forever and would not terminate. A pass that fails fewer + * payloads than the last one learned something; a pass that does not is as + * far as this archive gets, and the rest is a hole to be filled by another + * page or another archive. + * + * **The answer is the last pass, not the sum of them.** Accumulating would + * count a payload once per pass it survived and report failures the next pass + * went on to fix, so `failed > 0` would stop meaning "still missing". The + * final pass is the settled state: what these pages can apply, what they will + * never apply, and what is still waiting on something that has not arrived. + */ + suspend fun sweep( + database: MantraDatabase, + chatRoomId: String, + userPublicKey: HexKey, + ): Outcome { + val pages = database.marmotInnerEventDao() + .getByChatRoomAndKinds(chatRoomId, listOf(ArchiveEvent.KIND)) + .filter { addressedTo(it, userPublicKey) } + + if (pages.isEmpty()) return Outcome() + + var pass = Outcome() + var previousFailures = Int.MAX_VALUE + + while (true) { + pass = Outcome() + pages.forEach { pass += applyPage(database, chatRoomId, it) } + + if (pass.failed == 0 || pass.failed >= previousFailures) { + if (pass.failed > 0) { + logger.w( + "Stopped sweeping $chatRoomId with ${pass.failed} payload(s) still " + + "unapplied: nothing in the pages it holds satisfies them" + ) + } + break + } + + previousFailures = pass.failed + } + + logger.i( + "Swept ${pages.size} archive page(s) for $chatRoomId: " + + "${pass.applied} applied, ${pass.skipped} skipped, ${pass.failed} left" + ) + + return pass + } + + private fun addressedTo(stored: MarmotInnerEvent, userPublicKey: HexKey): Boolean = + ArchiveEvent( + id = stored.id, + pubKey = stored.publicKey, + createdAt = stored.createdAt.epochSeconds, + tags = stored.tags, + content = stored.content, + sig = "", + ).recipient().equals(userPublicKey, ignoreCase = true) + + /** + * One page, checked payload by payload and applied in dependency order. + * + * **Per payload, not per page.** A forged payload sitting beside honest ones + * costs itself and nothing else -- the same rule `MarmotInboundManager` uses + * for a forged direct message, and for the same reason: this runs inside the + * inbound transaction, and one bad event must not take the room down with it. + * Refusing the whole page would also let a single forgery deny an entire + * archive. + * + * The page's own framing is still all-or-nothing; see + * [ArchiveEvent.decodePage] for why those two are not in tension. + */ + private suspend fun applyPage( + database: MantraDatabase, + chatRoomId: String, + stored: MarmotInnerEvent, + ): Outcome { + val payloads = ArchiveEvent.decodePage(stored.content) + if (payloads == null) { + logger.w("Archive page ${stored.id.take(8)} does not read as a page; dropping it") + return Outcome() + } + + var outcome = Outcome() + + ArchiveEvent.inApplyOrder(payloads).forEach { payload -> + // The allowlist first, and it is not a formality. Verification admits + // an event to the apply path on the strength of the group's + // signature, which makes every kind the group has ever signed + // replayable by any member at any time -- a GroupKeyStateEvent from + // an earlier epoch passes the signature check perfectly. + if (!ArchiveEvent.isArchivable(payload.kind)) { + logger.w( + "Archive page ${stored.id.take(8)} carries kind ${payload.kind}, " + + "which an archive may not deliver; dropping ${payload.id.take(8)}" + ) + outcome += Outcome(skipped = 1) + return@forEach + } + + if (!GroupKeyStateEvent.isSignedByRoom(payload, chatRoomId)) { + logger.w( + "Archive page ${stored.id.take(8)} carries ${payload.id.take(8)}, " + + "which room $chatRoomId did not sign; dropping it" + ) + outcome += Outcome(skipped = 1) + return@forEach + } + + outcome += try { + // The chat line this returns is deliberately dropped rather than + // filed. ChatMessage has an autoGenerate primary key, so there is + // no id to dedupe on and every applied payload would mint a new + // row -- giving the recipient a synthetic transcript dated now, + // and another one on every pass of the sweep. The archive + // restores the work; the conversation is forward secret and stays + // gone. + ChatMessage.applyInnerEvent( + database = database, + groupId = chatRoomId, + event = payload, + marmotGroupEventId = stored.marmotGroupEventId, + marmotInnerEventId = stored.id, + senderPublicKey = payload.pubKey, + isUserMessage = false, + createdAt = stored.createdAt, + ) + Outcome(applied = 1) + } catch (error: Throwable) { + // Almost always a foreign key: this payload names a row that is + // in a page which has not arrived yet. Retryable, which is what + // the sweep is for, so it is counted rather than logged loudly. + logger.d( + "Could not apply ${payload.id.take(8)} from archive page " + + "${stored.id.take(8)} yet: ${error.message}" + ) + Outcome(failed = 1) + } + } + + return outcome + } + /** * The room's rows, rebuilt into the events they came from. * diff --git a/composeApp/src/jvmTest/kotlin/press/mantra/compose/managers/ArchiveApplyJvmTest.kt b/composeApp/src/jvmTest/kotlin/press/mantra/compose/managers/ArchiveApplyJvmTest.kt new file mode 100644 index 00000000..34614e9b --- /dev/null +++ b/composeApp/src/jvmTest/kotlin/press/mantra/compose/managers/ArchiveApplyJvmTest.kt @@ -0,0 +1,593 @@ +package press.mantra.compose.managers + +import androidx.room3.Room +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray +import com.vitorpamplona.quartz.nip01Core.crypto.EventHasher +import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate +import fr.acinq.bitcoin.ByteVector +import fr.acinq.bitcoin.ByteVector32 +import fr.acinq.bitcoin.PrivateKey +import fr.acinq.bitcoin.crypto.frost.Frost +import fr.acinq.bitcoin.crypto.frost.IndividualNonce +import fr.acinq.bitcoin.crypto.frost.KeyMaterial +import fr.acinq.bitcoin.crypto.frost.SecretNonce +import fr.acinq.bitcoin.crypto.frost.Session +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue +import kotlin.time.Instant +import kotlinx.coroutines.runBlocking +import press.mantra.compose.database.MantraDatabase +import press.mantra.compose.database.builder.getRoomDatabase +import press.mantra.compose.database.model.ChatMessage +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.extensions.toHex +import press.mantra.compose.nostr.archive.ArchiveEvent +import press.mantra.compose.nostr.archive.tags.ArchiveIdTag +import press.mantra.compose.nostr.archive.tags.ArchivePageTag +import press.mantra.compose.nostr.archive.tags.ArchiveRecipientTag +import press.mantra.compose.nostr.frost.GroupKeyStateEvent +import press.mantra.compose.nostr.nip30303.ArtifactEvent +import press.mantra.compose.nostr.nip30303.ChapterEvent +import press.mantra.compose.nostr.nip30303.ChunkEvent +import press.mantra.compose.nostr.nip30303.DialectEvent +import press.mantra.compose.nostr.nip30303.TranslationArtifactVersionEvent +import press.mantra.compose.nostr.nip30303.TranslationChapterEvent + +/** + * Two devices, two databases: one that did the work and one that arrived after + * it, with the pages between them carried by hand. + * + * This is the claim the whole feature is: a member who holds no share, took part + * in no signing session and cannot decrypt a word of the room's history ends up + * with the same rows as everybody else, and gets there without trusting the + * member who sent them. + * + * The negative cases matter more than the positive one. Verification is what + * makes an archive safe to accept from anybody, so what has to be proved is that + * it refuses -- a forged payload beside honest ones, and a genuine key state + * replayed out of its epoch, which passes every signature check there is and is + * stopped by nothing but the allowlist. + */ +class ArchiveApplyJvmTest { + /** The member who did the work. */ + private val sender: MantraDatabase = getRoomDatabase(Room.inMemoryDatabaseBuilder()) + + /** The member who arrived afterwards. No share, no sessions, no history. */ + private val receiver: MantraDatabase = getRoomDatabase(Room.inMemoryDatabaseBuilder()) + + @AfterTest + fun closeDbs() { + sender.close() + receiver.close() + } + + private val participants = 3 + private val threshold = 2 + + private val keyMaterial: KeyMaterial = Frost.trustedDealerKeygen( + thresholdSecretKey = PrivateKey( + ByteVector32("1c0ffee0000000000000000000000000000000000000000000000000000000a1") + ), + nParticipants = participants, + threshold = threshold + ) + + /** A second group entirely: its own key, its own quorum, its own rooms. */ + private val otherKeyMaterial: KeyMaterial = Frost.trustedDealerKeygen( + thresholdSecretKey = PrivateKey( + ByteVector32("2bada550000000000000000000000000000000000000000000000000000000b2") + ), + nParticipants = participants, + threshold = threshold + ) + + private val room: SharedKeyDerivation.Derived = SharedKeyDerivation.derive( + thresholdPublicKey = keyMaterial.thresholdPublicKey.value.toHex(), + path = SharedKeyDerivation.MARMOT_ADMIN_GROUP_PATH + ) + + private val chatRoomId = room.hex + + private val oldMember = "8".repeat(64) + private val newMember = "9".repeat(64) + + private fun groupSignature(material: KeyMaterial, eventId: String): String { + val cache = SharedKeyDerivation.derive( + thresholdPublicKey = material.thresholdPublicKey.value.toHex(), + path = SharedKeyDerivation.MARMOT_ADMIN_GROUP_PATH + ).cache + val message = ByteVector(eventId.hexToByteArray()) + val signerIds = listOf(0, 1) + + val nonces = signerIds.map { signerId -> + SecretNonce.generate( + sessionRandom = ByteVector32("a".repeat(63) + "${signerId + 1}"), + secretShare = material.secretShares[signerId], + publicShare = material.publicShares[signerId], + tweakedThresholdPublicKey = cache.tweakedPublicKey, + message = message, + extraInput = null + ) + } + + val signingSession = Session.create( + aggregatedNonce = IndividualNonce.aggregate(nonces.map { it.second }).right!!, + signerIds = signerIds.map { it.toUInt() }, + signerPublicShares = signerIds.map { material.publicShares[it] }, + nParticipants = participants, + threshold = threshold, + tweakCache = cache, + message = message + ) + + val partials = signerIds.mapIndexed { position, signerId -> + signingSession.sign( + nonces[position].first, + material.secretShares[signerId], + signerId.toUInt() + ).right!! + } + + return signingSession.aggregateSigs(partials).right!!.toByteArray().toHex() + } + + /** A template as a finished session leaves it, authored by [material]'s room. */ + private fun signedBy(material: KeyMaterial, template: EventTemplate<*>): Event { + val author = SharedKeyDerivation.derive( + thresholdPublicKey = material.thresholdPublicKey.value.toHex(), + path = SharedKeyDerivation.MARMOT_ADMIN_GROUP_PATH + ).hex + + val id = EventHasher.hashId( + pubKey = author, + createdAt = template.createdAt, + kind = template.kind, + tags = template.tags, + content = template.content + ) + + return Event( + id = id, + pubKey = author, + createdAt = template.createdAt, + kind = template.kind, + tags = template.tags, + content = template.content, + sig = groupSignature(material, id) + ) + } + + private fun signed(template: EventTemplate<*>) = signedBy(keyMaterial, template) + + private suspend fun seedRoom(db: MantraDatabase, user: String) { + val nostrEventId = "c".repeat(63) + if (user == newMember) "1" else "2" + db.nostrEventDao().upsert( + NostrEvent( + id = nostrEventId, + pubKey = user, + kind = 0, + tags = emptyArray(), + content = "{}", + sig = "0".repeat(128), + ) + ) + db.profileDao().upsert(Profile(publicKey = user, userName = "member", nostrEventId = nostrEventId)) + db.chatRoomDao().upsert( + ChatRoom( + id = chatRoomId, + userPublicKey = user, + subject = "#admins", + description = null, + mlsGroupState = null, + ) + ) + } + + private suspend fun apply(db: MantraDatabase, event: Event) { + ChatMessage.applyInnerEvent( + database = db, + groupId = chatRoomId, + event = event, + marmotGroupEventId = null, + marmotInnerEventId = null, + senderPublicKey = chatRoomId, + isUserMessage = false, + createdAt = Instant.fromEpochSeconds(event.createdAt) + ) + } + + /** The sender's room, with a full artifact and one translation scaffolded. */ + private suspend fun seedSenderWork() { + seedRoom(sender, oldMember) + + val dialect = signed( + DialectEvent.build(name = "isiZulu", country = "ZA", language = "zu", createdAt = 1_700_000_000L) + ).also { apply(sender, it) } + + val artifact = signed( + ArtifactEvent.build( + name = "In Detention", + url = "https://example.com/in-detention", + visibility = "private", + license = "cc", + dialectId = dialect.id, + versionLabel = "1.0", + createdAt = 1_700_000_010L + ) + ).also { apply(sender, it) } + + val version = assertNotNull( + sender.mantraArtifactVersionDao().getArtifactVersionsByArtifactId(artifact.id).firstOrNull() + ) + + val chapter = signed( + ChapterEvent.build( + artifactVersionId = version.id, + name = "Chapter One", + originalText = "He was a man of parts. He had many.", + index = 0, + wordCount = 8, + characterCount = 35, + createdAt = 1_700_000_020L + ) + ).also { apply(sender, it) } + + apply( + sender, + signed( + ChunkEvent.build( + chapterId = chapter.id, + text = "He was a man of parts.", + index = 0, + wordCount = 6, + characterCount = 22, + createdAt = 1_700_000_030L + ) + ) + ) + apply( + sender, + signed( + ChunkEvent.build( + chapterId = chapter.id, + text = "He had many.", + index = 1, + wordCount = 3, + characterCount = 12, + createdAt = 1_700_000_031L + ) + ) + ) + + val translationVersion = signed( + TranslationArtifactVersionEvent.build( + artifactVersionId = version.id, + dialectId = dialect.id, + name = "isiZulu", + visibility = "private", + license = "cc", + createdAt = 1_700_000_040L + ) + ).also { apply(sender, it) } + + apply( + sender, + signed( + TranslationChapterEvent.build( + translationArtifactVersionId = translationVersion.id, + chapterId = chapter.id, + index = 0, + createdAt = 1_700_000_050L + ) + ) + ) + } + + /** + * An archive page as it reaches a device: stored as an inner event first, + * exactly as the inbound path stores every payload it decrypts before + * dispatching on kind, then handed to the manager. + */ + private suspend fun deliver( + template: EventTemplate, + to: String = newMember, + from: String = oldMember, + ): ArchiveManager.Outcome { + val page = Event( + id = EventHasher.hashId( + pubKey = from, + createdAt = template.createdAt, + kind = template.kind, + tags = template.tags, + content = template.content + ), + pubKey = from, + createdAt = template.createdAt, + kind = template.kind, + tags = template.tags, + // An archive page is a member's own rumor. What carries the group's + // word is each payload's signature, not the envelope's. + content = template.content, + sig = "" + ) + + receiver.marmotInnerEventDao().upsert( + MarmotInnerEvent( + id = page.id, + publicKey = page.pubKey, + kind = page.kind, + createdAt = Instant.fromEpochSeconds(page.createdAt), + tags = page.tags, + content = page.content, + chatRoomId = chatRoomId, + ) + ) + + return ArchiveManager.receive(receiver, chatRoomId, to, page) + } + + private suspend fun senderArchive() = ArchiveManager.assemble(sender, chatRoomId, newMember) + + private suspend fun rowCounts(db: MantraDatabase): Map { + val dialects = db.mantraDialectDao().getDialectsByChatRoomId(chatRoomId) + val artifacts = db.mantraArtifactDao().getArtifactsByChatRoomId(chatRoomId) + val versions = artifacts.flatMap { + db.mantraArtifactVersionDao().getArtifactVersionsByArtifactId(it.id) + } + val chapters = versions.flatMap { db.mantraChapterDao().getChaptersByArtifactVersionId(it.id) } + val chunks = chapters.flatMap { db.mantraChunkDao().getChunksByChapterId(it.id) } + val translationVersions = versions.flatMap { + db.mantraTranslationArtifactVersionDao().getTranslationsByArtifactVersionId(it.id) + } + val translationChapters = translationVersions.flatMap { + db.mantraTranslationChapterDao() + .getTranslationChaptersByTranslationArtifactVersionId(it.id) + } + + return mapOf( + "dialects" to dialects.size, + "artifacts" to artifacts.size, + "versions" to versions.size, + "chapters" to chapters.size, + "chunks" to chunks.size, + "translationVersions" to translationVersions.size, + "translationChapters" to translationChapters.size, + ) + } + + @Test + fun `a member who was never there ends up holding what the group signed`() = runBlocking { + seedSenderWork() + seedRoom(receiver, newMember) + + assertEquals( + mapOf( + "dialects" to 0, "artifacts" to 0, "versions" to 0, "chapters" to 0, + "chunks" to 0, "translationVersions" to 0, "translationChapters" to 0, + ), + rowCounts(receiver), + "the new member starts with nothing, which is the whole problem" + ) + + senderArchive().forEach { deliver(it) } + + assertEquals(rowCounts(sender), rowCounts(receiver)) + + // Not merely the same shape: the same rows, with the group's signature on + // them. An archive that produced look-alike rows authored by the sender + // would pass a count and fail the only thing that matters. + val theirs = receiver.mantraArtifactDao().getArtifactsByChatRoomId(chatRoomId).single() + val ours = sender.mantraArtifactDao().getArtifactsByChatRoomId(chatRoomId).single() + assertEquals(ours.id, theirs.id) + assertEquals(ours.signature, theirs.signature) + assertEquals(chatRoomId, theirs.publicKey) + } + + @Test + fun `an archive files no chat lines`() = runBlocking { + seedSenderWork() + seedRoom(receiver, newMember) + + senderArchive().forEach { deliver(it) } + + assertEquals( + 0, + receiver.chatMessageDao().getChatMessagesByChatRoomId(chatRoomId).size, + "the archive restores the work; the conversation is forward secret and stays gone" + ) + } + + @Test + fun `pages arriving out of order still converge`() = runBlocking { + seedSenderWork() + seedRoom(receiver, newMember) + + // One page per payload, delivered backwards -- a translation chapter + // before the chapter it names, chunks before their chapter, the artifact + // last. Relays give no ordering guarantee, so this is the ordinary case + // rather than the adversarial one. + val payloads = senderArchive() + .flatMap { assertNotNull(ArchiveEvent.decodePage(it.content)) } + val archiveId = "e".repeat(64) + + val outcomes = payloads.reversed().mapIndexed { index, payload -> + deliver( + ArchiveEvent.build( + payloads = listOf(payload), + archiveId = archiveId, + index = index, + count = payloads.size, + recipient = newMember, + createdAt = 1_700_000_100L + index, + ) + ) + } + + // Without this the test could pass for the wrong reason -- if nothing + // ever failed to apply, the pages were not really out of order and the + // sweep was never the thing that saved them. + assertTrue( + outcomes.any { it.failed > 0 }, + "delivering backwards should have left payloads waiting on rows they depend on" + ) + assertEquals( + 0, + outcomes.last().failed, + "the last page completes the archive, so the pass it triggers should leave nothing" + ) + + assertEquals(rowCounts(sender), rowCounts(receiver)) + } + + @Test + fun `one forged payload costs itself and nothing else`() = runBlocking { + seedSenderWork() + seedRoom(receiver, newMember) + + val honest = signed( + DialectEvent.build(name = "isiZulu", country = "ZA", language = "zu", createdAt = 1_700_000_000L) + ) + + // Four ways to be a dishonest member, in one page beside one honest + // dialect. Every one of them is well-formed and would file a row without + // the check. + val forgeries = listOf( + // The room's id as author -- everybody knows it -- and a made-up + // signature. + Event( + id = EventHasher.hashId( + pubKey = chatRoomId, createdAt = 1_700_000_001L, kind = DialectEvent.KIND, + tags = honest.tags, content = "forged" + ), + pubKey = chatRoomId, createdAt = 1_700_000_001L, kind = DialectEvent.KIND, + tags = honest.tags, content = "forged", sig = "9".repeat(128) + ), + // A real quorum, of another group. + signedBy( + otherKeyMaterial, + DialectEvent.build(name = "seSotho", country = "ZA", language = "st", createdAt = 1_700_000_002L) + ), + // Edited after the group signed it. + Event( + id = honest.id, pubKey = honest.pubKey, createdAt = honest.createdAt, + kind = honest.kind, + tags = arrayOf(arrayOf("alt", "Dialect"), arrayOf("name", "tampered")), + content = honest.content, sig = honest.sig + ), + // A member's own rumor, which is what everything on the wire looks + // like today. + Event( + id = EventHasher.hashId( + pubKey = oldMember, createdAt = 1_700_000_003L, kind = DialectEvent.KIND, + tags = honest.tags, content = "unsigned" + ), + pubKey = oldMember, createdAt = 1_700_000_003L, kind = DialectEvent.KIND, + tags = honest.tags, content = "unsigned", sig = "" + ), + ) + + val outcome = deliver( + ArchiveEvent.build( + payloads = forgeries + honest, + archiveId = "f".repeat(64), + index = 0, + count = 1, + recipient = newMember, + createdAt = 1_700_000_100L, + ) + ) + + assertEquals(4, outcome.skipped) + assertEquals(1, outcome.applied) + + val dialects = receiver.mantraDialectDao().getDialectsByChatRoomId(chatRoomId) + assertEquals(1, dialects.size, "exactly the honest one") + assertEquals(honest.id, dialects.single().id) + } + + @Test + fun `a key state the room really signed cannot be replayed through an archive`() = runBlocking { + seedSenderWork() + seedRoom(receiver, newMember) + + // Signed by the room, verifying perfectly, and refused: the allowlist is + // the only thing between "the group signed this once" and "any member may + // re-point what this room signs with, at any time, forever". + val keyState = signed( + EventTemplate( + createdAt = 1_700_000_060L, + kind = GroupKeyStateEvent.KIND, + tags = GroupKeyStateEvent.assembleTags( + chatRoomId = chatRoomId, + dkgSessionId = "somebody-elses-ceremony", + ), + content = keyMaterial.thresholdPublicKey.value.toHex(), + ) + ) + assertTrue( + GroupKeyStateEvent.isSignedByRoom(keyState, chatRoomId), + "the point of this test is that verification passes" + ) + + // Built by hand: ArchiveEvent.build refuses this kind, which is the + // outbound half of the same rule. Only a hand-rolled page gets this far. + val page = EventTemplate( + createdAt = 1_700_000_100L, + kind = ArchiveEvent.KIND, + tags = arrayOf( + arrayOf("alt", ArchiveEvent.ALT_DESCRIPTION), + ArchiveIdTag.assemble("a".repeat(64)), + ArchivePageTag.assemble(0, 1), + ArchiveRecipientTag.assemble(newMember), + ), + content = ArchiveEvent.encodePage(listOf(keyState)), + ) + + val outcome = deliver(page) + + assertEquals(1, outcome.skipped) + assertEquals(0, outcome.applied) + assertNull(receiver.groupKeyStateDao().getByChatRoomId(chatRoomId)) + } + + @Test + fun `a member the archive is not addressed to applies none of it`() = runBlocking { + seedSenderWork() + seedRoom(receiver, newMember) + + val outcome = senderArchive().map { deliver(it, to = "7".repeat(64)) } + + assertEquals(0, outcome.sumOf { it.applied }) + assertEquals( + mapOf( + "dialects" to 0, "artifacts" to 0, "versions" to 0, "chapters" to 0, + "chunks" to 0, "translationVersions" to 0, "translationChapters" to 0, + ), + rowCounts(receiver), + "a bystander already holds the work; re-applying would only rewrite its provenance" + ) + } + + @Test + fun `applying the same archive twice changes nothing the second time`() = runBlocking { + seedSenderWork() + seedRoom(receiver, newMember) + + val archive = senderArchive() + archive.forEach { deliver(it) } + val afterFirst = rowCounts(receiver) + + archive.forEach { deliver(it) } + + assertEquals(afterFirst, rowCounts(receiver)) + assertEquals(0, receiver.chatMessageDao().getChatMessagesByChatRoomId(chatRoomId).size) + } +} diff --git a/docs/member-archive.md b/docs/member-archive.md index 73e9a1aa..2534b833 100644 --- a/docs/member-archive.md +++ b/docs/member-archive.md @@ -429,9 +429,20 @@ database.marmotInnerEventDao() ``` Re-apply every stored page for the room, oldest first, after each new page -arrives; repeat while a pass applies something it did not apply before; stop when -a pass applies nothing. Everything in it is an `upsert` keyed on the event id, so -a re-run is free and a converged archive costs one no-op pass. +arrives. Everything in it is an `upsert` keyed on the event id, so a re-run is +free and a converged archive costs one no-op pass. + +**Progress is falling failures, not rows written.** "Repeat while a pass applies +something new" is the obvious loop condition and it does not terminate: an upsert +succeeds every time, so every pass applies something forever. What strictly +decreases is the number of payloads that threw. A pass that fails fewer than the +last one learned something; a pass that does not is as far as these pages get. + +**And the answer is the last pass, not the sum of them.** Accumulating counts a +payload once per pass it survived and reports failures that a later pass went on +to fix, so `failed > 0` stops meaning "still missing" -- which is exactly the +question the caller is asking. Found by asserting that the page completing an +out-of-order archive leaves nothing behind, which failed against the sum. Only the recipient sweeps, which is what bounds it: the members who skip apply never build the list.