diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index 9f3f7e9901b..66e7c3e5cf5 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Ordered subscriptions wait for owed groups + +An ordered reader on a subscribed track now waits at a missing group while the feed may still deliver it, up to `maxDelay` by media time or wall clock. Local tracks still skip gaps, and the zero default waits for nothing. +With `maxDelay: null`, there is no subscriber deadline. Use a finite budget for bounded waiting: a moq-transport SUBSCRIBE signals no gaps, so an unlimited wait can last until the track ends. Publisher retention still bounds the budget. + ### `maxDelay: null` sets no subscriber limit ([#7403](https://github.com/moq-dev/moq/pull/7403)) A `Track.Subscription` whose `maxDelay` is `null` reaches back as far as the publisher's `Track.Info.maxAge` keeps, as `None` does in Rust. An omitted or `undefined` `maxDelay` is still zero. diff --git a/js/net/bench/drift.ts b/js/net/bench/drift.ts index 47c95ba7a82..8224eb80a67 100644 --- a/js/net/bench/drift.ts +++ b/js/net/bench/drift.ts @@ -1,6 +1,7 @@ /** Sweep publishers, readers, and retained history for a parked frame read's drift check. */ import { Manual, Milli } from "@moq/clock"; import { type Group, Time, Track } from "@moq/net"; +import { hooks } from "../src/internal.ts"; /** Read live frames after a retained window, checking every delivered payload. */ export async function run(): Promise { @@ -42,8 +43,10 @@ export async function run(): Promise { ); const sources: Group.Producer[] = []; const consumers: Group.Consumer[] = []; + const waiting: Track.Ordered[] = []; try { for (const track of tracks) { + hooks.receiveTrack(track); for (let sequence = 0; sequence < retained; sequence++) { const group = track.appendGroup(); group.writeFrame({ @@ -73,6 +76,10 @@ export async function run(): Promise { if (!group) throw new Error("live group missing"); consumers.push(group); } + // A frameless candidate beyond one owed group keeps the ordered + // cursor parked while frames advance its presentation edge. + track.createGroup(retained + 2); + waiting.push(...ordered); } let start = 0; for (let iteration = 0; iteration < iterations + warmup; iteration++) { @@ -85,6 +92,8 @@ export async function run(): Promise { iteration, ); for (const source of sources) source.writeFrame({ payload, timestamp }); + for (const reader of waiting) + if (reader.tryNextGroup()) throw new Error("an owed hole was skipped"); for (const frame of await Promise.all(pending)) if ( frame?.payload[0] !== 7 || diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index 05b72dbd2fd..17b776c54e8 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -1260,9 +1260,9 @@ test.each([ new Reader(undefined, encodeObjects([0]), version), ); - const ordered = track.ordered(); - for (const sequence of expected) expect((await ordered.nextGroup())?.sequence).toBe(sequence); - ordered.close(); + // This tests stream framing, independent of gaps below these groups. An ordered + // subscription with this budget would wait for the unreceived groups 0..2. + for (const sequence of expected) expect((await track.recvGroup())?.sequence).toBe(sequence); track.close(); }); diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 60b46de1242..d0f5dc33c91 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -1263,6 +1263,7 @@ export class Subscriber { ietf: version, }); const producer = staged.producer; + hooks.receiveTrack(producer); const tail = new Tail(); const subscription: Subscription = { track: producer, tail, timescale: ok.properties.timescale, fetchIds: [] }; // The subscription starts after Largest, so the joining FETCH or the fill brings its diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 81dc71b81c7..1a45ef84d7b 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -410,6 +410,8 @@ export const hooks: { * the drop. Mirrors the Rust `drop_groups`; assigned by `track.ts`. */ dropTrackGroups: (producer: Producer, start: number, end: number, error: Error) => void; + /** Mark a track as fed by a wire subscription, which owes missing groups. Assigned by `track.ts`. */ + receiveTrack: (producer: Producer) => void; /** Create a received group at its wire offset before publishing it to readers. Assigned by `track.ts`. */ receiveGroup: (producer: Producer, group: { sequence: number; start: number }) => GroupProducer; /** The producer's largest reported location and its subsequent changes. */ @@ -485,10 +487,10 @@ export const hooks: { * kept for it, so a resumed reader takes groups the plain way and drops from here. */ tryDrop: (subscriber: Subscriber) => (Recv & { kind: "drop" }) | undefined; - /** The sequence {@link tryNextGroup} (`ordered`) or {@link tryRecvGroup} would take next, without taking it. */ - peekGroup: (subscriber: Subscriber, ordered: boolean) => number | undefined; + /** The lowest buffered candidate, which an ordered take may wait below; `floor` passes sequences taken on other copies. */ + peekGroup: (subscriber: Subscriber, ordered: boolean, floor?: number) => number | undefined; /** Take the next group in sequence order without waiting, as {@link tryRecvGroup} does in arrival order. */ - tryNextGroup: (subscriber: Subscriber) => Recv; + tryNextGroup: (subscriber: Subscriber, floor?: number) => Recv; /** Wake once a subscriber's datagram cursor may read differently; fires at most once. */ datagramChanged: (subscriber: Subscriber, fn: () => void) => Dispose; /** Take the next datagram in arrival order without waiting. */ @@ -766,6 +768,9 @@ export const hooks: { bindGroupCompression: () => { throw new Error("group.ts not loaded"); }, + receiveTrack: () => { + throw new Error("track.ts not loaded"); + }, receiveGroup: () => { throw new Error("track.ts not loaded"); }, diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 8fd261c5abc..fb961bfecae 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -1163,6 +1163,7 @@ export class Subscriber { request: track.Request, state: SubscribeSetup, ): SubscribeEntry { + hooks.receiveTrack(producer); // Lite-06+ resolves where the feed starts only with SUBSCRIBE_START, so a relay serving // this copy holds its first group until the answer arrives. if (resolvesStart(this.version)) hooks.startTrack(producer, { requested: msg.startGroup }); diff --git a/js/net/src/lite/tail.test.ts b/js/net/src/lite/tail.test.ts index 7c470986f37..b0dba5db99b 100644 --- a/js/net/src/lite/tail.test.ts +++ b/js/net/src/lite/tail.test.ts @@ -197,20 +197,25 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver await fin(); // Group 0 has no header, so accounting still waits, and its stream may yet - // land behind group 1's. The arrival reader holds at the hole; the sequence - // reader is already past it and ends at the boundary. + // land behind group 1's. Both readers wait, with the ordered reader holding + // group 1 until its budget gives up on the owed hole. expect((await reader.recvGroup())?.sequence).toBe(1); let ended = false; const held = reader.recvGroup().finally(() => (ended = true)); - expect(await ordered.readString()).toBe("1.0"); - expect(await ordered.readString()).toBeUndefined(); + const orderedHeld = ordered.readString(); + let orderedReady = false; + void orderedHeld.then(() => (orderedReady = true)); + await settle(); + expect(orderedReady).toBe(false); await settle(); expect(ended).toBe(false); expect(reader.closed.peek()).toBeUndefined(); expect(hooks.subscriberEnd(reader)).toEqual({ group: 2, frame: 0 }); // The grace gives up on group 0, and the settled subscription ends the reader. + expect(await clock.drive(orderedHeld)).toBe("1.0"); expect(await clock.drive(held)).toBeUndefined(); + expect(await ordered.readString()).toBeUndefined(); expect(await reader.closed).toBeNull(); ordered.close(); } finally { diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index 74030c17fb0..9d4c530740d 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -1042,6 +1042,52 @@ test("an ordered reader takes a replaced copy's buffered group before the new co b.close(); }); +for (const resolution of ["arrival", "drop", "other copy", "budget"] as const) { + test(`an ordered resumed reader keeps an older copy's owed hole until ${resolution}`, async () => { + const clock = new Manual(); + const installed = clock.install(); + const front = new Front(); + const a = new BroadcastProducer(); + const b = new BroadcastProducer(); + const ta = copy(a); + const tb = copy(b); + hooks.receiveTrack(ta.producer); + hooks.receiveTrack(tb.producer); + front.serve(a.consume()); + const ordered = front.consumer + .track("seq") + .subscribe({ maxDelay: Milli(1000) }) + .ordered(); + try { + ta.write(4); + expect((await ordered.nextGroup())?.sequence).toBe(4); + ta.write(6); + tb.write(7); + hooks.startTrack(tb.producer, { start: 7, requested: 4 }); + front.serve(b.consume()); + await settle(); + const held = ordered.nextGroup(); + expect(await isPending(held)).toBe(true); + if (resolution === "arrival") ta.write(5); + if (resolution === "other copy") { + tb.write(5); + tb.write(6); + } + if (resolution === "drop") hooks.dropTrackGroups(ta.producer, 5, 6, new StreamError(StreamCode.Old)); + if (resolution === "budget") await clock.advance(Nano.fromMilli(Milli(1000))); + const expected = resolution === "arrival" || resolution === "other copy" ? [5, 6, 7] : [6, 7]; + expect((await held)?.sequence).toBe(expected[0]); + for (const sequence of expected.slice(1)) expect((await ordered.nextGroup())?.sequence).toBe(sequence); + } finally { + ordered.close(); + front.consumer.close(); + a.close(); + b.close(); + installed.close(); + } + }); +} + test("a late subscriber's first route is checked against properties an earlier source gave", async () => { const front = new Front(); const a = new BroadcastProducer(); diff --git a/js/net/src/resume.ts b/js/net/src/resume.ts index 7d28429123a..1c4f5ddc4b2 100644 --- a/js/net/src/resume.ts +++ b/js/net/src/resume.ts @@ -281,8 +281,8 @@ function subscribe(wire: BroadcastWire, name: string, subscription: track.Subscr } /** Take a copy's next group, in sequence or arrival order. */ -function take(copy: Copy, ordered: boolean): Recv { - return ordered ? hooks.tryNextGroup(copy.sub) : hooks.tryRecvGroup(copy.sub, false); +function take(copy: Copy, ordered: boolean, floor?: number): Recv { + return ordered ? hooks.tryNextGroup(copy.sub, floor) : hooks.tryRecvGroup(copy.sub, false); } /** The first group past where `copy` was replaced: 0 for one replaced before any group. */ @@ -650,18 +650,21 @@ class Reader implements ResumeSource { // lower group that truly arrives later. const next = this.#lowest(ordered, delivered); if (next) { - const recv = take(next, ordered); + const recv = take(next, ordered, this.#newest === undefined ? undefined : this.#newest + 1); if (recv.kind === "group") { this.#observe(next, recv.group.sequence); return { kind: "group", group: this.#handOut(next, recv.group) }; } + // The lowest buffered group keeps its place while its copy waits on a hole. + // A newer copy must not pass it and permanently discard it from sequence order. + if (ordered && recv.kind === "idle") return recv; } // The serving copy first: it is the one that keeps delivering. for (let index = this.#copies.length - 1; index >= 0; index--) { const copy = this.#copies[index]; if (!this.#feeds(copy)) continue; for (;;) { - const recv = take(copy, ordered); + const recv = take(copy, ordered, this.#newest === undefined ? undefined : this.#newest + 1); if (recv.kind === "group") { if (!this.#deliverable(copy, recv.group.sequence, delivered)) { recv.group.close(); @@ -745,7 +748,11 @@ class Reader implements ResumeSource { for (const copy of this.#copies) { if (!this.#feeds(copy)) continue; for (;;) { - const sequence = hooks.peekGroup(copy.sub, ordered); + const sequence = hooks.peekGroup( + copy.sub, + ordered, + this.#newest === undefined ? undefined : this.#newest + 1, + ); if (sequence === undefined) break; if (this.#deliverable(copy, sequence, delivered)) { // A tie goes to the newer copy, so the serving one hands out its first group and @@ -753,7 +760,7 @@ class Reader implements ResumeSource { if (!lowest || sequence <= lowest.sequence) lowest = { copy, sequence }; break; } - const recv = take(copy, ordered); + const recv = take(copy, ordered, this.#newest === undefined ? undefined : this.#newest + 1); if (recv.kind !== "group") break; recv.group.close(); } diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index c3ab1e5fed3..0be32df7ab4 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -59,6 +59,291 @@ function manualTime(initial: number) { }; } +// A session-fed copy owes holes; an in-process producer does not. +function receivedTrack(info: Parameters[0] = {}) { + const producer = new TrackProducer("test").accept(info); + hooks.receiveTrack(producer); + return producer; +} + +function putGroup(producer: TrackProducer, sequence: number, millis?: number) { + const group = producer.createGroup(sequence); + if (millis !== undefined) group.writeFrame(textFrame("x", Timestamp.fromMillis(millis))); + group.finish(); +} + +function drainOrdered(reader: import("./track.ts").Ordered) { + const sequences: number[] = []; + for (let group = reader.tryNextGroup(); group; group = reader.tryNextGroup()) { + sequences.push(group.sequence); + group.close(); + } + return sequences; +} + +test("ordered reads wait only for subscription-owed groups", () => { + const clock = manualTime(0); + try { + for (const received of [false, true]) { + const producer = received ? receivedTrack() : new TrackProducer("test"); + const reader = producer.subscribe({ maxDelay: Milli(1000) }).ordered(); + putGroup(producer, 0); + putGroup(producer, 2); + expect(drainOrdered(reader)).toEqual(received ? [0] : [0, 2]); + putGroup(producer, 1); + expect(drainOrdered(reader)).toEqual(received ? [1, 2] : []); + reader.close(); + producer.finish(); + producer.close(); + } + } finally { + clock.installed.close(); + } +}); + +test("an ordered owed hole wakes exactly at its wall budget", async () => { + const clock = manualTime(0); + const producer = receivedTrack(); + const reader = producer.subscribe({ maxDelay: Milli(50) }).ordered(); + try { + putGroup(producer, 0); + expect(drainOrdered(reader)).toEqual([0]); + putGroup(producer, 2); + let resolved = false; + const pending = reader.nextGroup().then((group) => { + resolved = true; + return group; + }); + await clock.advance(49); + expect(resolved).toBe(false); + await clock.advance(50); + expect(resolved).toBe(true); + expect((await pending)?.sequence).toBe(2); + for (let sequence = 3; sequence < 10; sequence++) { + putGroup(producer, sequence); + expect(drainOrdered(reader)).toEqual([sequence]); + } + } finally { + reader.close(); + producer.finish(); + producer.close(); + clock.installed.close(); + } +}); + +test("an ordered read skipping a stale group stops waiting on holes below it", () => { + const clock = manualTime(0); + const producer = receivedTrack(); + const reader = producer.subscribe({ maxDelay: Milli(50) }).ordered(); + // A wider subscriber keeps the ledger's gaps open past this reader's budget. + const wide = producer.subscribe({ maxDelay: Milli(1000) }); + try { + putGroup(producer, 0); + expect(drainOrdered(reader)).toEqual([0]); + putGroup(producer, 3); + // Group 2 arrives late: its successor makes it stale at 50 ms, while hole 1, + // succeeded by group 2, is spared until 90 ms. + clock.set(40); + putGroup(producer, 2); + expect(drainOrdered(reader)).toEqual([]); + // Skipping group 2 moves the cursor past hole 1, so group 3 waits on nothing. + clock.set(60); + expect(drainOrdered(reader)).toEqual([3]); + } finally { + reader.close(); + wide.close(); + producer.finish(); + producer.close(); + clock.installed.close(); + } +}); + +test("an ordered owed hole follows media time and a lower successor", async () => { + const clock = manualTime(0); + const producer = receivedTrack({ timescale: Timescale.MILLI }); + const reader = producer.subscribe({ maxDelay: Milli(1000) }).ordered(); + try { + putGroup(producer, 0, 0); + expect(drainOrdered(reader)).toEqual([0]); + const edge = producer.createGroup(4); + edge.writeFrame(textFrame("edge", Timestamp.fromMillis(4000))); + expect(drainOrdered(reader)).toEqual([]); + let resolved = false; + const pending = reader.nextGroup().then((group) => { + resolved = true; + return group; + }); + // The nearer successor puts group 1's reach behind the edge immediately. + putGroup(producer, 2, 2000); + await settle(); + expect(resolved).toBe(true); + expect((await pending)?.sequence).toBe(2); + expect(drainOrdered(reader)).toEqual([]); // Group 3 is still owed. + edge.writeFrame(textFrame("within", Timestamp.fromMillis(4999))); + expect(drainOrdered(reader)).toEqual([]); + const deadline = reader.nextGroup(); + edge.writeFrame(textFrame("deadline", Timestamp.fromMillis(5000))); + await settle(); + expect((await deadline)?.sequence).toBe(4); + edge.finish(); + } finally { + reader.close(); + producer.finish(); + producer.close(); + clock.installed.close(); + } +}); + +test("an ordered owed hole wakes when its ledger gap folds", async () => { + const clock = manualTime(0); + const producer = receivedTrack(); + const reader = producer.subscribe({ maxDelay: Milli(50) }).ordered(); + try { + putGroup(producer, 0); + expect(drainOrdered(reader)).toEqual([0]); + const opener = producer.createGroup(2); + await clock.advance(40); + putGroup(producer, 3); + opener.abort(new StreamError(StreamCode.Cancel)); + let resolved = false; + const pending = reader.nextGroup().then((group) => { + resolved = true; + return group; + }); + await clock.advance(49); + expect(resolved).toBe(false); + await clock.advance(50); + expect(resolved).toBe(true); + expect((await pending)?.sequence).toBe(2); + expect(drainOrdered(reader)).toEqual([3]); + } finally { + reader.close(); + producer.finish(); + producer.close(); + clock.installed.close(); + } +}); + +test("ordered reads pass accounted sequences and honor the feed and reader floors", () => { + const clock = manualTime(0); + const producer = receivedTrack(); + const reader = producer.subscribe({ maxDelay: null }).ordered(); + try { + putGroup(producer, 0); + producer.appendDatagram(undefined, enc.encode("d")); + putGroup(producer, 2); + expect(drainOrdered(reader)).toEqual([0, 2]); + hooks.dropTrackGroups(producer, 3, 4, new StreamError(StreamCode.Old)); + putGroup(producer, 4); + expect(drainOrdered(reader)).toEqual([4]); + producer.createGroup(5).abort(new StreamError(StreamCode.Cancel)); + putGroup(producer, 6); + expect(drainOrdered(reader)).toEqual([5, 6]); + hooks.startTrack(producer, { start: 8 }); + putGroup(producer, 8); + expect(drainOrdered(reader)).toEqual([8]); + putGroup(producer, 10); + expect(drainOrdered(reader)).toEqual([]); + reader.withRange({ start: { included: 10 } }); + expect(drainOrdered(reader)).toEqual([10]); + } finally { + reader.close(); + producer.finish(); + producer.close(); + clock.installed.close(); + } +}); + +test("ordered owed holes follow zero, narrowed, unlimited, and publisher budgets", async () => { + const clock = manualTime(0); + try { + for (const maxDelay of [Milli.zero, Milli(1000), null]) { + const producer = receivedTrack(); + const reader = producer.subscribe({ maxDelay }).ordered(); + putGroup(producer, 0); + expect(drainOrdered(reader)).toEqual([0]); + putGroup(producer, 2); + expect(drainOrdered(reader)).toEqual(maxDelay === 0 ? [2] : []); + if (maxDelay === null) { + await clock.advance(3_600_000); + expect(drainOrdered(reader)).toEqual([]); + putGroup(producer, 1); + expect(drainOrdered(reader)).toEqual([1, 2]); + } else if (maxDelay > 0) { + const pending = reader.nextGroup(); + reader.update({ maxDelay: Milli.zero }); + expect((await pending)?.sequence).toBe(2); + } + reader.close(); + producer.finish(); + producer.close(); + } + const producer = receivedTrack({ maxAge: Milli(50) }); + const reader = producer.subscribe({ maxDelay: null }).ordered(); + putGroup(producer, 0); + putGroup(producer, 2); + expect(drainOrdered(reader)).toEqual([0]); + const pending = reader.nextGroup(); + await clock.advance(3_600_050); + expect((await pending)?.sequence).toBe(2); + reader.close(); + producer.finish(); + producer.close(); + } finally { + clock.installed.close(); + } +}); + +test("ordered owed waits wake on a datagram, a drop, a declared start, or track close", async () => { + const clock = manualTime(0); + try { + for (const resolve of ["datagram", "drop", "start", "close"] as const) { + const producer = receivedTrack(); + const reader = producer.subscribe({ maxDelay: null }).ordered(); + putGroup(producer, 1); + // With no declared start the first arrival is the floor. + expect(drainOrdered(reader)).toEqual([1]); + hooks.startTrack(producer, { start: 0 }); + putGroup(producer, 3); + let ready = false; + const held = reader.nextGroup().then((group) => { + ready = true; + return group; + }); + await settle(); + expect(ready).toBe(false); + // A datagram shares the sequence namespace, so it accounts for the hole too. + if (resolve === "datagram") producer.insertDatagram(2, undefined, new Uint8Array([2])); + if (resolve === "drop") hooks.dropTrackGroups(producer, 2, 3, new StreamError(StreamCode.Old)); + if (resolve === "start") hooks.startTrack(producer, { start: 3 }); + if (resolve === "close") { + producer.finish(); + producer.close(); + } + await settle(); + expect(ready).toBe(true); + expect((await held)?.sequence).toBe(3); + reader.close(); + if (resolve !== "close") { + producer.finish(); + producer.close(); + } + } + const producer = receivedTrack(); + hooks.startTrack(producer, { start: 0 }); + const reader = producer.subscribe({ maxDelay: null }).ordered(); + putGroup(producer, 1); + expect(drainOrdered(reader)).toEqual([]); + putGroup(producer, 0); + expect(drainOrdered(reader)).toEqual([0, 1]); + reader.close(); + producer.finish(); + producer.close(); + } finally { + clock.installed.close(); + } +}); + test("wanted reflects subscriber demand and clears when the last one leaves", async () => { const producer = new TrackProducer("test"); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 290d25aea5b..9cb6a95da89 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -464,8 +464,7 @@ export class Consumer { * {@link Subscription.maxDelay} is what asks for data: delivery skips everything above * the floor that the budget convicts. Without options this is {@link Subscription.default}, * whose budget of zero delivers only the latest group; a larger one reaches back over what - * it can still use, and an options object without `maxDelay` reaches back as far as the - * publisher retains. + * it can still use, and `maxDelay: null` reaches back as far as the publisher retains. */ subscribe(options?: Subscription): Subscriber { return this.#broadcast.subscribe(this.name, options); @@ -542,6 +541,8 @@ class TrackState { // First timestamps mutate group state rather than track state, so held groups // watch this revision as well as arrivals when enforcing latency after handoff. timelineChanged = new Signal(0); + // Frames wake the sequence cursor only while an owed hole holds it. + waitingOrdered = false; // Bumped when a wire subscription declares or resolves where its live feed starts, so a // publisher holding its first group for that answer wakes. startChanged = new Signal(0); @@ -647,6 +648,7 @@ let reofferCached: ( let reofferSequences: (producer: Producer, sink: TrackState, sequences: readonly number[]) => void; // Whether a group may still land on the producer's live feed; assigned in its static block. let mayLand: (producer: Producer, land: Land) => boolean; +let owedGroup: (producer: Producer, start: number, end: number) => { sequence: number; folds?: number } | undefined; let sourceStart: (producer: Producer) => number | "pending" | undefined; // Actual object metadata outlives cached payloads, unlike empty group heads in model largest. let objectLargest: (producer: Producer) => Position | undefined; @@ -801,6 +803,8 @@ export class Producer { // The first group the wire subscription's live feed serves (see `hooks.startTrack`), unset // until its START resolves it, null for a resolved-empty range, and the requested floor. #served?: { start?: number | null; requested?: number }; + // Only copies fed by a wire subscription owe missing groups. + #subscribed = false; // The sequences the live feed delivered, as a group or a datagram: what a pending tail is // judged by. Eviction never touches it. A gap folds away a subscription's grace after a // later arrival opened it, as the session's tail does, so this holds only the gaps still @@ -813,6 +817,9 @@ export class Producer { } static { + hooks.receiveTrack = (producer) => { + producer.#subscribed = true; + }; hooks.receiveGroup = (producer, { sequence, start }) => producer.#createGroup(sequence, start); hooks.writeDatagram = (producer, datagram, inflated) => producer.#insertDatagram(datagram, inflated); sourceStart = (producer) => producer.#served && producer.#feedStart(); @@ -864,17 +871,28 @@ export class Producer { if (entry && !held.has(sequence)) producer.#mirror(entry, sink); } }; - mayLand = (producer, land) => producer.#mayLand(land); + mayLand = (producer, land) => producer.#landable(land) !== undefined; + owedGroup = (producer, start, end) => { + if (!producer.#subscribed) return undefined; + const feed = producer.#feedStart(); + const floor = feed === "pending" ? producer.#served?.requested : feed; + return producer.#landable({ + start: Math.max(start, floor ?? 0), + end, + floor: floor !== undefined, + grace: producer.#grace(), + }); + }; } - // Whether a group may still land from the live feed, by the rule a pending tail is judged + // The first group that may still land from the live feed, by the rule a pending tail is judged // by; see {@link Land}. Mirrors the Rust `TrackState::may_land`. - #mayLand(land: Land): boolean { - if (this.#state.closed.peek() !== undefined) return false; + #landable(land: Land): { sequence: number; folds?: number } | undefined { + if (this.#state.closed.peek() !== undefined) return undefined; let start = land.start; if (!land.floor) { const first = this.#arrivals.first(); - if (first === undefined) return false; + if (first === undefined) return undefined; start = Math.max(start, first); } start = Math.max(start, this.#served?.start ?? 0); @@ -883,13 +901,14 @@ export class Producer { const at = now(); const limit = Nano.fromMilli(land.grace); const drops = this.#state.drops.peek(); - return this.#arrivals - .gaps(start, end) - .some( - (gap) => - (gap.since === undefined || Instant.durationSince(at, gap.since) < limit) && - runGaps(drops, gap.start, gap.end).some((open) => !land.delivered?.covers(open.start, open.end)), - ); + for (const gap of this.#arrivals.gaps(start, end)) { + const folds = gap.since === undefined ? undefined : gap.since + limit; + if (folds !== undefined && at >= folds) continue; + for (const open of runGaps(drops, gap.start, gap.end)) { + if (!land.delivered?.covers(open.start, open.end)) return { sequence: open.start, folds }; + } + } + return undefined; } // Record `[start, end)` given up with `error`. An upstream drop (`open`) also ends the @@ -1008,8 +1027,7 @@ export class Producer { * {@link Subscription.maxDelay} is what asks for data: delivery skips everything above * the floor that the budget convicts. Without options this is {@link Subscription.default}, * whose budget of zero delivers only the latest group; a larger one reaches back over what - * it can still use, and an options object without `maxDelay` reaches back as far as the - * publisher retains. + * it can still use, and `maxDelay: null` reaches back as far as the publisher retains. */ subscribe(options: Subscription = Subscription.default()): Subscriber { const sink = new TrackState(options); @@ -1371,9 +1389,14 @@ export class Producer { // Follow the group's frames as they land, past the head `#advance` already named. const stop = hooks.groupProgress(group, (total) => { if (total > hooks.groupOffset(group)) this.#advanceObject({ group: group.sequence, frame: total - 1 }); + // A frame can change an owed hole's reach or drift edge without adding a group. + for (const sink of this.#sinks) + if (sink.waitingOrdered) sink.timelineChanged.update((revision) => revision + 1); }); void group.closed.then((closed) => { stop(); + for (const sink of this.#sinks) + if (sink.waitingOrdered) sink.timelineChanged.update((revision) => revision + 1); if (closed === null) this.#finishLargest(group.sequence); }); // Give held mirrors the new live edge before pruning their timeline entry, @@ -2196,8 +2219,14 @@ export class Subscriber { }; hooks.tryRecvGroup = (subscriber, dispositions) => subscriber.#tryRecvGroup(dispositions); hooks.tryDrop = (subscriber) => subscriber.#tryDrop(); - hooks.tryNextGroup = (subscriber) => subscriber.#tryNextGroup(); - hooks.peekGroup = (subscriber, ordered) => subscriber.#peekGroup(ordered); + hooks.tryNextGroup = (subscriber, floor) => { + subscriber.#nextSequence = Math.max(subscriber.#nextSequence, floor ?? 0); + return subscriber.#tryNextGroup(); + }; + hooks.peekGroup = (subscriber, ordered, floor) => { + if (ordered) subscriber.#nextSequence = Math.max(subscriber.#nextSequence, floor ?? 0); + return subscriber.#peekGroup(ordered); + }; hooks.forgetPastBudget = (subscriber, delivered) => subscriber.#forgetPastBudget(delivered); hooks.unreport = (subscriber, start, end) => { subscriber.#reported = removeRun(subscriber.#reported, start, end); @@ -2499,6 +2528,7 @@ export class Subscriber { })); } this.#closed = true; + this.#state.waitingOrdered = false; // Settle if we're first (the producer may already have); either way drop anything // still buffered. Groups parked at the withRange cap deliberately outlive a clean // producer close, so the subscriber leaving is what must release them: closing @@ -2715,6 +2745,14 @@ export class Subscriber { this.#state.received.changed(fn), this.#state.owed.changed(fn), ]; + if (this.#state.waitingOrdered) + dispose.push( + this.#state.timelineChanged.changed(fn), + this.#wall.revision.changed(fn), + this.#state.info.changed(fn), + // A datagram shares the sequence namespace, so it can fill the owed hole. + this.#state.datagrams.changed(fn), + ); return () => { for (const close of dispose) close(); }; @@ -2818,8 +2856,20 @@ export class Subscriber { } } - // The sequence #tryNextGroup (`ordered`) or #tryRecvGroup would take next: the lowest buffered - // group in range, not yet handed out, that the budget spares. + // A missing group the subscription still owes holds sequence order until it lands, + // is accounted for, or the same drift budget as a buffered group gives up on it. + #waitOwed(start: number, end: number, drift: Drift): boolean { + const producer = this.#state.producer; + if (start >= end || !producer) return false; + const owed = owedGroup(producer, start, end); + if (!owed || this.#staleAt(owed.sequence, drift, true)) return false; + if (owed.folds !== undefined) this.#wakeAt(owed.folds); + this.#state.waitingOrdered = true; + return true; + } + + // The lowest buffered candidate in range that the budget spares. An ordered take may + // still wait on an owed hole below it, so a resumed reader must keep its place. #peekGroup(ordered: boolean): number | undefined { const cursor = this.#cursor.peek(); const start = ordered ? Math.max(cursor.start, this.#nextSequence) : cursor.start; @@ -2827,7 +2877,9 @@ export class Subscriber { for (const group of this.#state.groups.peek()) { if (group.sequence < start) continue; if (cursor.end !== undefined && group.sequence >= cursor.end) return undefined; - if (!this.#delivered.has(group.sequence) && !this.#staleAt(group.sequence, drift)) return group.sequence; + if (!this.#delivered.has(group.sequence) && !this.#staleAt(group.sequence, drift)) { + return group.sequence; + } } return undefined; } @@ -2845,9 +2897,10 @@ export class Subscriber { } #tryNextBuffered(): Recv { + this.#state.waitingOrdered = false; const groups = this.#state.groups.peek(); const cursor = this.#cursor.peek(); - const start = Math.max(cursor.start, this.#nextSequence); + let start = Math.max(cursor.start, this.#nextSequence); while (groups.length > 0 && groups[0].sequence < start) groups.shift()?.close(); // One anchor for the whole pass, so walking a backlog off stays linear. const drift = this.#drift(); @@ -2855,14 +2908,20 @@ export class Subscriber { for (;;) { const group = groups[0]; if (!group || (cursor.end !== undefined && group.sequence >= cursor.end)) break; - groups.shift(); - this.#nextSequence = group.sequence + 1; // Every frame this group could still hold is past the budget, so keep // scanning: one pass walks a whole backlog off rather than replaying it. if (this.#staleAt(group.sequence, drift)) { + groups.shift(); + this.#nextSequence = group.sequence + 1; + // Holes below a skipped group are undeliverable, even ones the wall clock + // still spares because their successor arrived late. + start = this.#nextSequence; group.close(); continue; } + if (this.#waitOwed(start, group.sequence, drift)) return { kind: "idle" }; + groups.shift(); + this.#nextSequence = group.sequence + 1; this.#handOut(group.sequence, true); return { kind: "group", group: this.#guard(group) }; } @@ -3000,9 +3059,9 @@ export class Subscriber { * group already handed out too: a stalled group's pending frame read rejects once newer * content has pulled that far ahead. * - * A history or VOD read sets `maxDelay: null` (no limit). This reader skips a missing - * group rather than waiting for it, but the Rust reader waits for one a subscription still - * owes, within the budget, or until it resolves without one. An unlimited or long budget + * A history or VOD read sets `maxDelay: null` (no limit). A subscribed track waits for a + * missing group the feed still owes, within the budget, or until it resolves without one. + * Local tracks skip missing groups. An unlimited or long budget * only avoids blocking where the feed signals its gaps: moq-lite, whose publisher sends * `SUBSCRIBE_DROP` for every group it won't deliver, or an ordered moq-transport FETCH. * Over a moq-transport SUBSCRIBE, which signals no gaps, that wait can last until the diff --git a/quest/m1/README.md b/quest/m1/README.md index e9e6d9570c2..ca69e71f06f 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -179,7 +179,6 @@ per-frame CPU growth since #5926. - [A follower is told when the serving epochless route leaves](/quest/m1/epochless-leave-restart.md) - wire-level tests pin the restart a follower gets when the serving epochless route is withdrawn or dies while another remains - [Poll-first async](/quest/m1/poll-first/README.md) - library crates compile no coroutines by default: async wrappers are generated from `poll_*` and gated behind an `async` feature - [Authored tracks default to a 30 s max-age](/quest/m1/max-age-defaults.md) - raw and media tracks advertise 30 s unless their author chooses otherwise, with explicit unlimited in every binding -- [JS ordered reads wait only on owed groups](/quest/m1/js-ordered-owed.md) - @moq/net matches Rust's owed-group wait - [HLS export follows a resolution change](/quest/m1/hls-reinit-same-codec.md) - a decoder-identity change under the same codec string starts a new init and segment tag - [moq-quic deletes its remaining 0-RTT code](/quest/m1/quic-zero-rtt-leftovers.md) - only the code that drops a 0-RTT packet remains, with a test - [Gate the no-em-dash rule](/quest/m1/em-dash-gate.md) - `just check` fails on an em dash in any tracked text file diff --git a/quest/m1/js-ordered-owed.md b/quest/m1/js-ordered-owed.md deleted file mode 100644 index 8feb7a02cbe..00000000000 --- a/quest/m1/js-ordered-owed.md +++ /dev/null @@ -1,21 +0,0 @@ -# [S] JS ordered reads wait only on owed groups - -## Goal - -`@moq/net`'s ordered reader on a subscribed track waits at a hole while the feed may still deliver that group, bounded by its max delay, as Rust's does since [moq-dev/moq#7403](https://github.com/moq-dev/moq/pull/7403). It no longer skips the gap, and local tracks still skip. - -## Plan - -Decided 2026-10-30 (maintainer, follow-up planning): m1, for parity with Rust before the cut. Port #7403's rule rather than invent a JS one: - -- Only subscription-fed copies owe groups. -- An unlimited budget waits until the hole resolves, with no grace period. -- A finite budget gives up on the same media-time and wall-clock rule. - -Port #7403's Rust tests, using JS's manual clock. - -Public API: none. Wire: none. - -## Related - -- [Ordered reads wait only on owed groups](https://github.com/moq-dev/moq/pull/7403) - its PR, #7403, is the Rust rule this ports