Merge pull request #2638 from vitorpamplona/claude/fix-listener-test-timeout-HqseI

test(nests): fix flaky ReconnectingNestsListenerTest on macOS

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt

@@ -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
+                withTimeout(5_000L) { consumerSubscribed.await() }
-                // session before we emit / fail. With Dispatchers.Default
+
-                // the pumpJob's collectLatest doesn't run synchronously
+                // Wait for the pump to actually subscribe to first.frames
-                // off the launch site — yielding here lets it advance.
+                // 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
+                // The pump re-issuing the subscription against the new
-                // re-issued the subscription against the new session —
+                // session is observable via second.frames.subscriptionCount
-                // is observable directly via second.subscribeCount.
+                // (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 —
+                // Wait until the pump opened the underlying subscription
-                // collectLatest schedules the inner block on a child of
+                // AND committed liveHandleRef. subscribeCount alone
-                // scope, so it runs after we yield.
+                // 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")