From 15754209e5212aaa55dd39eee5025b7091d47357 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 19:35:32 -0800 Subject: [PATCH 1/6] chore(quest): claim quest/m1/js-ordered-owed.md Quest-Claim: 45cd66197d1df7cf From d68c978ed9b241aff8636c31290e892edc98e4d5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 19:43:03 -0800 Subject: [PATCH 2/6] fix(net): wait for subscription-owed groups in JS ordered reads Co-Authored-By: GPT-6.1 Sol --- quest/m1/README.md | 1 - quest/m1/js-ordered-owed.md | 21 --------------------- 2 files changed, 22 deletions(-) delete mode 100644 quest/m1/js-ordered-owed.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 55660210155..5a6b281379c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -185,7 +185,6 @@ per-frame CPU growth since #5926. - [JS join tests cover draft 20](/quest/m1/join-tests-draft20.md) - join.test.ts runs on every supported draft, not just 18 and 22 - [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 - [Audit retention defaults](/quest/m1/max-age-defaults.md) - every max-age default listed, with a proposed finite default for the maintainer -- [JS ordered reads wait only on owed groups](/quest/m1/js-ordered-owed.md) - @moq/net matches Rust's owed-group wait - [Binding announcements always carry captures](/quest/m1/ffi-required-captures.md) - moq-ffi and its generated bindings make `captures` a required list, as #7266 always sets it - [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 - [A JS reader finishes after a clean close](/quest/m1/js-close-range-finish.md) - a range ending below the cache delivers what's held and then ends 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 From 4ce37a18e16e435e441df797bd5143a22a0f1c34 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 19:43:14 -0800 Subject: [PATCH 3/6] fix(net): port subscription-owed waiting to the JS cursor Co-Authored-By: GPT-6.1 Sol --- js/net/MIGRATE.md | 5 + js/net/bench/drift.ts | 9 + js/net/src/ietf/subscriber.test.ts | 6 +- js/net/src/ietf/subscriber.ts | 1 + js/net/src/internal.ts | 5 + js/net/src/lite/subscriber.ts | 1 + js/net/src/lite/tail.test.ts | 13 +- js/net/src/track.test.ts | 256 +++++++++++++++++++++++++++++ js/net/src/track.ts | 84 +++++++--- 9 files changed, 350 insertions(+), 30 deletions(-) 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..3c9400d4f1b 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. */ @@ -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 9d804d38882..5f5f2151818 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/track.test.ts b/js/net/src/track.test.ts index 25a2d0602a4..6f3bf0913b4 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -59,6 +59,262 @@ 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 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 drop, a declared start, or track close", async () => { + const clock = manualTime(0); + try { + for (const resolve of ["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); + 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 00644fb663f..394dae6ebe6 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); @@ -638,6 +637,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; @@ -792,6 +792,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 @@ -804,6 +806,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(); @@ -855,17 +860,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); @@ -874,13 +890,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 @@ -999,8 +1016,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); @@ -1122,7 +1138,6 @@ export class Producer { if (entry.unverified) hooks.unverifyGroup(dst); entry.mirrors.set(sink, dst); timelineInsert(sink.timeline, dst); - void dst.readable().then(() => sink.timelineChanged.update((revision) => revision + 1)); sink.latest = Math.max(sink.latest ?? 0, dst.sequence); sink.received.set(Math.max(sink.received.peek(), dst.sequence + 1)); sink.groups.mutate((groups) => { @@ -1365,9 +1380,12 @@ 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) sink.timelineChanged.update((revision) => revision + 1); }); void group.closed.then((closed) => { stop(); + for (const sink of this.#sinks) 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, @@ -2701,6 +2719,9 @@ export class Subscriber { this.#state.groups.changed(fn), this.#state.drops.changed(fn), this.#state.startChanged.changed(fn), + this.#state.timelineChanged.changed(fn), + this.#wall.revision.changed(fn), + this.#state.info.changed(fn), this.#cursor.changed(fn), // A widened budget re-offers the groups it skipped as stale. this.#state.update.changed(fn), @@ -2812,6 +2833,17 @@ export class Subscriber { } } + // 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); + return true; + } + // The sequence #tryNextGroup (`ordered`) or #tryRecvGroup would take next: the lowest buffered // group in range, not yet handed out, that the budget spares. #peekGroup(ordered: boolean): number | undefined { @@ -2821,7 +2853,10 @@ 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)) { + if (ordered && this.#waitOwed(start, group.sequence, drift)) return undefined; + return group.sequence; + } } return undefined; } @@ -2849,14 +2884,17 @@ 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; 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) }; } @@ -2998,9 +3036,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 From 50aea47d898b50ca9bb0a84ba1f21ed8ae0fa0c0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 19:51:33 -0800 Subject: [PATCH 4/6] fix(net): preserve owed waits across route copies Co-Authored-By: GPT-6.1 Sol --- js/net/src/internal.ts | 6 ++--- js/net/src/resume.test.ts | 46 +++++++++++++++++++++++++++++++++++++++ js/net/src/resume.ts | 19 +++++++++++----- js/net/src/track.ts | 36 +++++++++++++++++++++--------- 4 files changed, 88 insertions(+), 19 deletions(-) diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 3c9400d4f1b..1a45ef84d7b 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -487,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. */ 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.ts b/js/net/src/track.ts index 394dae6ebe6..1f36f648063 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -532,6 +532,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); @@ -1138,6 +1140,7 @@ export class Producer { if (entry.unverified) hooks.unverifyGroup(dst); entry.mirrors.set(sink, dst); timelineInsert(sink.timeline, dst); + void dst.readable().then(() => sink.timelineChanged.update((revision) => revision + 1)); sink.latest = Math.max(sink.latest ?? 0, dst.sequence); sink.received.set(Math.max(sink.received.peek(), dst.sequence + 1)); sink.groups.mutate((groups) => { @@ -1381,11 +1384,13 @@ export class Producer { 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) sink.timelineChanged.update((revision) => revision + 1); + 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) sink.timelineChanged.update((revision) => revision + 1); + 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, @@ -2208,8 +2213,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); @@ -2511,6 +2522,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 @@ -2719,9 +2731,6 @@ export class Subscriber { this.#state.groups.changed(fn), this.#state.drops.changed(fn), this.#state.startChanged.changed(fn), - this.#state.timelineChanged.changed(fn), - this.#wall.revision.changed(fn), - this.#state.info.changed(fn), this.#cursor.changed(fn), // A widened budget re-offers the groups it skipped as stale. this.#state.update.changed(fn), @@ -2730,6 +2739,12 @@ 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), + ); return () => { for (const close of dispose) close(); }; @@ -2841,11 +2856,12 @@ export class Subscriber { 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 sequence #tryNextGroup (`ordered`) or #tryRecvGroup would take next: the lowest buffered - // group in range, not yet handed out, that the budget spares. + // 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; @@ -2854,7 +2870,6 @@ export class Subscriber { 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)) { - if (ordered && this.#waitOwed(start, group.sequence, drift)) return undefined; return group.sequence; } } @@ -2874,6 +2889,7 @@ 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); From a9339ef6e6c6e90628239bef9e8c3711a8d07c56 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 20:53:52 -0800 Subject: [PATCH 5/6] fix(net): wake owed ordered reads when a datagram fills the hole Co-Authored-By: Claude Opus 5.5 --- js/net/src/track.test.ts | 6 ++++-- js/net/src/track.ts | 2 ++ 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 6f3bf0913b4..e2a4bb839a9 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -267,10 +267,10 @@ test("ordered owed holes follow zero, narrowed, unlimited, and publisher budgets } }); -test("ordered owed waits wake on a drop, a declared start, or track close", async () => { +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 ["drop", "start", "close"] as const) { + for (const resolve of ["datagram", "drop", "start", "close"] as const) { const producer = receivedTrack(); const reader = producer.subscribe({ maxDelay: null }).ordered(); putGroup(producer, 1); @@ -285,6 +285,8 @@ test("ordered owed waits wake on a drop, a declared start, or track close", asyn }); 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") { diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 1f36f648063..43bb857d188 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -2744,6 +2744,8 @@ export class Subscriber { 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(); From 2236e48c0932d85b3dd426baca4965265e0b2557 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 09:57:20 -0800 Subject: [PATCH 6/6] fix(net): stop waiting on owed holes below a skipped stale group An ordered pass that skips a stale group moves its cursor past every hole below it, but kept checking those holes for owed waits. Wall-clock staleness is not monotone in sequence: a group arriving after its successor can go stale while a lower hole, succeeded by that late group, is still spared, so the next candidate parked on a hole it can no longer deliver. Co-Authored-By: Claude Opus 5.5 --- js/net/src/track.test.ts | 27 +++++++++++++++++++++++++++ js/net/src/track.ts | 5 ++++- 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index a9120f88a59..0be32df7ab4 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -131,6 +131,33 @@ test("an ordered owed hole wakes exactly at its wall budget", async () => { } }); +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 }); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 49b9ed8285c..9cb6a95da89 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -2900,7 +2900,7 @@ export class Subscriber { 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(); @@ -2913,6 +2913,9 @@ export class Subscriber { 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; }