diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 38621fa966..e6426429ab 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -1053,7 +1053,7 @@ A value of 0 means the whole group (default). A non-zero value is the absolute frame index + 1, matching `Group End`. MUST be 0 when `Group End` is 0, since an unbounded subscription has no end group to qualify. -`Group Start` and `Group End` are offset by 1 only so 0 can mean "absent"; every other group field in this document is a plain absolute sequence. +`Group End` and `Frame End` are offset by 1 so 0 can mean "absent". `Group Start` is an absolute sequence, and 0 is group 0. ## SUBSCRIBE_UPDATE A subscriber can modify a subscription with a SUBSCRIBE_UPDATE message. @@ -1150,9 +1150,13 @@ SUBSCRIBE_OK Message { Set to 0x0 to indicate a SUBSCRIBE_OK message. **Group**: -The absolute sequence number of the first group that will be delivered. -It MUST be greater than or equal to the requested start group; any groups in between are unavailable. -A subscriber that requested the latest group learns the resolved sequence here. +The absolute sequence number where delivery starts when the subscription resolves. +Group streams can arrive out of order, so it need not be the first group sent. +It MUST be greater than or equal to the requested `Group Start`. +This group is not a new floor. +A subscriber SHOULD NOT wait for a group between the requested floor and this group. +The publisher still delivers such a group if it is created or becomes available later, while it is within `Subscriber Max Age`. +A subscriber whose `Subscriber Max Age` resolves the start to the latest group learns that sequence here. There is no matching frame field, because the start frame is never in doubt: a partial group is only delivered when it was asked for, so the subscription starts either exactly where it asked or at the beginning of a later group (see [Positions](#positions)). The subscriber derives the start frame from `Group` and its own request: @@ -1346,6 +1350,8 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 +- SUBSCRIBE_OK `Group` names where delivery starts when the subscription resolves, which need not be the first group sent. It is not a new floor. A subscriber does not wait for a group between the requested floor and it, but a publisher still delivers one that arrives later within Subscriber Max Age. +- Corrected the SUBSCRIBE note that offset `Group Start` by 1. Only `Group End` and `Frame End` are offset so 0 can mean absent. `Group Start` is an absolute sequence, and 0 is group 0. - Assigned `moq-lite-07-wip` as this draft's protocol identifier until it is finalized as `moq-lite-07`. - Switched every variable-length integer, including SETUP parameter values, from QUIC's two-bit length prefix to moq-transport's leading-ones encoding, widening the range to 64 bits. - Hid routes with a `.`-prefixed segment below the requested prefix from announce discovery, and added the ANNOUNCE_REQUEST `Hidden` field to opt in. diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 17b29108c5..5a91fda6f9 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -164,8 +164,7 @@ test("integration: lite subscription options and updates reach the publisher", a })(); const remote = wireOf(client).consume(Path.from("test")); - // A floor of 1, not 0: a pre-06 wire folds a vacuous floor of 0 back to absent, since - // its encoding of group 0 means "replay from the beginning" instead. + // A floor of group 1, a concrete group on every draft. const subscriber = remote.track("video").subscribe({ priority: 3, maxAge: Milli(250), @@ -223,8 +222,7 @@ test("integration: lite carries a fractional maxAge as a whole millisecond", asy const remote = wireOf(client).consume(Path.from("test")); // A varint cannot encode 38.75, so an unrounded value fails the SUBSCRIBE outright and - // nothing resubscribes. The publisher must see the budget rounded up instead. A floor - // of 1, not 0: a pre-06 wire folds a vacuous floor of 0 back to absent. + // nothing resubscribes. The publisher must see the budget rounded up instead. const subscriber = remote.track("video").subscribe({ maxAge: Milli(38.75), groups: { start: { included: 1 } } }); // A failed subscribe never reaches the publisher, so race its closure to report the diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 64d50dd980..f46bfcb805 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -644,9 +644,9 @@ test("lite draft-05: a late-arriving older group is still served", async () => { } }); -// SUBSCRIBE_START promises nothing below the announced sequence will be delivered, so -// the floor is pinned there: a straggler below the first served group is dropped even -// though arrival-order serving would otherwise surface it. +// An unfloored pre-06 subscribe joins at the publisher's start. SUBSCRIBE_START makes +// that group the floor: a straggler below it is dropped even though arrival-order +// serving would otherwise surface it. test("lite draft-05: a straggler below the announced start group is not served", async () => { const sub = await servedSubscription(); try { @@ -668,6 +668,71 @@ test("lite draft-05: a straggler below the announced start group is not served", } }); +// An explicit floor stays where it was named. SUBSCRIBE_START still reports the first +// served group, and a later group at or above the floor is delivered. +test("lite draft-05: an explicit floor still serves a group below the first one", async () => { + const sub = await servedSubscription({ startGroup: 0 }); + try { + sub.serve(2); + expect(await sub.servedSequence()).toBe(2); + + const resp = await decodeSubscribeResponse(sub.client.reader, Version.DRAFT_05); + if (!("start" in resp)) throw new Error("expected SUBSCRIBE_START"); + expect(resp.start.group).toBe(2); + + sub.serve(0); + expect(await sub.servedSequence()).toBe(0); + } finally { + await sub.close(); + } +}); + +// Pre-06 an absent Group Start is the latest group. An update that drops a named floor +// re-pins there, so a group created below it afterwards is not served. +test("lite draft-05: an update dropping the floor re-pins to the latest group", async () => { + const sub = await servedSubscription({ startGroup: 0 }); + try { + sub.serve(2); + expect(await sub.servedSequence()).toBe(2); + + await replayUpdate({ priority: 0 }).encode(sub.client.writer, Version.DRAFT_05); + await flush(); + sub.serve(0); + sub.serve(3); + expect(await sub.servedSequence()).toBe(3); + } finally { + await sub.close(); + } +}); + +// On an empty track the re-pin has no latest group to name, so the first group the +// track produces is served, even below the dropped floor. +test("lite draft-05: an update dropping the floor on an empty track serves the next group", async () => { + const sub = await servedSubscription({ startGroup: 5 }); + try { + await replayUpdate({ priority: 0 }).encode(sub.client.writer, Version.DRAFT_05); + await flush(); + sub.serve(0); + expect(await sub.servedSequence()).toBe(0); + } finally { + await sub.close(); + } +}); + +// Lite-06 encodes a floor of group 0 as 0, including a subscribe that omitted Group Start. +// That is a floor, so a later group 0 is still served. +test("lite draft-06: an omitted group start still serves a group below the first one", async () => { + const sub = await servedSubscription({ version: Version.DRAFT_06 }); + try { + sub.serve(2); + expect(await sub.servedSequence()).toBe(2); + sub.serve(0); + expect(await sub.servedSequence()).toBe(0); + } finally { + await sub.close(); + } +}); + // A group pop is the linearization point. Once the publisher reaches the SUBSCRIBE_START write, // the group and its range have already been decided, so an update queued during the write applies // to the next pop rather than reaching backward into this one. @@ -687,7 +752,8 @@ test("lite draft-05: a group popped before a cap update is still served", async sub.release(); expect(await sub.servedSequence()).toBe(1); while (ranges.mock.calls.length === 0) await flush(); - expect(lastGroups(ranges)).toEqual({ start: undefined, end: { included: 0 } }); + // Dropping the named floor re-pins to the latest group, as an unfloored SUBSCRIBE does. + expect(lastGroups(ranges)).toEqual({ start: { included: 1 }, end: { included: 0 } }); } finally { ranges.mockRestore(); sub.release(); @@ -709,8 +775,9 @@ test("lite draft-06: a queued floor update applies before the next group pop", a await replayUpdate({ priority: 0, startGroup: 2 }).encode(sub.client.writer, Version.DRAFT_06); sub.serve(2); await flush(); - expect(ranges).toHaveBeenCalledTimes(1); - expect(lastGroups(ranges)).toEqual({ start: { included: 0 }, end: undefined }); + // Draft-06 group 0 is a floor, so the first group does not raise the cursor. + // The queued update is still waiting on the parked write. + expect(ranges).not.toHaveBeenCalled(); sub.release(); @@ -761,8 +828,9 @@ test("lite draft-06: a queued frame floor applies before the next group pop", as await replayUpdate({ priority: 0, startGroup: 1, startFrame: 2 }).encode(sub.client.writer, Version.DRAFT_06); await flush(); - expect(ranges).toHaveBeenCalledTimes(1); - expect(lastGroups(ranges)).toEqual({ start: { included: 0 }, end: undefined }); + // Draft-06 group 0 is a floor, so serving the first group does not move the cursor. + // The queued update is still waiting on the parked write. + expect(ranges).not.toHaveBeenCalled(); sub.release(); diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index d8d5c57004..83cd92bbb2 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -790,8 +790,17 @@ export class Publisher { case "update": { const update = control.update; console.debug(`subscribe update: broadcast=${broadcast} track=${track.name}`); + // `positionCursor` pins an unfloored pre-06 SUBSCRIBE to the latest + // group, so an update that drops a named floor re-pins there too. On + // an empty track the next group is the latest, whatever its sequence. + const start = + update.startGroup === undefined && + bounds.startGroup !== undefined && + !resolvesStart(this.version) + ? (track.latest() ?? 0) + : update.startGroup; hooks.replaceGroups(track, { - start: update.startGroup === undefined ? undefined : { included: update.startGroup }, + start: start === undefined ? undefined : { included: start }, end: update.endGroup === undefined ? undefined : { included: update.endGroup }, }); bounds.startGroup = update.startGroup; @@ -851,13 +860,17 @@ export class Publisher { if (emitRange && !startSent) { startSent = true; - // SUBSCRIBE_START promises nothing below this sequence will be delivered. - // Arrival-order serving could later surface a straggler below the first - // group, so pin the floor to what was announced. - hooks.replaceGroups(track, { - start: { included: group.sequence }, - end: bounds.endGroup === undefined ? undefined : { included: bounds.endGroup }, - }); + // SUBSCRIBE_START names the first group served now. A later group at or + // above an explicit floor is still delivered, so the cursor stays put. + // A pre-06 subscribe that named no group is the exception: an absent + // Group Start is the latest group, and this sequence becomes the floor. + // Lite-06 encodes a floor of group 0 as 0, so an omitted start is not pinned. + if (bounds.startGroup === undefined && !resolvesStart(this.version)) { + hooks.replaceGroups(track, { + start: { included: group.sequence }, + end: bounds.endGroup === undefined ? undefined : { included: bounds.endGroup }, + }); + } if ( !(await controls.response( encodeSubscribeResponse( diff --git a/js/net/src/lite/subscribe.test.ts b/js/net/src/lite/subscribe.test.ts index 4da1af0918..0fe349628f 100644 --- a/js/net/src/lite/subscribe.test.ts +++ b/js/net/src/lite/subscribe.test.ts @@ -111,14 +111,23 @@ test("Subscribe round-trips every option including startGroup 0", async () => { expect(resumed.startFrame).toBe(4); message.startFrame = 0; - // A pre-06 wire folds the vacuous floor back to absent: an explicit group 0 there - // would mean "replay from the beginning", which is not what a floor of 0 asks for. - const folded = await Subscribe.decode( + // A pre-06 wire encodes an explicit group 0 as sequence + 1, so it round-trips. + // Omitting the floor is still the latest group, not group 0. + const explicit = await Subscribe.decode( new Reader(undefined, await encodeMessage(Version.DRAFT_05, message), Version.DRAFT_05), Version.DRAFT_05, ); - expect(folded.startGroup).toBeUndefined(); - expect(folded.endGroup).toBe(9); + expect(explicit.startGroup).toBe(0); + expect(explicit.endGroup).toBe(9); + const absent = await Subscribe.decode( + new Reader( + undefined, + await encodeMessage(Version.DRAFT_05, new Subscribe({ ...message, startGroup: undefined })), + Version.DRAFT_05, + ), + Version.DRAFT_05, + ); + expect(absent.startGroup).toBeUndefined(); }); test("SubscribeUpdate round-trips every option including startGroup 0", async () => { @@ -137,13 +146,45 @@ test("SubscribeUpdate round-trips every option including startGroup 0", async () expect(got.startGroup).toBeUndefined(); expect(got.endGroup).toBe(12); - // The same fold as SUBSCRIBE on a pre-06 wire. - const folded = await SubscribeUpdate.decode( + // The same pre-06 encoding as SUBSCRIBE: explicit group 0 round-trips, and an + // omitted floor stays omitted. + const explicit = await SubscribeUpdate.decode( new Reader(undefined, await encodeMessage(Version.DRAFT_05, message), Version.DRAFT_05), Version.DRAFT_05, ); - expect(folded.startGroup).toBeUndefined(); - expect(folded.endGroup).toBe(12); + expect(explicit.startGroup).toBe(0); + expect(explicit.endGroup).toBe(12); + const absent = await SubscribeUpdate.decode( + new Reader( + undefined, + await encodeMessage(Version.DRAFT_05, new SubscribeUpdate({ ...message, startGroup: undefined })), + Version.DRAFT_05, + ), + Version.DRAFT_05, + ); + expect(absent.startGroup).toBeUndefined(); +}); + +// The same wire values as Rust's `encode_start_group`: pre-06 writes an omitted floor as 0 +// and group 0 as 1, while lite-06 writes 0 for both. +test("Group Start encodes an omitted floor and group 0 apart before draft-06", async () => { + for (const make of [ + (startGroup?: number) => + new Subscribe({ id: 1n, broadcast: Path.from("test"), track: "video", priority: 0, startGroup }), + (startGroup?: number) => new SubscribeUpdate({ priority: 0, startGroup }), + ]) { + const absent = await encodeMessage(Version.DRAFT_05, make(undefined)); + const zero = await encodeMessage(Version.DRAFT_05, make(0)); + expect(zero.byteLength).toBe(absent.byteLength); + const diff = [...absent.keys()].filter((i) => absent[i] !== zero[i]); + expect(diff).toHaveLength(1); + expect(absent[diff[0]]).toBe(0); + expect(zero[diff[0]]).toBe(1); + + expect(await encodeMessage(Version.DRAFT_06, make(0))).toEqual( + await encodeMessage(Version.DRAFT_06, make(undefined)), + ); + } }); test("SubscribeStart round-trips on draft-05", async () => { diff --git a/js/net/src/lite/subscribe.ts b/js/net/src/lite/subscribe.ts index 3d1a1f2711..8acc561be4 100644 --- a/js/net/src/lite/subscribe.ts +++ b/js/net/src/lite/subscribe.ts @@ -6,17 +6,16 @@ import { hasFrameBounds, hasGroupOrder, hasStreamCount, resolvesStart, Version } /** * Encode the `Group Start` field shared by SUBSCRIBE and SUBSCRIBE_UPDATE. * - * Lite-06 writes the raw floor (`undefined` and 0 are the same absence of a constraint), - * while a pre-06 wire encodes the sequence + 1 and gets a vacuous floor folded back to - * absent: an explicit group 0 there means "replay from the beginning", which is not what - * a floor of 0 asks for. + * Lite-06 writes the raw floor (`undefined` and 0 are both group 0). A pre-06 wire + * encodes the sequence + 1: omitting the floor is 0 (the latest group, where the + * publisher starts) and an explicit 0 is 1, replay from the beginning. */ async function encodeStartGroup(w: Writer, version: Version, startGroup?: number) { if (resolvesStart(version)) { await w.u53(startGroup ?? 0); return; } - await w.u53(startGroup !== undefined && startGroup > 0 ? startGroup + 1 : 0); + await w.u53(startGroup === undefined ? 0 : startGroup + 1); } /** diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index ba81f25468..c007208eb9 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -47,6 +47,7 @@ import { hasDatagrams, hasProbeRtt, hasStreamCount, + resolvesStart, restartSupported, Version, } from "./version.ts"; @@ -892,8 +893,8 @@ export class Subscriber { if ("start" in resp) { entry.start = resp.start.group; - // The groups the SUBSCRIBE asked for below it are unavailable, whatever the - // demand asks later. + // The groups the SUBSCRIBE asked for below it are not waited for, whatever the + // demand asks later. One that still arrives is delivered. if (entry.requested !== undefined) entry.tail.account(entry.requested, entry.start); } else if ("end" in resp) { if (entry.end !== undefined) throw new ProtocolViolation("duplicate SUBSCRIBE_END"); @@ -1012,9 +1013,14 @@ export class Subscriber { #sameSubscription(a: track.Subscription, b: track.Subscription): boolean { const ag = groupBounds(a.groups); const bg = groupBounds(b.groups); + // `groupBounds` reads an omitted start as 0. A pre-06 wire tells them apart: omitted + // joins at the publisher's start, and 0 is group 0. Lite-06 encodes both as 0. + const sameStart = + resolvesStart(this.version) || (a.groups?.start === undefined) === (b.groups?.start === undefined); return ( (a.priority ?? 0) === (b.priority ?? 0) && (a.maxAge ?? 0) === (b.maxAge ?? 0) && + sameStart && ag.start === bg.start && ag.end === bg.end ); diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 04ac64ae9b..7a858577e2 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -323,6 +323,15 @@ test("multiple subscriber options aggregate like Rust", async () => { expect(await none).toBeUndefined(); }); +// A resume floor above the live edge must not hide the latest group from a reader that +// joins after it: on a quiet track that group may be the only one for a long time. +test("an unfloored subscriber clears a resume floor above the live edge", () => { + const producer = new TrackProducer("test"); + producer.subscribe({ groups: { start: { included: 4 } } }); + producer.subscribe({}); + expect(producer.subscription.peek()?.groups?.start).toBeUndefined(); +}); + test("the producer aggregate is clamped without changing subscriber options", async () => { const producer = new TrackProducer("test"); const track = producer.subscribe({ maxAge: Milli(10_000) }); diff --git a/quest/m1/README.md b/quest/m1/README.md index 2a508b46f4..d417b770aa 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -45,7 +45,6 @@ QUIC studies there on that rule. - [WebTransport close upstream](/quest/m1/wt-close-upstream.md) - web-transport-moq delivers the close capsule itself, and moq-tokio's `CLOSE_LINGER` is deleted - [Browser close code](/quest/m1/browser-close-code.md) - a playwright case proves the page reads a relay's close code and reason, on every web-transport backend - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - every stream group in a lite subscription arrives or is dropped by name, and lite-07 drops its stream count for it -- [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does - [Parked reads wake](/quest/m1/parked-read-wakes.md) - a read parked on an evicted or aborted group wakes and re-judges, for resume successors and plain tracks alike - [Reader end parity](/quest/m1/reader-end-parity.md) - JS readers see a track's end once the newest group reaches the declared end, as Rust readers do - [JS cache window](/quest/m1/js-prune-cache-window.md) - js/net ages idle groups on a cache window and keeps max_age as media-time staleness, as Rust does diff --git a/quest/m1/lite-late-lower-group.md b/quest/m1/lite-late-lower-group.md deleted file mode 100644 index 451ed6ec01..0000000000 --- a/quest/m1/lite-late-lower-group.md +++ /dev/null @@ -1,63 +0,0 @@ -# [S] moq-lite delivers a late group above the subscriber's floor - -## Goal - -A moq-lite subscriber with an explicit floor receives every group at or above -that floor that its `max_age` still considers fresh, whatever order the -publisher created them in, as moq-transport already does. Today the lite -publisher silently drops a group created below the first group it served: the -publisher's `create_group` and `write_frame` succeed, and the subscriber sees a -clean end without it. - -## Plan - -`TrackRun::start` (`rs/moq-net/src/lite/publisher.rs`) sends `SUBSCRIBE_START` -for the first served group and then calls `raise_start_to(start)`, which -suppresses every lower group regardless of the subscription's floor and -budget. #4387 added it to resolve a relayed subscription's start from its -source; keep that fix (see the note in -[Track tail interop](/quest/m1/track-tail-interop.md)) while honoring an -explicit floor. - -Decided: - -- An explicit floor delivers late lower groups within `max_age`, matching - moq-transport, draft-ietf-moq-transport ("subscriptions without a filter pass - all Objects"), and moq-mux's floor handling (#3258). -- `start: None` joins where the publisher starts: the first served group - becomes the floor, so a group created below it is dropped by design. This is - what moq-mux's `container::Consumer` and the IETF publisher already assume. - Fix the `Subscription::start` docs in moq-net (which say `None` equals a - floor of group 0) and the matching js/net docs. -- `raise_start_to` applies only when the subscription has no explicit floor - (`start.is_none()`). With an explicit floor, `SUBSCRIBE_START` still reports - the first served group, but nothing below it is suppressed. -- Aggregation follows: `Subscription::poll_combined` - (`rs/moq-net/src/model/subscription.rs`) lets any `None` clear another - subscriber's explicit floor, so a relay forwards "join at the start" upstream - and the explicit-floor subscriber still loses group 0. An explicit floor - survives mixing with `None`; the loosest explicit floor wins. Each - subscriber's own cursor still filters what it sees. -- `start_floor_suppresses_late_lower_arrivals` stays as the `None` case: it - subscribes with `None`, receives group 7, raises the start to 7 as - `SUBSCRIBE_START` does, and group 5 is still suppressed. Add its explicit-floor - twin, which delivers group 5. - -If `drafts/draft-lcurley-moq-lite.md` describes `SUBSCRIBE_START` as an -implicit drop below the start regardless of the floor, update it in the same -PR. - -Test: port the reporter's deterministic repro -(`rs/moq-net/tests/late_lower_group.rs` on kidq330's fork, paused time and the -mock transport) for lite-05, lite-06, and lite-07-wip, direct and through a -relay, with the in-process and moq-transport controls. Add a mixed case through -a relay: a `None` subscriber already receiving group 1, then one with a floor -of group 0, and a fresh group 0 reaches only the second. - -## Closes - -- [#4595](https://github.com/moq-dev/moq/issues/4595) - moq-lite: a group created below the first served group is silently dropped - -## Related - -- [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - once publishers name every undelivered group, a group dropped under the None floor can be named instead of vanishing diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index d4080025d9..ba16afb3ca 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1502,10 +1502,9 @@ mod test { assert_eq!(drain(&mut declared), vec![0, 1, 2]); } - /// The run_track contract behind SUBSCRIBE_OK's implicit drop: once the first - /// served group resolves the start, the cursor floor rises to it, so a lower - /// group arriving late (unordered delivery) is never served after the range - /// was declared dropped. + /// An unfloored join adopts the first served group: the cursor rises to it, so a + /// group created below it is not served. An explicit floor does not; see + /// [`explicit_floor_delivers_a_late_lower_group`]. #[tokio::test] async fn start_floor_suppresses_late_lower_arrivals() { use futures::FutureExt; @@ -1540,6 +1539,39 @@ mod test { ); } + /// The explicit-floor twin of [`start_floor_suppresses_late_lower_arrivals`]. + /// The subscription named group 0, so group 5 is still inside the floor when + /// it is created after group 7. + #[tokio::test] + async fn explicit_floor_delivers_a_late_lower_group() { + let mut producer = track_producer("test"); + let mut subscriber = producer.subscribe( + track::Subscription::default() + .with_start(track::Position::group(0)) + .with_max_age(std::time::Duration::from_secs(60)), + ); + + let write = |producer: &mut track::Producer, sequence: u64| { + let mut group = producer.create_group(crate::group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::from_millis(1).unwrap(), b"x".to_vec()) + .unwrap(); + group.finish().unwrap(); + }; + + write(&mut producer, 7); + match recv_next(&mut subscriber, false, false).await.unwrap() { + Recv::Group(group) => assert_eq!(group.sequence, 7), + _ => panic!("expected the first group"), + } + + write(&mut producer, 5); + match recv_next(&mut subscriber, false, false).await.unwrap() { + Recv::Group(group) => assert_eq!(group.sequence, 5), + _ => panic!("expected the late group above the floor"), + } + } + #[tokio::test] async fn recv_next_drains_datagram_before_finished() { let mut producer = track_producer("test"); @@ -2300,7 +2332,9 @@ impl TrackRun { } // Feed the full update into the model subscriber so the producer's // aggregate reflects it (and a relay re-forwards it upstream). - let floored = self.track.subscription().start.is_some_and(|start| start.group > 0); + // Read first: `update` replaces these preferences. + let requested = self.track.subscription().start; + let floored = requested.is_some_and(|start| start.group > 0); let bounds = Bounds::from(&upd); let _ = self.track.update(crate::track::Subscription { priority: upd.priority, @@ -2309,9 +2343,15 @@ impl TrackRun { }); // Clearing a resume floor must also expose a finished group below it. // An unfloored cursor keeps its progress and never replays served groups. + // Pre-06 an absent start means the latest group: `position_cursor` pins an + // unfloored SUBSCRIBE there, so an update that drops a named floor re-pins too. + // On an empty track the next group is the latest, whatever its sequence. match upd.start_group { Some(start_group) => self.track.start_at(start_group), None if floored && self.ctx.version.resolves_start() => self.track.start_at(0), + None if requested.is_some() && !self.ctx.version.resolves_start() => { + self.track.start_at(self.track.latest().unwrap_or(0)) + } None => {} } self.track @@ -2441,12 +2481,19 @@ impl TrackRun { // Only the group: the subscriber derives the start frame from its own request // (see `lite::SubscribeStart`). writer.buffer(&lite::SubscribeResponse::Start(lite::SubscribeStart { group: start }))?; - // SUBSCRIBE_START is an implicit drop of everything below the resolved start (the - // subscriber records it as a permanent miss), so a lower group arriving late must - // not be served after all. A widening SUBSCRIBE_UPDATE re-lowers the floor, - // renegotiating the resolved start along with the demand. Raised, not assigned: an - // update that landed while the group was held may already have raised it past. - self.track.raise_start_to(start); + // SUBSCRIBE_START names where delivery starts. Groups already skipped between + // the requested floor and this group are not served. A later group at or above an + // explicit floor, still inside the subscriber's max age, is delivered, so the cursor + // stays at that floor. + // + // A pre-06 subscription that named no group is the exception: those drafts define an + // absent Group Start as the latest group, and the resolved start becomes the + // floor. Lite-06 encodes a floor of group 0 as 0, which decodes as no named floor; + // that 0 is group 0, so it is not pinned. Raised, not assigned: an update that + // landed while the group was held may already have raised it past. + if self.track.subscription().start.is_none() && !self.ctx.version.resolves_start() { + self.track.raise_start_to(start); + } self.serve(group); Ok(()) } @@ -3289,6 +3336,92 @@ mod serve_group_test { assert_eq!(*log.writes.lock().unwrap(), [0, 1, 0, 1, 2, 3, 2]); } + /// A lite-05 subscription's run loop and its count of group streams opened. + fn lite05_run(track: track::Subscriber) -> (TrackRun, Writer, Arc) { + let opens = Arc::::default(); + let track_priority = kio::Producer::new(0u8); + let ctx = Subscription { + session: SinkSession::new(Log::default()), + id: 0, + track_name: "test".into(), + priority: PriorityQueue::default(), + track_priority: track_priority.consume(), + track_priority_seen: 0, + version: Version::Lite05, + timescale: Some(crate::Timescale::default()), + opens: opens.clone(), + }; + let bounds = Bounds { + start_group: Some(0), + start_frame: 0, + end_group: None, + end_frame: None, + }; + let writer = Writer::new(SinkSend::new(Log::default()), Version::Lite05); + (TrackRun::new(ctx, track, bounds, track_priority), writer, opens) + } + + /// Pre-06 an absent Group Start is the latest group, so an update that drops a named + /// floor re-pins the cursor there: a group created below it afterwards is not served. + #[tokio::test] + async fn a_pre06_update_dropping_the_floor_repins_to_the_latest_group() { + let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let subscriber = track.subscribe( + track::Subscription::default() + .with_start(track::Position::group(0)) + .with_max_age(Duration::from_secs(60)), + ); + let (mut run, mut writer, opens) = lite05_run(subscriber); + let mut settle = |run: &mut TrackRun| { + let res = kio::wait(|waiter| run.poll(&mut writer, waiter)).now_or_never(); + assert!(res.is_none(), "the run ended"); + opens.opened.load(Ordering::Relaxed) + }; + + write_group(&mut track, 7, 7); + assert_eq!(settle(&mut run), 1); + + run.update(lite::SubscribeUpdate { + priority: 0, + max_age: Duration::from_secs(60), + start_group: None, + end_group: None, + start_frame: 0, + end_frame: None, + }); + write_group(&mut track, 5, 5); + assert_eq!(settle(&mut run), 1, "served a group below the latest"); + + write_group(&mut track, 8, 8); + assert_eq!(settle(&mut run), 2); + } + + /// On an empty track the re-pin has no latest group to name, so the first group the + /// track produces is served, even below the dropped floor. + #[tokio::test] + async fn a_pre06_update_dropping_the_floor_on_an_empty_track_serves_the_next_group() { + let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let subscriber = track.subscribe( + track::Subscription::default() + .with_start(track::Position::group(5)) + .with_max_age(Duration::from_secs(60)), + ); + let (mut run, mut writer, opens) = lite05_run(subscriber); + run.update(lite::SubscribeUpdate { + priority: 0, + max_age: Duration::from_secs(60), + start_group: None, + end_group: None, + start_frame: 0, + end_frame: None, + }); + + write_group(&mut track, 0, 0); + let res = kio::wait(|waiter| run.poll(&mut writer, waiter)).now_or_never(); + assert!(res.is_none(), "the run ended"); + assert_eq!(opens.opened.load(Ordering::Relaxed), 1, "group 0 was not served"); + } + /// A lite-07 run over a relay's track, whose upstream subscription still waits on the /// source's SUBSCRIBE_START, on a subscribe stream the test can push updates onto. struct RelayRun { diff --git a/rs/moq-net/src/lite/subscribe.rs b/rs/moq-net/src/lite/subscribe.rs index 4656e94521..22f753b5a4 100644 --- a/rs/moq-net/src/lite/subscribe.rs +++ b/rs/moq-net/src/lite/subscribe.rs @@ -151,10 +151,10 @@ fn canonical_start_group(version: Version, start_group: Option, start_frame /// Encode the `Group Start` field shared by SUBSCRIBE and SUBSCRIBE_UPDATE. /// -/// The inverse of [`decode_start_group`]: lite-06 writes the raw floor (`None` and -/// `Some(0)` are the same absence of a constraint), while a pre-06 wire gets `Some(0)` -/// folded back to absent. On those wires an explicit group 0 means "replay from the -/// beginning", which is not what a vacuous floor asks for. +/// The inverse of [`decode_start_group`]. Lite-06 writes the raw floor, so `None` and +/// `Some(0)` are the same group 0. A pre-06 wire encodes the sequence + 1: `None` is 0 +/// (the latest group, where the publisher starts) and `Some(0)` is 1, replay from the +/// beginning. fn encode_start_group( w: &mut W, version: Version, @@ -163,7 +163,7 @@ fn encode_start_group( if version.resolves_start() { return start_group.unwrap_or(0).encode(w, version); } - start_group.filter(|&group| group > 0).encode(w, version) + start_group.encode(w, version) } /// Decode the trailing `Frame Start` / `Frame End` pair shared by SUBSCRIBE, @@ -727,8 +727,8 @@ mod test { } /// Lite06 carries the raw floor; pre-06 wires encode the sequence + 1 with 0 meaning - /// the latest group. A vacuous floor folds to absent on the old wire, where an - /// explicit group 0 would mean "replay from the beginning" instead. + /// the latest group. An explicit group 0 is that sequence, so it round-trips as group 0 + /// rather than as the latest group. #[test] fn group_start_is_absolute_on_lite06() { let mut msg = subscribe_sample(); @@ -759,13 +759,41 @@ mod test { let got = Subscribe::decode_msg(&mut zero.as_slice(), Version::Lite06).unwrap(); assert_eq!(got.start_group, None); - // On the pre-06 wire the vacuous floor folds to absent (the latest group). - let mut folded = Vec::new(); - msg.encode_msg(&mut folded, Version::Lite05).unwrap(); - let got = Subscribe::decode_msg(&mut folded.as_slice(), Version::Lite05).unwrap(); + // On the pre-06 wire an explicit group 0 is sequence + 1, distinct from an + // absent start (the latest group). + let mut explicit = Vec::new(); + msg.encode_msg(&mut explicit, Version::Lite05).unwrap(); + let got = Subscribe::decode_msg(&mut explicit.as_slice(), Version::Lite05).unwrap(); + assert_eq!(got.start_group, Some(0)); + + msg.start_group = None; + let mut latest = Vec::new(); + msg.encode_msg(&mut latest, Version::Lite05).unwrap(); + assert_ne!(explicit, latest); + let got = Subscribe::decode_msg(&mut latest.as_slice(), Version::Lite05).unwrap(); assert_eq!(got.start_group, None); } + /// The same wire values as js/net's `encodeStartGroup`: pre-06 writes `None` as 0 and + /// `Some(0)` as 1, while lite-06 writes 0 for both. + #[test] + fn group_start_wire_values_match_js() { + for (version, start_group, wire) in [ + (Version::Lite05, None, 0u8), + (Version::Lite05, Some(0), 1), + (Version::Lite05, Some(4), 5), + (Version::Lite06, None, 0), + (Version::Lite06, Some(0), 0), + (Version::Lite06, Some(4), 4), + ] { + let mut buf = Vec::new(); + encode_start_group(&mut buf, version, start_group).unwrap(); + assert_eq!(buf, [wire], "{version:?} {start_group:?}"); + let got = decode_start_group(&mut buf.as_slice(), version).unwrap(); + assert_eq!(got, start_group.or(version.resolves_start().then_some(0))); + } + } + /// A subscription can resume partway through group 0 (a catalog never leaves it), so /// a `Frame Start` qualifying the zero floor must survive the round trip. #[test] diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 94af451d85..a50a331db2 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4259,8 +4259,9 @@ impl ServeLoop { let _ = self.serving.start_at(start.group); } active.served = Some(start.group); - // The groups the SUBSCRIBE asked for below it are - // unavailable, whatever the demand asks later. + // The groups the SUBSCRIBE asked for below it are not + // waited for, whatever the demand asks later. One that + // still arrives is delivered. if let Some(requested) = active.requested && let Ok(mut tail) = active.tail.write() { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 29bb0a8915..38d5344467 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -235,8 +235,8 @@ pub(crate) struct TrackState { closed: bool, // The first sequence the live feed serves, once the publisher declared one - // (the wire's SUBSCRIBE_START). Lower groups never arrive on their own; a - // fetch can still create them. + // (the wire's SUBSCRIBE_START). Lower groups are not waited for, though one + // may still arrive late or be fetched. start_sequence: Option, // Whether `start_sequence` is only the floor a subscription asked for, still @@ -1561,8 +1561,8 @@ impl Producer { /// Declare the first group the live feed serves (the wire's SUBSCRIBE_START, /// or the start the subscription itself requested): groups below `sequence` - /// will never arrive on their own, so a reader waiting for one fails over - /// instead of stalling. A fetch can still retrieve them. + /// are not waited for, so a reader waiting for one fails over instead of + /// stalling. One may still arrive late, and a fetch can still retrieve them. /// /// Scoped to the current subscription's demand, so a later declaration /// replaces this one in either direction: a re-subscription may start diff --git a/rs/moq-net/tests/late_lower_group.rs b/rs/moq-net/tests/late_lower_group.rs new file mode 100644 index 0000000000..124ec7cf2f --- /dev/null +++ b/rs/moq-net/tests/late_lower_group.rs @@ -0,0 +1,198 @@ +//! A subscriber with an explicit floor receives a group created below the first group +//! that was served, while it is still inside `max_age`. +//! +//! An unfloored pre-06 subscribe joins where the publisher starts, so that same group +//! is dropped. Lite-06 and later encode a floor of group 0 as 0, which is the only +//! spelling of both an omitted floor and group 0; the draft says that 0 is group 0. + +mod support; + +use std::time::Duration; + +use moq_net::track::{Info, Position, Subscription}; +use moq_net::{Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +/// Long enough that two groups written one after the other are both fresh. +const BUDGET: Duration = Duration::from_secs(60); + +/// A budget no group outlives, on the publisher. +const FOREVER: Duration = Duration::from_millis((1 << 53) - 1); + +const LITE: &[&str] = &["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"]; + +const TRANSPORT: &[&str] = &["moq-transport-14", "moq-transport-17", "moq-transport-22"]; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = + moq_net::origin::Producer::new(moq_net::origin::Config::new(moq_net::Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +fn write_group(track: &moq_net::track::Producer, sequence: u64) { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + for _ in 0..4 { + group.write_frame(Timestamp::now(), &b"x"[..]).unwrap(); + } + group.finish().unwrap(); +} + +async fn read_all(sub: &mut moq_net::track::Subscriber) -> Vec<(u64, usize)> { + let mut got = Vec::new(); + loop { + let next = tokio::time::timeout(TIMEOUT, sub.recv_group()) + .await + .expect("recv_group timed out"); + let Some(mut group) = next.expect("recv_group") else { + break; + }; + let mut frames = 0; + loop { + let frame = tokio::time::timeout(TIMEOUT, group.read_frame()) + .await + .expect("read_frame timed out") + .expect("read_frame"); + match frame { + Some(_) => frames += 1, + None => break, + } + } + got.push((group.sequence, frames)); + } + got +} + +/// Group 1, then a fresh group 0, over `version`. `relay` inserts one hop. +async fn session(version: &str, relay: bool, start: Option) -> Vec<(u64, usize)> { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast + .create_track("late", Info::default().with_max_age(FOREVER)) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let relay_origin = produce_origin(2); + let upstream = match relay { + true => { + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay_origin.clone()); + Some(connect_mock(options).await) + } + false => None, + }; + + let subscriber = produce_origin(3); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(match relay { + true => relay_origin.consume(), + false => publisher.consume(), + }); + options.client_subscribe = Some(subscriber.clone()); + let pair = connect_mock(options).await; + + let consumer = subscriber.consume(); + tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let remote = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + + let subscription = Subscription::default().with_max_age(BUDGET).with_start(start); + let reader = tokio::spawn(async move { + let mut sub = remote + .track("late") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + read_all(&mut sub).await + }); + + tokio::time::timeout(TIMEOUT, track.used()) + .await + .expect("no subscriber appeared") + .unwrap(); + // Idle time lets the subscription settle, and again lets SUBSCRIBE_START land, + // before the lower group is created. + tokio::time::sleep(Duration::from_millis(10)).await; + write_group(&track, 1); + tokio::time::sleep(Duration::from_millis(10)).await; + write_group(&track, 0); + track.finish().unwrap(); + + let got = tokio::time::timeout(TIMEOUT * 3, reader) + .await + .expect("reader hung") + .expect("reader panicked"); + drop((track, pair, upstream, relay_origin, broadcast, publisher, subscriber)); + got +} + +async fn in_process(start: Option) -> Vec<(u64, usize)> { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast + .create_track("late", Info::default().with_max_age(FOREVER)) + .unwrap(); + let subscription = Subscription::default().with_max_age(BUDGET).with_start(start); + let mut sub = track.subscribe(subscription); + let reader = tokio::spawn(async move { read_all(&mut sub).await }); + tokio::time::sleep(Duration::from_millis(10)).await; + write_group(&track, 1); + tokio::time::sleep(Duration::from_millis(10)).await; + write_group(&track, 0); + track.finish().unwrap(); + let got = tokio::time::timeout(TIMEOUT * 3, reader) + .await + .expect("reader hung") + .expect("reader panicked"); + drop((track, broadcast, publisher)); + got +} + +#[tokio::test] +async fn an_explicit_floor_receives_a_group_created_below_the_first_served_group() { + tokio::time::pause(); + let want = vec![(1, 4), (0, 4)]; + let mut failures = Vec::new(); + + let got = in_process(Some(Position::group(0))).await; + if got != want { + failures.push(format!("in-process: {got:?}")); + } + + for version in LITE.iter().chain(TRANSPORT) { + for relay in [false, true] { + let got = session(version, relay, Some(Position::group(0))).await; + if got != want { + failures.push(format!("{version} relay={relay}: {got:?}")); + } + } + } + assert!(failures.is_empty(), "{failures:#?}"); +} + +/// Pre-06, an omitted floor joins at the first served group. Group 0, created after +/// group 1 was served, is below that join. Lite-06 cannot spell this separately from +/// a floor of group 0; that case is the explicit-floor test. +#[tokio::test] +async fn an_unfloored_lite05_join_drops_a_group_created_below_the_first_served_group() { + tokio::time::pause(); + let want = vec![(1, 4)]; + let mut failures = Vec::new(); + for relay in [false, true] { + let got = session("moq-lite-05", relay, None).await; + if got != want { + failures.push(format!("relay={relay}: {got:?}")); + } + } + assert!(failures.is_empty(), "{failures:#?}"); +}