From a1031680d22fd2c32a03142f595cb0915ace50c3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 8 Oct 2026 16:29:57 -0700 Subject: [PATCH 1/9] quest: claim js-track-takeover Co-Authored-By: Claude Opus 5.5 From e995f52248e3241cc1241df75b5c8fa1cd4f600c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 8 Oct 2026 16:39:06 -0700 Subject: [PATCH 2/9] fix(net): JS createTrack takes over a queued request and continues sequences createTrack answers a request still queued for its name, and createTrack and insertTrack continue the name's group and datagram sequences within the broadcast, as Rust's create_track does. A broadcast remembers written sequences for up to 1024 names; unwritten, unheld names are swept and a new name past the bound is refused. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 2 +- js/net/src/broadcast.test.ts | 119 ++++++++++++++++++ js/net/src/broadcast.ts | 74 ++++++++--- js/net/src/internal.ts | 43 ++++++- js/net/src/track.ts | 35 ++++-- quest/m0/README.md | 1 - .../publish-catalog-restart.md | 14 +-- quest/m0/js-track-takeover.md | 30 ----- 8 files changed, 246 insertions(+), 72 deletions(-) delete mode 100644 quest/m0/js-track-takeover.md diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 8db354e665..6786e0d5ad 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -38,7 +38,7 @@ broadcast.announce(); - **One connection per URL, unless you opt out.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. Supplying your own transport options, discovery, or origin selects a private loop. - **GOAWAY moves the session.** The connection dials the replacement at once while the old session keeps serving its groups, up to `goaway.handover` (default 10s, or the relay's deadline when that is sooner). New requests keep opening on the old session until the replacement's route outranks it. An empty GOAWAY redials the same URL. A redirect stays on the same host unless `goaway.redirect` is `"follow"`. - **One send estimate per connection.** `Bandwidth.Allocator` divides it by track priority, max-min fair within a tier. An idle track claims nothing. Publishers reserve against it so their targets sum to the estimate. -- **One track per name.** Concurrent subscriptions share one request and producer. An on-demand producer that replaces an ended one continues the name's group and datagram sequences; only a new broadcast restarts them. `createTrack` and `insertTrack` start at 0. +- **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. A track that replaces an ended one, created or requested, continues the name's group and datagram sequences; only a new broadcast restarts them. A broadcast remembers the sequences of up to 1024 names, and creating a track under a new name past that throws. - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Unlike Rust, omitting it means untimed. - **Subscriber staleness is `maxDelay`.** It is media time. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts). diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index 39d66f044a..c6d80a0b93 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -3,6 +3,7 @@ import { Signal } from "@moq/signals"; import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import { GroupTooLarge, NotFound } from "./error.ts"; import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts"; +import { MAX_TRACK_SEQUENCES } from "./internal.ts"; import { Milli, Timescale, Timestamp } from "./time.ts"; import type { Request as TrackRequest } from "./track.ts"; import { Producer as TrackProducer } from "./track.ts"; @@ -92,6 +93,124 @@ test("dynamic track sequences continue across producer replacements", async () = nextGeneration.close(); }); +test("createTrack answers a queued request", async () => { + const broadcast = new BroadcastProducer(); + const pulled = wireOf(broadcast).requested(); + const info = broadcast.track("media").info(); + const subscriber = broadcast.track("media").subscribe().ordered(); + + // The track is created before any handler takes the queued request. + const producer = broadcast.createTrack("media", { timescale: Timescale.MILLI, priority: 3 }); + producer.writeString("hello"); + expect(await subscriber.readString()).toBe("hello"); + expect((await info).priority).toBe(3); + + // No handler ever sees the request, and the lookup that opened it leaves the track open. + expect(await pendingRequest(broadcast)).toBeUndefined(); + subscriber.close(); + await broadcast.demand().unused(); + expect(producer.closed.peek()).toBeUndefined(); + + // Later subscriptions read the created track, and the name stays taken. + const again = broadcast.track("media").subscribe().ordered(); + producer.writeString("again"); + expect(await again.readString()).toBe("again"); + expect(() => broadcast.createTrack("media")).toThrow("duplicate track"); + + broadcast.close(); + expect(await pulled).toBeUndefined(); +}); + +test("a taken request is not answered by insertTrack", async () => { + const broadcast = new BroadcastProducer(); + const pulled = wireOf(broadcast).requested(); + broadcast.track("media").subscribe(); + const request = await pulled; + if (!request) throw new Error("expected request"); + + // The handler holds it, so the name is taken until it answers. + expect(() => broadcast.createTrack("media")).toThrow("duplicate track"); + expect(() => broadcast.insertTrack(new TrackProducer("media"))).toThrow("duplicate track"); + request.reject(); + broadcast.close(); +}); + +test("created and requested tracks share a name's sequences", async () => { + const broadcast = new BroadcastProducer(); + const first = broadcast.createTrack("media", { timescale: Timescale.MILLI }); + expect(first.appendGroup().sequence).toBe(0); + first.writeGroup(new GroupProducer(8)); + first.insertDatagram(12, Timestamp.fromMillis(0), new Uint8Array()); + first.close(); + + // A replacement continues past every explicit write. + const second = broadcast.createTrack("media", { timescale: Timescale.MILLI }); + expect(second.appendGroup().sequence).toBe(13); + second.close(); + + // An on-demand producer continues the same namespace, and so does a later inserted track. + broadcast.removeTrack("media"); + const pulled = wireOf(broadcast).requested(); + const subscriber = broadcast.track("media").subscribe(); + const request = await pulled; + if (!request) throw new Error("expected request"); + const requested = request.accept({ timescale: Timescale.MILLI }); + expect(requested.appendDatagram(Timestamp.fromMillis(0), new Uint8Array())).toBe(14); + subscriber.close(); + requested.close(); + + const inserted = new TrackProducer("media").accept({ timescale: Timescale.MILLI }); + broadcast.insertTrack(inserted); + expect(inserted.appendGroup().sequence).toBe(15); + + // Re-inserting the same track keeps its place in the namespace. + broadcast.removeTrack("media"); + broadcast.insertTrack(inserted); + expect(inserted.appendGroup().sequence).toBe(16); + inserted.close(); + + // Only a new broadcast starts over. + const next = new BroadcastProducer(); + expect(next.createTrack("media", { timescale: Timescale.MILLI }).appendGroup().sequence).toBe(0); + broadcast.close(); + next.close(); +}); + +test("a broadcast bounds the names it remembers sequences for", async () => { + const broadcast = new BroadcastProducer(); + + // Names that never wrote leave nothing behind once their tracks close. + for (let i = 0; i < MAX_TRACK_SEQUENCES * 2; i++) { + broadcast.createTrack(`unused-${i}`).close(); + } + + // Written names are remembered, and so is an open track that has not written yet. + const open = broadcast.createTrack("open"); + for (let i = 1; i < MAX_TRACK_SEQUENCES; i++) { + const track = broadcast.createTrack(`written-${i}`); + track.appendGroup().close(); + track.close(); + } + expect(() => broadcast.createTrack("refused")).toThrow("too many track names"); + expect(() => broadcast.insertTrack(new TrackProducer("refused"))).toThrow("too many track names"); + + // A remembered name is still admitted, continuing its sequences. + expect(broadcast.createTrack("written-1").appendGroup().sequence).toBe(1); + + // An on-demand request past the bound is refused to its subscribers. + const pulled = wireOf(broadcast).requested(); + const subscriber = broadcast.track("refused").subscribe(); + const request = await pulled; + if (!request) throw new Error("expected request"); + expect(() => request.accept()).toThrow("too many track names"); + expect(await subscriber.closed).toBeInstanceOf(Error); + + // Closing the open track without a write frees its slot. + open.close(); + broadcast.createTrack("admitted"); + broadcast.close(); +}); + test("publishing-side subscriptions coalesce onto one request", async () => { const broadcast = new BroadcastProducer(); const pulled = wireOf(broadcast).requested(); diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index 8526ec6be0..77941c4c0d 100644 --- a/js/net/src/broadcast.ts +++ b/js/net/src/broadcast.ts @@ -8,7 +8,7 @@ import type * as Epoch from "./epoch.ts"; import { NotFound } from "./error.ts"; import type { Consumer as GroupConsumer } from "./group.ts"; import { Route } from "./hop.ts"; -import { hooks, type TrackSequence } from "./internal.ts"; +import { hooks, TrackSequences } from "./internal.ts"; import * as Path from "./path.ts"; import * as track from "./track.ts"; import { untilAborted } from "./util/abort.ts"; @@ -57,7 +57,8 @@ class BroadcastState { // publisher that has started pulling them. The counterpart of a live Rust `broadcast::Dynamic`; // without one, a track nobody publishes is `NotFound` rather than a request nobody will answer. served = false; - sequences = new Map(); + // Each name's sequence namespace, shared by every producer of the name, created or requested. + sequences = new TrackSequences(); // Live consumer handles sharing this state (see {@link Consumer.clone}). The broadcast // closes once the last one closes, so a shared consumer can be handed to several callers. consumers = 0; @@ -279,6 +280,31 @@ async function fetchGroup( } } +// Remove and return the request queued for `name`, if no handler has taken it yet. +function takeQueued(state: BroadcastState, name: string): track.Request | undefined { + const queued = state.requested.peek().find((request) => request.name === name); + if (!queued) return undefined; + state.requested.mutate((requests) => { + requests.splice(requests.indexOf(queued), 1); + }); + return queued; +} + +// Serve `producer` as the broadcast's track for its name. +function adopt(state: BroadcastState, producer: track.Producer): void { + const lookups = state.lookups.get(producer) ?? new Lookups(); + state.lookups.set(producer, lookups); + watchDemand(state, producer, lookups); + state.tracks.set(producer.name, producer); + + // A finished track keeps serving its cache, so only an abort evicts it. + void producer.closed.then((closed) => { + if (closed instanceof Error && state.tracks.get(producer.name) === producer) { + state.tracks.delete(producer.name); + } + }); +} + let makeDemand: (state: BroadcastState) => Demand; /** A watch-only view of demand for any track in a broadcast. */ @@ -357,33 +383,51 @@ export class Producer { return makeConsumer({ state: this.#state, path: this.#path }); } - /** Insert a track that is served directly, without an on-demand request round-trip. */ + /** + * Insert a track that is served directly, without an on-demand request round-trip. + * + * It continues the name's group and datagram sequences from any earlier track of the name in + * this broadcast. Throws if an open track or a request already holds the name, or if the + * broadcast cannot remember another track name's sequences. + */ insertTrack(track: track.Producer): void { if (this.#state.closed.peek() !== undefined) { throw new Error("broadcast is closed"); } // One logical track per name: an open inserted track or a live request already serves it. + // A queued request's subscribers wait on its own producer, so only createTrack can answer it. const live = [this.#state.tracks.get(track.name), this.#state.requests.get(track.name)]; if (live.some((existing) => existing && existing.closed.peek() === undefined)) { throw new Error(`duplicate track: ${track.name}`); } - const lookups = this.#state.lookups.get(track) ?? new Lookups(); - this.#state.lookups.set(track, lookups); - watchDemand(this.#state, track, lookups); - this.#state.tracks.set(track.name, track); - - // A finished track keeps serving its cache, so only an abort evicts it. - void track.closed.then((closed) => { - if (closed instanceof Error && this.#state.tracks.get(track.name) === track) { - this.#state.tracks.delete(track.name); - } - }); + hooks.bindSequence(track, this.#state.sequences.admit(track.name)); + adopt(this.#state, track); } - /** Create a track, insert it into the broadcast, and return its producer. */ + /** + * Create a track, insert it into the broadcast, and return its producer. + * + * A request still queued for the name is answered with this track, so its subscribers read + * from the returned producer. Otherwise it behaves as {@link insertTrack}. + */ createTrack(name: string, info: Partial = {}): track.Producer { + if (this.#state.closed.peek() !== undefined) { + throw new Error("broadcast is closed"); + } + + const request = takeQueued(this.#state, name); + if (request) { + const producer = request.accept(info); + // The application owns it now: a lookup that opened the request must not close it. + const lookups = this.#state.lookups.get(producer); + if (lookups) lookups.opened = false; + this.#state.requests.delete(name); + adopt(this.#state, producer); + return producer; + } + const producer = new track.Producer(name).accept(info); this.insertTrack(producer); return producer; diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 0fa1f7f737..1fa52438a4 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -114,13 +114,45 @@ export interface ReadGroupFrame { complete(): void; } -/** The next group or datagram sequence, shared by every dynamic producer that serves one broadcast track in turn. */ +/** The next group or datagram sequence, shared by every producer that serves one broadcast track in turn. */ export interface TrackSequence { next: number; + /** Open producers writing through this namespace. */ + holders: number; } -/** Per-track sequence namespaces owned by one broadcast generation. */ -export type TrackSequences = Map; +/** The most track names one broadcast remembers sequences for. */ +export const MAX_TRACK_SEQUENCES = 1024; + +/** + * Each name's sequence namespace for one broadcast generation, outliving the producers that + * serve it so a replacement continues where the last one stopped. Written namespaces are kept + * until the broadcast ends, since forgetting one would let a later producer reuse its sequences. + * Namespaces nothing holds or wrote are swept once the map fills, and a new name past that bound + * is refused: starting sequences over takes a new broadcast. + */ +export class TrackSequences { + #map = new Map(); + + /** The namespace for `name`, admitting the name if it is new. Throws once the bound is reached. */ + admit(name: string): TrackSequence { + const existing = this.#map.get(name); + if (existing) return existing; + + if (this.#map.size >= MAX_TRACK_SEQUENCES) { + for (const [key, sequence] of this.#map) { + if (sequence.next === 0 && sequence.holders === 0) this.#map.delete(key); + } + if (this.#map.size >= MAX_TRACK_SEQUENCES) { + throw new Error(`too many track names: ${name} would exceed ${MAX_TRACK_SEQUENCES}`); + } + } + + const sequence = { next: 0, holders: 0 }; + this.#map.set(name, sequence); + return sequence; + } +} /** Inputs for creating a package-internal track request. */ export interface TrackRequestOptions { @@ -140,6 +172,8 @@ export const hooks: { makeRequest: (options: TrackRequestOptions) => Request; /** Access the existing producer while a request awaits immutable wire metadata. */ pendingTrackProducer: (request: Request) => Producer; + /** Have a producer write through a broadcast's sequence namespace for its name. */ + bindSequence: (producer: Producer, sequence: TrackSequence) => void; /** * Take the next group the subscriber's cursor allows, without waiting; assigned by `track.ts`. * @@ -196,6 +230,9 @@ export const hooks: { pendingTrackProducer: () => { throw new Error("track.ts not loaded"); }, + bindSequence: () => { + throw new Error("track.ts not loaded"); + }, tryRecvGroup: () => { throw new Error("track.ts not loaded"); }, diff --git a/js/net/src/track.ts b/js/net/src/track.ts index fcb950141e..4747ec415c 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -242,10 +242,20 @@ export class Request { return this.subscription.priority ?? 0; } - /** Accept the request, committing the track's immutable {@link Info}. */ + /** + * Accept the request, committing the track's immutable {@link Info}. Throws, rejecting the + * request, if the broadcast cannot remember another track name's sequences. + */ accept(info: Partial = {}): Producer { this.#pending.delete(this); - bindProducer(this.name, this.#producer, this.#sequences); + let sequence: TrackSequence; + try { + sequence = this.#sequences.admit(this.name); + } catch (err) { + this.#producer.close(err instanceof Error ? err : new Error(String(err))); + throw err; + } + bindProducerSequence(this.#producer, sequence); return this.#producer.accept(info); } @@ -409,15 +419,6 @@ async function resolveInfo(state: TrackState): Promise { // so eviction can drop them together. type CachedGroup = { group: GroupProducer; mirrors: Map }; -function bindProducer(name: string, producer: Producer, sequences: TrackSequences): void { - let shared = sequences.get(name); - if (!shared) { - shared = { next: 0 }; - sequences.set(name, shared); - } - bindProducerSequence(producer, shared); -} - let bindProducerSequence: (producer: Producer, sequence: TrackSequence) => void; // The sequence-order cursor lives inside `Subscriber` (it shares the group buffer and the @@ -508,7 +509,8 @@ export class Producer { // The producer's own state is the source of truth (info/closed); subscribers // read mirrored sinks, never this state directly. #state = new TrackState(); - #sequence: TrackSequence = { next: 0 }; + // Until a broadcast binds it, a namespace of its own. An open producer holds its namespace. + #sequence: TrackSequence = { next: 0, holders: 1 }; // One past the highest group or datagram this producer received, like the Rust // `max_sequence`. The shared counter above can run ahead of it: earlier producers of // the same broadcast track advanced it too. @@ -545,9 +547,17 @@ export class Producer { } static { + // Anything written before binding still counts, so the namespace never goes backwards. bindProducerSequence = (producer, sequence) => { + const own = producer.#sequence; + if (own === sequence) return; + sequence.next = Math.max(sequence.next, own.next); producer.#sequence = sequence; + if (producer.#state.closed.peek() !== undefined) return; + own.holders--; + sequence.holders++; }; + hooks.bindSequence = bindProducerSequence; } /** @@ -1015,6 +1025,7 @@ export class Producer { // Not evicted: a reader already holding one keeps its frames and sees the abort. const open = abort ? this.#cache.filter((entry) => entry.group.closed.peek() === undefined) : []; closeTrackState(this.#state, abort); + this.#sequence.holders--; for (const { group } of this.#cache) group.close(abort); for (const entry of open) { this.#unlink(entry); diff --git a/quest/m0/README.md b/quest/m0/README.md index 1a36af6821..04cc561a80 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -70,6 +70,5 @@ remain and no release waits on them. - [FFI publisher stall](/quest/m0/ffi-publisher-stall.md) - every Go and Python publisher cell passes reliably once the serve budget lands, and a cell fails when a connection idles out - [web-transport releases the qmux fixes](/quest/m0/qmux-credit-upstream.md) - waiting on moq-dev/web-transport#412 and #413 to merge and ship, which qmux credit bumps to - [qmux credit](/quest/m0/qmux-credit.md) - qmux returns connection credit for dropped and stopped streams and delivers its close frame, on both lines -- [JS track takeover](/quest/m0/js-track-takeover.md) - JS `createTrack` answers a queued request and continues its sequences, as Rust does, so a re-announced `@moq/publish` catalog never restarts its groups - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - every first-party publisher that can restart mints a fresh route epoch, the newest wins a path, and only routes with the same epoch resume a subscription - [moq-noq 2.0.2 on main](/quest/m0/noq-2.0.2.md) - `main` carries the max datagram size fix `release` gets in 1.3.4 diff --git a/quest/m0/broadcast-epoch/publish-catalog-restart.md b/quest/m0/broadcast-epoch/publish-catalog-restart.md index 251f0a0c3d..8d1d38090c 100644 --- a/quest/m0/broadcast-epoch/publish-catalog-restart.md +++ b/quest/m0/broadcast-epoch/publish-catalog-restart.md @@ -22,15 +22,9 @@ re-announce. `js/publish/src/broadcast.ts` mints the epoch once and keeps the publisher across unannounce and announce on purpose, which points at continuing the sequence. -Decided 2026-10-08: the `@moq/net` fix belongs to -[JS track takeover](/quest/m0/js-track-takeover.md), so every JS publisher -continues a name's sequences; this quest does not fix `createTrack` itself. -Re-run the test once it lands and fix `@moq/publish` only if it still -restarts. +Decided 2026-10-08: the `@moq/net` fix landed separately: `createTrack` and +`insertTrack` now continue a name's sequences within a broadcast, so this +quest does not fix `createTrack` itself. Run the test against it and fix +`@moq/publish` only if it still restarts. Public API: none expected. Wire: none. - -## Required - -- [JS track takeover](/quest/m0/js-track-takeover.md) - JS `createTrack` and `insertTrack` continue a name's sequences, which fixes the restart at its source - diff --git a/quest/m0/js-track-takeover.md b/quest/m0/js-track-takeover.md deleted file mode 100644 index 84ca2b8c1b..0000000000 --- a/quest/m0/js-track-takeover.md +++ /dev/null @@ -1,30 +0,0 @@ -# [S] JS createTrack takes over a queued request - -## Goal - -In `@moq/net`, `createTrack` and `insertTrack` on a name with a queued or -pending request answer that request and continue the name's group and -datagram sequences, as Rust's `create_track` does, instead of throwing -`duplicate track`. The per-name sequence map stays bounded without forgetting -sequences that were written within the broadcast. - -## Plan - -Decided 2026-10-08: moved to m0, since -[publish catalog restart](/quest/m0/broadcast-epoch/publish-catalog-restart.md) -gates the release and its fix is this one. - -Decided 2026-10-06 while settling #4929: that PR coalesces JS publishing-side -subscriptions per name and accepts the `duplicate track` throw for now; this -quest brings JS to parity. Drop unused entries only when no track holds them -and they never saw a write. Written sequence state survives until the -broadcast ends: evicting it and restarting under the same name would reuse -content identities and stall resumed readers. - -Bound admission of additional names if retaining their sequence state would -exceed the map's bound; settle the refusal surface during implementation. -Starting sequences over requires a new broadcast. Test queued-request -takeover, explicit sequence advancement, replacement continuity, and bounded -name admission. Update the JS net docs for the behavior change. - -Public API: behavior change in `@moq/net` only. Wire: none. From df2bb1786437632bebab58529ad466acbb982d4d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 11:19:08 -0700 Subject: [PATCH 3/9] fix(net)!: drop insertTrack from @moq/net createTrack is the only way to add a track by hand. It still answers a queued request, continues the name's sequences, and refuses a new name past 1024 with a plain Error. A request a handler already pulled stays duplicate. Co-Authored-By: Grok 4.7 --- js/net/examples/publish.ts | 4 +- js/net/src/broadcast.test.ts | 73 +++---------------- js/net/src/broadcast.ts | 38 ++++------ .../publish-catalog-restart.md | 8 +- quest/m1/README.md | 1 - quest/m1/js-insert-track.md | 28 ------- 6 files changed, 30 insertions(+), 122 deletions(-) delete mode 100644 quest/m1/js-insert-track.md diff --git a/js/net/examples/publish.ts b/js/net/examples/publish.ts index b3bf6f23cb..e9c38f6d26 100644 --- a/js/net/examples/publish.ts +++ b/js/net/examples/publish.ts @@ -11,8 +11,8 @@ async function main() { const broadcast = origin.createBroadcast(Moq.Path.from("my-broadcast")); broadcast.announce(); - // Insert the "chat" track up front. A subscriber is served directly from this - // track, no requested() round-trip needed. Mirrors the Rust createTrack/insertTrack. + // Create the "chat" track up front. A subscriber is served directly from this + // track, no requested() round-trip needed. Mirrors Rust's create_track. // Each frame is stamped with the time it was written, so the track declares a timescale. void publishTrack(broadcast.createTrack("chat", { timescale: Moq.Time.Timescale.MILLI })); console.log("Published broadcast: my-broadcast"); diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index 4621149863..bd6b3c9a91 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -1,5 +1,4 @@ -import { expect, setSystemTime, spyOn, test } from "bun:test"; -import { Signal } from "@moq/signals"; +import { expect, setSystemTime, test } from "bun:test"; import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import { GroupTooLarge, NotFound } from "./error.ts"; import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts"; @@ -7,7 +6,6 @@ import { MAX_TRACK_SEQUENCES } from "./internal.ts"; import { textFrame } from "./mock.ts"; import { Milli, Timescale, Timestamp } from "./time.ts"; import type { Request as TrackRequest } from "./track.ts"; -import { Producer as TrackProducer } from "./track.ts"; import { wireOf } from "./wire.ts"; // The public API mints consumers internally (Producer.consume, the wire layers); tests act @@ -122,16 +120,15 @@ test("createTrack answers a queued request", async () => { expect(await pulled).toBeUndefined(); }); -test("a taken request is not answered by insertTrack", async () => { +test("createTrack does not take over a request a handler already pulled", async () => { const broadcast = new BroadcastProducer(); const pulled = wireOf(broadcast).requested(); broadcast.track("media").subscribe(); const request = await pulled; if (!request) throw new Error("expected request"); - // The handler holds it, so the name is taken until it answers. + // The handler holds it, so the name stays duplicate until it answers, as in Rust. expect(() => broadcast.createTrack("media")).toThrow("duplicate track"); - expect(() => broadcast.insertTrack(new TrackProducer("media"))).toThrow("duplicate track"); request.reject(); broadcast.close(); }); @@ -149,7 +146,7 @@ test("created and requested tracks share a name's sequences", async () => { expect(second.appendGroup().sequence).toBe(13); second.close(); - // An on-demand producer continues the same namespace, and so does a later inserted track. + // An on-demand producer continues the same namespace, and so does a later created track. broadcast.removeTrack("media"); const pulled = wireOf(broadcast).requested(); const subscriber = broadcast.track("media").subscribe(); @@ -160,15 +157,9 @@ test("created and requested tracks share a name's sequences", async () => { subscriber.close(); requested.close(); - const inserted = new TrackProducer("media").accept({ timescale: Timescale.MILLI }); - broadcast.insertTrack(inserted); - expect(inserted.appendGroup().sequence).toBe(15); - - // Re-inserting the same track keeps its place in the namespace. - broadcast.removeTrack("media"); - broadcast.insertTrack(inserted); - expect(inserted.appendGroup().sequence).toBe(16); - inserted.close(); + const later = broadcast.createTrack("media", { timescale: Timescale.MILLI }); + expect(later.appendGroup().sequence).toBe(15); + later.close(); // Only a new broadcast starts over. const next = new BroadcastProducer(); @@ -193,7 +184,6 @@ test("a broadcast bounds the names it remembers sequences for", async () => { track.close(); } expect(() => broadcast.createTrack("refused")).toThrow("too many track names"); - expect(() => broadcast.insertTrack(new TrackProducer("refused"))).toThrow("too many track names"); // A remembered name is still admitted, continuing its sequences. expect(broadcast.createTrack("written-1").appendGroup().sequence).toBe(1); @@ -355,14 +345,6 @@ test("a held info lookup ending after removeTrack leaves demand alone", async () await Promise.resolve(); expect(demand.used.peek()).toBe(false); - // Re-inserting the track counts it again. - broadcast.insertTrack(track); - expect(demand.used.peek()).toBe(false); - const resubscriber = track.subscribe(); - await Promise.resolve(); - expect(demand.used.peek()).toBe(true); - - resubscriber.close(); track.close(); broadcast.close(); }); @@ -491,8 +473,7 @@ test("consumer track subscriptions fan out and close independently", async () => test("subscribe serves a statically inserted track without a request", async () => { const broadcast = new BroadcastProducer(); - const track1 = new TrackProducer("track1").accept({ timescale: Timescale.MILLI }); - broadcast.insertTrack(track1); + const track1 = broadcast.createTrack("track1", { timescale: Timescale.MILLI }); track1.appendGroup().close(); // The track already exists, so subscribe resolves immediately (no requested()). @@ -503,8 +484,7 @@ test("subscribe serves a statically inserted track without a request", async () expect(await pendingRequest(broadcast)).toBeUndefined(); // A second static track behaves the same. - const track2 = new TrackProducer("track2").accept({ timescale: Timescale.MILLI }); - broadcast.insertTrack(track2); + const track2 = broadcast.createTrack("track2", { timescale: Timescale.MILLI }); const sub2 = broadcast.track("track2").subscribe().ordered(); track2.appendGroup().close(); @@ -615,10 +595,10 @@ test("createTrack commits info up front", async () => { expect(info.priority).toBe(3); }); -test("insertTrack rejects a duplicate live name", () => { +test("createTrack rejects a duplicate live name", () => { const broadcast = new BroadcastProducer(); broadcast.createTrack("dup", { timescale: Timescale.MILLI }); - expect(() => broadcast.insertTrack(new TrackProducer("dup").accept({ timescale: Timescale.MILLI }))).toThrow(); + expect(() => broadcast.createTrack("dup", { timescale: Timescale.MILLI })).toThrow("duplicate track"); }); test("a finished track is still served from its cache", async () => { @@ -808,38 +788,7 @@ test("removeTrack stops counting the removed track's demand at once", async () = broadcast.removeTrack("video"); expect(demand.used.peek()).toBe(false); - // Re-inserting the same track counts it again. - broadcast.insertTrack(video); - expect(demand.used.peek()).toBe(true); - subscriber.close(); video.close(); broadcast.close(); }); - -test("inserting a closed track leaves no demand watcher behind", () => { - const broadcast = new BroadcastProducer(); - const track = new TrackProducer("video").accept({ timescale: Timescale.MILLI }); - track.close(); - - // Count the listeners insertTrack attaches and never disposes. - let live = 0; - const subscribe = Signal.prototype.subscribe; - const spy = spyOn(Signal.prototype, "subscribe").mockImplementation(function (this: Signal, fn) { - live++; - const dispose = subscribe.call(this, fn); - return () => { - live--; - dispose(); - }; - }); - try { - broadcast.insertTrack(track); - } finally { - spy.mockRestore(); - } - - expect(live).toBe(0); - expect(broadcast.demand().used.peek()).toBe(false); - broadcast.close(); -}); diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index 77941c4c0d..e40eec9150 100644 --- a/js/net/src/broadcast.ts +++ b/js/net/src/broadcast.ts @@ -383,34 +383,14 @@ export class Producer { return makeConsumer({ state: this.#state, path: this.#path }); } - /** - * Insert a track that is served directly, without an on-demand request round-trip. - * - * It continues the name's group and datagram sequences from any earlier track of the name in - * this broadcast. Throws if an open track or a request already holds the name, or if the - * broadcast cannot remember another track name's sequences. - */ - insertTrack(track: track.Producer): void { - if (this.#state.closed.peek() !== undefined) { - throw new Error("broadcast is closed"); - } - - // One logical track per name: an open inserted track or a live request already serves it. - // A queued request's subscribers wait on its own producer, so only createTrack can answer it. - const live = [this.#state.tracks.get(track.name), this.#state.requests.get(track.name)]; - if (live.some((existing) => existing && existing.closed.peek() === undefined)) { - throw new Error(`duplicate track: ${track.name}`); - } - - hooks.bindSequence(track, this.#state.sequences.admit(track.name)); - adopt(this.#state, track); - } - /** * Create a track, insert it into the broadcast, and return its producer. * * A request still queued for the name is answered with this track, so its subscribers read - * from the returned producer. Otherwise it behaves as {@link insertTrack}. + * from the returned producer. A request a handler has already taken stays a duplicate. + * The track continues the name's group and datagram sequences from any earlier track of the + * name in this broadcast. Throws if an open track or a live request already holds the name, + * or if the broadcast cannot remember another track name's sequences. */ createTrack(name: string, info: Partial = {}): track.Producer { if (this.#state.closed.peek() !== undefined) { @@ -428,8 +408,16 @@ export class Producer { return producer; } + // One logical track per name: an open track or a live request already serves it. + // A request a handler already pulled stays duplicate, as in Rust. + const live = [this.#state.tracks.get(name), this.#state.requests.get(name)]; + if (live.some((existing) => existing && existing.closed.peek() === undefined)) { + throw new Error(`duplicate track: ${name}`); + } + const producer = new track.Producer(name).accept(info); - this.insertTrack(producer); + hooks.bindSequence(producer, this.#state.sequences.admit(name)); + adopt(this.#state, producer); return producer; } diff --git a/quest/m0/broadcast-epoch/publish-catalog-restart.md b/quest/m0/broadcast-epoch/publish-catalog-restart.md index 8d1d38090c..2f72262849 100644 --- a/quest/m0/broadcast-epoch/publish-catalog-restart.md +++ b/quest/m0/broadcast-epoch/publish-catalog-restart.md @@ -22,9 +22,9 @@ re-announce. `js/publish/src/broadcast.ts` mints the epoch once and keeps the publisher across unannounce and announce on purpose, which points at continuing the sequence. -Decided 2026-10-08: the `@moq/net` fix landed separately: `createTrack` and -`insertTrack` now continue a name's sequences within a broadcast, so this -quest does not fix `createTrack` itself. Run the test against it and fix -`@moq/publish` only if it still restarts. +Decided 2026-10-08: the `@moq/net` fix landed separately: `createTrack` now +continues a name's sequences within a broadcast, so this quest does not fix +`createTrack` itself. Run the test against it and fix `@moq/publish` only if +it still restarts. Public API: none expected. Wire: none. diff --git a/quest/m1/README.md b/quest/m1/README.md index 12a90f84eb..e99a582b0d 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -133,7 +133,6 @@ were done or not worth their cost. - [iroh keep-alive](/quest/m1/iroh-keep-alive.md) - the iroh backend honors `quic.keep_alive` - [Rust papercuts](/quest/m1/papercuts-rs.md) - HTTP refusals are counted, and a `u64::MAX` resume is unbounded - [JS papercuts](/quest/m1/papercuts-js.md) - IETF status 0 keeps the subgroup open, a muted rendition change settles, and bad element attributes warn -- [Drop insertTrack](/quest/m1/js-insert-track.md) - `createTrack` is `@moq/net`'s only way to add a track by hand, as in Rust - [JS subgroup heads race close](/quest/m1/js-head-race.md) - a stalled publisher no longer pins a subgroup reader past unsubscribe - [Front parking](/quest/m1/origin-front-parks.md) - an unroutable request waits on a front instead of re-asking on every route-table move - [Route wakes](/quest/m1/route-wakes.md) - a route change wakes only the fronts it can move, so pool churn stops scaling with served paths diff --git a/quest/m1/js-insert-track.md b/quest/m1/js-insert-track.md deleted file mode 100644 index 23435af2ae..0000000000 --- a/quest/m1/js-insert-track.md +++ /dev/null @@ -1,28 +0,0 @@ -# [XS] @moq/net drops insertTrack - -## Goal - -`BroadcastProducer.insertTrack` is gone from `@moq/net`; `createTrack` is the -only way to add a track by hand, mirroring Rust's `create_track` (Rust has -`create_track` and `reserve_track`, no insert). It has no production caller. - -## Plan - -Decided 2026-10-08, from js-track-takeover's -[#5068](https://github.com/moq-dev/moq/pull/5068) report: `insertTrack` takes -a caller-built producer, so it can't take over a queued request (the waiting -subscribers belong to the request's own producer) and keeps throwing -`duplicate track` where `createTrack` answers. Delete it rather than keep a -second, weaker path. Rejected: making it internal, and keeping it public with -docs. - -Inline the method into `createTrack`, move the tests that call it onto -`createTrack`, and fix the stale comment in `js/net/examples/publish.ts` -("Mirrors the Rust createTrack/insertTrack") and the `insertTrack` mention in -`doc/lib/js/net.md`. - -Public API: breaking (`insertTrack` removed). Wire: none. - -## Required - -- [JS createTrack takes over a queued request](/quest/m0/js-track-takeover.md) - edits the same `broadcast.ts` methods From 611c59063786bf40422776926ccc068045f974e0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 16:25:46 -0700 Subject: [PATCH 4/9] fix(net)!: the application owns a replacement track's start sequence An append on a producer that replaces an ended one under the same name no longer silently continues the name's sequences. It refuses (Rust Error::Duplicate, JS "duplicate sequence") when it would reuse a sequence the name already wrote; the application starts the replacement with an explicit create_group/writeGroup. Explicit writes stay unchecked so relays keep forwarding upstream numbering. @moq/publish now keeps each track for its network broadcast's life, so a re-announce or rendition change continues the same producer, and a maxAge change publishes a new network broadcast. Absorbs the publish catalog restart quest. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 2 +- doc/lib/js/publish.md | 6 +- doc/lib/rs/moq-net.md | 2 +- js/net/src/broadcast.test.ts | 31 +++--- js/net/src/broadcast.ts | 9 +- js/net/src/internal.ts | 10 +- js/net/src/track.ts | 64 +++++++------ js/publish/src/broadcast.test.ts | 94 ++++++++++++++++++- js/publish/src/broadcast.ts | 83 +++++++++------- quest/m0/broadcast-epoch/README.md | 1 - .../publish-catalog-restart.md | 30 ------ rs/moq-net/src/model/broadcast.rs | 38 +++++--- rs/moq-net/src/model/track.rs | 32 ++++--- rs/moq-net/tests/route_change.rs | 14 ++- 14 files changed, 272 insertions(+), 144 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/publish-catalog-restart.md diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 8b708ac865..d66299a367 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -39,7 +39,7 @@ broadcast.announce(); - **One connection per URL, unless you opt out.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. Supplying your own transport options, discovery, or origin selects a private loop. - **GOAWAY moves the session.** The connection dials the replacement at once while the old session keeps serving its groups, up to `goaway.handover` (default 10s, or the relay's deadline when that is sooner). New requests keep opening on the old session until the replacement's route outranks it, which is a restart without an epoch: a request on the old session ends when that session closes, so request the path again. An empty GOAWAY redials the same URL. A redirect stays on the same host unless `goaway.redirect` is `"follow"`. - **One send estimate per connection.** `Bandwidth.Allocator` divides it by track priority, max-min fair within a tier. An idle track claims nothing. Publishers reserve against it so their targets sum to the estimate. -- **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. A track that replaces an ended one, created or requested, continues the name's group and datagram sequences; only a new broadcast restarts them. A broadcast remembers the sequences of up to 1024 names, and creating a track under a new name past that throws. +- **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: a track that replaces an ended one, created or requested, throws on an append that would reuse a sequence the name already wrote, so it writes its first group with an explicit sequence. Only a new broadcast starts over. A broadcast remembers the sequences of up to 1024 names, and creating a track under a new name past that throws. - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Unlike Rust, omitting it means untimed. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` and `@moq/flate` producers. - **Subscriber staleness is `maxDelay`.** It is media time. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts). diff --git a/doc/lib/js/publish.md b/doc/lib/js/publish.md index 54c002ca8c..9375518fe2 100644 --- a/doc/lib/js/publish.md +++ b/doc/lib/js/publish.md @@ -66,8 +66,10 @@ timeline. The catalog's `clock` maps it to wall time. Stamp your own tracks ## Custom tracks `broadcast.net` is the underlying `Moq.Broadcast.Producer`, so an application -can serve its own tracks alongside the media. It is recreated on each -reconnect, so acquire it from an effect and reseed the track each time: +can serve its own tracks alongside the media. It is recreated when the origin, name, +`enabled`, or `maxAge` changes, so acquire it from an effect and create each +track once per `net`. A track name always means the same content, so a track +re-created on the same `net` throws on an append that would reuse its sequences: ```ts import * as Json from "@moq/json"; diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index c4ead51ffd..e4af60c165 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -18,7 +18,7 @@ What you use it for, beyond what the concept page already describes: - **An origin outlives the session.** Broadcasts are created on the origin, and a reconnect announces them again. Closing the session does not delete them. - **Requests can pin an epoch.** `consumer.request_broadcast(path, Some(epoch))` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs); `None` takes whichever route wins. The resolved consumer names its epoch in `info().epoch`, pinned or not. - **Follow restarts.** `consumer.follow(path)` reduces the routes covering a path to the one a request resolves (the most specific) and yields its start, update, restart, and end. A restart says another publisher instance now serves the path: request it again. A broadcast already resolved stays on the old instance until dropped or its route goes. -- **One track per name.** Concurrent subscriptions share one request and producer. A producer that replaces an ended one continues the name's group and datagram sequences; only a new broadcast restarts them. Readers of an aborted track resume onto the replacement only when the broadcast announces an epoch; otherwise their subscription ends with the error. +- **One track per name.** Concurrent subscriptions share one request and producer. The application owns sequence numbers: a producer that replaces an ended one gets `Error::Duplicate` from an append that would reuse a sequence the name already wrote, so it starts with `create_group` at an explicit sequence. Only a new broadcast starts over. Readers of an aborted track resume onto the replacement only when the broadcast announces an epoch; otherwise their subscription ends with the error. - **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast consumer is not demand. Holding a track consumer is, so drop one you are not reading. A shared fetch stays up until its last reader leaves. - **You drive the session, or `moq-tokio` does.** `connect` and `accept` return a session plus a driver that never reads the clock itself. `moq_net::time::run` polls it on tokio or in the browser. `moq-tokio` and `moq-wasm` hide that. A custom transport implements `moq_net::transport::poll`. - **A session caps what its peer can hold.** `session::Limits` (via `Client::with_limits` and `Server::with_limits`) bounds announces and subscriptions per session; the defaults suit a relay mesh, so lower them for untrusted peers. Past either, the session closes with `TOO_MANY_REQUESTS`. On moq-transport drafts 14 to 16 they also size the request-ID window, which every request counts against, so very low limits can starve it. Peer-declared lengths are capped before they are buffered. diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index bd6b3c9a91..15bfa71729 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -54,9 +54,9 @@ test("consumer dedupes repeat subscriptions onto one upstream request", async () expect((await pendingRequest(consumer))?.name).toBe("video"); }); -// Shared sequence namespace across producer replacements: the TypeScript equivalent of -// cloned Rust producers, so insertDatagram advances the next append for the replacement. -test("dynamic track sequences continue across producer replacements", async () => { +// A name keeps one sequence namespace across producer replacements. The application owns where a +// replacement starts: an append that would reuse an earlier producer's sequence throws. +test("a replacement producer refuses to append a sequence the name already wrote", async () => { const broadcast = new BroadcastProducer(); // Pull before subscribing: a broadcast nobody serves on demand answers NotFound instead. @@ -76,7 +76,14 @@ test("dynamic track sequences continue across producer replacements", async () = const secondRequest = await wireOf(broadcast).requested(); if (!secondRequest) throw new Error("expected second request"); const secondProducer = secondRequest.accept({ timescale: Timescale.MILLI }); - expect(secondProducer.appendGroup().sequence).toBe(13); + expect(() => secondProducer.appendGroup()).toThrow("duplicate sequence"); + expect(() => secondProducer.appendDatagram(Timestamp.fromMillis(0), new Uint8Array())).toThrow( + "duplicate sequence", + ); + + // An explicit first write picks the start, and appends continue from there. + secondProducer.writeGroup(new GroupProducer(13)); + expect(secondProducer.appendGroup().sequence).toBe(14); const nextGeneration = new BroadcastProducer(); const nextPull = wireOf(nextGeneration).requested(); @@ -141,24 +148,26 @@ test("created and requested tracks share a name's sequences", async () => { first.insertDatagram(12, Timestamp.fromMillis(0), new Uint8Array()); first.close(); - // A replacement continues past every explicit write. + // A replacement refuses to append past nothing, since that would reuse every explicit write. const second = broadcast.createTrack("media", { timescale: Timescale.MILLI }); - expect(second.appendGroup().sequence).toBe(13); + expect(() => second.appendGroup()).toThrow("duplicate sequence"); + second.writeGroup(new GroupProducer(13)); second.close(); - // An on-demand producer continues the same namespace, and so does a later created track. + // An on-demand producer checks the same namespace, and so does a later created track. broadcast.removeTrack("media"); const pulled = wireOf(broadcast).requested(); const subscriber = broadcast.track("media").subscribe(); const request = await pulled; if (!request) throw new Error("expected request"); const requested = request.accept({ timescale: Timescale.MILLI }); - expect(requested.appendDatagram(Timestamp.fromMillis(0), new Uint8Array())).toBe(14); + expect(() => requested.appendDatagram(Timestamp.fromMillis(0), new Uint8Array())).toThrow("duplicate sequence"); + requested.insertDatagram(14, Timestamp.fromMillis(0), new Uint8Array()); subscriber.close(); requested.close(); const later = broadcast.createTrack("media", { timescale: Timescale.MILLI }); - expect(later.appendGroup().sequence).toBe(15); + expect(() => later.appendGroup()).toThrow("duplicate sequence"); later.close(); // Only a new broadcast starts over. @@ -185,8 +194,8 @@ test("a broadcast bounds the names it remembers sequences for", async () => { } expect(() => broadcast.createTrack("refused")).toThrow("too many track names"); - // A remembered name is still admitted, continuing its sequences. - expect(broadcast.createTrack("written-1").appendGroup().sequence).toBe(1); + // A remembered name is still admitted, and still refuses to reuse its sequences. + expect(() => broadcast.createTrack("written-1").appendGroup()).toThrow("duplicate sequence"); // An on-demand request past the bound is refused to its subscribers. const pulled = wireOf(broadcast).requested(); diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index e40eec9150..600645edbc 100644 --- a/js/net/src/broadcast.ts +++ b/js/net/src/broadcast.ts @@ -170,7 +170,7 @@ function lookup(state: BroadcastState, name: string): track.Producer | undefined // weak-dedup: a requested producer is cached in `state.requests` so every subscription and info // lookup fans out from one request instead of opening another. Whoever serves it closes it once // unused (the consuming wire tears the upstream down when its last subscriber leaves), which evicts -// the entry so a later subscribe requests the track again, continuing its sequences. +// the entry so a later subscribe requests the track again. function logical(state: BroadcastState, name: string): { producer: track.Producer; requested: boolean } { if (state.closed.peek() !== undefined) { throw new Error("broadcast is closed"); @@ -388,9 +388,10 @@ export class Producer { * * A request still queued for the name is answered with this track, so its subscribers read * from the returned producer. A request a handler has already taken stays a duplicate. - * The track continues the name's group and datagram sequences from any earlier track of the - * name in this broadcast. Throws if an open track or a live request already holds the name, - * or if the broadcast cannot remember another track name's sequences. + * An append on the track throws if it would reuse a sequence an earlier track of the name + * wrote in this broadcast, so a replacement writes its first group explicitly. Throws if an + * open track or a live request already holds the name, or if the broadcast cannot remember + * another track name's sequences. */ createTrack(name: string, info: Partial = {}): track.Producer { if (this.#state.closed.peek() !== undefined) { diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 6b29447a78..e3c2468215 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -114,7 +114,7 @@ export interface ReadGroupFrame { complete(): void; } -/** The next group or datagram sequence, shared by every producer that serves one broadcast track in turn. */ +/** One past the highest sequence written under a broadcast track's name, by any of its producers. */ export interface TrackSequence { next: number; /** Open producers writing through this namespace. */ @@ -126,10 +126,10 @@ export const MAX_TRACK_SEQUENCES = 1024; /** * Each name's sequence namespace for one broadcast generation, outliving the producers that - * serve it so a replacement continues where the last one stopped. Written namespaces are kept - * until the broadcast ends, since forgetting one would let a later producer reuse its sequences. - * Namespaces nothing holds or wrote are swept once the map fills, and a new name past that bound - * is refused: starting sequences over takes a new broadcast. + * serve it so a replacement that would append a sequence already written is refused. Written + * namespaces are kept until the broadcast ends, since forgetting one would let a later producer + * reuse its sequences. Namespaces nothing holds or wrote are swept once the map fills, and a new + * name past that bound is refused: starting sequences over takes a new broadcast. */ export class TrackSequences { #map = new Map(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 297338b58c..33139cbd00 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -509,11 +509,12 @@ export class Producer { // The producer's own state is the source of truth (info/closed); subscribers // read mirrored sinks, never this state directly. #state = new TrackState(); - // Until a broadcast binds it, a namespace of its own. An open producer holds its namespace. + // The name's sequence namespace, which records every sequence any producer of the name + // wrote. Until a broadcast binds it, a namespace of its own. An open producer holds it. #sequence: TrackSequence = { next: 0, holders: 1 }; // One past the highest group or datagram this producer received, like the Rust - // `max_sequence`. The shared counter above can run ahead of it: earlier producers of - // the same broadcast track advanced it too. + // `max_sequence`, and where the next append lands. The namespace above can run ahead of + // it: earlier producers of the same broadcast track advanced it too. #received = 0; // Recently written source groups, retained for replay to late subscribers and @@ -551,7 +552,7 @@ export class Producer { bindProducerSequence = (producer, sequence) => { const own = producer.#sequence; if (own === sequence) return; - sequence.next = Math.max(sequence.next, own.next); + sequence.next = Math.max(sequence.next, producer.#received); producer.#sequence = sequence; if (producer.#state.closed.peek() !== undefined) return; own.holders--; @@ -841,7 +842,7 @@ export class Producer { const entry: CachedGroup = { group, mirrors: new Map() }; this.#cache.push(entry); this.#cached.set(group.sequence, entry); - this.#received = Math.max(this.#received, group.sequence + 1); + this.#advance(group.sequence); this.#state.latest = Math.max(this.#state.latest ?? 0, group.sequence); for (const sink of this.#sinks) this.#mirror(entry, sink); // Give held mirrors the new live edge before pruning their timeline entry, @@ -849,6 +850,27 @@ export class Producer { this.#prune(); } + // Record a written sequence, here and in the name's namespace. + #advance(sequence: number): void { + this.#received = Math.max(this.#received, sequence + 1); + this.#sequence.next = Math.max(this.#sequence.next, sequence + 1); + } + + // The sequence the next append takes: one past this producer's own writes. The + // application owns where a replacement of the name starts, so an append that would + // reuse a sequence an earlier producer wrote throws instead of renumbering. + #nextSequence(): number { + const next = this.#received; + this.#writable(next); + const written = this.#sequence.next; + if (next < written) { + throw new Error( + `duplicate sequence: ${this.name} already wrote up to ${written - 1}, so start a replacement with an explicit sequence`, + ); + } + return next; + } + // Hold a group to this track's timedness, once the track has committed it. #bind(group: GroupProducer): void { const info = this.#state.info.peek(); @@ -871,17 +893,17 @@ export class Producer { } /** - * Append a new group with the next sequence number. + * Append a new group with the next sequence number: one past this producer's own writes. + * + * Throws if an earlier producer of the name in this broadcast already wrote that sequence. + * A replacement picks where it starts by writing its first group with {@link writeGroup}. * * Its frames must match the track's {@link Info.timescale}: a timestamp on every frame of * a timed track and none on an untimed one, or the write throws {@link TimestampMismatch}. */ appendGroup(): GroupProducer { - const sequence = this.#sequence; - this.#writable(sequence.next); - const group = new GroupProducer(sequence.next); + const group = new GroupProducer(this.#nextSequence()); this.#bind(group); - sequence.next = group.sequence + 1; this.#publish(group); return group; @@ -910,19 +932,13 @@ export class Producer { this.#uncache(existing); } - // Only advance the shared counter upward (for appendGroup auto-increment). - const sequence = this.#sequence; - if (group.sequence >= sequence.next) { - sequence.next = group.sequence + 1; - } - this.#publish(group); } // Fan a datagram out to every live subscriber, dropping the oldest once the ring is full. // Late subscribers do NOT replay old datagrams (best-effort, unlike the group cache). #publishDatagram(datagram: Datagram): void { - this.#received = Math.max(this.#received, datagram.sequence + 1); + this.#advance(datagram.sequence); for (const sink of this.#sinks) { sink.received.set(this.#received); sink.datagrams.mutate((list) => { @@ -942,19 +958,17 @@ export class Producer { * keep datagram payloads small (e.g. a single audio frame). Datagrams are never delivered * over stream-only transports (the WebSocket fallback). A payload over * 65535 bytes (the QUIC datagram frame ceiling) throws. An origin publisher uses this; a - * relay preserving upstream numbering uses {@link insertDatagram}. + * relay preserving upstream numbering uses {@link insertDatagram}. Like {@link appendGroup}, + * throws if an earlier producer of the name already wrote that sequence. * * Pass `undefined` for the timestamp on an untimed track; a timestamp whose presence * disagrees with the track's {@link Info.timescale} throws {@link TimestampMismatch}. */ appendDatagram(timestamp: Timestamp | undefined, payload: Uint8Array): number { - const counter = this.#sequence; - const sequence = counter.next; - this.#writable(sequence); + const sequence = this.#nextSequence(); if (payload.byteLength > MAX_DATAGRAM_BYTES) throw new Error("datagram payload too large"); this.#checkDatagram(timestamp); - counter.next = sequence + 1; this.#publishDatagram({ sequence, timestamp, payload }); return sequence; } @@ -962,7 +976,7 @@ export class Producer { /** * Insert a datagram with an explicit sequence number. * - * Preserves the supplied sequence (advancing the shared counter if needed) so a relay can + * Preserves the supplied sequence (later appends continue past it) so a relay can * forward a datagram without renumbering it. The size and timestamp rules of * {@link appendDatagram} apply. Most origin publishers want {@link appendDatagram} instead. */ @@ -971,10 +985,6 @@ export class Producer { if (payload.byteLength > MAX_DATAGRAM_BYTES) throw new Error("datagram payload too large"); this.#checkDatagram(timestamp); - const counter = this.#sequence; - if (sequence >= counter.next) { - counter.next = sequence + 1; - } this.#publishDatagram({ sequence, timestamp, payload }); } diff --git a/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts index 30a8cf9be9..3849bb94a4 100644 --- a/js/publish/src/broadcast.test.ts +++ b/js/publish/src/broadcast.test.ts @@ -1,7 +1,7 @@ import { expect, test } from "bun:test"; import * as Catalog from "@moq/hang/catalog"; import * as Json from "@moq/json"; -import { Origin, Path, Track } from "@moq/net"; +import { Origin, Path, Time, Track } from "@moq/net"; import { Effect, Signal } from "@moq/signals"; import { Broadcast } from "./broadcast.ts"; @@ -182,6 +182,98 @@ test("serves no catalog until announced, so the first snapshot is the one at ann broadcast.close(); }); +// Regression: a re-announce re-created the catalog track, restarting its group sequence under the +// same epoch, so a resumed viewer mistook the new catalog for the old one. +test("a re-announce continues the catalog's group sequence", async () => { + const announce = new Signal(true); + const broadcast = new Broadcast({ + enabled: true, + origin: new Origin.Producer(), + name: Path.from("test.hang"), + announce, + }); + broadcast.video("video").config.set(videoConfig); + await settle(); + + const net = broadcast.net.peek(); + if (!net) throw new Error("expected a network producer once connected"); + const first = net.track(Broadcast.CATALOG_TRACK).subscribe().ordered(); + expect((await first.nextGroup())?.sequence).toBe(0); + + announce.set(false); + await settle(); + broadcast.audio("audio").config.set(audioConfig); + await settle(); + announce.set(true); + await settle(); + + // The same track keeps serving, and its newest group is past the first. + expect(broadcast.net.peek()).toBe(net); + const group = await first.nextGroup(); + expect(group?.sequence).toBeGreaterThan(0); + const catalog = (await group?.readJson()) as Catalog.Root; + expect(Object.keys(catalog.audio?.renditions ?? {})).toEqual(["audio"]); + + first.close(); + broadcast.close(); +}); + +// Regression: any rendition change re-created every rendition's track, so the next append restarted +// a live track's group sequence. +test("a rendition keeps its track across rendition changes", async () => { + const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name: Path.from("test.hang") }); + const video = broadcast.video("video"); + await settle(); + + const net = broadcast.net.peek(); + if (!net) throw new Error("expected a network producer once connected"); + const subscriber = net.track("video").subscribe(); + await settle(); + const track = video.track.peek(); + if (!track) throw new Error("expected the demanded track"); + track.appendGroup().close(); + + // Registering another rendition leaves this one's track alone. + const audio = broadcast.audio("audio"); + await settle(); + expect(video.track.peek()).toBe(track); + + // So does closing a rendition and registering its name again. + video.close(); + await settle(); + const again = broadcast.video("video"); + await settle(); + expect(again.track.peek()).toBe(track); + expect(track.appendGroup().sequence).toBe(1); + + audio.close(); + subscriber.close(); + broadcast.close(); +}); + +test("changing maxAge publishes a new network broadcast", async () => { + const maxAge = new Signal(Time.Milli(1000)); + const broadcast = new Broadcast({ + enabled: true, + origin: new Origin.Producer(), + name: Path.from("test.hang"), + maxAge, + }); + broadcast.video("video"); + await settle(); + const before = broadcast.net.peek(); + + // A track's retention is fixed at creation, so the tracks move to a fresh broadcast. + maxAge.set(Time.Milli(2000)); + await settle(); + const after = broadcast.net.peek(); + expect(after).toBeDefined(); + expect(after).not.toBe(before); + expect((await after?.track("video").info())?.maxAge).toBe(Time.Milli(2000)); + + broadcast.close(); +}); + test("keeps the current catalog snapshot for a reconnecting viewer", async () => { const real = performance.now.bind(performance); let now = real(); diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 55269b1330..47903119ce 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -18,8 +18,8 @@ export type BroadcastInput = { // Whether to announce the broadcast. Defaults to true. Until it is announced nobody can // see or subscribe to it. The flip rather than a gate on creating it: tracks can be - // populated while this is false, then announced once ready. The catalog is served only while - // announced, so its first snapshot is the one current at announce time. + // populated while this is false, then announced once ready. The catalog is served from the + // first announce, so its first snapshot is the one current at announce time. announce: Getter; // The broadcast name. @@ -39,7 +39,8 @@ export type BroadcastInput = { * full playlist window. * * A retention budget, not a delivery one, so lowering it does not reduce latency: it only - * shortens how far back a fetch can reach. + * shortens how far back a fetch can reach. A track's retention is fixed when it is created, + * so changing this publishes a new network broadcast. */ maxAge: Getter; }; @@ -65,7 +66,7 @@ export class Broadcast { // root sections (e.g. `scte35`) by locking it too. readonly catalog = new CatalogProducer(); - // The underlying network broadcast, recreated when the name or enabled state changes and + // The underlying network broadcast, recreated when the name, enabled state, or maxAge changes and // `undefined` in between. It lives in the origin rather than any session, so it spans // reconnects. Exposed so an application can serve its own tracks alongside the built-in // catalog/audio/video, e.g. `net.createTrack("meta.json")` plus a matching `catalog` section. @@ -143,7 +144,7 @@ export class Broadcast { #unregister(name: string): void { const track = this.#tracks.get(name); if (track) { - track.peek()?.close(); + // The network track stays with the broadcast, so registering the name again continues it. track.set(undefined); this.#tracks.delete(name); } @@ -215,6 +216,7 @@ export class Broadcast { const [_enabled, origin] = values; const name = effect.get(this.in.name); + const maxAge = effect.get(this.in.maxAge); if (Catalog.detectFormat(name) === undefined) { console.warn( `You should append .hang to broadcast name ${JSON.stringify(name)} to make the catalog format explicit.`, @@ -235,30 +237,42 @@ export class Broadcast { if (this.net.peek() === broadcast) this.net.set(undefined); }); + // Each track lives as long as the broadcast: a track name always means the same content, + // so a re-announce or a rendition change keeps writing to it instead of restarting its + // group sequence under the same epoch. + const tracks = new Map(); + effect.cleanup(() => { + for (const track of tracks.values()) track.close(); + }); + + let serving = false; effect.run((inner) => { if (!inner.get(this.in.announce)) { broadcast.unannounce(); return; } - // Serve the catalog only while announced. Nobody can find the broadcast before then, so an - // earlier catalog would only leave a stale first snapshot behind; this way the first one is - // whatever the catalog holds at announce time. - for (const [name, compression] of [ - [Broadcast.CATALOG_TRACK, false], - [Broadcast.CATALOG_TRACK_COMPRESSED, true], - ] as const) { - // A catalog may publish once and stay unchanged for the broadcast's whole life. Keep - // that sole closed snapshot replayable so a viewer arriving after the ordinary media - // retention window can still bootstrap. - const track = broadcast.createTrack(name, { - // Each catalog snapshot is stamped when written. - timescale: Moq.Time.Timescale.MILLI, - maxAge: Moq.Time.Milli(Number.MAX_SAFE_INTEGER), - priority: Catalog.PRIORITY.catalog, - }); - inner.cleanup(() => track.close()); - this.catalog.serve(track, inner, { compression }); + // Serve the catalog from the first announce. Nobody can find the broadcast before then, so + // an earlier catalog would only leave a stale first snapshot behind; this way the first one + // is whatever the catalog holds at announce time. + if (!serving) { + serving = true; + for (const [name, compression] of [ + [Broadcast.CATALOG_TRACK, false], + [Broadcast.CATALOG_TRACK_COMPRESSED, true], + ] as const) { + // A catalog may publish once and stay unchanged for the broadcast's whole life. Keep + // that sole closed snapshot replayable so a viewer arriving after the ordinary media + // retention window can still bootstrap. + const track = broadcast.createTrack(name, { + // Each catalog snapshot is stamped when written. + timescale: Moq.Time.Timescale.MILLI, + maxAge: Moq.Time.Milli(Number.MAX_SAFE_INTEGER), + priority: Catalog.PRIORITY.catalog, + }); + tracks.set(name, track); + this.catalog.serve(track, effect, { compression }); + } } broadcast.announce({ epoch }); @@ -266,21 +280,24 @@ export class Broadcast { // Static tracks fan out to every subscriber. Keep the encoder-facing handle demand-gated // so capture and encoding still stop when the final subscriber leaves. - effect.run((tracks) => { - const renditions = tracks.get(this.#renditions); - const maxAge = tracks.get(this.in.maxAge); + effect.run((inner) => { + const renditions = inner.get(this.#renditions); for (const rendition of Object.values(renditions)) { const signal = this.#tracks.get(rendition.name); if (!signal) continue; - const track = broadcast.createTrack( - rendition.name, - Container.trackInfo({ maxAge, priority: Catalog.PRIORITY[rendition.kind] }), - ); - tracks.cleanup(() => track.close()); - tracks.run((demand) => { - demand.set(signal, demand.get(track.demand().used) ? track : undefined); + let track = tracks.get(rendition.name); + if (!track) { + track = broadcast.createTrack( + rendition.name, + Container.trackInfo({ maxAge, priority: Catalog.PRIORITY[rendition.kind] }), + ); + tracks.set(rendition.name, track); + } + const served = track; + inner.run((demand) => { + demand.set(signal, demand.get(served.demand().used) ? served : undefined); }); } }); diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 18605c77e2..a16efbac13 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -125,7 +125,6 @@ End+Start. The child quests own their focused regressions. - [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - a delayed group is delivered once even beyond the old 1024-entry history - [Catalog references](/quest/m0/broadcast-epoch/catalog-references.md) - players and exporters pin siblings consistently, with optional reference epochs and documented legacy best effort -- [Publish catalog restart](/quest/m0/broadcast-epoch/publish-catalog-restart.md) - `@moq/publish` never reuses catalog group numbers under one name and epoch after a re-announce - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one - [Stats totals and prefix tracks](/quest/m0/broadcast-epoch/stats-split.md) - the same release retires the per-path stats maps for totals and on-demand prefix tracks (decided 2026-10-05) diff --git a/quest/m0/broadcast-epoch/publish-catalog-restart.md b/quest/m0/broadcast-epoch/publish-catalog-restart.md deleted file mode 100644 index 2f72262849..0000000000 --- a/quest/m0/broadcast-epoch/publish-catalog-restart.md +++ /dev/null @@ -1,30 +0,0 @@ -# [S] @moq/publish never reuses catalog group numbers under one epoch - -## Goal - -`@moq/publish` never publishes different content under one name and epoch. -After an unannounce and re-announce within one run, its catalog track either -continues its group sequence or the broadcast goes out under a fresh epoch, so -a viewer or relay never mistakes a restarted catalog for the old one. - -## Plan - -Found by #4970 (broadcast-epoch apps): `@moq/publish` re-creates its catalog -track after an unannounce and re-announce, so catalog group numbers may -restart under the same name and epoch. That breaks the rule that a broadcast, -track, or group name always means the same content. - -Decided 2026-10-07: verify it first with a test that unannounces and -re-announces a running publish and reads the catalog's group sequence. Fix it -only if it reproduces: continue the sequence, or mint a fresh epoch per -announce, whichever matches how the rest of `@moq/publish` treats a -re-announce. `js/publish/src/broadcast.ts` mints the epoch once and keeps the -publisher across unannounce and announce on purpose, which points at -continuing the sequence. - -Decided 2026-10-08: the `@moq/net` fix landed separately: `createTrack` now -continues a name's sequences within a broadcast, so this quest does not fix -`createTrack` itself. Run the test against it and fix `@moq/publish` only if -it still restarts. - -Public API: none expected. Wire: none. diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 2b5742f8ed..5a0f108a5e 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -99,9 +99,9 @@ struct BroadcastState { requests: Requests, track::Request>, // Each name's sequence namespace, outliving the producers that serve it: a name means - // the same content for the broadcast's whole life, so a replacement producer continues - // where the last one stopped instead of reusing its sequences. Entries nothing holds or - // wrote are swept as the map grows, so requests for names never served stay bounded. + // the same content for the broadcast's whole life, so a replacement producer that would + // append a sequence an earlier one wrote is refused. Entries nothing holds or wrote are + // swept as the map grows, so requests for names never served stay bounded. sequences: HashMap, track::Sequence>, // Set once the broadcast ends: `Producer::close()` or the last producer-side @@ -1357,16 +1357,16 @@ mod test { track2.assert_not_closed(); track2.assert_not_clone(&track1); - // The new consumer should receive the new group. - producer2.append_group().unwrap(); + // The new consumer should receive the new group, which continues past the first's. + producer2.create_group(crate::group::Info { sequence: 1 }).unwrap(); track2.assert_group(); } - /// A name keeps one sequence namespace for the broadcast's life: a replacement producer - /// appends past everything an earlier one wrote, explicit writes included, while a new - /// broadcast starts over. + /// A name keeps one sequence namespace for the broadcast's life. The application owns + /// where a replacement starts: an append that would reuse an earlier producer's sequence, + /// explicit writes included, is refused, while a new broadcast starts over. #[test] - fn replacement_continues_sequences() { + fn replacement_refuses_reused_sequences() { let producer = Info::new().produce(); let mut dynamic = producer.dynamic(); let consumer = producer.consume(); @@ -1382,14 +1382,28 @@ mod test { let track = consumer.track("media").unwrap(); let mut second = dynamic.assert_request().accept(None); - assert_eq!(second.append_group().unwrap().sequence, 13); + assert!(matches!(second.append_group(), Err(Error::Duplicate))); + assert!(matches!( + second.append_datagram(crate::Timestamp::ZERO, b"x"), + Err(Error::Duplicate) + )); + + // An explicit first write picks the start, and appends continue from there. + second.create_group(crate::group::Info { sequence: 13 }).unwrap(); assert_eq!(second.append_datagram(crate::Timestamp::ZERO, b"x").unwrap(), 14); second.finish().unwrap(); drop((second, track)); - // A track the producer creates itself continues the same namespace. + // A track the producer creates itself checks the same namespace. let third = producer.create_track("media", None).unwrap(); - assert_eq!(third.append_group().unwrap().sequence, 15); + assert!(matches!(third.append_group(), Err(Error::Duplicate))); + assert_eq!( + third + .create_group(crate::group::Info { sequence: 15 }) + .unwrap() + .sequence, + 15 + ); let next = Info::new().produce(); assert_eq!( diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 15eb458ffc..fe0b5edcec 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -162,13 +162,13 @@ impl Info { } /// The sequence namespace a track name keeps within its broadcast, shared by every -/// producer that serves the name, so a replacement appends after whatever an earlier one -/// wrote. Holds one past the highest sequence written. +/// producer that serves the name, so a replacement that would append a sequence an earlier +/// one wrote is refused. Holds one past the highest sequence written. #[derive(Clone, Default)] pub(crate) struct Sequence(Arc); impl Sequence { - /// The next sequence an append takes. + /// One past the highest sequence any producer of the name wrote. fn next(&self) -> u64 { self.0.load(Ordering::Relaxed) } @@ -263,8 +263,8 @@ pub(crate) struct TrackState { // datagrams, so it can run ahead of any cached group. max_sequence: Option, - // The name's sequence namespace within the broadcast, which an append continues - // even when an earlier producer of the name wrote past this one's `max_sequence`. + // The name's sequence namespace within the broadcast, which can run past this track's + // `max_sequence` when an earlier producer of the name wrote further. sequence: Sequence, // The sequence of the newest cached group: the live edge, protected from @@ -1121,14 +1121,20 @@ impl TrackState { self.sequence.advance(sequence); } - /// The sequence the next append takes: past this track's own writes and past every - /// earlier producer's of the same name. + /// The sequence the next append takes: one past this track's own writes. + /// + /// The application owns where a replacement of the name starts, so an append that + /// would reuse a sequence an earlier producer wrote is [`Error::Duplicate`] instead of + /// renumbering. fn next_sequence(&self) -> Result { - let own = match self.max_sequence { + let next = match self.max_sequence { Some(max) => max.checked_add(1).ok_or(coding::BoundsExceeded)?, None => 0, }; - Ok(own.max(self.sequence.next())) + if next < self.sequence.next() { + return Err(Error::Duplicate); + } + Ok(next) } /// Reject a sequence that is still cached; a dead (aborted or evicted) @@ -1801,7 +1807,10 @@ impl Producer { Ok(group) } - /// Create a new group with the next sequence number. + /// Create a new group with the next sequence number: one past this producer's own writes. + /// + /// [`Error::Duplicate`] if an earlier producer of the name in this broadcast already wrote + /// that sequence. A replacement picks where it starts with [`Self::create_group`]. pub fn append_group(&self) -> Result { let mut state = self.modify()?; let sequence = state.next_sequence()?; @@ -1830,7 +1839,8 @@ impl Producer { /// before 05, or stream-only transports like WebSocket) never deliver them. Keep payloads well under the 1200-byte minimum path MTU. /// A datagram is never cached or served by a fetch, so use a group for anything a late /// joiner needs. An origin publisher uses this; a relay preserving upstream numbering - /// uses [`Self::insert_datagram`]. + /// uses [`Self::insert_datagram`]. Like [`Self::append_group`], it is [`Error::Duplicate`] + /// if an earlier producer of the name already wrote that sequence. /// /// Pass `None` on an untimed track; a datagram that doesn't match its track is /// [`Error::TimestampMismatch`]. diff --git a/rs/moq-net/tests/route_change.rs b/rs/moq-net/tests/route_change.rs index 37eaee1283..07b559f429 100644 --- a/rs/moq-net/tests/route_change.rs +++ b/rs/moq-net/tests/route_change.rs @@ -856,9 +856,8 @@ async fn pair_behind_a_relay_ends_lite_06() { /// `P` serves `video` on demand and replaces its producer partway through the broadcast, /// the way a publisher restarts an encoder. The epoch lets `P`'s own front resume the /// logical track onto the replacement, so `A` and `R` never see the abort. The -/// replacement continues the track's sequence namespace, so its first group reaches -/// `R` at once instead of being skipped until a fresh counter caught up with what `R` -/// already read. +/// replacement continues the track's sequences explicitly, so its first group reaches +/// `R` at once instead of being skipped as one `R` already read. async fn producer_replaced(version: &str) { let version: Version = version.parse().unwrap(); let publisher = produce_origin(1); @@ -908,8 +907,13 @@ async fn producer_replaced(version: &str) { .expect("the track was not requested again") .unwrap(); - let mut group = second.append_group().unwrap(); - assert_eq!(group.sequence, 2, "{version}: the replacement restarted its sequences"); + // The application owns where the replacement starts: an append would reuse the first + // producer's sequences, so it is refused, and the replacement continues explicitly. + assert!( + matches!(second.append_group(), Err(Error::Duplicate)), + "{version}: the replacement restarted its sequences" + ); + let mut group = second.create_group(moq_net::group::Info { sequence: 2 }).unwrap(); for frame in 0..FRAMES { group.write_frame(Timestamp::ZERO, payload(2, frame)).unwrap(); assert_eq!(next(&mut rx).await, (2, payload(2, frame)), "{version}"); From a8b2a07ae3ffb35fb9a57b997b42a0b2f9e0c325 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 16:35:07 -0700 Subject: [PATCH 5/9] fix(publish): keep catalog tracks out of the rendition map A rendition named like a catalog track would otherwise reuse the catalog's producer instead of failing as a duplicate. Co-Authored-By: Claude Opus 5.5 --- js/publish/src/broadcast.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 47903119ce..7285644998 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -239,7 +239,8 @@ export class Broadcast { // Each track lives as long as the broadcast: a track name always means the same content, // so a re-announce or a rendition change keeps writing to it instead of restarting its - // group sequence under the same epoch. + // group sequence under the same epoch. Rendition tracks by name, so registering a name + // again finds its track; the catalog's stay out of reach of a colliding rendition. const tracks = new Map(); effect.cleanup(() => { for (const track of tracks.values()) track.close(); @@ -270,7 +271,7 @@ export class Broadcast { maxAge: Moq.Time.Milli(Number.MAX_SAFE_INTEGER), priority: Catalog.PRIORITY.catalog, }); - tracks.set(name, track); + effect.cleanup(() => track.close()); this.catalog.serve(track, effect, { compression }); } } From 62f06722981535faab1bebc74b3f15f2292db518 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 16:41:07 -0700 Subject: [PATCH 6/9] fix(net): reject a request whose accept fails Invalid info left the request's producer open, so a queued request answered by createTrack stranded its subscribers and held the name. Also document that Rendition.close leaves the network track with the broadcast. Co-Authored-By: Claude Opus 5.5 --- js/net/src/broadcast.test.ts | 14 ++++++++++++++ js/net/src/track.ts | 10 +++++----- js/publish/src/rendition.ts | 5 ++++- 3 files changed, 23 insertions(+), 6 deletions(-) diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index 15bfa71729..093cfb00c5 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -127,6 +127,20 @@ test("createTrack answers a queued request", async () => { expect(await pulled).toBeUndefined(); }); +test("createTrack with invalid info rejects the queued request it answers", async () => { + const broadcast = new BroadcastProducer(); + const pulled = wireOf(broadcast).requested(); + const subscriber = broadcast.track("media").subscribe(); + + expect(() => broadcast.createTrack("media", { priority: 256 })).toThrow(); + expect(await subscriber.closed).toBeInstanceOf(Error); + + // The refused request no longer holds the name. + broadcast.createTrack("media"); + broadcast.close(); + expect(await pulled).toBeUndefined(); +}); + test("createTrack does not take over a request a handler already pulled", async () => { const broadcast = new BroadcastProducer(); const pulled = wireOf(broadcast).requested(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 33139cbd00..9e85b9cce7 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -244,19 +244,19 @@ export class Request { /** * Accept the request, committing the track's immutable {@link Info}. Throws, rejecting the - * request, if the broadcast cannot remember another track name's sequences. + * request, if `info` is invalid or the broadcast cannot remember another track name's + * sequences. */ accept(info: Partial = {}): Producer { this.#pending.delete(this); - let sequence: TrackSequence; try { - sequence = this.#sequences.admit(this.name); + bindProducerSequence(this.#producer, this.#sequences.admit(this.name)); + return this.#producer.accept(info); } catch (err) { + // Refused, so its subscribers see the error instead of waiting on an open request. this.#producer.close(err instanceof Error ? err : new Error(String(err))); throw err; } - bindProducerSequence(this.#producer, sequence); - return this.#producer.accept(info); } /** Reject the request, closing the track optionally with an error. */ diff --git a/js/publish/src/rendition.ts b/js/publish/src/rendition.ts index 46769a4f99..c1bc8ebbbf 100644 --- a/js/publish/src/rendition.ts +++ b/js/publish/src/rendition.ts @@ -40,7 +40,10 @@ export class Rendition { this.#close = close; } - /** Unregister: removes the catalog entry and closes any active track. Idempotent. */ + /** + * Unregister: removes the catalog entry and detaches the track. The broadcast keeps the + * network track open, so registering the name again continues it. Idempotent. + */ close(): void { if (this.#closed) return; this.#closed = true; From f46a96dbe61d9b5b191c2589bd12de8a9cbf667e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 16:46:39 -0700 Subject: [PATCH 7/9] fix(publish): replace a closed rendition track and keep a name's kind A rendition track closed by an encoder error is recreated on re-register instead of reused, and registering a name under a different kind throws. Co-Authored-By: Claude Opus 5.5 --- js/publish/src/broadcast.test.ts | 34 ++++++++++++++++++++++++++++++++ js/publish/src/broadcast.ts | 13 +++++++++++- 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts index 3849bb94a4..2259e16346 100644 --- a/js/publish/src/broadcast.test.ts +++ b/js/publish/src/broadcast.test.ts @@ -246,11 +246,45 @@ test("a rendition keeps its track across rendition changes", async () => { expect(again.track.peek()).toBe(track); expect(track.appendGroup().sequence).toBe(1); + // The name keeps its kind. + again.close(); + expect(() => broadcast.audio("video")).toThrow("registered as video"); + audio.close(); subscriber.close(); broadcast.close(); }); +test("a rendition whose track closed gets a replacement", async () => { + const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name: Path.from("test.hang") }); + const video = broadcast.video("video"); + await settle(); + + const net = broadcast.net.peek(); + if (!net) throw new Error("expected a network producer once connected"); + const first = net.track("video").subscribe(); + await settle(); + const track = video.track.peek(); + if (!track) throw new Error("expected the demanded track"); + + // An encoder error closes the track before it wrote anything. + track.close(new Error("encoder failed")); + video.close(); + await settle(); + + const again = broadcast.video("video"); + await settle(); + const second = net.track("video").subscribe(); + await settle(); + const replacement = again.track.peek(); + expect(replacement).toBeDefined(); + expect(replacement).not.toBe(track); + + first.close(); + second.close(); + broadcast.close(); +}); + test("changing maxAge publishes a new network broadcast", async () => { const maxAge = new Signal(Time.Milli(1000)); const broadcast = new Broadcast({ diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 7285644998..905602c238 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -87,6 +87,9 @@ export class Broadcast { // A static network track is exposed here only while at least one subscriber uses it. readonly #tracks = new Map>(); + // The kind each rendition name was first registered as, for the broadcast's whole life. + readonly #kinds = new Map(); + #signals = new Effect(); constructor(props?: Inputs) { @@ -131,6 +134,12 @@ export class Broadcast { if (this.#renditions.peek()[name]) { throw new Error(`rendition already registered: ${name}`); } + // A name keeps its track, so it keeps its kind: different media takes a different name. + const previous = this.#kinds.get(name); + if (previous !== undefined && previous !== kind) { + throw new Error(`rendition ${name} was registered as ${previous}, not ${kind}`); + } + this.#kinds.set(name, kind); const track = new Signal(undefined); this.#tracks.set(name, track); @@ -288,8 +297,10 @@ export class Broadcast { const signal = this.#tracks.get(rendition.name); if (!signal) continue; + // A track that closed (an encoder error) is replaced. Its appends then throw rather + // than reuse the closed track's sequences. let track = tracks.get(rendition.name); - if (!track) { + if (!track || track.closed.peek() !== undefined) { track = broadcast.createTrack( rendition.name, Container.trackInfo({ maxAge, priority: Catalog.PRIORITY[rendition.kind] }), From 054b563c329e9107ae74543119103a8ab5273a7e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 16:51:31 -0700 Subject: [PATCH 8/9] fix(publish): scope a rendition name's kind to its network broadcast Co-Authored-By: Claude Opus 5.5 --- js/publish/src/broadcast.test.ts | 21 +++++++++++++++++++++ js/publish/src/broadcast.ts | 9 ++++++--- 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts index 2259e16346..86799f55ba 100644 --- a/js/publish/src/broadcast.test.ts +++ b/js/publish/src/broadcast.test.ts @@ -255,6 +255,27 @@ test("a rendition keeps its track across rendition changes", async () => { broadcast.close(); }); +test("a rendition name may change kind on a new network broadcast", async () => { + const name = new Signal(Path.from("one.hang")); + const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name }); + const video = broadcast.video("x"); + await settle(); + const net = broadcast.net.peek(); + if (!net) throw new Error("expected a network producer once connected"); + const subscriber = net.track("x").subscribe(); + await settle(); + video.close(); + expect(() => broadcast.audio("x")).toThrow("registered as video"); + + // A new network broadcast holds no track for the name, so any kind may take it. + name.set(Path.from("two.hang")); + await settle(); + expect(() => broadcast.audio("x")).not.toThrow(); + + subscriber.close(); + broadcast.close(); +}); + test("a rendition whose track closed gets a replacement", async () => { const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name: Path.from("test.hang") }); const video = broadcast.video("video"); diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 905602c238..dc9772ae75 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -87,8 +87,8 @@ export class Broadcast { // A static network track is exposed here only while at least one subscriber uses it. readonly #tracks = new Map>(); - // The kind each rendition name was first registered as, for the broadcast's whole life. - readonly #kinds = new Map(); + // The kind of each rendition track in the current network broadcast. A new one starts over. + #kinds = new Map(); #signals = new Effect(); @@ -139,7 +139,6 @@ export class Broadcast { if (previous !== undefined && previous !== kind) { throw new Error(`rendition ${name} was registered as ${previous}, not ${kind}`); } - this.#kinds.set(name, kind); const track = new Signal(undefined); this.#tracks.set(name, track); @@ -251,8 +250,11 @@ export class Broadcast { // group sequence under the same epoch. Rendition tracks by name, so registering a name // again finds its track; the catalog's stay out of reach of a colliding rendition. const tracks = new Map(); + const kinds = new Map(); + this.#kinds = kinds; effect.cleanup(() => { for (const track of tracks.values()) track.close(); + if (this.#kinds === kinds) this.#kinds = new Map(); }); let serving = false; @@ -306,6 +308,7 @@ export class Broadcast { Container.trackInfo({ maxAge, priority: Catalog.PRIORITY[rendition.kind] }), ); tracks.set(rendition.name, track); + kinds.set(rendition.name, rendition.kind); } const served = track; inner.run((demand) => { From fcf9cf490112361fced86f32dd9e767263675e0e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 18:51:02 -0700 Subject: [PATCH 9/9] fix(net,publish): drop the JS name cap; replace a closed track on a new broadcast @moq/net no longer refuses a broadcast's 1025th track name. Like Rust, it sweeps names that never wrote and no producer holds as the record grows. @moq/publish moves to a new network broadcast when a closed rendition track must be replaced, since a replacement under the same name would restart its sequences. Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 2 +- doc/lib/js/publish.md | 3 +- js/net/src/broadcast.test.ts | 48 +++++++++++++------------------- js/net/src/broadcast.ts | 5 ++-- js/net/src/internal.ts | 24 ++++++++-------- js/net/src/track.ts | 5 ++-- js/publish/src/broadcast.test.ts | 16 +++++++++-- js/publish/src/broadcast.ts | 18 ++++++++---- 8 files changed, 66 insertions(+), 55 deletions(-) diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index b87d1b9369..bcb6bc0d5c 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -40,7 +40,7 @@ broadcast.announce(); - **One connection per URL, unless you opt out.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. Supplying your own transport options, discovery, or origin selects a private loop. - **GOAWAY moves the session.** The connection dials the replacement at once while the old session keeps serving its groups, up to `goaway.handover` (default 10s, or the relay's deadline when that is sooner). New requests keep opening on the old session until the replacement's route outranks it, which is a restart without an epoch: a request on the old session ends when that session closes, so request the path again. An empty GOAWAY redials the same URL. A redirect stays on the same host unless `goaway.redirect` is `"follow"`. - **One send estimate per connection.** `Bandwidth.Allocator` divides it by track priority, max-min fair within a tier. An idle track claims nothing. Publishers reserve against it so their targets sum to the estimate. -- **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: a track that replaces an ended one, created or requested, throws on an append that would reuse a sequence the name already wrote, so it writes its first group with an explicit sequence. Only a new broadcast starts over. A broadcast remembers the sequences of up to 1024 names, and creating a track under a new name past that throws. +- **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: a track that replaces an ended one, created or requested, throws on an append that would reuse a sequence the name already wrote, so it writes its first group with an explicit sequence. Only a new broadcast starts over. - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Unlike Rust, omitting it means untimed. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` and `@moq/flate` producers. - **Subscriber staleness is `maxDelay`.** It is media time. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts). diff --git a/doc/lib/js/publish.md b/doc/lib/js/publish.md index 9375518fe2..5aaa3752ff 100644 --- a/doc/lib/js/publish.md +++ b/doc/lib/js/publish.md @@ -67,7 +67,8 @@ timeline. The catalog's `clock` maps it to wall time. Stamp your own tracks `broadcast.net` is the underlying `Moq.Broadcast.Producer`, so an application can serve its own tracks alongside the media. It is recreated when the origin, name, -`enabled`, or `maxAge` changes, so acquire it from an effect and create each +`enabled`, or `maxAge` changes, or when a failed rendition track must be +replaced, so acquire it from an effect and create each track once per `net`. A track name always means the same content, so a track re-created on the same `net` throws on an append that would reuse its sequences: diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index 093cfb00c5..7ce0177414 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -2,7 +2,7 @@ import { expect, setSystemTime, test } from "bun:test"; import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import { GroupTooLarge, NotFound } from "./error.ts"; import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts"; -import { MAX_TRACK_SEQUENCES } from "./internal.ts"; +import { TrackSequences } from "./internal.ts"; import { textFrame } from "./mock.ts"; import { Milli, Timescale, Timestamp } from "./time.ts"; import type { Request as TrackRequest } from "./track.ts"; @@ -191,37 +191,29 @@ test("created and requested tracks share a name's sequences", async () => { next.close(); }); -test("a broadcast bounds the names it remembers sequences for", async () => { - const broadcast = new BroadcastProducer(); +test("names that never wrote leave nothing behind", () => { + const sequences = new TrackSequences(); + sequences.get("written").next = 1; + const held = sequences.get("held"); + held.holders = 1; - // Names that never wrote leave nothing behind once their tracks close. - for (let i = 0; i < MAX_TRACK_SEQUENCES * 2; i++) { - broadcast.createTrack(`unused-${i}`).close(); - } + // A peer asking for arbitrary names cannot grow the map: unused names are swept as it grows. + for (let i = 0; i < 1000; i++) sequences.get(`unused-${i}`); + expect(sequences.size).toBeLessThan(64); - // Written names are remembered, and so is an open track that has not written yet. - const open = broadcast.createTrack("open"); - for (let i = 1; i < MAX_TRACK_SEQUENCES; i++) { - const track = broadcast.createTrack(`written-${i}`); - track.appendGroup().close(); - track.close(); - } - expect(() => broadcast.createTrack("refused")).toThrow("too many track names"); - - // A remembered name is still admitted, and still refuses to reuse its sequences. - expect(() => broadcast.createTrack("written-1").appendGroup()).toThrow("duplicate sequence"); + // Written and held names survive every sweep. + expect(sequences.get("written").next).toBe(1); + expect(sequences.get("held")).toBe(held); +}); - // An on-demand request past the bound is refused to its subscribers. - const pulled = wireOf(broadcast).requested(); - const subscriber = broadcast.track("refused").subscribe(); - const request = await pulled; - if (!request) throw new Error("expected request"); - expect(() => request.accept()).toThrow("too many track names"); - expect(await subscriber.closed).toBeInstanceOf(Error); +test("a written name refuses to reuse its sequences after many other names", () => { + const broadcast = new BroadcastProducer(); + const written = broadcast.createTrack("written"); + written.appendGroup().close(); + written.close(); + for (let i = 0; i < 1000; i++) broadcast.createTrack(`unused-${i}`).close(); - // Closing the open track without a write frees its slot. - open.close(); - broadcast.createTrack("admitted"); + expect(() => broadcast.createTrack("written").appendGroup()).toThrow("duplicate sequence"); broadcast.close(); }); diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index 600645edbc..34acea2ae1 100644 --- a/js/net/src/broadcast.ts +++ b/js/net/src/broadcast.ts @@ -390,8 +390,7 @@ export class Producer { * from the returned producer. A request a handler has already taken stays a duplicate. * An append on the track throws if it would reuse a sequence an earlier track of the name * wrote in this broadcast, so a replacement writes its first group explicitly. Throws if an - * open track or a live request already holds the name, or if the broadcast cannot remember - * another track name's sequences. + * open track or a live request already holds the name. */ createTrack(name: string, info: Partial = {}): track.Producer { if (this.#state.closed.peek() !== undefined) { @@ -417,7 +416,7 @@ export class Producer { } const producer = new track.Producer(name).accept(info); - hooks.bindSequence(producer, this.#state.sequences.admit(name)); + hooks.bindSequence(producer, this.#state.sequences.get(name)); adopt(this.#state, producer); return producer; } diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index bb363d298f..a3b6133914 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -121,37 +121,39 @@ export interface TrackSequence { holders: number; } -/** The most track names one broadcast remembers sequences for. */ -export const MAX_TRACK_SEQUENCES = 1024; - /** * Each name's sequence namespace for one broadcast generation, outliving the producers that * serve it so a replacement that would append a sequence already written is refused. Written * namespaces are kept until the broadcast ends, since forgetting one would let a later producer - * reuse its sequences. Namespaces nothing holds or wrote are swept once the map fills, and a new - * name past that bound is refused: starting sequences over takes a new broadcast. + * reuse its sequences. Namespaces nothing holds or wrote are swept as the map grows, like the + * Rust map at capacity, so requests for names never served stay bounded. */ export class TrackSequences { #map = new Map(); + // The size that triggers the next sweep, doubling past what survives so sweeps stay amortized. + #sweepAt = 16; - /** The namespace for `name`, admitting the name if it is new. Throws once the bound is reached. */ - admit(name: string): TrackSequence { + /** The namespace for `name`, created if the name is new. */ + get(name: string): TrackSequence { const existing = this.#map.get(name); if (existing) return existing; - if (this.#map.size >= MAX_TRACK_SEQUENCES) { + if (this.#map.size >= this.#sweepAt) { for (const [key, sequence] of this.#map) { if (sequence.next === 0 && sequence.holders === 0) this.#map.delete(key); } - if (this.#map.size >= MAX_TRACK_SEQUENCES) { - throw new Error(`too many track names: ${name} would exceed ${MAX_TRACK_SEQUENCES}`); - } + this.#sweepAt = Math.max(16, this.#map.size * 2); } const sequence = { next: 0, holders: 0 }; this.#map.set(name, sequence); return sequence; } + + /** How many names hold a namespace. */ + get size(): number { + return this.#map.size; + } } /** Inputs for creating a package-internal track request. */ diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 9e85b9cce7..6d53703d14 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -244,13 +244,12 @@ export class Request { /** * Accept the request, committing the track's immutable {@link Info}. Throws, rejecting the - * request, if `info` is invalid or the broadcast cannot remember another track name's - * sequences. + * request, if `info` is invalid. */ accept(info: Partial = {}): Producer { this.#pending.delete(this); try { - bindProducerSequence(this.#producer, this.#sequences.admit(this.name)); + bindProducerSequence(this.#producer, this.#sequences.get(this.name)); return this.#producer.accept(info); } catch (err) { // Refused, so its subscribers see the error instead of waiting on an open request. diff --git a/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts index 86799f55ba..49fbaeb7e5 100644 --- a/js/publish/src/broadcast.test.ts +++ b/js/publish/src/broadcast.test.ts @@ -276,7 +276,9 @@ test("a rendition name may change kind on a new network broadcast", async () => broadcast.close(); }); -test("a rendition whose track closed gets a replacement", async () => { +// A closed track can't be continued, and a replacement under its name would restart its sequences, +// so the replacement goes out on a new network broadcast. +test("replacing a closed rendition track publishes a new network broadcast", async () => { const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name: Path.from("test.hang") }); const video = broadcast.video("video"); await settle(); @@ -288,18 +290,26 @@ test("a rendition whose track closed gets a replacement", async () => { const track = video.track.peek(); if (!track) throw new Error("expected the demanded track"); - // An encoder error closes the track before it wrote anything. + // The encoder wrote, then failed. + track.appendGroup().close(); track.close(new Error("encoder failed")); video.close(); await settle(); const again = broadcast.video("video"); await settle(); - const second = net.track("video").subscribe(); + const next = broadcast.net.peek(); + expect(next).toBeDefined(); + expect(next).not.toBe(net); + if (!next) throw new Error("expected a new network producer"); + + // The replacement starts the name over on the new broadcast. + const second = next.track("video").subscribe(); await settle(); const replacement = again.track.peek(); expect(replacement).toBeDefined(); expect(replacement).not.toBe(track); + expect(replacement?.appendGroup().sequence).toBe(0); first.close(); second.close(); diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index dc9772ae75..65a4a77bb3 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -66,8 +66,8 @@ export class Broadcast { // root sections (e.g. `scte35`) by locking it too. readonly catalog = new CatalogProducer(); - // The underlying network broadcast, recreated when the name, enabled state, or maxAge changes and - // `undefined` in between. It lives in the origin rather than any session, so it spans + // The underlying network broadcast, recreated when the name, enabled state, or maxAge changes, + // or when a closed rendition track must be replaced, and `undefined` in between. It lives in the origin rather than any session, so it spans // reconnects. Exposed so an application can serve its own tracks alongside the built-in // catalog/audio/video, e.g. `net.createTrack("meta.json")` plus a matching `catalog` section. // Reacquire it via an effect, since a rename swaps in a fresh producer. @@ -90,6 +90,9 @@ export class Broadcast { // The kind of each rendition track in the current network broadcast. A new one starts over. #kinds = new Map(); + // Bumped to publish a new network broadcast, when a closed rendition track must be replaced. + readonly #restart = new Signal(0); + #signals = new Effect(); constructor(props?: Inputs) { @@ -225,6 +228,7 @@ export class Broadcast { const name = effect.get(this.in.name); const maxAge = effect.get(this.in.maxAge); + effect.get(this.#restart); if (Catalog.detectFormat(name) === undefined) { console.warn( `You should append .hang to broadcast name ${JSON.stringify(name)} to make the catalog format explicit.`, @@ -299,10 +303,14 @@ export class Broadcast { const signal = this.#tracks.get(rendition.name); if (!signal) continue; - // A track that closed (an encoder error) is replaced. Its appends then throw rather - // than reuse the closed track's sequences. + // A track that closed (an encoder error) can't be continued, and a replacement under + // the same name would restart its sequences, so the tracks move to a new broadcast. let track = tracks.get(rendition.name); - if (!track || track.closed.peek() !== undefined) { + if (track?.closed.peek() !== undefined) { + this.#restart.update((restarts) => restarts + 1); + return; + } + if (!track) { track = broadcast.createTrack( rendition.name, Container.trackInfo({ maxAge, priority: Catalog.PRIORITY[rendition.kind] }),