From 9f542a6ff40de244a8ef2e775d0249cc2bc7981b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 4 May 2026 18:28:42 +0000
Subject: [PATCH] fix(nests): retry SUBSCRIBE on relay-FIN instead of
 permanently giving up
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When a listener joins a Nest before the speaker starts publishing,
moq-rs FINs the SUBSCRIBE bidi immediately without sending
SubscribeOk or SubscribeDrop. MoqLiteSession.subscribe surfaces that
as a MoqLiteSubscribeException("subscribe stream FIN before reply"),
which the listener wrapper translated to MoqProtocolException via
MoqLiteNestsListener.wrapSubscription.

The reconnecting wrapper's inner re-issue loop then did:

    val handle = try { opener(listener) } catch (...) { null } ?: break

— breaking out of the loop forever. The outer collectLatest only
re-runs on listener swap (session reconnect), so once the first
SUBSCRIBE failed the audio path stayed dead until the user manually
disconnected and reconnected. With the typical join-order being
"listener taps Join before speaker taps Start", this hit nearly
every nest.

The kdoc on pumpAnnounceWatch claims "moq-lite supports subscribe-
before-announce, so a subscribe issued during the gap … attaches
cleanly when the new publisher comes up" — true for some publisher-
cycle gaps mid-broadcast, but verifiably false for the cold-start
case where the publisher hasn't existed yet on the relay's view of
the namespace. Production trace (commit 283e776):

    14:23:29.556 subscribe id=0 track='audio/data': SUBSCRIBE bytes flushed
    14:23:29.597 subscribe id=0: bidi closed BEFORE any response parsed
    14:23:29.617 ReconnectingHandle.opener threw MoqProtocolException — pump breaks
    14:23:33.604 announce update status=Active suffix=…  (4 s later)
    14:23:34.218 publish suffix=…  (speaker starts publishing)

Fix: instead of `break`, treat opener-throw as a transient failure
and retry after SUBSCRIBE_RETRY_BACKOFF_MS (1 s). The outer
collectLatest still cancels the inner loop on listener swap, and
unsubscribeAction.cancel() still tears the pump down on consumer
release, so the retry loop is bounded by both the session and the
caller. 1 s is well over moq-rs's typical announce-propagation
window (< 200 ms in traces) and short enough that the listener
attaches within a second of the speaker actually publishing.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ
---
 .../nestsclient/ReconnectingNestsListener.kt  | 31 +++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
index ebe165c3c..40ee46549 100644
--- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
+++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
@@ -353,11 +353,29 @@ private class ReconnectingHandle(
                                 // suspend re-checked active state.
                                 throw ce
                             } catch (t: Throwable) {
+                                // The opener can throw for a transient
+                                // reason — most commonly: the listener
+                                // joined the room before the speaker started
+                                // publishing, the announce-watch hasn't yet
+                                // surfaced an Active for this broadcast, and
+                                // the relay FINs the SUBSCRIBE bidi without
+                                // a SubscribeOk/SubscribeDrop reply. The
+                                // earlier shape returned `null` and `break`'d
+                                // out of the inner re-issue loop, which made
+                                // a single pre-publisher subscribe permanent
+                                // — the audio never recovered when the
+                                // speaker did come up. Retry with the same
+                                // backoff used between publisher cycles.
                                 Log.w("NestRx") {
-                                    "ReconnectingHandle.opener threw ${t::class.simpleName}: ${t.message} — pump breaks"
+                                    "ReconnectingHandle.opener threw ${t::class.simpleName}: ${t.message} " +
+                                        "— retrying after ${SUBSCRIBE_RETRY_BACKOFF_MS}ms"
                                 }
                                 null
-                            } ?: break
+                            }
+                        if (handle == null) {
+                            delay(SUBSCRIBE_RETRY_BACKOFF_MS)
+                            continue
+                        }
                         liveHandleRef.set(handle)
                         try {
                             handle.objects.collect { frames.emit(it) }
@@ -410,6 +428,15 @@ private class ReconnectingHandle(
         // gone publisher doesn't spin the relay with re-subscribes.
         private const val RESUBSCRIBE_BACKOFF_MS = 100L
 
+        // Backoff after an opener throws (typically: subscribe-before-
+        // announce arrives at the relay before the publisher exists,
+        // and the relay FINs the bidi without a SubscribeOk/Drop). One
+        // second is well over moq-rs's typical announce-propagation
+        // latency (< 200 ms in production traces), so the next retry
+        // usually succeeds; long enough that a never-arriving publisher
+        // doesn't hammer the relay either.
+        private const val SUBSCRIBE_RETRY_BACKOFF_MS = 1_000L
+
         private val SYNTH_OK =
             SubscribeOk(
                 subscribeId = -1L,