diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventDao.kt index 3a23775b..a6894d90 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventDao.kt @@ -34,6 +34,35 @@ interface UnsignedNostrEventDao { @Query("SELECT * FROM UnsignedNostrEvent WHERE kind = 0") suspend fun getLocalAccounts(): List - @Query("SELECT * FROM UnsignedNostrEvent WHERE pubKey = :publicKey AND signedAt IS NULL ORDER BY kind ASC") - fun observeUnsignedNostrEvents(publicKey: String): Flow + /** + * Everything this key has queued and not yet signed, lowest kind first. + * + * A backlog, not a head, for the reason spelled out on + * `MarmotInnerEventDao.observeUnprocessedMarmotInnerEvents`: the only exit from this + * queue is a successful publish stamping `signedAt`, in the transaction + * `NostrDao.commitPublishedNostrEvent` wraps. Nothing else clears it, so a row that + * cannot be published stays, and while the notary was handed one row at a time it was + * the whole queue. The comment on `commitPublishedNostrEvent` is the account of that + * happening: an indexing throw rolled `signedAt` back and nothing queued afterwards -- + * the MLS key package included -- was ever signed. + * + * `kind ASC` is kept because it is load-bearing: kind 0 sorts first, and + * NavigationViewModel holds the user on a screen until their profile is announced. It + * also puts the key package (30443) last, which is what the `commitPublishedNostrEvent` + * comment means by "enqueued last". Between them sits the relay feeds list (10012) -- + * the one row of the account burst carrying `privateTags`, and so the only one whose + * publish runs a NIP-44 encryption first. A throw there takes both relay lists (10050, + * 10051) and the key package with it, which is to say everything a peer needs to reach + * this user. That is the shape of the next stall, not a hypothetical one. + * + * `id ASC` breaks the tie. It is the autogenerated row id, so two events of one kind + * are published in the order they were queued -- for a replaceable kind that is the + * difference between the newest one standing and an older one overwriting it. + */ + @Query( + "SELECT * FROM UnsignedNostrEvent " + + "WHERE pubKey = :publicKey AND signedAt IS NULL " + + "ORDER BY kind ASC, id ASC" + ) + fun observeUnsignedNostrEvents(publicKey: String): Flow> } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt index 54ec9490..dd51024c 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt @@ -100,7 +100,7 @@ class DatabaseNostrRepository( database.connectionDao().delete(connection) } - override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow { + override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow> { return database.unsignedNostrEventDao().observeUnsignedNostrEvents(publicKey) } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt index c96ccc72..5372e220 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt @@ -34,7 +34,14 @@ interface NostrRepository { suspend fun deleteConnection(connection: Connection) - suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow + /** + * The whole unsigned backlog for [publicKey], lowest kind first. + * + * See `UnsignedNostrEventDao.observeUnsignedNostrEvents`: a row that cannot be + * published used to be the entire queue. Callers work through it in order, and a row + * that fails costs only itself. + */ + suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow> suspend fun observePendingBroadcastNostrEventRequests(): Flow @@ -195,7 +202,7 @@ interface NostrRepository { TODO("Not yet implemented") } - override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow { + override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow> { TODO("Not yet implemented") } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt index 6875c636..2f69969a 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt @@ -109,6 +109,24 @@ class NotaryViewModel( } } + /** + * Sign and publish everything queued for this key, not just the lowest kind. + * + * The last of the notary's three queues to be a single row, and the one already + * observed to stall: `NostrDao.commitPublishedNostrEvent` carries the account of an + * indexing throw rolling `signedAt` back, after which nothing queued behind that row + * was ever signed. That was fixed by moving indexing out of the transaction, which + * closed the one known way in and left the queue as narrow as it was. + * + * `UnsignedNostrEvent.equals` is a plain value comparison, so this was the frozen + * variant rather than the retrying one: the failed row's re-emission compared equal to + * the last, `distinctUntilChanged` dropped it, and the collector saw nothing again for + * the life of the session. Both go, for the reason given on + * [observeUnprocessedMarmotInnerEvents]. + * + * Serially and in the query's order: kind 0 is what NavigationViewModel is waiting on, + * and a replaceable kind queued twice must go out oldest first or the older one wins. + */ private suspend fun observeUnsignedNostrEvents( keyPair: KeyPair ) { @@ -118,8 +136,11 @@ class NotaryViewModel( logger.i { "observeUnsignedNostrEvents: ${keyPair.pubKey.toHexKey()}" } nostrRepository.observeUnsignedNostrEvents( publicKey = keyPair.pubKey.toHexKey() - ).distinctUntilChanged().collect { unsignedNostrEventOrNull -> - unsignedNostrEventOrNull?.let { unsignedNostrEvent -> + ).collect { unsignedNostrEvents -> + unsignedNostrEvents.forEach { unsignedNostrEvent -> + // Per row: a publish that throws rolls back its own transaction and + // nothing else, so the row stays queued for the next pass and the rest of + // the account's events still go out. guardNotarization("unsigned event ${unsignedNostrEvent.id} (kind ${unsignedNostrEvent.kind})") { logger.d("Unsigned: ${unsignedNostrEvent.kind}") val event = tempSigner.signNormal( diff --git a/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventQueueJvmTest.kt b/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventQueueJvmTest.kt new file mode 100644 index 00000000..0764b909 --- /dev/null +++ b/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventQueueJvmTest.kt @@ -0,0 +1,243 @@ +package press.mantra.compose.database.dao + +import androidx.room3.Room +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.toHexKey +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip51Lists.encryption.PrivateTagsInContent +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.runBlocking +import press.mantra.compose.database.MantraDatabase +import press.mantra.compose.database.builder.getRoomDatabase +import press.mantra.compose.database.model.NostrEvent +import press.mantra.compose.database.model.UnsignedNostrEvent +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 + +/** + * The notary's unsigned queue, and what an event it cannot publish does to the rest. + * + * The last of the three queues to hand back one row at a time, and the only one already + * observed to stall: the comment on `NostrDao.commitPublishedNostrEvent` records an indexing + * throw rolling `signedAt` back, after which nothing queued behind that row was ever signed -- + * "including the MLS key package, which is enqueued last". That was closed by moving indexing + * into its own transaction (covered in [NostrDaoJvmTest]), which fixed the one known way in + * and left the queue as narrow as it was. + * + * These cover the width instead. `signedAt` is the only exit, so a row that cannot be + * published stays; what matters is that it is no longer the only row the notary is shown. The + * kinds used are the ones a real account queues -- 0 metadata, 3 contacts, 10012 relay feeds + * (the one carrying `privateTags`, and so the only one whose publish encrypts first), 10050 + * and 10051 relay lists, 30443 key package -- because their sort order is the whole reason a + * stall in the middle of that burst leaves a user nobody can reach. + * + * What is asserted here is the DAO's half: that the backlog arrives whole and in a stable + * order, and that publishing some rows neither disturbs nor depends on the others. The loop + * that walks it lives in NotaryViewModel, which wants an ActiveWallet flow to stand up; the + * [drain] below is shaped like it but is this test's own, so it pins the queue rather than the + * collector. + */ +class UnsignedNostrEventQueueJvmTest { + + private val db: MantraDatabase = getRoomDatabase( + Room.inMemoryDatabaseBuilder() + ) + + @AfterTest + fun closeDb() = db.close() + + private val keyPair = KeyPair() + private val author = keyPair.pubKey.toHexKey() + private val signer = NostrSignerSync(keyPair) + private val relays = listOf("wss://one.example", "wss://two.example") + + private suspend fun queue( + kind: Int, + content: String, + pubKey: String = author, + privateTags: Array>? = null, + ): Long = db.unsignedNostrEventDao().upsert( + UnsignedNostrEvent( + pubKey = pubKey, + kind = kind, + tags = emptyArray(), + privateTags = privateTags, + content = content, + createdAt = Instant.fromEpochSeconds(1_000), + ) + ) + + private suspend fun backlog(publicKey: String = author) = db.unsignedNostrEventDao() + .observeUnsignedNostrEvents(publicKey) + .first() + + /** Sign and publish one row exactly as the notary does, private tags and all. */ + private suspend fun publish(unsigned: UnsignedNostrEvent) { + val event = signer.signNormal( + createdAt = unsigned.createdAt.epochSeconds, + kind = unsigned.kind, + tags = unsigned.tags, + content = unsigned.privateTags?.let { PrivateTagsInContent.encryptNip44(it, signer) } + ?: unsigned.content, + ) + + db.nostrDao().publishNostrEvent( + unsignedNostrEvent = unsigned, + nostrEvent = NostrEvent( + id = event.id, + pubKey = event.pubKey, + kind = event.kind, + tags = event.tags, + content = event.content, + createdAt = Instant.fromEpochSeconds(event.createdAt), + sig = event.sig, + unsignedNostrEventId = unsigned.id, + ), + relayURLs = relays, + activeKeyPair = keyPair, + ) + } + + /** + * A pass over the backlog with [failKinds] standing in for rows the notary cannot publish. + * A real failure writes nothing -- `commitPublishedNostrEvent` is one transaction -- so + * skipping the row leaves the database in the state a throw would have. + */ + private suspend fun drain(failKinds: Set = emptySet()): List { + val failed = mutableListOf() + backlog().forEach { unsigned -> + if (unsigned.kind in failKinds) { + failed += unsigned.kind + } else { + publish(unsigned) + } + } + return failed + } + + @Test + fun `the queue hands back the whole backlog, lowest kind first`() = runBlocking { + queue(kind = 30_443, content = "key package") + queue(kind = 0, content = "metadata") + queue(kind = 10_050, content = "dm relays") + queue(kind = 3, content = "contacts") + + assertEquals( + listOf(0, 3, 10_050, 30_443), + backlog().map { it.kind }, + "the notary is handed the backlog to work through, not just its lowest kind", + ) + } + + /** + * `id` is the autogenerated row id, so this is the order they were queued in. For a + * replaceable kind it is the difference between the newest version standing and an older + * one being published last and winning. + */ + @Test + fun `events of one kind come back in the order they were queued`() = runBlocking { + queue(kind = 1, content = "first") + queue(kind = 1, content = "second") + queue(kind = 1, content = "third") + + assertEquals(listOf("first", "second", "third"), backlog().map { it.content }) + assertEquals( + listOf("first", "second", "third"), + backlog().map { it.content }, + "the order changed between passes", + ) + } + + @Test + fun `only this key's events are in its backlog`() = runBlocking { + val stranger = KeyPair().pubKey.toHexKey() + queue(kind = 0, content = "ours") + queue(kind = 0, content = "theirs", pubKey = stranger) + + assertEquals(listOf("ours"), backlog().map { it.content }) + assertEquals(listOf("theirs"), backlog(stranger).map { it.content }) + } + + @Test + fun `a published event leaves the queue`() = runBlocking { + queue(kind = 0, content = "metadata") + + drain() + + assertEquals(emptyList(), backlog(), "a signed row is still being offered") + } + + /** + * The one that matters, and the shape of the stall this queue is most exposed to: 10012 is + * the only row of the account burst carrying private tags, and it sorts ahead of both relay + * lists a peer needs to find this user and ahead of the key package they need to invite + * them. A queue that yields only its head would offer 10012 forever and publish none of the + * three. + */ + @Test + fun `an event that cannot be published no longer hides the ones behind it`() = runBlocking { + queue(kind = 0, content = "metadata") + queue(kind = 10_012, content = "relay feeds", privateTags = arrayOf(arrayOf("relay", "wss://private.example"))) + queue(kind = 10_050, content = "dm relays") + queue(kind = 10_051, content = "key package relays") + queue(kind = 30_443, content = "key package") + + assertEquals(listOf(10_012), drain(failKinds = setOf(10_012))) + + assertEquals( + listOf(10_012), + backlog().map { it.kind }, + "the stopper stays queued and nothing else does", + ) + assertEquals( + listOf(0, 10_050, 10_051, 30_443), + db.unsignedNostrEventDao().getAUnsignedNostrEvents() + .filter { it.pubKey == author && it.signedAt != null } + .map { it.kind } + .sorted(), + "the events either side of the stopper were not published", + ) + } + + /** + * Published means queued for the relays. A row that got `signedAt` but no + * BroadcastNostrEventRequest is still an event no relay ever sees, which is the same + * silence in a different place -- see 0211764. + */ + @Test + fun `an event past the stopper is queued for every relay`() = runBlocking { + queue(kind = 10_012, content = "relay feeds", privateTags = arrayOf(arrayOf("relay", "wss://private.example"))) + queue(kind = 30_443, content = "key package") + + drain(failKinds = setOf(10_012)) + + val keyPackage = assertNotNull( + db.unsignedNostrEventDao().getAUnsignedNostrEvents().singleOrNull { it.kind == 30_443 } + ) + assertNotNull(keyPackage.signedAt) + val queued = db.broadcastNostrEventRequestDao().getAllBroadcastNostrEventRequests() + .filter { it.unsignedNostrEventId == keyPackage.id } + assertEquals(relays.toSet(), queued.map { it.relayURL }.toSet()) + assertTrue(queued.all { it.status == "pending" }, "pending is the only status the broadcaster looks at") + } + + /** A skipped row must be left exactly as it was, or the next pass has nothing to retry. */ + @Test + fun `the stopper is not marked signed`() = runBlocking { + queue(kind = 10_012, content = "relay feeds") + queue(kind = 30_443, content = "key package") + + drain(failKinds = setOf(10_012)) + + assertNull( + backlog().single().signedAt, + "a row still on the queue cannot also be signed", + ) + } +}