From d24953d98ca825b74f68094727d75b749f0033a2 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 9 May 2026 14:51:54 +0000 Subject: [PATCH] =?UTF-8?q?feat(nestsclient):=20subscriber-driven=20Probe?= =?UTF-8?q?=20API=20(MoqLiteSession.probe)=20=E2=80=94=20moq-lite=20Lite-0?= =?UTF-8?q?3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lite-03 audit L3: completes the subscriber-side Probe surface to mirror kixelated's `Subscriber::run_probe_stream` (`rs/moq-lite/src/lite/subscriber.rs`). `MoqLiteSession.probe()` opens a bidi, writes `ControlType::Probe` (varint 4), and returns a `MoqLiteProbeHandle` whose `updates` flow yields each size-prefixed `MoqLiteProbe` message the publisher pushes. The handle's `close()` FINs the bidi and cancels the pump. `updates` is a `MutableSharedFlow(replay=8)` so a collector that attaches after the publisher's first emit doesn't miss it (matches the same shape the announce-watch uses). No production consumer for the API today — Amethyst's nests listener doesn't run ABR on a fixed-rate Opus encoder. The API exists to round out the moq-lite Lite-03 surface: a diagnostic tool can now read a publisher's bitrate without subscribing to its data track. The publisher-side handler (existing) was already correct: it writes one `MoqLiteProbe` (32 kbps Opus voice hint) and FINs. Regression test: `MoqLiteSessionTest.subscriber_probe_writes_control_type_and_decodes_publisher_bitrate`. Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (L3). https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq --- .../nestsclient/moq/lite/MoqLiteHandles.kt | 22 +++++++ .../nestsclient/moq/lite/MoqLiteSession.kt | 61 +++++++++++++++++++ .../moq/lite/MoqLiteSessionTest.kt | 44 +++++++++++++ 3 files changed, 127 insertions(+) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt index 9f69f1f00..86ecbbee8 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteHandles.kt @@ -54,6 +54,28 @@ class MoqLiteAnnouncesHandle internal constructor( suspend fun close() = close.invoke() } +/** + * Active probe handle returned by [MoqLiteSession.probe]. Mirrors + * kixelated's `Subscriber::run_probe_stream` + * (`rs/moq-lite/src/lite/subscriber.rs`): subscriber opens a bidi, + * writes `ControlType::Probe`, then reads size-prefixed + * [MoqLiteProbe] messages indefinitely until the publisher FINs or + * the consumer calls [close]. + * + * [updates] is a cold flow that emits every Probe message the + * publisher pushes — typically a single bitrate hint at session + * start, but a publisher running an ABR codec MAY emit multiple + * over time. For a fixed-rate Opus producer (which is what + * Amethyst's nests speaker is) the publisher emits one and FINs; + * the flow then completes naturally and consumers see end-of-flow. + */ +class MoqLiteProbeHandle internal constructor( + val updates: Flow, + private val close: suspend () -> Unit, +) { + suspend fun close() = close.invoke() +} + /** Thrown when subscribe is rejected (Drop) or the response stream dies. */ class MoqLiteSubscribeException( message: String, 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 4749f5f15..40282f6bf 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 @@ -691,6 +691,67 @@ class MoqLiteSession internal constructor( sub.frames.close() } + /** + * Open a Probe bidi against the relay / publisher (audit L3). + * Mirrors kixelated's `Subscriber::run_probe_stream` + * (`rs/moq-lite/src/lite/subscriber.rs`): subscriber writes + * `ControlType::Probe` (varint 4), then reads size-prefixed + * [MoqLiteProbe] messages until the publisher FINs. + * + * The publisher reciprocates by writing one or more [MoqLiteProbe] + * messages indicating its expected outbound bitrate. Amethyst's + * own publisher path is fixed-rate Opus and emits exactly one + * Probe before FIN, but third-party (e.g. browser-side + * `kixelated/moq` watcher → relay-served broadcast) publishers + * MAY emit multiple over time as their ABR estimates change. + * + * The returned handle's `updates` flow is cold — collecting it + * starts the read pump on this session's scope. On collector + * cancellation OR [MoqLiteProbeHandle.close], the bidi FINs and + * the pump exits. + * + * No application-level use of probe data exists in nests today + * (we don't run ABR), but the API completes the moq-lite Lite-03 + * protocol surface. Useful for diagnostic tools that want to + * read a publisher's bitrate without subscribing to its data + * track. + */ + suspend fun probe(): MoqLiteProbeHandle { + ensureOpen() + val bidi = transport.openBidiStream() + bidi.write(Varint.encode(MoqLiteControlType.Probe.code)) + val updates = + MutableSharedFlow( + replay = 8, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + val pump = + scope.launch { + val buffer = MoqLiteFrameBuffer() + try { + bidi.incoming().collect { chunk -> + buffer.push(chunk) + while (true) { + val payload = buffer.readSizePrefixed() ?: break + updates.emit(MoqLiteCodec.decodeProbe(payload)) + } + } + } catch (ce: CancellationException) { + throw ce + } catch (_: Throwable) { + // Probe bidi died — best-effort silent. The + // updates flow simply stops emitting. + } + } + return MoqLiteProbeHandle( + updates = updates, + close = { + runCatching { bidi.finish() } + pump.cancelAndJoin() + }, + ) + } + // ==================================================================== // Publisher side // ==================================================================== diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt index cf09542b8..3e419fee4 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt @@ -940,6 +940,50 @@ class MoqLiteSessionTest { session.close() } + @Test + fun subscriber_probe_writes_control_type_and_decodes_publisher_bitrate() = + runBlocking { + // Lite-03 audit L3: the subscriber side opens a Probe bidi + // by writing ControlType=Probe and reads size-prefixed + // MoqLiteProbe messages from the publisher. Mirrors + // kixelated's `Subscriber::run_probe_stream` + // (`rs/moq-lite/src/lite/subscriber.rs`). + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + // Server side: read the subscriber's control byte, then + // write two Probe messages. + val serverPump = + async { + val bidi = serverSide.peerOpenedBidiStreams().first() + val buf = MoqLiteFrameBuffer() + // Read the leading ControlType varint. The session + // writes it as a single chunk via bidi.write. + val ctChunk = withTimeout(2_000) { bidi.incoming().first() } + buf.push(ctChunk) + val ct = buf.readVarint() + assertEquals(MoqLiteControlType.Probe.code, ct, "subscriber writes ControlType=Probe") + + // Push two probes (different bitrates) then FIN. + bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 32_000L))) + bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 64_000L))) + bidi.finish() + } + + val handle = session.probe() + // Drain both probes from the cold flow. SharedFlow with + // replay=8 means we can collect after the publisher has + // already pushed without losing emissions. + val received = withTimeout(2_000) { handle.updates.take(2).toList() } + assertEquals(2, received.size) + assertEquals(32_000L, received[0].bitrate) + assertEquals(64_000L, received[1].bitrate) + + serverPump.await() + handle.close() + session.close() + } + @Test fun unsubscribe_FINs_the_subscribe_bidi() = runBlocking {