fix(relay): forward ephemeral events to live subscribers (NIP-01)
LiveEventStore.query had a race window between emitting EOSE and registering as a SharedFlow collector — any event emitted in that window was lost because newEventStream has replay=0. The race was usually masked by SQLite write latency for persisted kinds, but it fired reliably for ephemeral kinds (20000-29999) where store.insert is a no-op. Per NIP-01 ephemeral events are not persisted but MUST still reach matching active subscriptions. Fixed by registering the live collector before signalling EOSE via Flow.onSubscription. Adds two tests: one proves an ephemeral event reaches an active subscriber, the other proves it isn't persisted (a follow-up REQ returns zero events).
This commit is contained in:
@@ -311,6 +311,64 @@ class Nip01ComplianceTest {
|
|||||||
client.unsubscribe("live-2")
|
client.unsubscribe("live-2")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// -- Ephemeral events (NIP-01: kinds 20000-29999) ----------------------
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Ephemeral events MUST be forwarded to active subscriptions whose
|
||||||
|
* filters match, even though the relay does not persist them.
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
fun ephemeralEventForwardedToActiveSubscription() =
|
||||||
|
runBlocking {
|
||||||
|
val ch = Channel<Event>(UNLIMITED)
|
||||||
|
val gotEose = Channel<Unit>(UNLIMITED)
|
||||||
|
client.subscribe(
|
||||||
|
"eph-1",
|
||||||
|
mapOf(relayUrl to listOf(Filter(kinds = listOf(20_001)))),
|
||||||
|
object : SubscriptionListener {
|
||||||
|
override fun onEvent(
|
||||||
|
event: Event,
|
||||||
|
isLive: Boolean,
|
||||||
|
relay: NormalizedRelayUrl,
|
||||||
|
forFilters: List<Filter>?,
|
||||||
|
) {
|
||||||
|
ch.trySend(event)
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun onEose(
|
||||||
|
relay: NormalizedRelayUrl,
|
||||||
|
forFilters: List<Filter>?,
|
||||||
|
) {
|
||||||
|
gotEose.trySend(Unit)
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
withTimeout(5000) { gotEose.receive() }
|
||||||
|
|
||||||
|
hub.getOrCreate(relayUrl).publish(fakeEvent(70, kind = 20_001, content = "ephemeral-payload"))
|
||||||
|
|
||||||
|
val received = withTimeout(5000) { ch.receive() }
|
||||||
|
assertEquals("ephemeral-payload", received.content)
|
||||||
|
assertEquals(20_001, received.kind)
|
||||||
|
client.unsubscribe("eph-1")
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Ephemeral events MUST NOT be persisted. A REQ issued after the
|
||||||
|
* event was published returns nothing.
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
fun ephemeralEventIsNotStoredAndDoesNotShowOnFollowupReq() =
|
||||||
|
runBlocking {
|
||||||
|
// Publish ephemeral first — no live subscriber listening.
|
||||||
|
hub.getOrCreate(relayUrl).publish(fakeEvent(71, kind = 20_002, content = "vanish"))
|
||||||
|
|
||||||
|
// Late subscriber: should see EOSE with no events.
|
||||||
|
val (events, eose) = collectUntilEose(Filter(kinds = listOf(20_002)))
|
||||||
|
assertTrue(eose, "EOSE must fire for an ephemeral kind even if zero events match")
|
||||||
|
assertEquals(0, events.size, "Ephemeral events must not be persisted")
|
||||||
|
}
|
||||||
|
|
||||||
// -- Multi-relay --------------------------------------------------------
|
// -- Multi-relay --------------------------------------------------------
|
||||||
|
|
||||||
/** A single client can hold subscriptions against multiple relays simultaneously. */
|
/** A single client can hold subscriptions against multiple relays simultaneously. */
|
||||||
|
|||||||
+17
-11
@@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
|||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
import kotlinx.coroutines.channels.BufferOverflow
|
import kotlinx.coroutines.channels.BufferOverflow
|
||||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||||
|
import kotlinx.coroutines.flow.onSubscription
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A reactive event store that combines historical data retrieval with live event streaming.
|
* A reactive event store that combines historical data retrieval with live event streaming.
|
||||||
@@ -56,18 +57,23 @@ class LiveEventStore(
|
|||||||
onEach: (Event) -> Unit,
|
onEach: (Event) -> Unit,
|
||||||
onEose: () -> Unit,
|
onEose: () -> Unit,
|
||||||
) {
|
) {
|
||||||
// 1. Replay stored events matching filters.
|
// Order matters: register the live collector BEFORE replaying
|
||||||
store.query(filters, onEach)
|
// stored events and signalling EOSE. Otherwise an event emitted
|
||||||
|
// between EOSE and `collect` is lost because [newEventStream] has
|
||||||
// 2. Signal end of stored events.
|
// replay=0. The race is only occasionally visible for kinds the
|
||||||
onEose()
|
// store persists (insert latency masks it) but fires reliably for
|
||||||
|
// ephemeral kinds (20000-29999) where insert is a no-op — and
|
||||||
// 3. Stream live events until cancelled.
|
// ephemeral events MUST still reach matching live subscribers per
|
||||||
newEventStream.collect { newEvent ->
|
// NIP-01.
|
||||||
if (filters.any { it.match(newEvent) }) {
|
newEventStream
|
||||||
onEach(newEvent)
|
.onSubscription {
|
||||||
|
store.query(filters, onEach)
|
||||||
|
onEose()
|
||||||
|
}.collect { newEvent ->
|
||||||
|
if (filters.any { it.match(newEvent) }) {
|
||||||
|
onEach(newEvent)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
suspend fun count(filters: List<Filter>) = store.count(filters)
|
suspend fun count(filters: List<Filter>) = store.count(filters)
|
||||||
|
|||||||
Reference in New Issue
Block a user