mantra/mantra-kmp
1
0
Fork 0
Code Issues 2 Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

fix: widen the last of the notary's queues, the one already known to stall

Browse Source 
9107b81 left the unsigned Nostr event queue alone on the grounds that it
carries account traffic rather than group messages. It has the same defect, and
unlike the other two it is not a latent one: NostrDao carries a written account
of it having already happened.

**The queue.** UnsignedNostrEventDao.observeUnsignedNostrEvents selected every
unsigned row for a key and returned `Flow<UnsignedNostrEvent?>`, so Room handed
back the first and dropped the rest:

    SELECT * FROM UnsignedNostrEvent
    WHERE pubKey = :publicKey AND signedAt IS NULL
    ORDER BY kind ASC

The only exit is a successful publish. NostrDao.commitPublishedNostrEvent
stamps `signedAt`, stores the signed event and queues a
BroadcastNostrEventRequest per relay, all in one @Transaction. Nothing else
clears it -- no attempt count, no failure status, no sweep -- so a row that
cannot be published is selected again at the head of every later emission.

**It has already happened.** The comment on commitPublishedNostrEvent is the
report: indexing used to share that transaction, so any throw in it rolled
`signedAt` back, "leaving the notary to re-select the same unsigned row forever
and never sign anything queued behind it, including the MLS key package that
goes last". That was closed by giving indexing its own transaction, and
NostrDaoJvmTest pins it. What it did not close is the queue: it fixed the one
known way to produce a stuck row and left the queue as narrow as it was, so the
next way in has the same consequence.

**The frozen variant.** Both other queues sat behind distinctUntilChanged too,
and 9107b81 recorded that they failed by different mechanics depending on
whether the row's `equals` was honest. UnsignedNostrEvent.equals is a plain
value comparison with no @Ignore'd Logger in it, so this is the worse one: the
stuck row's re-emission compared equal to the last, was dropped as no change,
and the collector saw nothing again for the life of the session. Not a retry
loop that never advances -- a collector that has stopped, while rows keep
piling up behind a head nobody is looking at.

**What sits behind the head.** The kinds matter here in a way they did not for
the other two, because `kind ASC` is not an arbitrary order. An account queues
0 metadata, 3 contacts, 10007 search relays, 10012 relay feeds, 10050 DM
relays, 10051 key package relays, and later 30443, the MLS key package -- which
is what "goes last" means, since 30443 is the highest of them.

Sitting in the middle is 10012, the one row of that burst carrying
`privateTags`, and therefore the only one whose publish runs a NIP-44
encryption before it signs. A throw there takes 10050, 10051 and 30443 with
it: both relay lists a peer needs to find this user, and the key package they
need to invite them into a marmot group. The device looks fine to its owner --
the profile is announced, the gate has opened -- and is unreachable to everyone
else. That is the shape of the next stall rather than a hypothetical one, which
is why it is written on the query.

**The fix.** Same as the other two. The query returns the backlog,
NotaryViewModel walks it serially and keeps guardNotarization per row, and
distinctUntilChanged goes. A publish that throws rolls back its own transaction
and nothing else, so the row stays queued for the next pass while the rest of
the account's events go out.

`kind ASC` is kept, and now says why: kind 0 sorts first and NavigationViewModel
holds the user on "announcing your profile" until it lands, so the order is
load-bearing rather than incidental. `id ASC` is added as the tiebreak -- it is
the autogenerated row id, so two events of one kind publish in the order they
were queued, which for a replaceable kind is the difference between the newest
version standing and an older one being published last and winning.

Nothing about the navigation gate changes. It reads the kind-0 LocalAccount's
relations, not the queue's shape, and kind 0 is still the first row of the
first pass.

With this the notary has no single-row queue left. The one remaining
`distinctUntilChanged` in it, on observeActiveMarmotKeyPackageBundle, is
correct: that flow is a state observation -- null means "no active bundle,
make one" -- not a backlog, and re-running the creation on every unrelated
emission is exactly what it is there to prevent.

**Tests.** UnsignedNostrEventQueueJvmTest, seven of them, Room-backed. The
account's real kinds are used rather than an inert one, because their sort
order is the whole reason a stall in the middle of that burst leaves a user
nobody can reach. "an event that cannot be published no longer hides the ones
behind it" fails the 10012 row and asserts the metadata ahead of it and both
relay lists and the key package behind it are all signed, with the stopper
still queued and still unsigned; "an event past the stopper is queued for every
relay" then checks the key package got a pending BroadcastNostrEventRequest per
relay, since per 0211764 a row with `signedAt` and no request is the same
silence in a different place. The rest pin the backlog arriving whole and
lowest-kind first, one kind's rows keeping their insertion order across two
passes, and one key's queue not containing another's.

**Not covered.** The drain in the test is shaped like NotaryViewModel's loop
but is the test's own, so these pin the DAO's half -- the backlog arrives
whole, publishing some rows neither disturbs nor depends on the others -- and
not the collector. Standing that up wants an ActiveWallet StateFlow and the
whole ViewModel with it. A row that can never be published is still never
published; as with 9107b81 it is only no longer contagious, and nothing yet
tells the user which of their events is stuck or why.

Verified: :composeApp:compileDebugKotlinAndroid succeeds; 518 tests pass, 511
before these seven.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
2026-09-06 19:04:27 +02:00
parent 9107b81c99
commit 950deb2288
5 changed files with 307 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventDao.kt
View File

@@ -34,6 +34,35 @@ interface UnsignedNostrEventDao {
@Query("SELECT * FROM UnsignedNostrEvent WHERE kind = 0")
suspend fun getLocalAccounts(): List<LocalAccount>
@Query("SELECT * FROM UnsignedNostrEvent WHERE pubKey = :publicKey AND signedAt IS NULL ORDER BY kind ASC")
fun observeUnsignedNostrEvents(publicKey: String): Flow<UnsignedNostrEvent?>
/**
* 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<List<UnsignedNostrEvent>>
}

2
composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt
View File

@@ -100,7 +100,7 @@ class DatabaseNostrRepository(
database.connectionDao().delete(connection)
}
override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow<UnsignedNostrEvent?> {
override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow<List<UnsignedNostrEvent>> {
return database.unsignedNostrEventDao().observeUnsignedNostrEvents(publicKey)
}

11
composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt
View File

@@ -34,7 +34,14 @@ interface NostrRepository {
suspend fun deleteConnection(connection: Connection)
suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow<UnsignedNostrEvent?>
/**
* 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<List<UnsignedNostrEvent>>
suspend fun observePendingBroadcastNostrEventRequests(): Flow<LocalBroadcastNostrEventRequest?>
@@ -195,7 +202,7 @@ interface NostrRepository {
TODO("Not yet implemented")
}
override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow<UnsignedNostrEvent?> {
override suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow<List<UnsignedNostrEvent>> {
TODO("Not yet implemented")
}

25
composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt
View File

@@ -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<Event>(

243
composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/UnsignedNostrEventQueueJvmTest.kt Normal file
View File

@@ -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<MantraDatabase>()
)
@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<Array<String>>? = 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<Event>(
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<Int> = emptySet()): List<Int> {
val failed = mutableListOf<Int>()
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",
)
}
}