nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

test(nests): fix flaky ReconnectingNestsListenerTest on macOS

Browse Source 
`subscribeSpeaker_survives_session_swap` and
`unsubscribe_before_session_swap_releases_handle` had two race
conditions exposed only on macOS scheduling:

1. ScriptedListener.subscribeCount is incremented inside
   opener(listener) — BEFORE the wrapper's pump reaches
   `handle.objects.collect { frames.emit(it) }`. Waiting on
   subscribeCount alone doesn't prove the pump is collecting
   from first.frames. An emit upstream before the pump's collect
   registers is silently dropped.

2. The wrapper's outer `frames` SharedFlow is replay=0. The
   `scope.async { take(2).toList() }` consumer might not have
   subscribed by the time the test thread races into the first
   emit, dropping the value.

Both are fixed by waiting for actual subscription registration:
- first/second.frames.subscriptionCount.first { it > 0 } — flips
  to 1 only after the pump's collect lands, which is strictly
  after liveHandleRef.set(handle).
- onSubscription + CompletableDeferred — fires after the
  consumer's collector is registered against the SharedFlow.
This commit is contained in:
Claude
2026-04-29 03:16:23 +00:00
parent 6c8c5eb7a5
commit 099da6e7b0
1 changed files with 47 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt
+47 -14
View File
@@ -28,15 +28,18 @@ import com.vitorpamplona.nestsclient.transport.WebTransportSession
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.onSubscription
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.launch
@@ -175,19 +178,41 @@ class ReconnectingNestsListenerTest {
val handle = reconnecting.subscribeSpeaker("speaker-pubkey")
// Signal when the consumer has actually subscribed to
// the outer frames flow. The wrapper's frames SharedFlow
// is replay=0, so an emit before the consumer registers
// its subscription is silently dropped — observed on
// macOS where the async coroutine isn't scheduled
// before the test thread races into the first emit.
val consumerSubscribed = CompletableDeferred<Unit>()
// The wrapper backs handle.objects with a SharedFlow
// (frames.asSharedFlow); the cast lets us use
// SharedFlow.onSubscription, which fires AFTER the
// collector is registered (Flow.onStart fires before).
@Suppress("UNCHECKED_CAST")
val objectsAsShared = handle.objects as SharedFlow<MoqObject>
val collected =
scope.async {
withTimeout(5_000L) {
handle.objects.take(2).toList()
objectsAsShared
.onSubscription { consumerSubscribed.complete(Unit) }
.take(2)
.toList()
}
}
// Wait for the pump to actually subscribe to the first
// session before we emit / fail. With Dispatchers.Default
// the pumpJob's collectLatest doesn't run synchronously
// off the launch site — yielding here lets it advance.
withTimeout(5_000L) { consumerSubscribed.await() }
// Wait for the pump to actually subscribe to first.frames
// before we emit upstream. first.subscribeCount is
// incremented inside opener(first) — BEFORE the pump
// reaches `handle.objects.collect` — so subscribeCount
// alone doesn't prove the pump is listening.
// subscriptionCount on first.frames flips to 1 only once
// the pump's collect actually registers.
withTimeout(5_000L) {
while (first.subscribeCount == 0) kotlinx.coroutines.delay(5)
first.frames.subscriptionCount.first { it > 0 }
}
first.frames.emit(frame(byteArrayOf(0x01)))
@@ -198,11 +223,12 @@ class ReconnectingNestsListenerTest {
// Both ScriptedListeners report the same `room` and
// `negotiatedMoqVersion`, so we can't differentiate
// first.Connected from second.Connected via state alone.
// The actual postcondition we care about — the pump
// re-issued the subscription against the new session —
// is observable directly via second.subscribeCount.
// The pump re-issuing the subscription against the new
// session is observable via second.frames.subscriptionCount
// (same rationale as first.frames above — wait for the
// collect, not just the subscribeSpeaker call).
withTimeout(5_000L) {
while (second.subscribeCount == 0) kotlinx.coroutines.delay(5)
second.frames.subscriptionCount.first { it > 0 }
}
second.frames.emit(frame(byteArrayOf(0x02)))
@@ -251,11 +277,18 @@ class ReconnectingNestsListenerTest {
val handle = reconnecting.subscribeSpeaker("speaker-pubkey")
// Wait until the pump opened the underlying subscription —
// collectLatest schedules the inner block on a child of
// scope, so it runs after we yield.
// Wait until the pump opened the underlying subscription
// AND committed liveHandleRef. subscribeCount alone
// increments inside opener(first), which runs BEFORE
// the pump sets liveHandleRef + starts collecting from
// first.frames. If we unsubscribe while liveHandleRef
// is still null the inner unsubscribe never fires and
// the assertion below hangs/fails on macOS.
// first.frames.subscriptionCount flips to 1 only once
// the pump's collect actually registers, which is
// strictly after liveHandleRef.set(handle).
withTimeout(5_000L) {
while (first.subscribeCount == 0) kotlinx.coroutines.yield()
first.frames.subscriptionCount.first { it > 0 }
}
assertTrue(first.subscribeCount >= 1, "pump should have opened the underlying sub")