diff --git a/composeApp/build.gradle.kts b/composeApp/build.gradle.kts index 7d1eb406..69a2d9c8 100644 --- a/composeApp/build.gradle.kts +++ b/composeApp/build.gradle.kts @@ -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) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt index 04145fc1..e3492a7f 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt @@ -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, @@ -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) = 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): List = + 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): List> = + 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 = - 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) = 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() 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.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 -> { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/LiveSubscriptionTransport.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/LiveSubscriptionTransport.kt new file mode 100644 index 00000000..a27defe2 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/LiveSubscriptionTransport.kt @@ -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 + + /** @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() +} 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 d4a1be20..a260a1d7 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 @@ -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 = mutableSetOf() private val relayMutex = Mutex() 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 8b18428f..d901402a 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 @@ -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 { + override 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) { + 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 diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt index 153a6d4e..d876c11b 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt @@ -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. diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt index d751989f..99262b0d 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt @@ -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, 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 23740028..2a6fb364 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 @@ -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 diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/ReconnectBackoff.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/ReconnectBackoff.kt new file mode 100644 index 00000000..7adc6ceb --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/ReconnectBackoff.kt @@ -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 +} diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt index fab0fb71..c4fe428c 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt @@ -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, ) diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionPlanTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionPlanTest.kt new file mode 100644 index 00000000..8761f337 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionPlanTest.kt @@ -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, + ) + ) +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionReconcileTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionReconcileTest.kt new file mode 100644 index 00000000..0eb46c50 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/LiveSubscriptionReconcileTest.kt @@ -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>, + nostr: RecordingNostrRepository = RecordingNostrRepository(), + isForeground: MutableStateFlow = 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(extraBufferCapacity = 32) +} + +private class FakeTransport : LiveSubscriptionTransport { + val opened = mutableListOf() + val updated = mutableListOf() + val closed = mutableListOf() + var reconnects = 0 + + data class ClosedSubscription(val subId: String, val relayUrl: String) + + override suspend fun openLiveSubscription(reqCommand: ReqCmd, relayUrl: String): Flow = + 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>, +) : ChatRepository by ChatRepository.NO_OP_CHAT_REPOSITORY { + + override suspend fun observeChatRoomListByPublicKey(publicKey: String): Flow> = rooms + + override suspend fun getChatRoomListByPublicKey(publicKey: String): List = rooms.value +} + +private class RecordingNostrRepository : NostrRepository by NostrRepository.NO_OP_NOSTR_REPOSITORY { + /** (event id, relay it arrived from) */ + val saved = mutableListOf>() + val queued = mutableListOf() + + override suspend fun saveNostrEvent( + nostrEvent: NostrEvent, + relayURL: String, + synchronizationRelayURLs: List, + level: Int, + activeKeyPair: KeyPair, + ) { + saved += nostrEvent.id to relayURL + } + + override suspend fun queueNegentropySynchronizeRequest( + negentropySynchronizeRequests: List, + ) { + queued += negentropySynchronizeRequests + } +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayBackPressureTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayBackPressureTest.kt new file mode 100644 index 00000000..25a13e0d --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayBackPressureTest.kt @@ -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("")) + } +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayPoolSubscriptionTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayPoolSubscriptionTest.kt new file mode 100644 index 00000000..decc69cd --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/network/relays/RelayPoolSubscriptionTest.kt @@ -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() + 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() + 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() + 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() + private val reopenCallbacks = mutableMapOf() + + 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() + var failSends = false + + private val _incomingMessages = MutableSharedFlow(extraBufferCapacity = 64) + override val incomingMessages: SharedFlow = _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) + } +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/network/sockets/ReconnectBackoffTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/network/sockets/ReconnectBackoffTest.kt new file mode 100644 index 00000000..0c4cb389 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/network/sockets/ReconnectBackoffTest.kt @@ -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) + } +} diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index e51ecfde..9d846a9f 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -63,6 +63,7 @@ kermit = { module = "co.touchlab:kermit", version.ref = "kermit" } kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotlin" } kotlin-testJunit = { module = "org.jetbrains.kotlin:kotlin-test-junit", version.ref = "kotlin" } kotlinx-coroutinesSwing = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-swing", version.ref = "kotlinx-coroutines" } +kotlinx-coroutinesTest = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "kotlinx-coroutines" } kotlinx-datetime = { module = "org.jetbrains.kotlinx:kotlinx-datetime", version.ref = "kotlinx-datetime" } kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinxSerialization" } kotlinx-serialization-cbor = { module = "org.jetbrains.kotlinx:kotlinx-serialization-cbor", version.ref = "kotlinxSerialization" }