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

test: cover the long-running sync, and open the seams needed to do it

Browse Source 
The six commits that built the live chat sync added no tests. Everything they
touch fails silently by nature — a filter that drops messages, a subscription
that stops being replayed, a group whose id never reaches the `#h` tag — so the
symptom is always "some messages didn't arrive", days later, on someone else's
phone. 46 tests, in four files.

**What is covered**

RelayPoolSubscriptionTest (13) — the pool's half of surviving a dropped socket.
A query is retained and replayed on reconnect; a closed one is forgotten and
stops the socket reconnecting for it; closing one of two leaves the other alone;
a negentropy exchange is never replayed (its rounds are stateful, so resuming
one reconciles against a conversation the relay is no longer having); an update
to a live subscription replaces what gets replayed, including when the send
itself fails; dropping a relay or closing the pool forgets what they carried;
replay is scoped to the relay that reconnected. Plus the semantic the whole
change rests on, asserted in both directions: a live subscription keeps
delivering after EOSE, a one-shot query still ends at it.

LiveSubscriptionReconcileTest (12) — the requirement this all exists for: the
group filter follows group membership with nobody calling a subscribe function.
Joining widens the filter *in place* rather than reopening (a reopen would drop
the live tail of every other group in that chunk); leaving drops one; leaving
everything closes the subscription; churn inside the debounce window collapses
to one update; a NIP-17 room never becomes a group subscription. Then the
collect loop: events stored against the relay they came from, an event after
EOSE still stored, a CLOSED reopened once the back-off elapses and not before,
and a rate-limited CLOSED waiting far longer — but still coming back.
Backgrounding closes and foregrounding rebuilds, reconnects, and queues the
catch-up.

LiveSubscriptionPlanTest (11) — the filter and planning rules, led by the one
most likely to be "tidied up" later: the gift wrap filter carries no `since`,
because NIP-59 randomizes created_at into the past and a `since` near the
present silently drops new messages.

RelayBackPressureTest (4) and ReconnectBackoffTest (6) — the two pure decisions.
Which CLOSED reasons mean "ease off", and the backoff arithmetic including the
exponent clamp: 2.0.pow(4000) is Infinity and Duration * Double throws on it, so
without it a socket failing long enough turned its reconnect loop into a crash
loop, at the point the network was least likely to recover unaided.

**Seams opened to get there**, each a readability win on its own terms:

  - NostrSocketClientFactory becomes an interface with DefaultNostrSocketClientFactory
    behind it, so the pool can be driven by a fake socket.
  - RelayPool takes its CoroutineScope, so the replay a reconnect triggers can be
    observed rather than raced.
  - LiveSubscriptionManager depends on a new LiveSubscriptionTransport (4
    methods) rather than RelaysSocketManager, which observes the active wallet in
    its init and cannot be stood up in a test at all.
  - Its pure planning helpers move to the companion as `internal`, and its
    launches inherit the caller's dispatcher instead of pinning Dispatchers.IO.
    SynchronizationViewModel already launches observe() on IO, so nothing moves —
    but a coroutine that picks its own dispatcher cannot be driven by a test
    scheduler.
  - reconnectDelay is extracted to ReconnectBackoff.kt with jitter as a
    parameter, so the arithmetic can be pinned without randomness.
  - endsLiveSubscription names the live-subscription termination rule next to
    isTerminalFor, which is the one-shot rule. Having both named makes the
    difference between them reviewable rather than implicit.

kotlinx-coroutines-test is added to commonTest: the pool's bookkeeping is all
suspend functions and there is no runBlocking in a common source set.

The tests were checked by mutation, not just by passing — reintroducing a
`since`, making EOSE terminal, dropping the leftGroupAt filter and removing
retention from query() each produce failures.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
2026-09-05 23:31:11 +02:00
parent c8c962e4f4
commit f5eb744ca7
16 changed files with 1333 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
composeApp/build.gradle.kts
View File

@@ -121,6 +121,9 @@ kotlin {
}
commonTest.dependencies {
implementation(libs.kotlin.test)
// runTest: the relay pool's bookkeeping is all suspend functions, and there is no
// runBlocking in a common source set.
implementation(libs.kotlinx.coroutinesTest)
}
jvmMain.dependencies {
implementation(compose.desktop.currentOs)

147
composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt
View File

@@ -10,9 +10,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.IO
import kotlinx.coroutines.Job
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.cancelAndJoin
@@ -30,10 +28,12 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import press.mantra.compose.database.model.NegentropySynchronizeRequest
import press.mantra.compose.database.model.NostrEvent
import press.mantra.compose.database.model.intermdiate.LocalChatRoom
import press.mantra.compose.database.model.types.SynchronizationFilter
import press.mantra.compose.network.relays.RelaysSocketManager
import press.mantra.compose.network.relays.LiveSubscriptionTransport
import press.mantra.compose.network.relays.isRelayBackPressure
import press.mantra.compose.network.sockets.NostrIncomingMessage
import press.mantra.compose.network.sockets.endsLiveSubscription
import press.mantra.compose.nostr.Relays
import press.mantra.compose.repository.ChatRepository
import press.mantra.compose.repository.NostrRepository
@@ -63,7 +63,7 @@ import kotlin.time.Duration.Companion.seconds
* (commonly ~20 per connection), which they comfortably are.
*/
class LiveSubscriptionManager(
private val relaysSocketManager: RelaysSocketManager,
private val relaysSocketManager: LiveSubscriptionTransport,
private val nostrRepository: NostrRepository,
private val chatRepository: ChatRepository,
private val isForeground: StateFlow<Boolean>,
@@ -128,6 +128,69 @@ class LiveSubscriptionManager(
* only survives into the plain-REQ fallback.
*/
private const val CATCH_UP_LIMIT = 50
/**
* 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`.
*/
internal fun giftWrapFilter(publicKey: HexKey) = Filter(
kinds = listOf(GiftWrapEvent.KIND),
tags = mapOf("p" to listOf(publicKey)),
limit = INITIAL_HISTORY_LIMIT,
)
/**
* Marmot group messages for a chunk of the groups we belong to.
*
* Unlike gift wraps these carry honest timestamps (`MarmotOutboundDao` stamps them
* with `TimeUtils.now()`), so a `since` watermark would be safe here. It is still not
* used: `limit` already bounds the initial burst, and a watermark would have to be
* recomputed every time the chunk's membership changed.
*/
internal fun groupFilter(groupIds: List<HexKey>) = Filter(
kinds = listOf(GroupEvent.KIND),
tags = mapOf("h" to groupIds),
limit = INITIAL_GROUP_HISTORY_LIMIT,
)
/**
* The groups a live subscription should be watching.
*
* An MLS group is a room with group state; a NIP-17 room has none and is served by
* the gift wrap subscription instead. A room we have left or deleted keeps its
* history locally but must stop pulling new messages.
*
* Sorted so the same membership in a different row order is the same value, and a
* re-emit that changes nothing costs nothing downstream.
*/
internal fun groupIdsFrom(rooms: List<LocalChatRoom>): List<HexKey> =
rooms
.filter { it.chatRoom.mlsGroupState != null }
.filter { it.chatRoom.leftGroupAt == null && it.chatRoom.deletedAt == null }
.map { it.chatRoom.id }
.sorted()
internal fun groupChunks(groupIds: List<HexKey>): List<List<HexKey>> =
groupIds.chunked(MAX_GROUPS_PER_SUBSCRIPTION)
internal fun groupSubId(index: Int) = "$GROUP_SUB_ID_PREFIX$index"
internal fun subscriptionKey(relayUrl: String, subId: String) = "$relayUrl|$subId"
/**
* The chunk index a [subscriptionKey] refers to, or [Int.MAX_VALUE] for anything that
* does not parse — which reads as "past the end", so an unrecognised key is closed
* rather than kept open forever.
*/
internal fun subscriptionIndex(key: String) =
key.substringAfterLast(GROUP_SUB_ID_PREFIX).toIntOrNull() ?: Int.MAX_VALUE
}
/**
@@ -160,6 +223,11 @@ class LiveSubscriptionManager(
}
}
/**
* Everything below inherits the caller's dispatcher rather than pinning Dispatchers.IO.
* `SynchronizationViewModel` already launches [observe] on IO, so nothing has moved — but
* a coroutine that picks its own dispatcher cannot be driven by a test scheduler.
*/
private suspend fun runWhileForeground(publicKey: HexKey, keyPair: KeyPair): Unit = coroutineScope {
logger.i("Foregrounded; opening live subscriptions for $publicKey")
@@ -171,10 +239,10 @@ class LiveSubscriptionManager(
// Live subscriptions cover the window we are online for; this covers the gap we were
// not. Neither subsumes the other.
launch(Dispatchers.IO) { queueCatchUpSynchronization(publicKey) }
launch { queueCatchUpSynchronization(publicKey) }
Relays.DefaultDMRelayList.forEach { relay ->
launch(Dispatchers.IO) {
launch {
runLiveSubscription(
subId = GIFT_WRAP_SUB_ID,
relayUrl = relay.url,
@@ -184,7 +252,7 @@ class LiveSubscriptionManager(
}
}
launch(Dispatchers.IO) { followGroupMembership(publicKey = publicKey, keyPair = keyPair) }
launch { followGroupMembership(publicKey = publicKey, keyPair = keyPair) }
}
/**
@@ -241,37 +309,7 @@ class LiveSubscriptionManager(
}
private suspend fun liveGroupIds(publicKey: HexKey): List<HexKey> =
chatRepository.getChatRoomListByPublicKey(publicKey).toLiveGroupIds()
/**
* 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,
)
/**
* Marmot group messages for a chunk of the groups we belong to.
*
* Unlike gift wraps these carry honest timestamps (`MarmotOutboundDao` stamps them with
* `TimeUtils.now()`), so a `since` watermark would be safe here. It is still not used:
* `limit` already bounds the initial burst, and a watermark would have to be recomputed
* every time the chunk's membership changed.
*/
private fun groupFilter(groupIds: List<HexKey>) = Filter(
kinds = listOf(GroupEvent.KIND),
tags = mapOf("h" to groupIds),
limit = INITIAL_GROUP_HISTORY_LIMIT,
)
groupIdsFrom(chatRepository.getChatRoomListByPublicKey(publicKey))
/**
* Keeps the group subscriptions matching the groups we are actually in.
@@ -291,7 +329,7 @@ class LiveSubscriptionManager(
val subscriptions = mutableMapOf<String, Job>()
chatRepository.observeChatRoomListByPublicKey(publicKey)
.map { rooms -> rooms.toLiveGroupIds() }
.map { rooms -> groupIdsFrom(rooms) }
.distinctUntilChanged()
.debounce(GROUP_CHANGE_DEBOUNCE)
.collect { groupIds ->
@@ -318,7 +356,7 @@ class LiveSubscriptionManager(
keyPair: KeyPair,
scope: CoroutineScope,
) {
val chunks = groupIds.chunked(MAX_GROUPS_PER_SUBSCRIPTION)
val chunks = groupChunks(groupIds)
groupIdChunks = chunks
logger.i("Live group membership: ${groupIds.size} group(s) over ${chunks.size} subscription(s)")
@@ -327,7 +365,7 @@ class LiveSubscriptionManager(
val relayUrl = relay.url
chunks.forEachIndexed { index, chunk ->
val subId = "$GROUP_SUB_ID_PREFIX$index"
val subId = groupSubId(index)
val key = subscriptionKey(relayUrl = relayUrl, subId = subId)
if (subscriptions[key]?.isActive == true) {
@@ -336,7 +374,7 @@ class LiveSubscriptionManager(
relayUrl = relayUrl,
)
} else {
subscriptions[key] = scope.launch(Dispatchers.IO) {
subscriptions[key] = scope.launch {
runLiveSubscription(
subId = subId,
relayUrl = relayUrl,
@@ -369,26 +407,6 @@ class LiveSubscriptionManager(
}
}
/**
* An MLS group is a room with group state; a NIP-17 room has none and is served by the
* gift wrap subscription instead. A room we have left or deleted keeps its history
* locally but must stop pulling new messages.
*
* Sorted so the same membership in a different row order is the same value, and a
* re-emit that changes nothing costs nothing downstream.
*/
private fun List<press.mantra.compose.database.model.intermdiate.LocalChatRoom>.toLiveGroupIds() =
this
.filter { it.chatRoom.mlsGroupState != null }
.filter { it.chatRoom.leftGroupAt == null && it.chatRoom.deletedAt == null }
.map { it.chatRoom.id }
.sorted()
private fun subscriptionKey(relayUrl: String, subId: String) = "$relayUrl|$subId"
private fun subscriptionIndex(key: String) =
key.substringAfterLast(GROUP_SUB_ID_PREFIX).toIntOrNull() ?: Int.MAX_VALUE
/**
* Keeps one subscription open on one relay, re-opening it if the relay ends it.
*
@@ -476,10 +494,7 @@ class LiveSubscriptionManager(
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
!message.endsLiveSubscription()
}.collect { message ->
when (message) {
is NostrIncomingMessage.EventMessage -> {

28
composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/LiveSubscriptionTransport.kt Normal file
View File

@@ -0,0 +1,28 @@
package press.mantra.compose.network.relays
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
import kotlinx.coroutines.flow.Flow
import press.mantra.compose.network.sockets.NostrIncomingMessage
/**
* The slice of the relay layer a subscription that stays open needs.
*
* Narrower than [RelaysSocketManager] on purpose. That class observes the active wallet and
* starts collecting relay lists in its `init`, so it cannot be stood up in a test at all —
* which would otherwise leave the reconcile loop that keeps group subscriptions matching group
* membership with no coverage but the pure helpers underneath it.
*/
interface LiveSubscriptionTransport {
/** @see RelayPool.openLiveSubscription */
suspend fun openLiveSubscription(reqCommand: ReqCmd, relayUrl: String): Flow<NostrIncomingMessage>
/** @see RelayPool.updateLiveSubscription */
suspend fun updateLiveSubscription(reqCommand: ReqCmd, relayUrl: String)
/** @see RelayPool.closeLiveSubscription */
suspend fun closeLiveSubscription(subId: String, relayUrl: String)
/** @see RelayPool.reconnectAll */
suspend fun reconnectAll()
}

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

@@ -42,6 +42,12 @@ import kotlin.time.Duration.Companion.milliseconds
class RelayPool(
private val nostrSocketClientFactory: press.mantra.compose.network.sockets.NostrSocketClientFactory,
private val cachingImportRepository: press.mantra.compose.repository.CachingImportRepository,
/**
* Where this pool's own fire-and-forget work runs: status updates, the socket closes that
* a relay-list change triggers, and the subscription replay a reconnect triggers.
* Injectable so a test can drive that work deterministically rather than racing it.
*/
private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO),
) {
val logger = Logger.withTag("RelayPool")
@@ -49,8 +55,6 @@ class RelayPool(
const val PUBLISH_TIMEOUT = 30_000
}
private val scope = CoroutineScope(Dispatchers.IO)
val relays: MutableSet<press.mantra.compose.network.dto.RelayDTO> = mutableSetOf()
private val relayMutex = Mutex()

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

@@ -33,7 +33,7 @@ class RelaysSocketManager(
private val nostrSocketClientFactory: press.mantra.compose.network.sockets.NostrSocketClientFactory,
private val cachingImportRepository: press.mantra.compose.repository.CachingImportRepository,
private val relayRepository: press.mantra.compose.repository.RelayRepository,
) {
) : LiveSubscriptionTransport {
val logger = Logger.withTag("RelaysSocketManager")
private val scope = CoroutineScope(Dispatchers.IO)
private val relayPoolsMutex = Mutex()
@@ -126,8 +126,7 @@ class RelaysSocketManager(
)
}
/** @see RelayPool.reconnectAll */
suspend fun reconnectAll() = relayPool.reconnectAll()
override suspend fun reconnectAll() = relayPool.reconnectAll()
fun tryConnectingToAllRelays() {
relayPool.relays.forEach {
@@ -147,24 +146,21 @@ class RelaysSocketManager(
}
/** @see RelayPool.openLiveSubscription */
suspend fun openLiveSubscription(reqCommand: ReqCmd, relayUrl: String): Flow<press.mantra.compose.network.sockets.NostrIncomingMessage> {
override 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) {
override suspend fun updateLiveSubscription(reqCommand: ReqCmd, relayUrl: String) {
return relayPool.updateLiveSubscription(
reqCommand = reqCommand,
relayUrl = relayUrl
)
}
/** @see RelayPool.closeLiveSubscription */
suspend fun closeLiveSubscription(subId: String, relayUrl: String) {
override suspend fun closeLiveSubscription(subId: String, relayUrl: String) {
return relayPool.closeLiveSubscription(
subId = subId,
relayUrl = relayUrl

13
composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt
View File

@@ -28,6 +28,19 @@ fun NostrIncomingMessage.isTerminalFor(id: String): Boolean =
(this is NostrIncomingMessage.ClosedMessage && subscriptionId == id) ||
(this is NostrIncomingMessage.NegentropyError && subscriptionId == id)
/**
* True for the message that ends a subscription meant to stay OPEN.
*
* Only CLOSED. EOSE emphatically does not end one: it is the boundary between the stored
* history a relay had and the live tail it will now stream, and treating it as an end is
* exactly what makes a subscription a poll. Compare [isTerminalFor], which is the rule for a
* one-shot request, where EOSE is the whole point.
*
* No subscription-id argument, unlike [isTerminalFor]: a live collector is already filtered by
* id before it gets here.
*/
fun NostrIncomingMessage.endsLiveSubscription(): Boolean = this is NostrIncomingMessage.ClosedMessage
/**
* Completes the flow once the subscription is over, emitting the terminal
* message first so callers still see the EOSE/CLOSED/NEG-ERR that ended it.

30
composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt
View File

@@ -19,7 +19,25 @@ internal val defaultSocketsHttpClient by lazy {
}
object NostrSocketClientFactory {
/**
* How [press.mantra.compose.network.relays.RelayPool] gets a socket for a relay.
*
* An interface rather than the object directly, so the pool's subscription bookkeeping — what
* it retains, when it replays, when it lets a socket stop reconnecting — can be exercised
* against a fake instead of against a real websocket.
*/
interface NostrSocketClientFactory {
fun create(
wssUrl: String,
incomingCompressionEnabled: Boolean = false,
onSocketConnectionOpened: SocketConnectionOpenedCallback? = null,
onSocketConnectionClosed: SocketConnectionClosedCallback? = null,
onSocketConnectionReopened: SocketConnectionReopenedCallback? = null,
): NostrSocketClient
}
object DefaultNostrSocketClientFactory : NostrSocketClientFactory {
fun create(
wssUrl: String,
@@ -39,12 +57,12 @@ object NostrSocketClientFactory {
)
}
fun create(
override fun create(
wssUrl: String,
incomingCompressionEnabled: Boolean = false,
onSocketConnectionOpened: SocketConnectionOpenedCallback? = null,
onSocketConnectionClosed: SocketConnectionClosedCallback? = null,
onSocketConnectionReopened: SocketConnectionReopenedCallback? = null,
incomingCompressionEnabled: Boolean,
onSocketConnectionOpened: SocketConnectionOpenedCallback?,
onSocketConnectionClosed: SocketConnectionClosedCallback?,
onSocketConnectionReopened: SocketConnectionReopenedCallback?,
) = create(
httpClient = defaultSocketsHttpClient,
wssUrl = wssUrl,

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

@@ -32,10 +32,6 @@ import okio.buffer
import okio.use
import kotlin.concurrent.Volatile
import kotlin.coroutines.cancellation.CancellationException
import kotlin.math.min
import kotlin.math.pow
import kotlin.random.Random
import kotlin.time.Duration
import kotlin.time.Duration.Companion.seconds
import kotlin.uuid.ExperimentalUuidApi
import kotlin.uuid.Uuid
@@ -58,17 +54,7 @@ internal class NostrSocketClientImpl(
private val MAX_RECONNECT_DELAY = 60.seconds
/**
* Bounds the exponent so a socket that has been failing for hours cannot overflow
* the doubling into `Infinity`, which `Duration * Double` rejects outright.
*/
private const val MAX_RECONNECT_EXPONENT = 16
/**
* Up to this fraction of the delay is added at random. Every relay in the pool
* drops at once when the network does, and without jitter they would all come
* back in lockstep for as long as the outage lasts.
*/
/** Fraction of the delay added at random. See [reconnectDelay]. */
private const val RECONNECT_JITTER = 0.25
/**
@@ -276,7 +262,12 @@ internal class NostrSocketClientImpl(
reconnectJob = scope.launch {
while (isActive && autoReconnect && !closedByClient) {
val attempt = ++reconnectAttempts
val wait = reconnectDelay(attempt)
val wait = reconnectDelay(
attempt = attempt,
initialDelay = INITIAL_RECONNECT_DELAY,
maxDelay = MAX_RECONNECT_DELAY,
jitterFraction = RECONNECT_JITTER,
)
logger.i { "Reconnecting to $socketUrl in $wait (attempt $attempt)" }
delay(wait)
@@ -291,13 +282,6 @@ internal class NostrSocketClientImpl(
}
}
private fun reconnectDelay(attempt: Int): Duration {
val doublings = 2.0.pow(min(attempt - 1, MAX_RECONNECT_EXPONENT))
val backoff = minOf(INITIAL_RECONNECT_DELAY * doublings, MAX_RECONNECT_DELAY)
return backoff + backoff * RECONNECT_JITTER * Random.nextDouble()
}
override suspend fun close() {
val session = wsMutex.withLock {
closedByClient = true

37
composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/ReconnectBackoff.kt Normal file
View File

@@ -0,0 +1,37 @@
package press.mantra.compose.network.sockets
import kotlin.math.min
import kotlin.math.pow
import kotlin.random.Random
import kotlin.time.Duration
/**
* Bounds the exponent so a socket that has been failing for hours cannot overflow the doubling
* into `Infinity`, which `Duration * Double` rejects outright — turning a reconnect loop into a
* crash loop at roughly the point the network is least likely to come back on its own.
*/
internal const val MAX_RECONNECT_EXPONENT = 16
/**
* How long to wait before reconnect [attempt]: [initialDelay] doubled once per previous
* failure, capped at [maxDelay], plus up to [jitterFraction] of that again at random.
*
* The jitter is not decoration. Every relay in the pool drops at the same moment when the
* network does, so without it they all come back in lockstep for as long as the outage lasts,
* and each round of that is a simultaneous burst of connection attempts.
*
* [jitter] is a parameter rather than an inline `Random.nextDouble()` so the arithmetic can be
* pinned in a test; callers leave it defaulted.
*/
internal fun reconnectDelay(
attempt: Int,
initialDelay: Duration,
maxDelay: Duration,
jitterFraction: Double,
jitter: Double = Random.nextDouble(),
): Duration {
val doublings = 2.0.pow(min(attempt - 1, MAX_RECONNECT_EXPONENT).coerceAtLeast(0))
val backoff = minOf(initialDelay * doublings, maxDelay)
return backoff + backoff * jitterFraction * jitter
}

4
composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt
View File

@@ -14,7 +14,7 @@ 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.network.sockets.DefaultNostrSocketClientFactory
import press.mantra.compose.repository.CachingImportRepository
import press.mantra.compose.repository.ChatRepository
import press.mantra.compose.repository.NostrRepository
@@ -68,7 +68,7 @@ class SynchronizationViewModel(
val relaysSocketManager = RelaysSocketManager(
activeWalletStateFlow = activeWalletStateFlow,
nostrSocketClientFactory = NostrSocketClientFactory,
nostrSocketClientFactory = DefaultNostrSocketClientFactory,
cachingImportRepository = CachingImportRepository.NO_OP_CACHING_IMPORT_REPOSITORY,
relayRepository = relayRepository,
)

175
composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionPlanTest.kt Normal file
View File

@@ -0,0 +1,175 @@
package press.mantra.compose.managers
import com.vitorpamplona.quartz.marmot.mip03GroupMessages.GroupEvent
import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
import press.mantra.compose.database.model.ChatRoom
import press.mantra.compose.database.model.intermdiate.LocalChatRoom
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.GIFT_WRAP_SUB_ID
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.GROUP_SUB_ID_PREFIX
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.giftWrapFilter
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.groupChunks
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.groupFilter
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.groupIdsFrom
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.groupSubId
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.subscriptionIndex
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.subscriptionKey
import press.mantra.compose.network.sockets.NostrIncomingMessage
import press.mantra.compose.network.sockets.endsLiveSubscription
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertNull
import kotlin.test.assertTrue
import kotlin.time.Instant
class LiveSubscriptionPlanTest {
private val me = "a".repeat(64)
// --- filters ----------------------------------------------------------------------
/**
* The one that will look like an oversight to whoever reads it next.
*
* NIP-59 randomizes a gift wrap's `created_at` into the past, and our own outbound path
* stamps them with `TimeUtils.randomWithTwoDays()` — so a wrap published this second can
* carry a timestamp two days old. A `since` anywhere near the present silently drops a
* large share of genuinely new messages, and the symptom is "some DMs just never arrive":
* no error, no log, nothing to grep for.
*/
@Test
fun `the gift wrap filter carries no since`() {
assertNull(giftWrapFilter(me).since)
}
@Test
fun `the gift wrap filter asks for wraps addressed to us`() {
val filter = giftWrapFilter(me)
assertEquals(listOf(GiftWrapEvent.KIND), filter.kinds)
assertEquals(mapOf("p" to listOf(me)), filter.tags)
assertEquals(100, filter.limit)
}
/**
* `limit` is not a cap on the subscription — NIP-01 scopes it to the stored events a relay
* sends before EOSE, explicitly not to the stream after it. It bounds what a reconnect
* costs; it does not bound what arrives live.
*/
@Test
fun `the group filter asks for the given groups, and also carries no since`() {
val filter = groupFilter(listOf("group-a", "group-b"))
assertEquals(listOf(GroupEvent.KIND), filter.kinds)
assertEquals(mapOf("h" to listOf("group-a", "group-b")), filter.tags)
assertEquals(500, filter.limit)
assertNull(filter.since)
}
// --- which groups are watched -----------------------------------------------------
@Test
fun `only MLS rooms are watched`() {
val rooms = listOf(
room(id = "mls", mlsGroupState = "state"),
// A NIP-17 room has no group state; its messages arrive as gift wraps instead, so
// h-tag subscribing to it would ask for events that do not exist.
room(id = "nip17", mlsGroupState = null),
)
assertEquals(listOf("mls"), groupIdsFrom(rooms))
}
/**
* A room we left keeps its history locally and must stop pulling new messages. Getting
* this wrong is not merely wasteful: it means still receiving from a group we are no
* longer a member of.
*/
@Test
fun `rooms we have left or deleted are not watched`() {
val rooms = listOf(
room(id = "here", mlsGroupState = "state"),
room(id = "left", mlsGroupState = "state", leftGroupAt = Instant.fromEpochSeconds(10)),
room(id = "gone", mlsGroupState = "state", deletedAt = Instant.fromEpochSeconds(10)),
)
assertEquals(listOf("here"), groupIdsFrom(rooms))
}
/**
* Sorted, so the same membership in a different row order is the same value and the
* `distinctUntilChanged` upstream of the reconcile does not re-send a filter that has not
* actually changed.
*/
@Test
fun `group ids come out sorted`() {
val rooms = listOf("c", "a", "b").map { room(id = it, mlsGroupState = "state") }
assertEquals(listOf("a", "b", "c"), groupIdsFrom(rooms))
}
// --- chunking and subscription ids -------------------------------------------------
@Test
fun `groups are chunked to bound the size of one filter's tag array`() {
val chunks = groupChunks((1..250).map { "group-$it" })
assertEquals(3, chunks.size)
assertEquals(listOf(100, 100, 50), chunks.map { it.size })
}
@Test
fun `no groups means no subscriptions`() {
assertTrue(groupChunks(emptyList()).isEmpty())
}
@Test
fun `a subscription key round-trips back to its chunk index`() {
val key = subscriptionKey(relayUrl = "wss://relay.example.com", subId = groupSubId(3))
assertEquals("wss://relay.example.com|${GROUP_SUB_ID_PREFIX}3", key)
assertEquals(3, subscriptionIndex(key))
}
/**
* `subscriptionIndex` decides which subscriptions get closed when membership shrinks —
* anything at or past the new chunk count goes. Unparseable therefore has to read as "past
* the end", so a key nobody recognises is closed rather than kept open forever.
*/
@Test
fun `an unrecognisable key sorts past the end so it gets closed`() {
assertEquals(Int.MAX_VALUE, subscriptionIndex("wss://relay.example.com|$GIFT_WRAP_SUB_ID"))
assertEquals(Int.MAX_VALUE, subscriptionIndex("nonsense"))
}
// --- what ends a live subscription -------------------------------------------------
@Test
fun `EOSE does not end a live subscription, CLOSED does`() {
assertFalse(NostrIncomingMessage.EoseMessage(subscriptionId = "s").endsLiveSubscription())
assertFalse(NostrIncomingMessage.EventMessage(subscriptionId = "s").endsLiveSubscription())
assertFalse(NostrIncomingMessage.NoticeMessage(message = "hi").endsLiveSubscription())
assertTrue(
NostrIncomingMessage.ClosedMessage(subscriptionId = "s", message = "rate-limited")
.endsLiveSubscription()
)
}
private fun room(
id: String,
mlsGroupState: String?,
leftGroupAt: Instant? = null,
deletedAt: Instant? = null,
) = LocalChatRoom(
chatRoom = ChatRoom(
id = id,
userPublicKey = me,
subject = null,
description = null,
mlsGroupState = mlsGroupState,
leftGroupAt = leftGroupAt,
deletedAt = deletedAt,
)
)
}

428
composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionReconcileTest.kt Normal file
View File

@@ -0,0 +1,428 @@
package press.mantra.compose.managers
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.advanceTimeBy
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import press.mantra.compose.database.model.ChatRoom
import press.mantra.compose.database.model.NegentropySynchronizeRequest
import press.mantra.compose.database.model.NostrEvent
import press.mantra.compose.database.model.intermdiate.LocalChatRoom
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.GIFT_WRAP_SUB_ID
import press.mantra.compose.managers.LiveSubscriptionManager.Companion.groupSubId
import press.mantra.compose.network.relays.LiveSubscriptionTransport
import press.mantra.compose.network.sockets.NostrIncomingMessage
import press.mantra.compose.repository.ChatRepository
import press.mantra.compose.repository.NostrRepository
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
import kotlin.time.Duration
import kotlin.time.Duration.Companion.minutes
import kotlin.time.Duration.Companion.seconds
import kotlin.time.Instant
/**
* The requirement this whole change exists to meet: the group subscription has to follow group
* membership, without anyone remembering to call a subscribe function when a group is joined.
*
* The failure this guards against is entirely silent. A group whose id never makes it into the
* `#h` filter is not an error anywhere — it is a conversation that simply never delivers, on a
* screen that looks exactly like an empty one.
*/
@OptIn(ExperimentalCoroutinesApi::class)
class LiveSubscriptionReconcileTest {
private val keyPair = KeyPair()
private val me = keyPair.pubKey.toHexString()
@Test
fun `opens a group subscription for the groups we are in`() = runTest {
val transport = FakeTransport()
val rooms = MutableStateFlow(listOf(room("group-a"), room("group-b")))
val running = start(transport, rooms)
settle()
val opened = transport.opened.single { it.reqCommand.subId == groupSubId(0) }
assertEquals(
mapOf("h" to listOf("group-a", "group-b")),
opened.reqCommand.filters.single().tags,
)
running.cancelAndJoin()
}
/**
* The loop that closes: a Welcome arrives on the gift wrap subscription, a ChatRoom row is
* written, the room list re-emits, and the filter widens — with no chat screen involved.
*
* It has to widen *in place*. Closing and re-opening would drop the live tail of every
* group already in that chunk for as long as the round trip takes, so joining one group
* would briefly stop delivery on all the others.
*/
@Test
fun `joining a group widens the filter without reopening the subscription`() = runTest {
val transport = FakeTransport()
val rooms = MutableStateFlow(listOf(room("group-a")))
val running = start(transport, rooms)
settle()
val opensBefore = transport.opened.count { it.reqCommand.subId == groupSubId(0) }
rooms.value = listOf(room("group-a"), room("group-new"))
settle()
val updated = transport.updated.last { it.reqCommand.subId == groupSubId(0) }
assertEquals(
mapOf("h" to listOf("group-a", "group-new")),
updated.reqCommand.filters.single().tags,
)
assertEquals(
opensBefore,
transport.opened.count { it.reqCommand.subId == groupSubId(0) },
"widening must not re-open the subscription",
)
running.cancelAndJoin()
}
@Test
fun `leaving a group drops it from the filter`() = runTest {
val transport = FakeTransport()
val rooms = MutableStateFlow(listOf(room("group-a"), room("group-b")))
val running = start(transport, rooms)
settle()
rooms.value = listOf(room("group-a"), room("group-b", leftGroupAt = Instant.fromEpochSeconds(1)))
settle()
val updated = transport.updated.last { it.reqCommand.subId == groupSubId(0) }
assertEquals(mapOf("h" to listOf("group-a")), updated.reqCommand.filters.single().tags)
running.cancelAndJoin()
}
@Test
fun `leaving every group closes the subscription`() = runTest {
val transport = FakeTransport()
val rooms = MutableStateFlow(listOf(room("group-a")))
val running = start(transport, rooms)
settle()
rooms.value = emptyList()
settle()
assertTrue(transport.closed.any { it.subId == groupSubId(0) })
running.cancelAndJoin()
}
/**
* Membership churn inside the debounce window must collapse. Joining a group writes the
* room, its participants and placeholder profiles in quick succession, each of which
* re-emits the list — so without this a single join re-sends the filter several times.
*/
@Test
fun `rapid membership changes collapse into one update`() = runTest {
val transport = FakeTransport()
val rooms = MutableStateFlow(listOf(room("group-a")))
val running = start(transport, rooms)
settle()
val updatesBefore = transport.updated.size
rooms.value = listOf(room("group-a"), room("group-b"))
advanceTimeBy(100)
rooms.value = listOf(room("group-a"), room("group-b"), room("group-c"))
advanceTimeBy(100)
rooms.value = listOf(room("group-a"), room("group-b"), room("group-c"), room("group-d"))
settle()
assertEquals(1, transport.updated.size - updatesBefore, "expected one update, not three")
running.cancelAndJoin()
}
@Test
fun `a NIP-17 room never becomes a group subscription`() = runTest {
val transport = FakeTransport()
val rooms = MutableStateFlow(listOf(room("dm-room", mlsGroupState = null)))
val running = start(transport, rooms)
settle()
assertTrue(transport.opened.none { it.reqCommand.subId.startsWith("live-groups-") })
running.cancelAndJoin()
}
// --- what arrives on a subscription -------------------------------------------------
@Test
fun `events are stored against the relay they arrived from`() = runTest {
val transport = FakeTransport()
val nostr = RecordingNostrRepository()
val running = start(transport, MutableStateFlow(emptyList()), nostr)
settle()
val subscription = transport.opened.single { it.reqCommand.subId == GIFT_WRAP_SUB_ID }
subscription.messages.emit(
NostrIncomingMessage.EventMessage(subscriptionId = GIFT_WRAP_SUB_ID, nostrEvent = event("ev-1"))
)
settle()
val saved = nostr.saved.single()
assertEquals("ev-1", saved.first)
assertEquals(subscription.relayUrl, saved.second)
running.cancelAndJoin()
}
/** EOSE is a marker, not an end: what follows it is the whole point. */
@Test
fun `an event after EOSE is still stored`() = runTest {
val transport = FakeTransport()
val nostr = RecordingNostrRepository()
val running = start(transport, MutableStateFlow(emptyList()), nostr)
settle()
val subscription = transport.opened.single { it.reqCommand.subId == GIFT_WRAP_SUB_ID }
subscription.messages.emit(NostrIncomingMessage.EoseMessage(subscriptionId = GIFT_WRAP_SUB_ID))
subscription.messages.emit(
NostrIncomingMessage.EventMessage(subscriptionId = GIFT_WRAP_SUB_ID, nostrEvent = event("after-eose"))
)
settle()
assertEquals(listOf("after-eose"), nostr.saved.map { it.first })
running.cancelAndJoin()
}
@Test
fun `a relay that closes the subscription gets it re-opened`() = runTest {
val transport = FakeTransport()
val running = start(transport, MutableStateFlow(emptyList()))
settle()
val first = transport.opened.single { it.reqCommand.subId == GIFT_WRAP_SUB_ID }
first.messages.emit(
NostrIncomingMessage.ClosedMessage(subscriptionId = GIFT_WRAP_SUB_ID, message = "shutting down")
)
settle(1.seconds)
assertEquals(
1,
transport.opened.count { it.reqCommand.subId == GIFT_WRAP_SUB_ID },
"should still be inside the back-off, not hammering the relay",
)
settle(30.seconds)
assertEquals(
2,
transport.opened.count { it.reqCommand.subId == GIFT_WRAP_SUB_ID },
"a CLOSED subscription should have been re-opened once the back-off elapsed",
)
running.cancelAndJoin()
}
/**
* Back-pressure is the one refusal that must NOT be answered promptly — opening another
* subscription is exactly what the relay just asked us to stop doing.
*/
@Test
fun `a rate-limited close waits far longer before re-opening`() = runTest {
val transport = FakeTransport()
val running = start(transport, MutableStateFlow(emptyList()))
settle()
transport.opened.single { it.reqCommand.subId == GIFT_WRAP_SUB_ID }.messages.emit(
NostrIncomingMessage.ClosedMessage(
subscriptionId = GIFT_WRAP_SUB_ID,
message = "rate-limited: too many concurrent REQs",
)
)
settle(30.seconds)
assertEquals(
1,
transport.opened.count { it.reqCommand.subId == GIFT_WRAP_SUB_ID },
"must not re-open within the window a plain failure would have",
)
// ...but it is a back-off, not a give-up.
settle(5.minutes)
assertEquals(
2,
transport.opened.count { it.reqCommand.subId == GIFT_WRAP_SUB_ID },
"the subscription should come back once the relay has had its breathing room",
)
running.cancelAndJoin()
}
// --- lifecycle -----------------------------------------------------------------------
@Test
fun `backgrounding closes the subscriptions and foregrounding rebuilds them`() = runTest {
val transport = FakeTransport()
val foreground = MutableStateFlow(true)
val running = start(transport, MutableStateFlow(emptyList()), isForeground = foreground)
settle()
val openedWhileForeground = transport.opened.size
assertTrue(openedWhileForeground > 0)
foreground.value = false
settle()
assertTrue(transport.closed.any { it.subId == GIFT_WRAP_SUB_ID }, "should have said goodbye")
foreground.value = true
settle()
assertTrue(transport.opened.size > openedWhileForeground, "should have re-subscribed")
running.cancelAndJoin()
}
/**
* A live subscription covers the window we are connected for; it cannot answer for the gap
* we were away. That is negentropy's job, and returning to the foreground is precisely
* when it needs asking.
*/
@Test
fun `foregrounding reconnects and queues a catch-up reconciliation`() = runTest {
val transport = FakeTransport()
val nostr = RecordingNostrRepository()
val running = start(transport, MutableStateFlow(listOf(room("group-a"))), nostr)
settle()
assertEquals(1, transport.reconnects, "the socket must not be trusted after a gap")
val purposes = nostr.queued.map { it.purpose }.toSet()
assertTrue("chat" in purposes, "gift wraps should be reconciled")
assertTrue("mlsMessages" in purposes, "group events should be reconciled")
running.cancelAndJoin()
}
// --- harness -------------------------------------------------------------------------
/**
* Advances virtual time by exactly [duration] and runs what that makes due.
*
* Deliberately not `advanceUntilIdle()`: that runs until nothing is scheduled at all,
* which means it fast-forwards through *any* pending delay — including the five-minute
* back-off this suite needs to assert has NOT elapsed. A test that cannot tell "waited"
* from "did not wait" cannot test a back-off at all.
*/
private fun kotlinx.coroutines.test.TestScope.settle(duration: Duration = 1.seconds) {
advanceTimeBy(duration)
runCurrent()
}
private fun kotlinx.coroutines.test.TestScope.start(
transport: FakeTransport,
rooms: MutableStateFlow<List<LocalChatRoom>>,
nostr: RecordingNostrRepository = RecordingNostrRepository(),
isForeground: MutableStateFlow<Boolean> = MutableStateFlow(true),
) = launch {
LiveSubscriptionManager(
relaysSocketManager = transport,
nostrRepository = nostr,
chatRepository = FakeChatRepository(rooms),
isForeground = isForeground,
).observe(keyPair)
}
private fun room(
id: String,
mlsGroupState: String? = "state",
leftGroupAt: Instant? = null,
) = LocalChatRoom(
chatRoom = ChatRoom(
id = id,
userPublicKey = me,
subject = null,
description = null,
mlsGroupState = mlsGroupState,
leftGroupAt = leftGroupAt,
)
)
private fun event(id: String) = NostrEvent(
id = id,
pubKey = me,
createdAt = Instant.fromEpochSeconds(1_000),
kind = 1059,
tags = emptyArray(),
content = "",
sig = "",
)
}
private class OpenedSubscription(val reqCommand: ReqCmd, val relayUrl: String) {
val messages = MutableSharedFlow<NostrIncomingMessage>(extraBufferCapacity = 32)
}
private class FakeTransport : LiveSubscriptionTransport {
val opened = mutableListOf<OpenedSubscription>()
val updated = mutableListOf<OpenedSubscription>()
val closed = mutableListOf<ClosedSubscription>()
var reconnects = 0
data class ClosedSubscription(val subId: String, val relayUrl: String)
override suspend fun openLiveSubscription(reqCommand: ReqCmd, relayUrl: String): Flow<NostrIncomingMessage> =
OpenedSubscription(reqCommand, relayUrl).also { opened += it }.messages.asSharedFlow()
override suspend fun updateLiveSubscription(reqCommand: ReqCmd, relayUrl: String) {
updated += OpenedSubscription(reqCommand, relayUrl)
}
override suspend fun closeLiveSubscription(subId: String, relayUrl: String) {
closed += ClosedSubscription(subId, relayUrl)
}
override suspend fun reconnectAll() {
reconnects++
}
}
private class FakeChatRepository(
private val rooms: StateFlow<List<LocalChatRoom>>,
) : ChatRepository by ChatRepository.NO_OP_CHAT_REPOSITORY {
override suspend fun observeChatRoomListByPublicKey(publicKey: String): Flow<List<LocalChatRoom>> = rooms
override suspend fun getChatRoomListByPublicKey(publicKey: String): List<LocalChatRoom> = rooms.value
}
private class RecordingNostrRepository : NostrRepository by NostrRepository.NO_OP_NOSTR_REPOSITORY {
/** (event id, relay it arrived from) */
val saved = mutableListOf<Pair<String, String>>()
val queued = mutableListOf<NegentropySynchronizeRequest>()
override suspend fun saveNostrEvent(
nostrEvent: NostrEvent,
relayURL: String,
synchronizationRelayURLs: List<String>,
level: Int,
activeKeyPair: KeyPair,
) {
saved += nostrEvent.id to relayURL
}
override suspend fun queueNegentropySynchronizeRequest(
negentropySynchronizeRequests: List<NegentropySynchronizeRequest>,
) {
queued += negentropySynchronizeRequests
}
}

58
composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayBackPressureTest.kt Normal file
View File

@@ -0,0 +1,58 @@
package press.mantra.compose.network.relays
import kotlin.test.Test
import kotlin.test.assertFalse
import kotlin.test.assertTrue
/**
* Pins which CLOSED reasons mean "ease off".
*
* The classification decides what happens next and is expensive to get wrong in both
* directions: a refusal read as a transient failure is answered by opening another
* subscription, which is how one "too many concurrent REQs" becomes a flood of them, while an
* unsupported filter read as back-pressure leaves a subscription shut for five minutes over a
* problem no amount of waiting fixes.
*/
class RelayBackPressureTest {
@Test
fun `recognises the NIP-01 machine-readable prefix`() {
assertTrue(isRelayBackPressure("rate-limited: slow down there chief"))
}
@Test
fun `recognises the free-text forms relays actually send`() {
listOf(
"ERROR: too many concurrent REQs",
"rate limit exceeded",
"Please slow down",
"TOO MANY SUBSCRIPTIONS",
"maximum concurrent subscriptions reached",
).forEach { reason ->
assertTrue(isRelayBackPressure(reason), "expected back-pressure: $reason")
}
}
@Test
fun `does not read a refusal we cannot wait out as back-pressure`() {
listOf(
"auth-required: we can't serve DMs to unauthenticated users",
"unsupported: filter contains unknown tag",
"invalid: filter is empty",
"error: negentropy disabled",
"blocked: you are not allowed to write here",
).forEach { reason ->
assertFalse(isRelayBackPressure(reason), "did not expect back-pressure: $reason")
}
}
/**
* A CLOSED with no reason at all is common. Treating it as back-pressure would mean the
* least informative refusal produced the longest possible outage.
*/
@Test
fun `a missing or empty reason is not back-pressure`() {
assertFalse(isRelayBackPressure(null))
assertFalse(isRelayBackPressure(""))
}
}

392
composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayPoolSubscriptionTest.kt Normal file
View File

@@ -0,0 +1,392 @@
package press.mantra.compose.network.relays
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.runTest
import kotlinx.serialization.json.JsonObject
import press.mantra.compose.network.dto.RelayDTO
import press.mantra.compose.network.sockets.NostrIncomingMessage
import press.mantra.compose.network.sockets.NostrSocketClient
import press.mantra.compose.network.sockets.NostrSocketClientFactory
import press.mantra.compose.network.sockets.SocketConnectionClosedCallback
import press.mantra.compose.network.sockets.SocketConnectionOpenedCallback
import press.mantra.compose.network.sockets.SocketConnectionReopenedCallback
import press.mantra.compose.repository.CachingImportRepository
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertTrue
/**
* The pool's half of surviving a dropped socket.
*
* A subscription that outlives its socket only works if the pool remembers what that socket was
* carrying and hands it back on reconnect. None of that is observable from the outside — no
* return value changes, nothing throws — so a regression here looks like "messages stopped
* arriving after a tunnel", hours later, on someone else's phone.
*
* The other half is that a relay answers a repeated REQ on an existing subscription id by
* replacing that subscription's filter, which is what makes replay a send rather than a
* close-and-reopen.
*/
@OptIn(ExperimentalCoroutinesApi::class)
class RelayPoolSubscriptionTest {
private val relayUrl = "wss://relay.example.com"
private val otherRelayUrl = "wss://other.example.com"
private fun req(subId: String, kind: Int = 1) =
ReqCmd(subId = subId, filters = listOf(Filter(kinds = listOf(kind))))
@Test
fun `a query is retained, and replayed when its socket comes back`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.query(req("sub-1"), relayUrl)
val socket = factory.only()
assertEquals(1, socket.sent.size, "the REQ should have gone out once")
assertTrue(socket.autoReconnect, "a relay carrying a subscription is worth reconnecting")
socket.sent.clear()
factory.reopen(relayUrl)
assertEquals(1, socket.sent.size, "the REQ should have been replayed")
assertTrue(socket.sent.single().contains("sub-1"))
}
@Test
fun `a closed query is forgotten, and the socket stops reconnecting for it`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.query(req("sub-1"), relayUrl)
pool.closeQuery(CloseCmd(subId = "sub-1"), relayUrl)
val socket = factory.only()
assertFalse(socket.autoReconnect, "nothing is open on this relay any more")
socket.sent.clear()
factory.reopen(relayUrl)
assertTrue(socket.sent.isEmpty(), "a closed subscription must not come back on reconnect")
}
/**
* Closing one of two must not drop the other — the bug this guards is a shared socket
* quietly losing its remaining subscription because a sibling finished first.
*/
@Test
fun `closing one subscription leaves the rest of that relay's alone`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.query(req("sub-1"), relayUrl)
pool.query(req("sub-2", kind = 7), relayUrl)
pool.closeQuery(CloseCmd(subId = "sub-1"), relayUrl)
val socket = factory.only()
assertTrue(socket.autoReconnect, "sub-2 is still open")
socket.sent.clear()
factory.reopen(relayUrl)
assertEquals(1, socket.sent.size)
assertTrue(socket.sent.single().contains("sub-2"))
}
/**
* Negentropy is stateful: NEG-OPEN carries a fingerprint of the local set and every round
* depends on the last. Replaying one mid-exchange would reconcile against a conversation
* the relay is no longer having, so an interrupted exchange is abandoned and re-queued
* instead.
*/
@Test
fun `a negentropy exchange is never replayed`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.negentropySync(
NegOpenCmd(subId = "neg-1", filter = Filter(kinds = listOf(1)), initialMessage = "6100"),
relayUrl,
)
val socket = factory.only()
assertEquals(1, socket.sent.size, "NEG-OPEN still goes out")
socket.sent.clear()
factory.reopen(relayUrl)
assertTrue(socket.sent.isEmpty(), "a half-finished reconciliation must not be resumed")
}
@Test
fun `a live subscription is retained like any other`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.openLiveSubscription(req("live-giftwrap"), relayUrl)
val socket = factory.only()
assertTrue(socket.autoReconnect)
socket.sent.clear()
factory.reopen(relayUrl)
assertTrue(socket.sent.single().contains("live-giftwrap"))
}
/**
* Widening a live subscription has to replace what gets replayed, not just what is on the
* wire now. Retaining the old filter would mean a reconnect quietly restored a
* subscription the caller had already moved on from — a group you just joined going
* silent the first time you walked through a tunnel.
*/
@Test
fun `updating a live subscription replaces what a reconnect replays`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.openLiveSubscription(req("live-groups-0", kind = 445), relayUrl)
pool.updateLiveSubscription(
ReqCmd(
subId = "live-groups-0",
filters = listOf(Filter(kinds = listOf(445), tags = mapOf("h" to listOf("group-b")))),
),
relayUrl,
)
val socket = factory.only()
socket.sent.clear()
factory.reopen(relayUrl)
val replayed = socket.sent.single()
assertTrue(replayed.contains("group-b"), "expected the current filter, got: $replayed")
}
/**
* An update that cannot be sent must still be recorded. Throwing instead would leave the
* previous filter retained, which is the one outcome worse than not sending at all.
*/
@Test
fun `an update whose send fails is still retained for the reconnect`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.openLiveSubscription(req("live-groups-0", kind = 445), relayUrl)
val socket = factory.only()
socket.failSends = true
pool.updateLiveSubscription(
ReqCmd(
subId = "live-groups-0",
filters = listOf(Filter(kinds = listOf(445), tags = mapOf("h" to listOf("group-b")))),
),
relayUrl,
)
socket.failSends = false
socket.sent.clear()
factory.reopen(relayUrl)
assertTrue(socket.sent.single().contains("group-b"))
}
@Test
fun `dropping a relay forgets what it was carrying`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.query(req("sub-1"), relayUrl)
pool.changeRelays(listOf(RelayDTO(url = otherRelayUrl, read = true, write = true)))
val dropped = factory.forUrl(relayUrl)
dropped.sent.clear()
factory.reopen(relayUrl)
assertTrue(dropped.sent.isEmpty(), "a relay we removed should not be re-subscribed")
}
@Test
fun `closing the pool forgets everything`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.query(req("sub-1"), relayUrl)
pool.closePool()
val socket = factory.only()
socket.sent.clear()
factory.reopen(relayUrl)
assertTrue(socket.sent.isEmpty())
}
/** Retention is per relay: one relay's reconnect must not re-send another's REQ. */
@Test
fun `replay is scoped to the relay that reconnected`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
pool.query(req("sub-here"), relayUrl)
pool.query(req("sub-there"), otherRelayUrl)
factory.forUrl(relayUrl).sent.clear()
factory.forUrl(otherRelayUrl).sent.clear()
factory.reopen(relayUrl)
assertTrue(factory.forUrl(relayUrl).sent.single().contains("sub-here"))
assertTrue(factory.forUrl(otherRelayUrl).sent.isEmpty())
}
/**
* The semantic the whole change rests on. For a one-shot request EOSE is the end; for a
* live one it is only the boundary between the history a relay had stored and the tail it
* will now stream. Ending there is precisely what made every chat sync a poll.
*/
@Test
fun `a live subscription keeps delivering after EOSE`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
val received = mutableListOf<NostrIncomingMessage>()
val flow = pool.openLiveSubscription(req("live-giftwrap"), relayUrl)
val collecting = launch(UnconfinedTestDispatcher(testScheduler)) {
flow.collect { received += it }
}
val socket = factory.only()
socket.deliver(NostrIncomingMessage.EoseMessage(subscriptionId = "live-giftwrap"))
socket.deliver(NostrIncomingMessage.EventMessage(subscriptionId = "live-giftwrap"))
assertEquals(2, received.size, "the event after EOSE should still have arrived")
assertTrue(collecting.isActive, "a live subscription ends when its collector stops, not at EOSE")
collecting.cancel()
}
/** The contrast, so the two rules cannot silently converge. */
@Test
fun `a one-shot query still ends at EOSE`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
val received = mutableListOf<NostrIncomingMessage>()
val flow = pool.query(req("sub-1"), relayUrl)
val collecting = launch(UnconfinedTestDispatcher(testScheduler)) {
flow.collect { received += it }
}
val socket = factory.only()
socket.deliver(NostrIncomingMessage.EoseMessage(subscriptionId = "sub-1"))
socket.deliver(NostrIncomingMessage.EventMessage(subscriptionId = "sub-1"))
assertEquals(1, received.size, "nothing should arrive after the EOSE that ended it")
assertFalse(collecting.isActive, "the collector should have completed")
}
/**
* A NOTICE carries no subscription id, so the socket hands it to every collector. It is
* admitted deliberately — it is the only signal some relays give for "negentropy disabled"
* — but it must stay advisory, never terminal, or one relay's complaint would tear down
* every unrelated subscription on that socket.
*/
@Test
fun `a NOTICE reaches a live subscription without ending it`() = runTest {
val factory = FakeSocketClientFactory()
val pool = pool(factory)
val received = mutableListOf<NostrIncomingMessage>()
val flow = pool.openLiveSubscription(req("live-giftwrap"), relayUrl)
val collecting = launch(UnconfinedTestDispatcher(testScheduler)) {
flow.collect { received += it }
}
factory.only().deliver(NostrIncomingMessage.NoticeMessage(message = "restricted: slow"))
assertEquals(1, received.size)
assertTrue(collecting.isActive)
collecting.cancel()
}
private fun kotlinx.coroutines.test.TestScope.pool(factory: FakeSocketClientFactory) =
RelayPool(
nostrSocketClientFactory = factory,
cachingImportRepository = CachingImportRepository.NO_OP_CACHING_IMPORT_REPOSITORY,
// Unconfined so the pool's launched work — status updates, and the replay a
// reconnect triggers — has run by the time the call that started it returns.
scope = CoroutineScope(UnconfinedTestDispatcher(testScheduler)),
)
}
private class FakeSocketClientFactory : NostrSocketClientFactory {
private val clients = mutableMapOf<String, FakeNostrSocketClient>()
private val reopenCallbacks = mutableMapOf<String, SocketConnectionReopenedCallback?>()
override fun create(
wssUrl: String,
incomingCompressionEnabled: Boolean,
onSocketConnectionOpened: SocketConnectionOpenedCallback?,
onSocketConnectionClosed: SocketConnectionClosedCallback?,
onSocketConnectionReopened: SocketConnectionReopenedCallback?,
): NostrSocketClient {
reopenCallbacks[wssUrl] = onSocketConnectionReopened
return FakeNostrSocketClient(wssUrl).also { clients[wssUrl] = it }
}
fun only(): FakeNostrSocketClient = clients.values.single()
fun forUrl(url: String): FakeNostrSocketClient = clients.getValue(url)
/** Stands in for a socket that dropped and re-established its session. */
fun reopen(url: String) {
reopenCallbacks.getValue(url)?.invoke(url)
}
}
private class FakeNostrSocketClient(override val socketUrl: String) : NostrSocketClient {
val sent = mutableListOf<String>()
var failSends = false
private val _incomingMessages = MutableSharedFlow<NostrIncomingMessage>(extraBufferCapacity = 64)
override val incomingMessages: SharedFlow<NostrIncomingMessage> = _incomingMessages.asSharedFlow()
override var autoReconnect: Boolean = false
override suspend fun close() = Unit
override suspend fun ensureSocketConnectionOrThrow() = Unit
override suspend fun sendMESSAGE(text: String, ensureSessionBeforeSend: Boolean) {
if (failSends) throw IllegalStateException("socket is down")
sent += text
}
override suspend fun sendAUTH(signedEvent: JsonObject) = Unit
override suspend fun sendCLOSE(subscriptionId: String) = Unit
override suspend fun sendCOUNT(data: JsonObject): String = "unused"
override suspend fun sendEVENT(signedEvent: JsonObject) = Unit
override suspend fun sendREQ(subscriptionId: String, data: JsonObject) = Unit
/** Stands in for a message arriving on the wire. */
suspend fun deliver(message: NostrIncomingMessage) {
_incomingMessages.emit(message)
}
}

73
composeApp/src/commonTest/kotlin/press/mantra/compose/network/sockets/ReconnectBackoffTest.kt Normal file
View File

@@ -0,0 +1,73 @@
package press.mantra.compose.network.sockets
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
import kotlin.time.Duration.Companion.seconds
class ReconnectBackoffTest {
private val initial = 1.seconds
private val max = 60.seconds
private fun delay(attempt: Int, jitter: Double = 0.0) =
reconnectDelay(
attempt = attempt,
initialDelay = initial,
maxDelay = max,
jitterFraction = 0.25,
jitter = jitter,
)
@Test
fun `doubles once per previous failure`() {
assertEquals(1.seconds, delay(attempt = 1))
assertEquals(2.seconds, delay(attempt = 2))
assertEquals(4.seconds, delay(attempt = 3))
assertEquals(8.seconds, delay(attempt = 4))
}
@Test
fun `stops growing at the cap`() {
assertEquals(max, delay(attempt = 7))
assertEquals(max, delay(attempt = 50))
}
/**
* The reason [MAX_RECONNECT_EXPONENT] exists. `2.0.pow(4000)` is `Infinity`, and
* `Duration * Double` throws on it — so without the clamp a socket that had been failing
* for long enough turned its own reconnect loop into a crash loop, at roughly the moment
* the network was least likely to recover unaided.
*/
@Test
fun `survives an attempt count large enough to overflow the doubling`() {
assertEquals(max, delay(attempt = 4_000))
assertEquals(max, delay(attempt = Int.MAX_VALUE))
}
/** Defensive: a caller that starts counting at zero should still get a usable delay. */
@Test
fun `treats a non-positive attempt as the first one`() {
assertEquals(initial, delay(attempt = 0))
assertEquals(initial, delay(attempt = -3))
}
@Test
fun `adds at most the jitter fraction on top, never subtracts`() {
assertEquals(4.seconds, delay(attempt = 3, jitter = 0.0))
assertEquals(5.seconds, delay(attempt = 3, jitter = 1.0))
val midway = delay(attempt = 3, jitter = 0.5)
assertTrue(midway > 4.seconds && midway < 5.seconds, "expected 4s..5s, got $midway")
}
/**
* Jitter is applied after the cap, so a capped delay still spreads: relays all drop at the
* same moment when the network does, and identical waits would bring them back in lockstep
* for as long as the outage lasted.
*/
@Test
fun `jitter still spreads a capped delay`() {
assertTrue(delay(attempt = 50, jitter = 1.0) > max)
}
}