diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index ecbeb240c8..d43a9f6541 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. 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. 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. Omitting it means untimed, as in Rust. 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..5aaa3752ff 100644 --- a/doc/lib/js/publish.md +++ b/doc/lib/js/publish.md @@ -66,8 +66,11 @@ 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, 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: ```ts import * as Json from "@moq/json"; diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 864a652382..58e54483a6 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -19,7 +19,7 @@ What you use it for, beyond what the concept page already describes: - **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. - **Rank several CDNs.** Dial each with its own subscribe handle, `origin.clone().with_preference(origin::Preference::new(rank))` (higher wins; 0 is neutral and a negative rank demotes), all on one origin. Requests use the preferred link that has a route and fail over to the next; see [multiple CDNs](/concept/moq-lite#multiple-cdns). - **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/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 02a93b52e1..7ce0177414 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -1,12 +1,11 @@ -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"; +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"; -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 @@ -55,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. @@ -77,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(); @@ -93,6 +99,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.writeFrame(textFrame("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.writeFrame(textFrame("again")); + expect(await again.readString()).toBe("again"); + expect(() => broadcast.createTrack("media")).toThrow("duplicate track"); + + broadcast.close(); + 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(); + broadcast.track("media").subscribe(); + const request = await pulled; + if (!request) throw new Error("expected request"); + + // The handler holds it, so the name stays duplicate until it answers, as in Rust. + expect(() => broadcast.createTrack("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 refuses to append past nothing, since that would reuse every explicit write. + const second = broadcast.createTrack("media", { timescale: Timescale.MILLI }); + expect(() => second.appendGroup()).toThrow("duplicate sequence"); + second.writeGroup(new GroupProducer(13)); + second.close(); + + // 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())).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()).toThrow("duplicate sequence"); + later.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("names that never wrote leave nothing behind", () => { + const sequences = new TrackSequences(); + sequences.get("written").next = 1; + const held = sequences.get("held"); + held.holders = 1; + + // 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 and held names survive every sweep. + expect(sequences.get("written").next).toBe(1); + expect(sequences.get("held")).toBe(held); +}); + +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(); + + expect(() => broadcast.createTrack("written").appendGroup()).toThrow("duplicate sequence"); + broadcast.close(); +}); + test("publishing-side subscriptions coalesce onto one request", async () => { const broadcast = new BroadcastProducer(); const pulled = wireOf(broadcast).requested(); @@ -236,14 +360,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(); }); @@ -372,8 +488,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()). @@ -384,8 +499,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(); @@ -496,10 +610,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 () => { @@ -689,38 +803,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 8526ec6be0..34acea2ae1 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; @@ -169,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"); @@ -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,35 +383,41 @@ 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. */ - insertTrack(track: track.Producer): void { + /** + * 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. 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. + */ + createTrack(name: string, info: Partial = {}): track.Producer { 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. - 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 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 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); - } - }); - } + // 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}`); + } - /** Create a track, insert it into the broadcast, and return its producer. */ - createTrack(name: string, info: Partial = {}): track.Producer { const producer = new track.Producer(name).accept(info); - this.insertTrack(producer); + 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 985eee8472..94c83af6fa 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -114,13 +114,47 @@ export interface ReadGroupFrame { complete(): void; } -/** The next group or datagram sequence, shared by every dynamic 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. */ + holders: number; } -/** Per-track sequence namespaces owned by one broadcast generation. */ -export type TrackSequences = Map; +/** + * 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 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`, created if the name is new. */ + get(name: string): TrackSequence { + const existing = this.#map.get(name); + if (existing) return existing; + + if (this.#map.size >= this.#sweepAt) { + for (const [key, sequence] of this.#map) { + if (sequence.next === 0 && sequence.holders === 0) this.#map.delete(key); + } + 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. */ export interface TrackRequestOptions { @@ -151,6 +185,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`. * @@ -215,6 +251,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 369baa017d..57630978b2 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -249,11 +249,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 `info` is invalid. + */ accept(info: Partial = {}): Producer { this.#pending.delete(this); - bindProducer(this.name, this.#producer, this.#sequences); - return this.#producer.accept(info); + try { + 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. + this.#producer.close(err instanceof Error ? err : new Error(String(err))); + throw err; + } } /** Reject the request, closing the track optionally with an error. */ @@ -416,15 +425,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 @@ -515,10 +515,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(); - #sequence: TrackSequence = { next: 0 }; + // 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 @@ -553,9 +555,17 @@ export class Producer { static { hooks.endTrack = (producer, end) => producer.#endPending(end); + // 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, producer.#received); producer.#sequence = sequence; + if (producer.#state.closed.peek() !== undefined) return; + own.holders--; + sequence.holders++; }; + hooks.bindSequence = bindProducerSequence; } /** @@ -839,7 +849,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, @@ -847,6 +857,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(); @@ -875,18 +906,18 @@ 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); this.#cap(group); - sequence.next = group.sequence + 1; this.#publish(group); return group; @@ -916,19 +947,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) => { @@ -948,19 +973,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; } @@ -968,7 +991,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. */ @@ -977,10 +1000,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 }); } @@ -1053,6 +1072,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/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts index 30a8cf9be9..49fbaeb7e5 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,163 @@ 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); + + // The name keeps its kind. + again.close(); + expect(() => broadcast.audio("video")).toThrow("registered as video"); + + audio.close(); + subscriber.close(); + 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(); +}); + +// 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(); + + 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"); + + // 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 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(); + 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..65a4a77bb3 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,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 or enabled state 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. @@ -86,6 +87,12 @@ export class Broadcast { // A static network track is exposed here only while at least one subscriber uses it. readonly #tracks = new Map>(); + // 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) { @@ -130,6 +137,11 @@ 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}`); + } const track = new Signal(undefined); this.#tracks.set(name, track); @@ -143,7 +155,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 +227,8 @@ export class Broadcast { const [_enabled, origin] = values; 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.`, @@ -235,30 +249,46 @@ 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. 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; 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, + }); + effect.cleanup(() => track.close()); + this.catalog.serve(track, effect, { compression }); + } } broadcast.announce({ epoch }); @@ -266,21 +296,31 @@ 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); + // 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?.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] }), + ); + tracks.set(rendition.name, track); + kinds.set(rendition.name, rendition.kind); + } + const served = track; + inner.run((demand) => { + demand.set(signal, demand.get(served.demand().used) ? served : undefined); }); } }); 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; diff --git a/quest/m0/README.md b/quest/m0/README.md index 8edcdcd9e0..e795ee09dc 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -63,5 +63,4 @@ remain and no release waits on them. - [TS duration fidelity](/quest/m0/ts-duration-fidelity.md) - Interop's TS compliance captures the whole round-tripped stream again, fixing a regression that turns Interop red on main - [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 diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 24985e11b5..af1292acff 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 - [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) ## Related 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 251f0a0c3d..0000000000 --- a/quest/m0/broadcast-epoch/publish-catalog-restart.md +++ /dev/null @@ -1,36 +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 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. - -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 264d9f2f04..0000000000 --- a/quest/m0/js-track-takeover.md +++ /dev/null @@ -1,34 +0,0 @@ -# [S] JS createTrack takes over a queued request - -## Goal - -In `@moq/net`, `createTrack` 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 - -Reconciled 2026-10-10: `insertTrack` cannot take over a request-owned -producer and remains out of scope. Its already-decided deletion belongs to -[Drop insertTrack](/quest/m1/js-insert-track.md), after this release fix. - -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. diff --git a/quest/m1/README.md b/quest/m1/README.md index 686e198ddf..f187cec1ee 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -144,7 +144,6 @@ folded into Dropped sources. - [Front deadlines](/quest/m1/front-deadline-index.md) - a front's per-event cost stops growing with its track count: an expiry index and per-track wakes, proven by a churn benchmark - [io_uring handshake deadline](/quest/m1/listener-deadlines.md) - the io_uring workers apply `listen.timeout` to the handshake - [Rust papercuts](/quest/m1/papercuts-rs.md) - HTTP refusals are counted, and a `u64::MAX` resume is unbounded -- [Drop insertTrack](/quest/m1/js-insert-track.md) - `createTrack` is `@moq/net`'s only way to add a track by hand, as in Rust - [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 - [Publish channel count](/quest/m1/publish-audio-channel-count.md) - forcing a channel count on an Audio.Capture stops costing the subscriber gaps of silence 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 diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index e537218d26..196f2667b2 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 @@ -1359,16 +1359,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(); @@ -1384,14 +1384,28 @@ mod test { let track = consumer.track("media").unwrap(); let mut second = dynamic.assert_request().accept(crate::track::Info::timed()); - 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", crate::track::Info::timed()).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 6ee2d2caaf..9b086da865 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -170,13 +170,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) } @@ -271,8 +271,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 @@ -1130,14 +1130,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) @@ -1840,7 +1846,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()?; @@ -1870,7 +1879,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 8ea059c8ed..d18be569e9 100644 --- a/rs/moq-net/tests/route_change.rs +++ b/rs/moq-net/tests/route_change.rs @@ -861,9 +861,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); @@ -913,8 +912,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}"); diff --git a/test/interop/clients/rust/tail.rs b/test/interop/clients/rust/tail.rs index ea02eb815a..769398d739 100644 --- a/test/interop/clients/rust/tail.rs +++ b/test/interop/clients/rust/tail.rs @@ -10,91 +10,91 @@ const BYTES: usize = 256 * 1024; #[tokio::main] async fn main() -> anyhow::Result<()> { - let args: Vec<_> = std::env::args().collect(); - ensure!( - args.len() == 4, - "usage: interop-tail publish|subscribe URL BROADCAST" - ); - let url: url::Url = args[2].parse()?; - let mut config = moq_tokio::connect::Config::default(); - config.websocket.enabled = Some(false); - let client = config.init(Default::default())?.with_reconnect(false); - let origin = moq_tokio::origin::spawn(); + let args: Vec<_> = std::env::args().collect(); + ensure!( + args.len() == 4, + "usage: interop-tail publish|subscribe URL BROADCAST" + ); + let url: url::Url = args[2].parse()?; + let mut config = moq_tokio::connect::Config::default(); + config.websocket.enabled = Some(false); + let client = config.init(Default::default())?.with_reconnect(false); + let origin = moq_tokio::origin::spawn(); - match args[1].as_str() { - "publish" => { - let broadcast = origin.create_broadcast(&args[3])?; - let mut track = broadcast.create_track( - "tail", - moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI), - )?; - broadcast.announce(Default::default())?; - let _session = client - .with_publisher(&origin) - .connect(url) - .established() - .await?; - track.demand().used().await?; - // Declare the end before the final group's payload is even available. - track.finish_at(moq_net::track::Position::group(GROUPS))?; - for sequence in 0..GROUPS { - let mut group = track.append_group()?; - group.write_frame(moq_net::Timestamp::ZERO, vec![sequence as u8; BYTES])?; - group.finish()?; - } - // A finished track is not a drained transport. Keep the session alive - // until the harness confirms the subscriber checked every byte and EOF. - let mut ack = String::new(); - tokio::io::BufReader::new(tokio::io::stdin()) - .read_line(&mut ack) - .await?; - ensure!( - ack == "clean end\n", - "reader did not acknowledge a clean end" - ); - broadcast.close(); - eprintln!("tail acknowledged"); - } - "subscribe" => { - let consumer = origin.consume(); - let _session = client - .with_subscriber(origin) - .connect(url) - .established() - .await?; - let broadcast = consumer.routed_broadcast(&args[3]).await?; - let options = moq_net::track::Subscription::default() - .with_max_delay(Duration::from_secs(5)) - .with_groups(0..); - let mut track = broadcast.track("tail")?.subscribe(Some(options)).await?; - let mut seen = Vec::new(); - while let Some(mut group) = track.recv_group().await? { - let sequence = group.sequence; - let frame = group.read_frame().await?.context("empty group")?; - ensure!(sequence < GROUPS, "unexpected group {sequence}"); - ensure!(frame.payload.len() == BYTES, "truncated group {sequence}"); - ensure!( - frame.payload.iter().all(|byte| *byte == sequence as u8), - "corrupt group {sequence}" - ); - ensure!( - group.read_frame().await?.is_none(), - "extra frame in group {sequence}" - ); - seen.push(sequence); - } - seen.sort_unstable(); - ensure!( - seen == (0..GROUPS).collect::>(), - "wrong groups: {seen:?}" - ); - ensure!( - track.finished().await? == moq_net::track::Position::group(GROUPS), - "wrong declared end" - ); - eprintln!("tail clean end=4 groups=0,1,2,3 bytes=1048576"); - } - _ => anyhow::bail!("unknown role: {}", args[1]), - } - Ok(()) + match args[1].as_str() { + "publish" => { + let broadcast = origin.create_broadcast(&args[3])?; + let mut track = broadcast.create_track( + "tail", + moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI), + )?; + broadcast.announce(Default::default())?; + let _session = client + .with_publisher(&origin) + .connect(url) + .established() + .await?; + track.demand().used().await?; + // Declare the end before the final group's payload is even available. + track.finish_at(moq_net::track::Position::group(GROUPS))?; + for sequence in 0..GROUPS { + let mut group = track.append_group()?; + group.write_frame(moq_net::Timestamp::ZERO, vec![sequence as u8; BYTES])?; + group.finish()?; + } + // A finished track is not a drained transport. Keep the session alive + // until the harness confirms the subscriber checked every byte and EOF. + let mut ack = String::new(); + tokio::io::BufReader::new(tokio::io::stdin()) + .read_line(&mut ack) + .await?; + ensure!( + ack == "clean end\n", + "reader did not acknowledge a clean end" + ); + broadcast.close(); + eprintln!("tail acknowledged"); + } + "subscribe" => { + let consumer = origin.consume(); + let _session = client + .with_subscriber(origin) + .connect(url) + .established() + .await?; + let broadcast = consumer.routed_broadcast(&args[3]).await?; + let options = moq_net::track::Subscription::default() + .with_max_delay(Duration::from_secs(5)) + .with_groups(0..); + let mut track = broadcast.track("tail")?.subscribe(Some(options)).await?; + let mut seen = Vec::new(); + while let Some(mut group) = track.recv_group().await? { + let sequence = group.sequence; + let frame = group.read_frame().await?.context("empty group")?; + ensure!(sequence < GROUPS, "unexpected group {sequence}"); + ensure!(frame.payload.len() == BYTES, "truncated group {sequence}"); + ensure!( + frame.payload.iter().all(|byte| *byte == sequence as u8), + "corrupt group {sequence}" + ); + ensure!( + group.read_frame().await?.is_none(), + "extra frame in group {sequence}" + ); + seen.push(sequence); + } + seen.sort_unstable(); + ensure!( + seen == (0..GROUPS).collect::>(), + "wrong groups: {seen:?}" + ); + ensure!( + track.finished().await? == moq_net::track::Position::group(GROUPS), + "wrong declared end" + ); + eprintln!("tail clean end=4 groups=0,1,2,3 bytes=1048576"); + } + _ => anyhow::bail!("unknown role: {}", args[1]), + } + Ok(()) }