feat(chat): keep a gift wrap subscription open for as long as the app is active
The first live subscription: kind 1059 p-tagged to us, on every DM relay, open until the wallet changes or the process ends. That covers direct messages and the Marmot Welcome events that make us a member of a group — so a DM now lands whether or not anyone has the chat list open, and a group invite is noticed without waiting for the next screen to schedule a sync. Nothing is removed from the queue yet. ChatRoomListViewModel and ChatMessageListViewModel still schedule what they always did; this runs beside them and the duplicate events cost a no-op storeNostrEvent each. The screens stop doing it in the last step, once groups are covered too. **No `since`.** NIP-59 randomizes a wrap's created_at into the past, and our own outbound path stamps them with TimeUtils.randomWithTwoDays() — so a wrap published right now can carry a timestamp two days old, and a `since` anywhere near the present would silently drop it. "Some messages just never arrive" is the worst failure mode to debug, and re-receiving a wrap costs one no-op write. **`limit` instead.** NIP-01 scopes limit to the initial query — the stored events a relay sends before EOSE — and explicitly not to the stream that follows, so limit=100 bounds what a reconnect costs without touching the live tail. Filling in the rest of the history stays negentropy's job. **What ends a subscription, and what doesn't.** CLOSED does, and is retried with backoff from 5s to 5m, jumping straight to the maximum when the reason parses as back-pressure — answering "too many subscriptions" by promptly opening another is how one refusal becomes a flood. EOSE does not end anything: it is the boundary between stored history and the live tail, and treating it as an end is precisely what made every sync a poll. It does reset the backoff, since a subscription that got that far was working and whatever ended it is a new problem. A dropped socket needs nothing from this loop at all — the pool replays the REQ on reconnect and the collector, attached to the socket client rather than to a session, simply starts receiving again. A NOTICE is logged and otherwise ignored. It carries no subscription id, so it reaches every collector on the socket; acting on one here would let an unrelated relay complaint tear this subscription down. Events are saved inline rather than in a launched coroutine, so they land in the order the relay sent them and the socket's buffer does the back-pressure, instead of fanning a burst out into a coroutine per event. Wiring: LiveSubscriptionManager is built by SynchronizationViewModel because it has to share that class's RelaysSocketManager — a second one would mean a second RelayPool and a duplicate socket per relay — and is launched inside the same per-wallet supervisorScope as the three pumps, so a wallet switch tears it down. It deliberately does not take from the pumps' Semaphore(4): a subscription that never finishes would hold a permit forever and permanently halve backfill throughput. The comment there now says so. isBackPressure moves out of SynchronizationViewModel to network/relays/RelayBackPressure.kt, unchanged, now that two callers need it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
@@ -0,0 +1,245 @@
|
||||
package press.mantra.compose.managers
|
||||
|
||||
import co.touchlab.kermit.Logger
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.IO
|
||||
import kotlinx.coroutines.NonCancellable
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.currentCoroutineContext
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.transformWhile
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.withContext
|
||||
import press.mantra.compose.database.model.NostrEvent
|
||||
import press.mantra.compose.network.relays.RelaysSocketManager
|
||||
import press.mantra.compose.network.relays.isRelayBackPressure
|
||||
import press.mantra.compose.network.sockets.NostrIncomingMessage
|
||||
import press.mantra.compose.nostr.Relays
|
||||
import press.mantra.compose.repository.NostrRepository
|
||||
import kotlin.time.Duration
|
||||
import kotlin.time.Duration.Companion.minutes
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
/**
|
||||
* Holds the chat subscriptions open for as long as it runs, so relays push new events at us
|
||||
* instead of us asking for them once per screen.
|
||||
*
|
||||
* See `docs/long-running-sync.md`. The short version: every chat sync in this app used to be
|
||||
* a pull — a screen queues a request row, a pump drains it, the relay answers, the
|
||||
* subscription is closed — which is structurally incapable of delivering a message that
|
||||
* arrives one second after EOSE.
|
||||
*
|
||||
* This deliberately does NOT share `SynchronizationViewModel`'s subscription semaphore. That
|
||||
* budget is four permits across all relays, sized for requests that finish; a subscription
|
||||
* that never finishes would hold one forever and permanently halve the backfill throughput.
|
||||
* The two budgets have to be read together against what a relay will actually tolerate
|
||||
* (commonly ~20 per connection), which they comfortably are.
|
||||
*/
|
||||
class LiveSubscriptionManager(
|
||||
private val relaysSocketManager: RelaysSocketManager,
|
||||
private val nostrRepository: NostrRepository,
|
||||
) {
|
||||
private val logger = Logger.withTag(TAG)
|
||||
|
||||
companion object {
|
||||
private const val TAG = "LiveSubscriptionManager"
|
||||
|
||||
/**
|
||||
* Stable, and the same on every relay. A subscription id is scoped to its socket, and
|
||||
* a relay answers a repeated REQ on an existing id by replacing that subscription's
|
||||
* filter — which is what lets a filter be widened later without a close-and-reopen.
|
||||
*/
|
||||
const val GIFT_WRAP_SUB_ID = "live-giftwrap"
|
||||
|
||||
/**
|
||||
* How much history to ask for when the subscription is opened.
|
||||
*
|
||||
* NIP-01 scopes `limit` to the initial query — the stored events a relay sends before
|
||||
* EOSE — and explicitly not to the stream that follows. So this bounds what a
|
||||
* reconnect costs without touching the live tail at all. Filling in the rest of the
|
||||
* history is negentropy's job, not this subscription's.
|
||||
*/
|
||||
private const val INITIAL_HISTORY_LIMIT = 100
|
||||
|
||||
private val INITIAL_REOPEN_DELAY = 5.seconds
|
||||
|
||||
private val MAX_REOPEN_DELAY = 5.minutes
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs until cancelled. Cancellation is the only exit: the caller scopes this to the
|
||||
* active wallet, so a wallet switch tears every subscription down and the new wallet's
|
||||
* call builds its own.
|
||||
*/
|
||||
suspend fun observe(keyPair: KeyPair): Unit = coroutineScope {
|
||||
val publicKey = keyPair.pubKey.toHexKey()
|
||||
logger.i("Opening live subscriptions for $publicKey")
|
||||
|
||||
Relays.DefaultDMRelayList.forEach { relay ->
|
||||
launch(Dispatchers.IO) {
|
||||
runLiveSubscription(
|
||||
subId = GIFT_WRAP_SUB_ID,
|
||||
relayUrl = relay.url,
|
||||
filters = listOf(giftWrapFilter(publicKey)),
|
||||
keyPair = keyPair,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Every gift wrap addressed to us: direct messages, and the Marmot Welcome events that
|
||||
* make us a member of a group.
|
||||
*
|
||||
* There is deliberately no `since`. NIP-59 randomizes a wrap's `created_at` into the past
|
||||
* — our own outbound path stamps them with `TimeUtils.randomWithTwoDays()` — so a wrap
|
||||
* published right now can carry a timestamp two days old, and a `since` anywhere near the
|
||||
* present would silently drop it. "Some messages just never arrive" is the worst failure
|
||||
* mode to debug, and re-receiving a wrap costs one no-op `storeNostrEvent`.
|
||||
*/
|
||||
private fun giftWrapFilter(publicKey: HexKey) = Filter(
|
||||
kinds = listOf(GiftWrapEvent.KIND),
|
||||
tags = mapOf("p" to listOf(publicKey)),
|
||||
limit = INITIAL_HISTORY_LIMIT,
|
||||
)
|
||||
|
||||
/**
|
||||
* Keeps one subscription open on one relay, re-opening it if the relay ends it.
|
||||
*
|
||||
* A socket that merely drops needs nothing from here: the pool replays the REQ on
|
||||
* reconnect and the collector below, which is attached to the socket client rather than
|
||||
* to a session, simply starts receiving again. This loop is for the other case — a relay
|
||||
* that answered with CLOSED, which is terminal for that subscription and needs a new one.
|
||||
*/
|
||||
private suspend fun runLiveSubscription(
|
||||
subId: String,
|
||||
relayUrl: String,
|
||||
filters: List<Filter>,
|
||||
keyPair: KeyPair,
|
||||
) {
|
||||
var reopenDelay = INITIAL_REOPEN_DELAY
|
||||
|
||||
try {
|
||||
while (currentCoroutineContext().isActive) {
|
||||
val end = try {
|
||||
collectUntilClosed(
|
||||
subId = subId,
|
||||
relayUrl = relayUrl,
|
||||
filters = filters,
|
||||
keyPair = keyPair,
|
||||
)
|
||||
} catch (error: CancellationException) {
|
||||
throw error
|
||||
} catch (error: Throwable) {
|
||||
logger.e("Live subscription $subId on $relayUrl failed", error)
|
||||
SubscriptionEnd(reason = error.message, reachedEose = false)
|
||||
}
|
||||
|
||||
// A subscription that got as far as EOSE was working. Whatever ended it is a
|
||||
// new problem, not a continuing one, so it does not inherit the old backoff.
|
||||
if (end.reachedEose) reopenDelay = INITIAL_REOPEN_DELAY
|
||||
|
||||
reopenDelay = if (isRelayBackPressure(end.reason)) {
|
||||
// Answering "you have too many subscriptions" by opening another one
|
||||
// promptly is how one refusal becomes a flood of them.
|
||||
logger.w("$relayUrl is rate limiting $subId; backing off to the maximum")
|
||||
MAX_REOPEN_DELAY
|
||||
} else {
|
||||
minOf(reopenDelay * 2, MAX_REOPEN_DELAY)
|
||||
}
|
||||
|
||||
logger.w("Re-opening $subId on $relayUrl in $reopenDelay (ended: ${end.reason})")
|
||||
delay(reopenDelay)
|
||||
}
|
||||
} finally {
|
||||
// NonCancellable so a cancelled manager still releases the retained REQ, which is
|
||||
// what tells the socket it no longer has a reason to reconnect.
|
||||
withContext(NonCancellable) {
|
||||
runCatching { relaysSocketManager.closeLiveSubscription(subId, relayUrl) }
|
||||
.onFailure { logger.w(throwable = it) { "Failed to close $subId on $relayUrl" } }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private data class SubscriptionEnd(
|
||||
/** The relay's CLOSED reason, or the failure that ended the collection. */
|
||||
val reason: String?,
|
||||
/** Whether the relay got as far as sending EOSE before it ended. */
|
||||
val reachedEose: Boolean,
|
||||
)
|
||||
|
||||
private suspend fun collectUntilClosed(
|
||||
subId: String,
|
||||
relayUrl: String,
|
||||
filters: List<Filter>,
|
||||
keyPair: KeyPair,
|
||||
): SubscriptionEnd {
|
||||
var reason: String? = null
|
||||
var reachedEose = false
|
||||
|
||||
relaysSocketManager.openLiveSubscription(
|
||||
reqCommand = ReqCmd(subId = subId, filters = filters),
|
||||
relayUrl = relayUrl,
|
||||
).transformWhile { message ->
|
||||
emit(message)
|
||||
// CLOSED is the only message that ends a live subscription. EOSE emphatically
|
||||
// does not: it is the boundary between the stored history and the live tail,
|
||||
// and treating it as an end is exactly what makes a sync a poll.
|
||||
message !is NostrIncomingMessage.ClosedMessage
|
||||
}.collect { message ->
|
||||
when (message) {
|
||||
is NostrIncomingMessage.EventMessage -> {
|
||||
message.nostrEvent?.let { save(it, relayUrl, keyPair) }
|
||||
}
|
||||
|
||||
is NostrIncomingMessage.EventsMessage -> {
|
||||
message.nostrEvents.forEach { save(it, relayUrl, keyPair) }
|
||||
}
|
||||
|
||||
is NostrIncomingMessage.EoseMessage -> {
|
||||
reachedEose = true
|
||||
logger.i("$subId on $relayUrl has caught up; now live")
|
||||
}
|
||||
|
||||
is NostrIncomingMessage.ClosedMessage -> {
|
||||
reason = message.message
|
||||
logger.w("Relay closed $subId ($relayUrl): ${message.message}")
|
||||
}
|
||||
|
||||
is NostrIncomingMessage.NoticeMessage -> {
|
||||
// Advisory only. A NOTICE carries no subscription id, so it is delivered
|
||||
// to every collector on the socket — acting on one here would let an
|
||||
// unrelated relay complaint tear this subscription down.
|
||||
logger.d("Notice while running $subId ($relayUrl): ${message.message}")
|
||||
}
|
||||
|
||||
else -> logger.d("Unhandled message on $subId ($relayUrl): $message")
|
||||
}
|
||||
}
|
||||
|
||||
return SubscriptionEnd(reason = reason, reachedEose = reachedEose)
|
||||
}
|
||||
|
||||
/**
|
||||
* Saved inline rather than in a launched coroutine, on purpose. Sequential writes keep
|
||||
* the events in the order the relay sent them and let the socket's buffer do the
|
||||
* back-pressure, instead of fanning a burst out into a coroutine per event.
|
||||
*/
|
||||
private suspend fun save(nostrEvent: NostrEvent, relayUrl: String, keyPair: KeyPair) {
|
||||
nostrRepository.saveNostrEvent(
|
||||
nostrEvent = nostrEvent,
|
||||
relayURL = relayUrl,
|
||||
synchronizationRelayURLs = listOf(relayUrl),
|
||||
level = 0,
|
||||
activeKeyPair = keyPair,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
package press.mantra.compose.network.relays
|
||||
|
||||
/**
|
||||
* Whether a relay's CLOSED/NOTICE reason means "you are asking for too much" rather than
|
||||
* "I cannot serve this".
|
||||
*
|
||||
* The difference decides what to do next, and getting it wrong is expensive in both
|
||||
* directions: answering back-pressure by immediately opening another subscription is what
|
||||
* turns one refusal into a flood of them, while treating an unsupported filter as
|
||||
* back-pressure leaves a request unserved for minutes.
|
||||
*
|
||||
* NIP-01 gives `rate-limited:` as the machine-readable prefix; the free-text forms are what
|
||||
* relays actually send.
|
||||
*/
|
||||
fun isRelayBackPressure(reason: String?): Boolean {
|
||||
val text = reason?.lowercase() ?: return false
|
||||
|
||||
return text.startsWith("rate-limited") ||
|
||||
text.contains("rate limit") ||
|
||||
text.contains("too many") ||
|
||||
text.contains("concurrent") ||
|
||||
text.contains("slow down")
|
||||
}
|
||||
@@ -8,8 +8,10 @@ import press.mantra.compose.database.model.BroadcastNostrEventRequest
|
||||
import press.mantra.compose.database.model.SynchronizeNostrEventRequest
|
||||
import press.mantra.compose.database.model.types.SynchronizationFilter
|
||||
import press.mantra.compose.network.dto.toRelayDTO
|
||||
import press.mantra.compose.managers.LiveSubscriptionManager
|
||||
import press.mantra.compose.network.relays.RelayPool
|
||||
import press.mantra.compose.network.relays.RelaysSocketManager
|
||||
import press.mantra.compose.network.relays.isRelayBackPressure
|
||||
import press.mantra.compose.network.sockets.NostrIncomingMessage
|
||||
import press.mantra.compose.network.sockets.NostrSocketClientFactory
|
||||
import press.mantra.compose.repository.CachingImportRepository
|
||||
@@ -70,6 +72,16 @@ class SynchronizationViewModel(
|
||||
relayRepository = relayRepository,
|
||||
)
|
||||
|
||||
/**
|
||||
* The chat subscriptions that stay open while a wallet is active. Built here because it
|
||||
* has to share this class's RelaysSocketManager — a second one would mean a second
|
||||
* RelayPool and a duplicate socket per relay.
|
||||
*/
|
||||
private val liveSubscriptionManager = LiveSubscriptionManager(
|
||||
relaysSocketManager = relaysSocketManager,
|
||||
nostrRepository = nostrRepository,
|
||||
)
|
||||
|
||||
companion object {
|
||||
private const val TAG = "SynchronizationViewModel"
|
||||
|
||||
@@ -150,6 +162,10 @@ class SynchronizationViewModel(
|
||||
* delivered to every collector on the socket, logged it once per open collector.
|
||||
* Requests queue on this semaphore instead, so the backlog still drains, just not
|
||||
* all at once.
|
||||
*
|
||||
* Live subscriptions are deliberately outside this budget (see LiveSubscriptionManager):
|
||||
* a subscription that never finishes would hold a permit forever. The two have to be
|
||||
* read together against what a relay tolerates per connection, not separately.
|
||||
*/
|
||||
private val subscriptionSlots = Semaphore(MAX_CONCURRENT_SUBSCRIPTIONS)
|
||||
|
||||
@@ -167,21 +183,6 @@ class SynchronizationViewModel(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a relay's CLOSED/NOTICE reason means "you are asking for too much"
|
||||
* rather than "I cannot serve this". NIP-01 gives `rate-limited:` as the
|
||||
* machine-readable prefix; the free-text forms are what relays actually send.
|
||||
*/
|
||||
private fun isBackPressure(reason: String?): Boolean {
|
||||
val text = reason?.lowercase() ?: return false
|
||||
|
||||
return text.startsWith("rate-limited") ||
|
||||
text.contains("rate limit") ||
|
||||
text.contains("too many") ||
|
||||
text.contains("concurrent") ||
|
||||
text.contains("slow down")
|
||||
}
|
||||
|
||||
init {
|
||||
scope.launch {
|
||||
activeWalletStateFlow.collectLatest { activeWallet ->
|
||||
@@ -213,6 +214,10 @@ class SynchronizationViewModel(
|
||||
launch(Dispatchers.IO) { observePendingBroadcastNostrEventRequests(keyPair) }
|
||||
launch(Dispatchers.IO) { observePendingSyncNostrEventRequests(keyPair) }
|
||||
launch(Dispatchers.IO) { observePendingNegentropySynchronizeRequests(keyPair) }
|
||||
// Runs until cancelled rather than draining a queue, but it belongs
|
||||
// here for the same reason the pumps do: it needs the active wallet's
|
||||
// key pair, and a wallet switch must tear it down.
|
||||
launch(Dispatchers.IO) { liveSubscriptionManager.observe(keyPair) }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -512,7 +517,7 @@ class SynchronizationViewModel(
|
||||
// is telling us to ease off: answering back-pressure
|
||||
// by opening another subscription is what turned one
|
||||
// refusal into a flood of them.
|
||||
if (isBackPressure(nostrIncomingMessage.message)) {
|
||||
if (isRelayBackPressure(nostrIncomingMessage.message)) {
|
||||
logger.w("Relay ${negentropySynchronizeRequest.relayURL} is rate limiting; not retrying this request")
|
||||
} else if (negentropySynchronizeRequest.purpose != "mlsMessages") {
|
||||
nostrRepository.queueSynchronizeNostrEvent(
|
||||
|
||||
Reference in New Issue
Block a user