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 27cfca37..67da567f 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 @@ -518,28 +518,55 @@ class DatabaseNostrRepository( synchronizationRelayURLs: List, 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, + 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( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt index 16b1d44b..09752bbd 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt @@ -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 { + 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 { addRelaysIfMissing( setOf( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt index fe9bc561..256828de 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt @@ -144,6 +144,30 @@ class RelaysSocketManager( } + /** @see RelayPool.openLiveSubscription */ + suspend fun openLiveSubscription(reqCommand: ReqCmd, relayUrl: String): Flow { + 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, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt index aa4442d8..23740028 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt @@ -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() + /** + * 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( + 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) { 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 44171146..c96ccc72 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt @@ -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, + level: Int, + activeKeyPair: KeyPair + ) + suspend fun queueSynchronizeNostrEvent( synchronizeNostrEventRequests: List, ) @@ -280,6 +297,16 @@ interface NostrRepository { TODO("Not yet implemented") } + override suspend fun saveNostrEvent( + nostrEvent: NostrEvent, + relayURL: String, + synchronizationRelayURLs: List, + level: Int, + activeKeyPair: KeyPair + ) { + TODO("Not yet implemented") + } + override suspend fun queueSynchronizeNostrEvent(synchronizeNostrEventRequests: List) { TODO("Not yet implemented") }