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

feat(relays): plumbing for subscriptions that are meant to stay open

Browse Source 
Three pieces, none of which opens a subscription yet — the manager that does
lands next.

**A subscription that survives EOSE.** RelayPool.openLiveSubscription is query()
without completeOnSubscriptionEnd. For a one-shot request EOSE is the end: it is
what finishes the collector, frees the subscription slot and triggers the CLOSE.
For a live subscription EOSE is only the boundary between stored history and the
live tail, and the relay owes us nothing further to mark an end — so the flow
ends when, and only when, the caller stops collecting.

The returned flow is attached to the socket CLIENT's message flow rather than to
a session, so it survives a drop: the socket reconnects, the previous commit's
replay re-sends the REQ, and the same collector starts receiving again with
nothing rebuilt. updateLiveSubscription re-sends under the same subscription id
to widen or narrow the filter in place — a relay answers a repeated REQ on an
existing id by replacing that subscription's filter, so there is no
close-and-reopen and the collector never notices.

**Ordered, buffered inbound.** The socket's incomingMessages was a rendezvous
SharedFlow emitted into from a coroutine launched per message. Two consequences a
short request/response collector never noticed, and a permanent one would:

  - messages reached collectors in whatever order those coroutines happened to be
    scheduled, and
  - an emit with every collector busy blocked on the slowest of them.

It now has a 256-message buffer and is emitted into inline, on the reader, in
wire order. That also lets the 75ms sleep before every EOSE go: it existed to
hope that the events preceding an EOSE had already been delivered by their own
coroutines, which ordering now guarantees outright.

The buffer is the back-pressure boundary — a collector may fall 256 behind
before it slows its socket's reader down. Deep enough to absorb the burst
between a REQ and its EOSE while a collector writes each event to SQLite, not so
deep that a stuck collector is invisible.

**A saveNostrEvent that does not need a request row.** Both existing overloads
take the row an event was fetched for, because they also record which request
produced it and flip that row to "processed". A live subscription has no row and
never finishes, so the new overload carries relayURL and level itself and goes
straight to storeNostrEvent. Everything downstream — indexing, gift wrap
unwrapping, Welcome handling, MLS decryption — is unchanged.

Fixed while adding it: the negentropy overload was writing outside
storeNostrEventMutex while the other one held it. storeNostrEvent reads an event
and then writes it and its indexes, so two of those interleaving is a lost
update. Survivable while a single queue was the only writer; not survivable with
a live subscription writing alongside a backfill.

Deviation from docs/long-running-sync.md worth noting: the doc proposed one
permanent collector per socket dispatching by subscription id prefix. With the
inbound flow now buffered and only a handful of live subscriptions per relay, a
collector per subscription has the same properties for less machinery. If the
live subscription count per relay ever grows, the router is the next move.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
2026-09-05 16:05:19 +02:00
parent c3b0c8671b
commit 3ade53607a
5 changed files with 185 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -518,28 +518,55 @@ class DatabaseNostrRepository(
synchronizationRelayURLs: List<String>,
activeKeyPair: KeyPair
) {
logger.d("saveNostrEvent: $nostrEvent")
// Under the same lock as the other overloads. storeNostrEvent reads the event, then
// writes it and its indexes; two of those interleaving is a lost update. This path
// used to skip the lock, which was survivable only while a single queue was the one
// thing writing — a live subscription writing alongside a backfill is not that.
storeNostrEventMutex.withLock {
logger.d("saveNostrEvent: $nostrEvent")
try {
database.nostrDao().storeNostrEvent(
nostrEvent,
relayURL = negentropySynchronizeRequest.relayURL,
synchronizationRelayURLs = synchronizationRelayURLs,
level = negentropySynchronizeRequest.level,
activeKeyPair = activeKeyPair
)
database.negentropySynchronizeRequestDao().upsert(
negentropySynchronizeRequest.copy(
status = "processed"
try {
database.nostrDao().storeNostrEvent(
nostrEvent,
relayURL = negentropySynchronizeRequest.relayURL,
synchronizationRelayURLs = synchronizationRelayURLs,
level = negentropySynchronizeRequest.level,
activeKeyPair = activeKeyPair
)
)
} catch (e: Throwable) {
logger.e("Failed to save nostr event $nostrEvent.id", e)
database.negentropySynchronizeRequestDao().upsert(
negentropySynchronizeRequest.copy(
status = "processed"
)
)
} catch (e: Throwable) {
logger.e("Failed to save nostr event ${nostrEvent.id}", e)
}
}
}
override suspend fun saveNostrEvent(
nostrEvent: NostrEvent,
relayURL: String,
synchronizationRelayURLs: List<String>,
level: Int,
activeKeyPair: KeyPair
) {
storeNostrEventMutex.withLock {
logger.d("saveNostrEvent (live): $nostrEvent")
try {
database.nostrDao().storeNostrEvent(
nostrEvent,
relayURL = relayURL,
synchronizationRelayURLs = synchronizationRelayURLs,
level = level,
activeKeyPair = activeKeyPair
)
} catch (e: Throwable) {
logger.e("Failed to save nostr event ${nostrEvent.id}", e)
}
}
}
override suspend fun queueSynchronizeNostrEvent(

59
composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt
View File

@@ -309,6 +309,65 @@ class RelayPool(
}
}
/**
* Opens a subscription that is meant to stay open, and returns a flow that never
* completes on its own.
*
* The difference from [query] is the absence of `completeOnSubscriptionEnd`. For a
* one-shot request EOSE is the end — it is what finishes the collector, frees the
* subscription slot and triggers the CLOSE. For a live subscription EOSE is only the
* boundary between the stored history and the live tail, and the relay owes us nothing
* further to mark the end. The flow therefore ends when, and only when, the caller
* stops collecting it.
*
* The returned flow is attached to the socket CLIENT's message flow, not to a session,
* so it survives a drop: the socket reconnects, [replayRetainedRequests] re-sends this
* REQ, and the same collector starts receiving again without anything being rebuilt.
*/
suspend fun openLiveSubscription(reqCommand: ReqCmd, relayUrl: String): Flow<press.mantra.compose.network.sockets.NostrIncomingMessage> {
addRelaysIfMissing(
setOf(
NormalizedRelayUrl(relayUrl).url.toRelayDTO()
)
)
val nostrSocketClient = socketClientFor(relayUrl)
?: throw press.mantra.compose.exceptions.NetworkException("$relayUrl is not connected")
val filterRequest = OptimizedJsonMapper.toJson(reqCommand)
// Attached before the send: the socket's message flow is buffered but not replayed,
// so a collector subscribing after the first events arrived would miss them.
val eventFlow = nostrSocketClient.incomingMessages.filterBySubscriptionId(id = reqCommand.subId)
retainRequest(relayUrl = relayUrl, subId = reqCommand.subId, message = filterRequest)
nostrSocketClient.sendMESSAGE(filterRequest)
return eventFlow
}
/**
* Narrows or widens a live subscription in place.
*
* A relay answers a repeated REQ on an existing subscription id by replacing that
* subscription's filter, so this never closes anything: the collector opened by
* [openLiveSubscription] keeps running and simply starts matching the new filter.
* Re-retained too, so a reconnect replays the new filter rather than the old one.
*/
suspend fun updateLiveSubscription(reqCommand: ReqCmd, relayUrl: String) {
val nostrSocketClient = socketClientFor(relayUrl)
?: throw press.mantra.compose.exceptions.NetworkException("$relayUrl is not connected")
val filterRequest = OptimizedJsonMapper.toJson(reqCommand)
retainRequest(relayUrl = relayUrl, subId = reqCommand.subId, message = filterRequest)
nostrSocketClient.sendMESSAGE(filterRequest)
}
/** Ends a live subscription: forgets the retained REQ and tells the relay to stop. */
suspend fun closeLiveSubscription(subId: String, relayUrl: String) =
closeQuery(CloseCmd(subId = subId), relayUrl)
suspend fun negentropySync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow<press.mantra.compose.network.sockets.NostrIncomingMessage> {
addRelaysIfMissing(
setOf(

24
composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt
View File

@@ -144,6 +144,30 @@ class RelaysSocketManager(
}
/** @see RelayPool.openLiveSubscription */
suspend fun openLiveSubscription(reqCommand: ReqCmd, relayUrl: String): Flow<press.mantra.compose.network.sockets.NostrIncomingMessage> {
return relayPool.openLiveSubscription(
reqCommand = reqCommand,
relayUrl = relayUrl
)
}
/** @see RelayPool.updateLiveSubscription */
suspend fun updateLiveSubscription(reqCommand: ReqCmd, relayUrl: String) {
return relayPool.updateLiveSubscription(
reqCommand = reqCommand,
relayUrl = relayUrl
)
}
/** @see RelayPool.closeLiveSubscription */
suspend fun closeLiveSubscription(subId: String, relayUrl: String) {
return relayPool.closeLiveSubscription(
subId = subId,
relayUrl = relayUrl
)
}
suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) {
return relayPool.closeQuery(
closeCmd = closeCmd,

44
composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt
View File

@@ -36,7 +36,6 @@ import kotlin.math.min
import kotlin.math.pow
import kotlin.random.Random
import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds
import kotlin.uuid.ExperimentalUuidApi
import kotlin.uuid.Uuid
@@ -71,6 +70,14 @@ internal class NostrSocketClientImpl(
* back in lockstep for as long as the outage lasts.
*/
private const val RECONNECT_JITTER = 0.25
/**
* How far a collector may lag before its socket's reader is made to wait for it.
* Deep enough to absorb the burst a relay sends between a REQ and its EOSE while a
* collector writes each event to SQLite; not so deep that a genuinely stuck
* collector is invisible.
*/
private const val INBOUND_BUFFER = 256
}
/** Outcome of a connect attempt, so the caller can tell a first connect from a repair. */
@@ -103,7 +110,19 @@ internal class NostrSocketClientImpl(
@Volatile
override var autoReconnect: Boolean = false
private val _incomingMessages = MutableSharedFlow<NostrIncomingMessage>()
/**
* Buffered on purpose. This used to be a rendezvous flow emitted into from a coroutine
* launched per message, which had two consequences a short-lived request/response
* collector never noticed and a permanent one would: messages reached collectors in
* whatever order those coroutines were scheduled, and an emit with every collector busy
* blocked on the slowest of them.
*
* With a buffer, a collector may fall [INBOUND_BUFFER] messages behind before it slows
* the socket reader down, and the reader emits in wire order.
*/
private val _incomingMessages = MutableSharedFlow<NostrIncomingMessage>(
extraBufferCapacity = INBOUND_BUFFER,
)
override val incomingMessages = _incomingMessages.asSharedFlow()
override val socketUrl = wssUrl.cleanWebSocketUrl()
@@ -179,13 +198,13 @@ internal class NostrSocketClientImpl(
is Frame.Text -> {
val text = frame.readText()
logLargeText(text = text, url = socketUrl, incoming = true)
processIncomingMessage(text = text)
emitIncomingMessage(text = text)
}
is Frame.Binary -> {
val decompressedMessage = decompressMessage(frame.data)
logLargeText(text = decompressedMessage, url = socketUrl, incoming = true)
processIncomingMessage(text = decompressedMessage)
emitIncomingMessage(text = decompressedMessage)
}
is Frame.Close -> {
@@ -302,15 +321,14 @@ internal class NostrSocketClientImpl(
}
}
private fun processIncomingMessage(text: String) {
text.parseIncomingMessage()?.let {
scope.launch {
if (it is NostrIncomingMessage.EoseMessage) {
delay(75.milliseconds)
}
_incomingMessages.emit(value = it)
}
}
/**
* Emits inline, on the reader, so messages reach collectors in the order the relay sent
* them. That replaces a 75ms delay this used to sleep before every EOSE — a way of
* hoping the events that preceded it had already been delivered by their own coroutines.
* Ordering is now a property rather than a race that usually resolved in time.
*/
private suspend fun emitIncomingMessage(text: String) {
text.parseIncomingMessage()?.let { _incomingMessages.emit(value = it) }
}
override suspend fun sendMESSAGE(text: String, ensureSessionBeforeSend: Boolean) {

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

@@ -107,6 +107,23 @@ interface NostrRepository {
activeKeyPair: KeyPair
)
/**
* Save a nostrEvent that arrived on a subscription rather than in answer to a queued
* request.
*
* The other two overloads take the request row an event was fetched for, because they
* also have to record which request produced it and flip that row to "processed". A
* live subscription has no row and never finishes, so it carries the relay and level
* itself.
*/
suspend fun saveNostrEvent(
nostrEvent: NostrEvent,
relayURL: String,
synchronizationRelayURLs: List<String>,
level: Int,
activeKeyPair: KeyPair
)
suspend fun queueSynchronizeNostrEvent(
synchronizeNostrEventRequests: List<SynchronizeNostrEventRequest>,
)
@@ -280,6 +297,16 @@ interface NostrRepository {
TODO("Not yet implemented")
}
override suspend fun saveNostrEvent(
nostrEvent: NostrEvent,
relayURL: String,
synchronizationRelayURLs: List<String>,
level: Int,
activeKeyPair: KeyPair
) {
TODO("Not yet implemented")
}
override suspend fun queueSynchronizeNostrEvent(synchronizeNostrEventRequests: List<SynchronizeNostrEventRequest>) {
TODO("Not yet implemented")
}