test: cover the outbound broadcast queue and its stale sweep
Nothing else drains BroadcastNostrEventRequest, so a row this DAO fails to hand back is an event that never reaches any relay -- and the failure is silent, because a queue returning nothing is indistinguishable from an empty one. That already happened. The observer's predicate carried a `createdAt > :now` bound whose `now` was evaluated once, when the Flow was built. Instants persist at second resolution, so it hid every broadcast enqueued during the observer's own start second -- the entire profile-creation burst -- plus everything a previous session had left pending. There is a test here for exactly that shape: a row enqueued before the observer existed has to come back. The stale sweep, which is the other half of not losing events. A request is flipped to "processing" before a publish is attempted, and a timeout or a dropped socket leaves it there; nothing observes "processing" or "failed", so those rows are dead weight until the sweep requeues them. Covered: both stale statuses flip to "pending" and are counted; a row already pending is not touched, so the returned count is not inflated by work that was never stale. Covered separately, because it is the reason the sweep is bounded at all: a row newer than the cutoff is left alone. A publish running right now holds its row in "processing", and requeueing that would hand the same event to a second publish while the first is still in flight. The bound is `<=`, so a row stamped exactly on the cutoff second is swept -- asserted, since that is the boundary the second resolution of these timestamps makes common rather than rare. Also covered: the queue drains oldest first; a "processing" row is not handed out as pending work; and getFirstBroadcastNostrEventRequestByNostrEventId returns the oldest of an event's per-relay rows rather than the only one, since an event is queued once per target relay. One test is deliberately kept despite not being able to fail, and says so in its own comment. `requests sharing a timestamp drain in insertion order` pins the observable order of a same-second burst, which is what callers depend on -- but deleting the `, id ASC` tiebreak leaves it passing, because `id` is an autoGenerate primary key and therefore the rowid, so sqlite's unspecified ordering already coincides with it under this plan. That coincidence is the argument for keeping the explicit tiebreak rather than against it: it is not contractual, and an index or a different plan can change it. Recording the limit in the test seemed better than implying a guard that is not there. Verified by mutation: reversing the drain order fails the oldest-first test. Removing only the tiebreak fails nothing, which is how the limitation above was found rather than assumed. Both mutations were reverted; no production source is touched by this commit. 9 tests. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
@@ -0,0 +1,201 @@
|
||||
package press.mantra.compose.database.dao
|
||||
|
||||
import androidx.room3.Room
|
||||
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.BroadcastNostrEventRequest
|
||||
import press.mantra.compose.database.model.NostrEvent
|
||||
import kotlin.test.AfterTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* The outbound queue. Nothing else drains this table, so a row it fails to hand back is an
|
||||
* event that is never sent to any relay -- and the failure is silent, because a queue that
|
||||
* returns nothing looks exactly like a queue that is empty.
|
||||
*
|
||||
* That is not hypothetical here. The observer's predicate used to carry a `createdAt > :now`
|
||||
* bound whose `now` was evaluated once, when the Flow was built. Instants persist at second
|
||||
* resolution, so it hid every broadcast enqueued during the observer's own start second -- the
|
||||
* whole profile-creation burst -- along with everything a previous session had left pending.
|
||||
*/
|
||||
class BroadcastNostrEventRequestDaoJvmTest {
|
||||
|
||||
private val db: MantraDatabase = getRoomDatabase(
|
||||
Room.inMemoryDatabaseBuilder<MantraDatabase>()
|
||||
)
|
||||
|
||||
@AfterTest
|
||||
fun closeDb() = db.close()
|
||||
|
||||
private val author = "a".repeat(64)
|
||||
|
||||
private suspend fun queue(
|
||||
eventId: String,
|
||||
status: String = "pending",
|
||||
createdAt: Instant = Instant.fromEpochSeconds(1_000),
|
||||
relayURL: String = "wss://relay.example",
|
||||
): Long {
|
||||
val id = eventId.padEnd(64, '0')
|
||||
// BroadcastNostrEventRequest.nostrEventId is a foreign key onto NostrEvent: a broadcast
|
||||
// cannot be queued for an event that was never stored.
|
||||
if (db.nostrEventDao().getNostrEventById(id) == null) {
|
||||
db.nostrEventDao().upsert(
|
||||
NostrEvent(
|
||||
id = id,
|
||||
pubKey = author,
|
||||
kind = 1,
|
||||
tags = emptyArray(),
|
||||
content = "outbound",
|
||||
sig = "0".repeat(128),
|
||||
)
|
||||
)
|
||||
}
|
||||
return db.broadcastNostrEventRequestDao().upsert(
|
||||
BroadcastNostrEventRequest(
|
||||
nostrEventId = id,
|
||||
relayURL = relayURL,
|
||||
status = status,
|
||||
createdAt = createdAt,
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
private suspend fun statusOf(requestId: Long): String =
|
||||
db.broadcastNostrEventRequestDao().getAllBroadcastNostrEventRequests()
|
||||
.single { it.id == requestId }.status
|
||||
|
||||
private suspend fun pendingHead() =
|
||||
db.broadcastNostrEventRequestDao().observeBroadcastNostrEventRequestsByStatus("pending").first()
|
||||
|
||||
/**
|
||||
* A request is flipped to "processing" before the publish is attempted, and a timeout or a
|
||||
* dropped socket leaves it there. Nothing observes "processing" or "failed", so without
|
||||
* this sweep those rows are dead weight and the events never go out.
|
||||
*/
|
||||
@Test
|
||||
fun `an interrupted publish is requeued on startup`() = runBlocking {
|
||||
val processing = queue("1", status = "processing")
|
||||
val failed = queue("2", status = "failed")
|
||||
|
||||
val changed = db.broadcastNostrEventRequestDao()
|
||||
.requeueStaleBroadcastNostrEventRequests(Instant.fromEpochSeconds(2_000))
|
||||
|
||||
assertEquals(2, changed, "both stale rows should have been requeued")
|
||||
assertEquals("pending", statusOf(processing))
|
||||
assertEquals("pending", statusOf(failed))
|
||||
}
|
||||
|
||||
/** A row already pending is not stale; touching it would inflate the reported count. */
|
||||
@Test
|
||||
fun `a pending request is left alone by the sweep`() = runBlocking {
|
||||
queue("1", status = "pending")
|
||||
|
||||
val changed = db.broadcastNostrEventRequestDao()
|
||||
.requeueStaleBroadcastNostrEventRequests(Instant.fromEpochSeconds(2_000))
|
||||
|
||||
assertEquals(0, changed)
|
||||
}
|
||||
|
||||
/**
|
||||
* The reason the sweep is bounded at all. A publish running right now holds its row in
|
||||
* "processing"; requeueing that would hand the same event to a second publish while the
|
||||
* first is still in flight.
|
||||
*/
|
||||
@Test
|
||||
fun `a request newer than the cutoff is left for the process that owns it`() = runBlocking {
|
||||
val inFlight = queue("1", status = "processing", createdAt = Instant.fromEpochSeconds(3_000))
|
||||
|
||||
val changed = db.broadcastNostrEventRequestDao()
|
||||
.requeueStaleBroadcastNostrEventRequests(Instant.fromEpochSeconds(2_000))
|
||||
|
||||
assertEquals(0, changed)
|
||||
assertEquals("processing", statusOf(inFlight))
|
||||
}
|
||||
|
||||
/** The bound is `<=`, so a row stamped exactly on the cutoff second is swept. */
|
||||
@Test
|
||||
fun `a request stamped exactly on the cutoff is requeued`() = runBlocking {
|
||||
val onCutoff = queue("1", status = "processing", createdAt = Instant.fromEpochSeconds(2_000))
|
||||
|
||||
val changed = db.broadcastNostrEventRequestDao()
|
||||
.requeueStaleBroadcastNostrEventRequests(Instant.fromEpochSeconds(2_000))
|
||||
|
||||
assertEquals(1, changed)
|
||||
assertEquals("pending", statusOf(onCutoff))
|
||||
}
|
||||
|
||||
/**
|
||||
* The regression the observer's comment describes. A row enqueued before the observer was
|
||||
* built -- a previous session's leftovers -- has to come back. The old bound compared
|
||||
* against a `now` captured when the Flow was created, so it returned nothing here and the
|
||||
* queue looked empty forever.
|
||||
*/
|
||||
@Test
|
||||
fun `the queue hands back work enqueued before the observer existed`() = runBlocking {
|
||||
val old = queue("1", createdAt = Instant.fromEpochSeconds(1_000))
|
||||
|
||||
val head = assertNotNull(pendingHead(), "a previous session's pending row was not returned")
|
||||
|
||||
assertEquals(old, head.broadcastNostrEventRequest.id)
|
||||
}
|
||||
|
||||
/** Oldest first: the queue drains in the order things were enqueued. */
|
||||
@Test
|
||||
fun `the queue hands back the oldest pending request first`() = runBlocking {
|
||||
queue("2", createdAt = Instant.fromEpochSeconds(3_000))
|
||||
val oldest = queue("1", createdAt = Instant.fromEpochSeconds(1_000))
|
||||
|
||||
assertEquals(oldest, assertNotNull(pendingHead()).broadcastNostrEventRequest.id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Instants persist at second resolution, so a burst enqueued in one second is all one
|
||||
* timestamp, and the head of the queue is decided by the `id ASC` tiebreak.
|
||||
*
|
||||
* Note what this test does and does not do. It pins the observable drain order, which is
|
||||
* what a caller depends on. It cannot fail if the tiebreak is deleted: `id` is an
|
||||
* autoGenerate primary key and therefore the rowid, so sqlite's own unspecified ordering
|
||||
* already coincides with it under this plan -- removing `, id ASC` leaves every assertion
|
||||
* here passing. That coincidence is exactly why the explicit tiebreak is worth keeping:
|
||||
* it is not contractual, and an index or a different query plan can change it.
|
||||
*/
|
||||
@Test
|
||||
fun `requests sharing a timestamp drain in insertion order`() = runBlocking {
|
||||
val sameSecond = Instant.fromEpochSeconds(1_000)
|
||||
val first = queue("1", createdAt = sameSecond)
|
||||
queue("2", createdAt = sameSecond)
|
||||
queue("3", createdAt = sameSecond)
|
||||
|
||||
assertEquals(first, assertNotNull(pendingHead()).broadcastNostrEventRequest.id)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the queue is empty when nothing is pending`() = runBlocking {
|
||||
queue("1", status = "processing")
|
||||
|
||||
assertNull(pendingHead(), "a processing row must not be handed out as pending work")
|
||||
}
|
||||
|
||||
/**
|
||||
* One event is queued once per target relay, so this returns the oldest of several rows
|
||||
* rather than the only one.
|
||||
*/
|
||||
@Test
|
||||
fun `the first request for an event is the oldest of its per-relay rows`() = runBlocking {
|
||||
val eventId = "1".padEnd(64, '0')
|
||||
val oldest = queue("1", createdAt = Instant.fromEpochSeconds(1_000), relayURL = "wss://one.example")
|
||||
queue("1", createdAt = Instant.fromEpochSeconds(2_000), relayURL = "wss://two.example")
|
||||
|
||||
val found = assertNotNull(
|
||||
db.broadcastNostrEventRequestDao().getFirstBroadcastNostrEventRequestByNostrEventId(eventId)
|
||||
)
|
||||
|
||||
assertEquals(oldest, found.id)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user