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

debug(nests): expose silent failures inside subscribe / announce / reissue

Browse Source 
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
This commit is contained in:
Claude
2026-05-04 17:16:28 +00:00
parent 62bedf1372
commit 9b6e09838e
2 changed files with 24 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
+6 -1
View File
@@ -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)
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+18 -1
View File
@@ -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<MoqLiteAnnounce>(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"),
)