fix(nestsclient): SubscribeOk narrows startGroup to publisher.nextSequence — moq-lite Lite-03
Lite-03 audit L2: when accepting a SUBSCRIBE, the publisher MAY narrow the subscriber's `startGroup` / `endGroup` request bounds. Pre-fix we always echoed `null/null`, which lost the diagnostic "which group am I about to start sending?" information — particularly useful for hot-swap continuations where the publisher's [MoqLitePublisherHandle.nextSequence] is non-zero (the seeded `startSequence` from the previous moq-lite session). The fix narrows `startGroup` to `targetPublisher.nextSequence`. The subscriber decodes this as "the next group on this subscription will be sequence N" and can log / surface the join-point. `endGroup` stays null because live audio rooms have no end in sight; the subscriber's request bound is honoured implicitly when the publisher closes. Regression test: `MoqLiteSessionTest.publisher_subscribeOk_narrows_startGroup_to_next_sequence`. Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (L2). https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq
This commit is contained in:
+17
-1
@@ -1028,17 +1028,33 @@ class MoqLiteSession internal constructor(
|
||||
targetPublisher.registerInboundSubscription(sub)
|
||||
inboundSub = sub
|
||||
inboundSubPublisher = targetPublisher
|
||||
// Audit L2: narrow `startGroup` to the
|
||||
// publisher's [MoqLitePublisherHandle.nextSequence]
|
||||
// — the exact sequence of the next group we'll
|
||||
// open. Spec lets the publisher narrow the
|
||||
// subscriber's bounds; pre-fix we always echoed
|
||||
// `null/null`, which loses the diagnostic
|
||||
// information of "which group are you about to
|
||||
// start sending?" Useful for hot-swap
|
||||
// continuations where `nextSequence` is non-zero
|
||||
// (the seeded `startSequence` from the previous
|
||||
// session). `endGroup` stays null because we're
|
||||
// a live broadcast with no end in sight; the
|
||||
// subscriber's request `endGroup` is honoured
|
||||
// implicitly when the publisher closes.
|
||||
val narrowedStart = targetPublisher.nextSequence
|
||||
bidi.write(
|
||||
MoqLiteCodec.encodeSubscribeOk(
|
||||
MoqLiteSubscribeOk(
|
||||
priority = sub.priority,
|
||||
ordered = sub.ordered,
|
||||
maxLatencyMillis = sub.maxLatencyMillis,
|
||||
startGroup = null,
|
||||
startGroup = narrowedStart,
|
||||
endGroup = null,
|
||||
),
|
||||
),
|
||||
)
|
||||
Log.d("NestTx") { "SUBSCRIBE_OK id=${sub.id} narrowed startGroup=$narrowedStart" }
|
||||
dispatched = true
|
||||
}
|
||||
|
||||
|
||||
+49
@@ -342,6 +342,55 @@ class MoqLiteSessionTest {
|
||||
session.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun publisher_subscribeOk_narrows_startGroup_to_next_sequence() =
|
||||
runBlocking {
|
||||
// Lite-03 audit L2: when accepting a SUBSCRIBE, the publisher
|
||||
// narrows the SubscribeOk's `startGroup` from the subscriber's
|
||||
// request to its own [MoqLitePublisherHandle.nextSequence],
|
||||
// communicating the exact sequence the next group will
|
||||
// carry. Particularly useful for hot-swap continuations
|
||||
// (publisher seeded with a non-zero startSequence).
|
||||
val (clientSide, serverSide) = FakeWebTransport.pair()
|
||||
val session = MoqLiteSession.client(clientSide, pumpScope)
|
||||
|
||||
val publisher =
|
||||
session.publish(
|
||||
broadcastSuffix = "speakerPubkey",
|
||||
track = "audio/data",
|
||||
startSequence = 100L,
|
||||
)
|
||||
|
||||
val subBidi = serverSide.openBidiStream()
|
||||
subBidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
|
||||
subBidi.write(
|
||||
MoqLiteCodec.encodeSubscribe(
|
||||
MoqLiteSubscribe(
|
||||
id = 5L,
|
||||
broadcast = "speakerPubkey",
|
||||
track = "audio/data",
|
||||
priority = 0x80,
|
||||
ordered = true,
|
||||
maxLatencyMillis = 0L,
|
||||
// Subscriber asks "from latest" — publisher is
|
||||
// free to narrow.
|
||||
startGroup = null,
|
||||
endGroup = null,
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
val ackChunk = withTimeout(2_000) { subBidi.incoming().first() }
|
||||
val resp = MoqLiteCodec.decodeSubscribeResponse(ackChunk)
|
||||
val ok = (resp as MoqLiteCodec.SubscribeResponse.Ok).ok
|
||||
assertEquals(100L, ok.startGroup, "publisher narrows startGroup to its nextSequence")
|
||||
// endGroup stays null — live broadcast has no end in sight.
|
||||
assertEquals(null, ok.endGroup)
|
||||
|
||||
publisher.close()
|
||||
session.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun publisher_startSequence_seeds_first_group_for_hot_swap_continuation() =
|
||||
runBlocking {
|
||||
|
||||
Reference in New Issue
Block a user