From 9b6e09838eec503d52634ca51ef715f22e755ee8 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 4 May 2026 17:16:28 +0000 Subject: [PATCH] debug(nests): expose silent failures inside subscribe / announce / reissue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 instrumentation after phone-B trace showed silence between SUBSCRIBE_OK-expected and the first audio frame. Three additions: - MoqLiteSession.subscribe: log the SUBSCRIBE-bytes-flushed transition, every chunk arriving on the response bidi (chunks=N), and any exception or natural close on bidi.incoming() that previously swallowed the cause. - MoqLiteSession.announce: same treatment for the AnnouncePlease bidi — chunks=N, natural-close, and exception path now visible. - ReconnectingNestsListener.reissuingSubscribe: previously had a `catch (_: Throwable) { null }` black hole on the re-issue pump (line 354); now logs the throwable's class + message before the pump breaks. Also logs natural-close of handle.objects so a publisher cycle vs a real failure are distinguishable in the trace. These narrow the listener-stuck-on-spinner diagnosis from "something downstream of subscribe never fires" to which exact step inside `MoqLiteSession.subscribe` is silent: the write, the response read, or the framing. https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ --- .../nestsclient/ReconnectingNestsListener.kt | 7 ++++++- .../nestsclient/moq/lite/MoqLiteSession.kt | 19 ++++++++++++++++++- 2 files changed, 24 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 7fcaf1a15..ebe165c3c 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -25,6 +25,7 @@ import com.vitorpamplona.nestsclient.moq.SubscribeHandle import com.vitorpamplona.nestsclient.moq.SubscribeOk import com.vitorpamplona.nestsclient.transport.WebTransportFactory import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner +import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.channels.BufferOverflow @@ -351,12 +352,16 @@ private class ReconnectingHandle( // one extra iteration before the next // suspend re-checked active state. throw ce - } catch (_: Throwable) { + } catch (t: Throwable) { + Log.w("NestRx") { + "ReconnectingHandle.opener threw ${t::class.simpleName}: ${t.message} — pump breaks" + } null } ?: break liveHandleRef.set(handle) try { handle.objects.collect { frames.emit(it) } + Log.d("NestRx") { "ReconnectingHandle.objects flow ended naturally — pump will re-issue after ${RESUBSCRIBE_BACKOFF_MS}ms" } } finally { if (liveHandleRef.get() === handle) liveHandleRef.set(null) } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 5979f190a..e0f8180ce 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -116,24 +116,31 @@ class MoqLiteSession internal constructor( suspend fun announce(prefix: String): MoqLiteAnnouncesHandle { ensureOpen() val bidi = transport.openBidiStream() + Log.d("NestRx") { "announce(prefix='$prefix'): bidi opened, writing AnnouncePlease" } bidi.write(Varint.encode(MoqLiteControlType.Announce.code)) bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix))) + Log.d("NestRx") { "announce(prefix='$prefix'): AnnouncePlease flushed, awaiting Active updates" } val updates = MutableSharedFlow(replay = 0, extraBufferCapacity = 64) val pump = scope.launch { val buffer = MoqLiteFrameBuffer() + var chunksSeen = 0 try { bidi.incoming().collect { chunk -> + chunksSeen += 1 + Log.d("NestRx") { "announce(prefix='$prefix'): bidi chunk #$chunksSeen size=${chunk.size}" } buffer.push(chunk) while (true) { val payload = buffer.readSizePrefixed() ?: break updates.emit(MoqLiteCodec.decodeAnnounce(payload)) } } + Log.d("NestRx") { "announce(prefix='$prefix'): bidi.incoming() ended naturally after $chunksSeen chunks" } } catch (ce: CancellationException) { throw ce - } catch (_: Throwable) { + } catch (t: Throwable) { + Log.w("NestRx") { "announce(prefix='$prefix'): bidi.incoming() threw ${t::class.simpleName}: ${t.message} (chunks=$chunksSeen)" } // Flow terminated (peer FIN or transport close). // The Announce stream's emit-side just stops; consumers // see an end-of-flow. @@ -201,8 +208,10 @@ class MoqLiteSession internal constructor( endGroup = endGroup, ) val bidi = transport.openBidiStream() + Log.d("NestRx") { "subscribe id=$id broadcast='$broadcast' track='$track': bidi opened, writing SUBSCRIBE bytes" } bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) bidi.write(MoqLiteCodec.encodeSubscribe(request)) + Log.d("NestRx") { "subscribe id=$id: SUBSCRIBE bytes flushed, awaiting response" } // Single long-running collector for the bidi's whole lifetime. // The collector parses the SubscribeResponse inline, signals it @@ -252,9 +261,14 @@ class MoqLiteSession internal constructor( scope.launch { val responseBuffer = MoqLiteFrameBuffer() var responseParsed = false + var chunksSeen = 0 try { bidi.incoming().collect { chunk -> + chunksSeen += 1 if (!responseParsed) { + Log.d("NestRx") { + "subscribe id=$id: bidi chunk #$chunksSeen size=${chunk.size} (response not yet parsed)" + } responseBuffer.push(chunk) val typeCode = responseBuffer.readVarint() ?: return@collect val body = responseBuffer.readSizePrefixed() ?: return@collect @@ -270,13 +284,16 @@ class MoqLiteSession internal constructor( // moq-lite leaves the bidi idle post-Ok. The signal // we care about is the flow's natural completion. } + Log.d("NestRx") { "subscribe id=$id: bidi.incoming() flow ended naturally after $chunksSeen chunks (responseParsed=$responseParsed)" } } catch (ce: CancellationException) { if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(ce) throw ce } catch (t: Throwable) { + Log.w("NestRx") { "subscribe id=$id: bidi.incoming() threw ${t::class.simpleName}: ${t.message} (chunks=$chunksSeen)" } if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(t) } if (!responseDeferred.isCompleted) { + Log.w("NestRx") { "subscribe id=$id: bidi closed BEFORE any response parsed (chunks=$chunksSeen)" } responseDeferred.completeExceptionally( MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id"), )