fix(nestsclient): STOP_SENDING on group uni when subscription already canceled — moq-lite Lite-03

Lite-03 audit M2: when a group's uni stream arrives for a `subscribeId` whose subscription has already been removed (typical case: listener canceled / unsubscribed before the publisher observed it), the listener MUST signal the publisher to abandon in-flight retransmits via `STOP_SENDING(applicationErrorCode)` on that uni stream. Pre-fix, `drainOneGroup` silently dropped every frame (`droppedNoSub++`) and let the publisher keep pushing until natural FIN — wasted bandwidth on bytes the listener would discard, plus relay queue pressure on the per-subscriber forward pipeline that already sits behind the production stream-cliff. Wire the latched `stopSending(MoqLiteStreamCancelCode.SUBSCRIPTION_GONE)` on the first frame that observes `sub == null`. Latched (one-shot) so concurrent frames in the same drain don't re-fire — RFC 9000 §3.5 says subsequent STOP_SENDINGs are ignored anyway, but it saves the syscall and keeps the trace clean. New error-code constant `MoqLiteStreamCancelCode.SUBSCRIPTION_GONE = 0x10L` lives next to `MoqLiteSubscribeDropCode` in `MoqLiteMessages.kt`. moq-lite leaves application error codes undefined; we follow the same "small non-zero varint" convention. Test seam: `ChannelWriteStream` is now a public class (was private) with a `peerStopSendingCode` accessor, so a test that holds the writer side (`serverSide.openUniStream() as ChannelWriteStream`) can assert the listener's stopSending code without racing for the peer-side `FakeReadStream` reference. Regression test: `MoqLiteSessionTest.listener_stopSending_group_uni_when_subscription_already_canceled`. Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (M2). https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq
2026-05-09 14:38:25 +00:00
parent c7a49d1679
commit 4368875346
4 changed files with 124 additions and 9 deletions
@@ -172,6 +172,30 @@ object MoqLiteSubscribeDropCode {
    const val BROADCAST_DOES_NOT_EXIST: Long = 0x05L
 }

+/**
+ * Application error codes a listener passes to
+ * [com.vitorpamplona.nestsclient.transport.WebTransportReadStream.stopSending]
+ * when canceling a group's uni stream. moq-lite leaves these
+ * application-defined; we follow the same convention as
+ * [MoqLiteSubscribeDropCode] (small non-zero varints) so a future
+ * cross-protocol mapping is straightforward.
+ *
+ * Used by [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession.drainOneGroup]
+ * when a group arrives for a subscription that's already been
+ * removed: rather than silently dropping every frame the publisher
+ * pushes (and letting the publisher waste bandwidth on
+ * retransmits), we `stopSending` the uni stream so the publisher
+ * abandons it.
+ */
+object MoqLiteStreamCancelCode {
+    /**
+     * The receiving subscription has been canceled / unsubscribed
+     * since this group started. The publisher should abandon any
+     * pending retransmits.
+     */
+    const val SUBSCRIPTION_GONE: Long = 0x10L
+}
+
 /**
 * Header at the start of a Group uni stream. After the
 * [MoqLiteDataType.Group] type byte, the publisher writes one
@@ -604,6 +604,15 @@ class MoqLiteSession internal constructor(
        var frameCount = 0
        var droppedNoSub = 0
        var trySendFailures = 0
+        // Audit M2: once we observe the subscription is gone, fire
+        // STOP_SENDING(SUBSCRIPTION_GONE) once on the uni stream so
+        // the publisher abandons any in-flight retransmits. Pre-fix
+        // we silently dropped every frame; the publisher kept pushing
+        // and the relay kept buffering bytes the listener would never
+        // read. Latched so concurrent frames don't fire it N times
+        // (RFC 9000 §3.5: subsequent STOP_SENDING calls are ignored
+        // by the peer, but we save the syscall).
+        var stopSendingFired = false
        try {
            stream.incoming().collect { chunk ->
                buffer.push(chunk)
@@ -630,6 +639,17 @@ class MoqLiteSession internal constructor(
                    val sub = state.withLock { subscriptionsBySubscribeId[subscribeId] }
                    if (sub == null) {
                        droppedNoSub += 1
+                        if (!stopSendingFired) {
+                            stopSendingFired = true
+                            Log.w("NestRx") {
+                                "drainOneGroup#$streamSeq subId=$subscribeId no longer subscribed — stopSending(SUBSCRIPTION_GONE)"
+                            }
+                            NestsTrace.emit("group_stop_sending") {
+                                "\"stream_seq\":$streamSeq,\"sub_id\":$subscribeId,\"group_seq\":$groupSequence," +
+                                    "\"code\":${MoqLiteStreamCancelCode.SUBSCRIPTION_GONE}"
+                            }
+                            runCatching { stream.stopSending(MoqLiteStreamCancelCode.SUBSCRIPTION_GONE) }
+                        }
                    } else {
                        val sent =
                            sub.frames.trySend(
@@ -641,16 +661,13 @@ class MoqLiteSession internal constructor(
                        if (!sent.isSuccess) trySendFailures += 1
                    }
                    frameCount += 1
-                    // If the subscription has been closed already we
-                    // silently drop the frame — the publisher hasn't
-                    // observed the unsubscribe yet (its uni streams
-                    // are independent of our bidi FIN).
                }
            }
-            Log.d("NestRx") { "drainOneGroup#$streamSeq FIN subId=$subscribeId groupSeq=$groupSequence frames=$frameCount droppedNoSub=$droppedNoSub trySendFail=$trySendFailures" }
+            Log.d("NestRx") { "drainOneGroup#$streamSeq FIN subId=$subscribeId groupSeq=$groupSequence frames=$frameCount droppedNoSub=$droppedNoSub trySendFail=$trySendFailures stopSent=$stopSendingFired" }
            NestsTrace.emit("group_fin") {
                "\"stream_seq\":$streamSeq,\"sub_id\":$subscribeId,\"group_seq\":$groupSequence," +
-                    "\"frames\":$frameCount,\"dropped_no_sub\":$droppedNoSub,\"try_send_fail\":$trySendFailures"
+                    "\"frames\":$frameCount,\"dropped_no_sub\":$droppedNoSub,\"try_send_fail\":$trySendFailures," +
+                    "\"stop_sent\":$stopSendingFired"
            }
        } catch (ce: CancellationException) {
            throw ce
@@ -329,8 +329,13 @@ class FakeReadStream internal constructor(
 * [FakeWebTransport.openUniStream] (production: locally-opened uni
 * stream that the paired peer reads via incomingUniStreams) and any
 * test that wants to drive uni-stream bytes through a known channel.
+ *
+ * Public so tests that hold the writer reference (e.g. via
+ * `serverSide.openUniStream()`) can introspect the peer-side
+ * `stopSending` code via [peerStopSendingCode] without racing the
+ * listener's uni-stream pump on the read side.
 */
-private class ChannelWriteStream(
+class ChannelWriteStream internal constructor(
    private val channel: Channel<ByteArray>,
    /**
     * Optional shared cell with the peer-side reader. When present,
@@ -338,6 +343,12 @@ private class ChannelWriteStream(
     */
    private val priorityCell: java.util.concurrent.atomic.AtomicInteger? = null,
    private val resetCell: java.util.concurrent.atomic.AtomicLong? = null,
+    /**
+     * Cell that records the *peer's* `stopSending(code)` call, since
+     * the writer is on the other side of the stream from the
+     * `stopSending` caller. Same cell that
+     * [FakeReadStream.lastStopSendingCode] reads.
+     */
    private val stopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
 ) : WebTransportWriteStream {
    override suspend fun write(chunk: ByteArray) {
@@ -360,8 +371,15 @@ private class ChannelWriteStream(
        priorityCell?.set(priority)
    }

-    @Suppress("unused")
-    private val unusedStopSendingCellTether: java.util.concurrent.atomic.AtomicLong? = stopSendingCell
+    /**
+     * Code the peer-side reader passed to
+     * [WebTransportReadStream.stopSending], or `null` if not called.
+     * Lets a test that holds the writer side inspect the listener's
+     * group-cancel without racing for the peer-side
+     * [WebTransportReadStream] reference.
+     */
+    val peerStopSendingCode: Long?
+        get() = stopSendingCell?.get()?.takeIf { it != NO_CODE }
 }

 /**
@@ -20,6 +20,7 @@
 */
 package com.vitorpamplona.nestsclient.moq.lite

+import com.vitorpamplona.nestsclient.transport.ChannelWriteStream
 import com.vitorpamplona.nestsclient.transport.FakeBidiStream
 import com.vitorpamplona.nestsclient.transport.FakeReadStream
 import com.vitorpamplona.nestsclient.transport.FakeWebTransport
@@ -835,6 +836,61 @@ class MoqLiteSessionTest {
            session.close()
        }

+    @Test
+    fun listener_stopSending_group_uni_when_subscription_already_canceled() =
+        runBlocking {
+            // Lite-03 audit M2: when a group's uni stream arrives for a
+            // subscribe id that was canceled before the first frame
+            // landed, the listener MUST stopSending() the uni stream so
+            // the publisher abandons retransmits instead of wasting
+            // bandwidth. Pre-fix the listener silently dropped every
+            // frame; the publisher kept pushing until natural FIN.
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            // Set up a subscription so the uni-stream pump is active.
+            val peer =
+                async {
+                    val (bidi, _) = nextSubscribeBidi(serverSide)
+                    bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
+                    bidi
+                }
+            val handle = session.subscribe("speakerX", "audio/data")
+            val subBidi = peer.await()
+
+            // Cancel the subscription before any group flows.
+            handle.unsubscribe()
+            // Drain the subscribe bidi's FIN so the peer's pump
+            // stays clean — not strictly required for the assertion
+            // but keeps the test isolated.
+            subBidi.incoming().toList()
+
+            // Now the peer (= the relay) opens a uni stream for that
+            // dead subscribe id and writes one frame.
+            val uni = serverSide.openUniStream() as ChannelWriteStream
+            uni.write(Varint.encode(MoqLiteDataType.Group.code))
+            uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handle.id, sequence = 0L)))
+            uni.write(framePayload(byteArrayOf(0x01)))
+
+            // Spin until the listener's drainOneGroup has processed
+            // the first frame, found no matching subscription, and
+            // fired stopSending. We hold the writer side, so reading
+            // [peerStopSendingCode] doesn't race the listener's
+            // pump for the peer-side [FakeReadStream] reference.
+            val deadline = System.currentTimeMillis() + 2_000
+            while (uni.peerStopSendingCode == null && System.currentTimeMillis() < deadline) {
+                kotlinx.coroutines.delay(10)
+            }
+            assertEquals(
+                MoqLiteStreamCancelCode.SUBSCRIPTION_GONE as Long?,
+                uni.peerStopSendingCode,
+                "listener must stopSending(SUBSCRIPTION_GONE) when a group arrives for a canceled subscription",
+            )
+
+            uni.finish()
+            session.close()
+        }
+
    @Test
    fun unsubscribe_FINs_the_subscribe_bidi() =
        runBlocking {