diff --git a/demo/web/src/stats.ts b/demo/web/src/stats.ts index 954e554629..924efff05b 100644 --- a/demo/web/src/stats.ts +++ b/demo/web/src/stats.ts @@ -136,22 +136,26 @@ discovery.run((effect) => { const node = Net.Path.stripPrefix(prefix, path); if (!node) continue; - if (entry.kind !== "end") { - if (subs.has(node)) continue; - const ne = new Signals.Effect(); - subs.set(node, ne); - subscribeNode(ne, origin, path, node); - } else { + // The same node re-priced or failed over: its counters carry on. + if (entry.kind === "update") continue; + if (entry.kind === "start" && subs.has(node)) continue; + + // An end, or a restart: another instance at the path counts from zero, so neither + // series may splice across it. + if (entry.kind !== "start") { subs.get(node)?.close(); subs.delete(node); nodeStats.mutate((s) => { delete s[node]; }); - // A restart ends the old epoch and starts a new one at the same path, - // counting from zero, so neither series may splice across it. history.delete(node); clusterMembership = ""; } + if (entry.kind === "end") continue; + + const ne = new Signals.Effect(); + subs.set(node, ne); + subscribeNode(ne, origin, path, node); } }); }); diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 35384e504e..652af7df57 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -127,8 +127,17 @@ default, `auto`, sizes the buffer to how unevenly audio arrives, with the same [algorithm](/concept/audio-jitter) as the browser player. A duration fixes the delay and is also how long a stalled group is waited on before it is skipped. Each role follows the catalog, switching rendition when the publisher retires -the one playing. Playback is behind the `play` feature, since it pulls in -windowing and audio-device dependencies: +the one playing. + +The broadcast's announcement is its online signal, followed until the window +closes. Playback starts when the name is announced. A +[restarted publisher](#publisher-runs) starts it over on the new run, with a +fresh catalog, decoders, and clock. When the announcement ends, what is playing +finishes and the player waits for the name to return. The log names each run's +epoch. A catalog with nothing this build can play still ends the player. + +Playback is behind the `play` feature, since it pulls in windowing and +audio-device dependencies: ```bash cargo install moq-cli --no-default-features --features "iroh,noq,websocket,play" diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index db24d9c9ee..e3c41b51c2 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -35,7 +35,7 @@ broadcast.announce(); - **The origin holds the broadcasts, not the connection.** Closing a session unannounces them and leaves them created for the next one. A broadcast is invisible, locally and remotely, until `announce()`. `dynamic(prefix, route)` claims a prefix and yields each requested path to accept or reject. `accept` throws on a broadcast a session delivered, since JS apps do not proxy; copy its tracks into a broadcast you produce instead. - **Requests can pin an epoch.** `consumer.request(path, { epoch })` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs) and reports `unroutable` otherwise. The resolved broadcast's `epoch` names the route it came through, pinned or not. -- **Follow restarts.** An announce event of kind `"restart"` says another publisher instance now serves the prefix: request the path again. A request already resolved stays on the old instance, and ends with an `unroutable` error once that instance stops serving: it never moves onto another instance, since without a shared epoch nothing says the new one holds the same bytes. A blind answer (a session without discovery) ends with its session for the same reason, so only announcements can trigger following a restart. +- **Follow restarts.** `origin.follow(path)` reduces the routes covering a path to the one a request resolves (the most specific) and streams its `start`, `update`, `restart`, and `end`. A `"restart"` says another publisher instance now serves the path: request it again. A request already resolved stays on the old instance, and ends with an `unroutable` error once that instance stops serving: it never moves onto another instance, since without a shared epoch nothing says the new one holds the same bytes. A blind answer (a session without discovery) ends with its session for the same reason, so only announcements can trigger following a restart. - **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. diff --git a/doc/lib/js/watch.md b/doc/lib/js/watch.md index cd5a324dd2..a6451e3575 100644 --- a/doc/lib/js/watch.md +++ b/doc/lib/js/watch.md @@ -47,6 +47,14 @@ buffering indicator, and a stats panel. `error` the origin refused the broadcast, `el.broadcast.out.error` says why, and the player does not ask again until it is pointed elsewhere or re-enabled. +The broadcast's announcement is its online signal: the player starts when a +route serving the name is announced, goes `offline` when none is left, and +waits for one to return for as long as it is mounted. When the exact route +ends while a covering prefix still serves the name, it moves to the prefix. A +name outside the origin's scope is an `error`. A restarted publisher at the same name is followed at +once, with a fresh catalog, and video, audio, and the clock start over on its +timeline instead of waiting for it to catch up to the old one. + ## Binding from a framework `import "@moq/watch/element"` registers `` when the module diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index e118b38f0b..e28a622d05 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -17,7 +17,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.** `AnnounceEvent::Restart` says another publisher instance now serves the prefix: request the path again. A broadcast already resolved stays on the old instance until dropped or its route goes. +- **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. - **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`. diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index f85d83911d..7efdf2f73c 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -5,7 +5,7 @@ */ import { type GetPromise, Once, Signal } from "@moq/signals"; import type { Route } from "./hop.js"; -import type * as Path from "./path.js"; +import * as Path from "./path.js"; /** * A route over a prefix, delivered inside an {@link Event}. @@ -163,3 +163,64 @@ export class Consumer { closeState(this.#state, abort); } } + +/** + * A reducer from the announcements of the routes covering `path` to those of the one route + * serving it: the most specific. Another route taking over is a `restart`, or an `update` when + * both carry the same epoch, since those serve the same bytes. Routes beneath `path` serve + * other broadcasts and are skipped, as is any change no request sees. + * + * While nothing serves the path, the rest of a batch (one change to the table) folds into one + * `start`, so the replay a late follower begins with (a covering prefix, then the exact path + * beneath it) starts on the route serving the path rather than starting and restarting. Once + * something serves it, each change comes through on its own: an `end` followed by a `start` is + * a gap that ended any request on the old route, even when both routes carry one epoch. + * Synchronous, so a followed stream delivers in step with an announced one. Mirrors + * `announce::Follow` in rs/moq-net. + * + * @internal + */ +export function follower(path: Path.Valid): (events: Event[]) => Event[] { + // Every route standing over the path, by prefix. + const covering = new Map(); + const serving = () => { + let best: Announce | undefined; + for (const announce of covering.values()) { + if (!best || announce.prefix.length > best.prefix.length) best = announce; + } + return best; + }; + + // Apply one route's event, returning what it means for the path alone. + const fold = ({ kind, ...announce }: Event): Event | undefined => { + if (!Path.hasPrefix(announce.prefix, path)) return undefined; + const before = serving(); + if (kind === "end") covering.delete(announce.prefix); + else covering.set(announce.prefix, announce); + const after = serving(); + + if (!before) return after && { kind: "start", ...after }; + if (!after) return { kind: "end", ...before }; + if (before.prefix !== after.prefix) { + const same = before.route.epoch !== undefined && before.route.epoch === after.route.epoch; + return { kind: same ? "update" : "restart", ...after }; + } + // A route less specific than the serving one changing is skipped: no request sees it. + if (after.prefix !== announce.prefix) return undefined; + return { kind: kind === "restart" ? "restart" : "update", ...after }; + }; + + return (events) => { + const reduced: Event[] = []; + let idle = !serving(); + for (const event of events) { + const change = fold(event); + if (idle || !change) continue; + reduced.push(change); + idle = change.kind === "end"; + } + const after = serving(); + if (idle && after) reduced.push({ kind: "start", ...after }); + return reduced; + }; +} diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 45fbe8b65f..e087b6a18f 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -459,7 +459,7 @@ export class Publisher { const onWire = (route: Route): Route => (hasRouteCost(this.version) ? route : { ...route, cost: Cost.zero }); const announce = async (suffix: Path.Valid, route: Route) => { - console.debug(`announce: broadcast=${suffix} active=true`); + console.debug(`announce: broadcast=${suffix} active=true epoch=${route.epoch}`); if (hasAnnounceId(this.version)) announceIds.set(suffix, nextAnnounceId++); await encodeAnnounceBroadcast( stream.writer, @@ -495,7 +495,7 @@ export class Publisher { await announce(suffix, route); return; } - console.debug(`announce: broadcast=${suffix} restart=true`); + console.debug(`announce: broadcast=${suffix} restart=true epoch=${route.epoch}`); await encodeAnnounceBroadcast( stream.writer, { status: "restart", id, epoch: route.epoch, hops: wireHops(route), cost: route.cost }, diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index e1c5991d3f..afbab2cca2 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -510,7 +510,7 @@ export class Subscriber { advertised.set(path, { live: true, route, captures }); if (epoch) this.#epochs.set(path, epoch); else this.#epochs.delete(path); - console.debug(`announced: broadcast=${path} restart=true`); + console.debug(`announced: broadcast=${path} restart=true epoch=${epoch}`); announced.append({ prefix: path, captures, kind: "restart", route }); continue; } @@ -533,7 +533,7 @@ export class Subscriber { if (epoch) this.#epochs.set(path, epoch); else this.#epochs.delete(path); - console.debug(`announced: broadcast=${path} active=true`); + console.debug(`announced: broadcast=${path} active=true epoch=${epoch}`); announced.append({ prefix: path, captures, kind: "start", route }); } diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 821e1c6648..f790460e9d 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -331,6 +331,138 @@ test("announced streams the table under a scope with origin-relative paths", asy a.close(); }); +test("follow reports the route serving the path", async () => { + const origin = new Producer(); + const follow = origin.consume().follow(Path.from("pool/job")); + const epoch = Epoch.mint(); + const next = async () => { + const event = await follow.next(); + return event && { kind: event.kind, prefix: event.prefix }; + }; + + // A prefix above the path covers it. + const pool = origin.dynamic(Path.from("pool"), { epoch }); + expect(await next()).toEqual({ kind: "start", prefix: Path.from("pool") }); + + // A route beneath the path serves another broadcast, so the next event is the exact path's. + const beneath = publish(origin, Path.from("pool/job/thumbnail")); + + // The path itself, from the same instance, takes over without a restart. + const exact = origin.createBroadcast(Path.from("pool/job")); + exact.announce({ epoch }); + expect(await next()).toEqual({ kind: "update", prefix: Path.from("pool/job") }); + + // The covering prefix no longer serves the path, so its re-price is not seen: the next + // event is another instance at the path. + pool.update({ ...pool.route, cost: 5n }); + exact.announce({ epoch: Epoch.mint() }); + expect(await next()).toEqual({ kind: "restart", prefix: Path.from("pool/job") }); + + // It goes, so the prefix serves the path again: another instance. + exact.close(); + expect(await next()).toEqual({ kind: "restart", prefix: Path.from("pool") }); + + pool.update({ ...pool.route, cost: 9n }); + expect(await next()).toEqual({ kind: "update", prefix: Path.from("pool") }); + + pool.close(); + expect(await next()).toEqual({ kind: "end", prefix: Path.from("pool") }); + + // The stream ends with the origin. + origin.close(); + expect(await follow.next()).toBeUndefined(); + beneath.close(); +}); + +test("follow restarts onto a more specific route without an epoch", async () => { + const origin = new Producer(); + const follow = origin.consume().follow(Path.from("pool/job")); + + const pool = origin.dynamic(Path.from("pool")); + expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool") }); + + const exact = publish(origin, Path.from("pool/job")); + expect(await follow.next()).toMatchObject({ kind: "restart", prefix: Path.from("pool/job") }); + + follow.close(); + exact.close(); + pool.close(); + origin.close(); +}); + +test("follow ignores a route scoped beneath the path", async () => { + const origin = new Producer(); + // A cheaper dynamic at the same prefix that claims only `room/*/chat`, so it never serves the path. + const chat = origin.scope(Path.empty(), new Path.Patterns([Path.Pattern.parse("room/*/chat")])); + const dynamic = chat.dynamic(Path.from("room/alice")); + const follow = origin.consume().follow(Path.from("room/alice")); + + const first = origin.createBroadcast(Path.from("room/alice")); + first.announce({ cost: 5n }); + expect(await follow.next()).toMatchObject({ kind: "start", route: { cost: 5n } }); + + // A republish is another instance of the route that serves the path. + const second = origin.createBroadcast(Path.from("room/alice")); + second.announce({ cost: 5n }); + first.close(); + expect(await follow.next()).toMatchObject({ kind: "restart", route: { cost: 5n } }); + + follow.close(); + second.close(); + dynamic.close(); + origin.close(); +}); + +test("follow starts on the serving route when joining late", async () => { + const origin = new Producer(); + const pool = origin.dynamic(Path.from("pool")); + const exact = publish(origin, Path.from("pool/job")); + + const follow = origin.consume().follow(Path.from("pool/job")); + expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool/job") }); + + // Nothing else is queued: the next event is the exact route ending, a restart onto the prefix. + exact.close(); + expect(await follow.next()).toMatchObject({ kind: "restart", prefix: Path.from("pool") }); + + follow.close(); + pool.close(); + origin.close(); +}); + +test("follow keeps a gap between routes of one epoch", async () => { + const origin = new Producer(); + const epoch = Epoch.mint(); + const exact = origin.createBroadcast(Path.from("pool/job")); + exact.announce({ epoch }); + const follow = origin.consume().follow(Path.from("pool/job")); + expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool/job") }); + + // Both changes land before the table is read again, but the path went unserved in between, + // which ended any request on it: a start of its own, not an update. + exact.close(); + const pool = origin.dynamic(Path.from("pool"), { epoch }); + expect(await follow.next()).toMatchObject({ kind: "end", prefix: Path.from("pool/job") }); + expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool") }); + + follow.close(); + pool.close(); + origin.close(); +}); + +test("follow refuses a path no pattern can spell", () => { + const origin = new Producer(); + expect(() => origin.consume().follow(Path.from("camera*main"))).toThrow(); + origin.close(); +}); + +test("follow refuses a path outside the scope", () => { + const origin = new Producer(); + const scoped = origin.scope(Path.empty(), new Path.Patterns([Path.Pattern.subtree(Path.from("pool/job/cam"))])); + expect(() => scoped.consume().follow(Path.from("pool/job"))).toThrow(); + origin.close(); +}); + test("hidden paths need an opt-in or a scope naming the dot segment", async () => { const origin = new Producer(); const consumer = origin.consume(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index a697275f7e..f67bd9fdd5 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -858,6 +858,9 @@ export interface Table { /** The available broadcasts under `scope`, as a live stream; see {@link Consumer.announced}. */ announced(scope?: Path.Pattern, options?: announce.Options): announce.Consumer; + /** The announcements of the route serving `path`, as a live stream; see {@link Consumer.follow}. */ + follow(path: Path.Valid): announce.Consumer; + /** Advertise a prefix and serve requests under it; see {@link Producer.dynamic}. */ dynamic( prefix: Path.Valid, @@ -1222,6 +1225,11 @@ export class Producer implements Table { return this.#reader.announced(scope, options); } + /** The announcements of the route serving `path`, as a live stream; see {@link Consumer.follow}. */ + follow(path: Path.Valid): announce.Consumer { + return this.#reader.follow(path); + } + /** Close the origin, every broadcast it still routes, and its announcement streams. Idempotent. */ close(abort?: Error) { if (this.#state.closed.peek() !== undefined) return; @@ -1619,6 +1627,29 @@ export class Consumer { return producer.consume(); } + /** + * The announcements of the routes covering `path`, reduced to the one serving it: the most + * specific, which is the one a {@link request} resolves. + * + * This is how a player follows a broadcast across publisher restarts: play on `start`, drop + * everything and request the path afresh on `restart`, and stop on `end`. An `update` is + * the same publisher instance re-priced or failed over, which subscriptions already ride + * out. Another route taking over is a `restart`, or an `update` when both carry the same + * epoch. Throws when the path is outside this consumer's scope, or when no pattern can spell + * it (a segment containing `*`). Close it when done. + */ + follow(path: Path.Valid): announce.Consumer { + this.#scope.path(path); + const producer = new announce.Producer(); + // Scoped to the path itself, which still sees every route whose claim covers it, so a + // route claiming only paths beneath it never masks the one that serves it, just as a + // request skips it. A path no pattern can spell (a segment containing `*`) throws here. + const patterns = this.#scope.patterns(Path.Pattern.literal(path)); + // Hiding narrows discovery, not lookup, so a hidden path follows like any other. + void this.#runAnnounced(producer, patterns, true, announce.follower(path)); + return producer.consume(); + } + /** One snapshot shared by map readers and announcement-stream diffing. */ #listed(patterns: Path.Patterns, hidden: boolean): Map { const next = new Map(); @@ -1653,7 +1684,18 @@ export class Consumer { return next; } - async #runAnnounced(producer: announce.Producer, patterns: Path.Patterns, hidden: boolean): Promise { + async #runAnnounced( + producer: announce.Producer, + patterns: Path.Patterns, + hidden: boolean, + reduce: (events: announce.Event[]) => announce.Event[] = (events) => events, + ): Promise { + // Each pass's changes, reduced together so a follower sees one table change as one event. + const batch: announce.Event[] = []; + const append = (event: announce.Event) => { + batch.push(event); + }; + // Keyed by the presented path (from the origin, not the scope), valued by identity // plus route. Diffing the instance rather than mere presence means a republish emits a // restart; a re-price of the same instance emits an update. @@ -1670,7 +1712,7 @@ export class Consumer { for (const [path, snap] of active) { if (!next.has(path)) - producer.append({ + append({ prefix: path, captures: snap.captures, kind: "end", @@ -1680,14 +1722,15 @@ export class Consumer { for (const [path, snap] of next) { const prev = active.get(path); if (!prev) { - producer.append({ prefix: path, captures: snap.captures, kind: "start", route: snap.route }); + append({ prefix: path, captures: snap.captures, kind: "start", route: snap.route }); } else if (!sameInstance(prev, snap)) { - producer.append({ prefix: path, captures: snap.captures, kind: "restart", route: snap.route }); + append({ prefix: path, captures: snap.captures, kind: "restart", route: snap.route }); } else if (!routesEqual(prev.route, snap.route)) { - producer.append({ prefix: path, captures: snap.captures, kind: "update", route: snap.route }); + append({ prefix: path, captures: snap.captures, kind: "update", route: snap.route }); } } active = next; + for (const event of reduce(batch.splice(0))) producer.append(event); await Signal.race(this.#state.local, this.#state.advertisedLocal, this.#state.routes, producer.closed); if (producer.closed.peek() !== undefined) return; diff --git a/js/watch/src/audio/decoder.test.ts b/js/watch/src/audio/decoder.test.ts index 73276c30f9..c74243259f 100644 --- a/js/watch/src/audio/decoder.test.ts +++ b/js/watch/src/audio/decoder.test.ts @@ -420,12 +420,16 @@ describe("Decoder across a rendition's absence", () => { playback.remove(); await microtasks(); + const reanchor = spyOn(Sync.prototype, "reset"); playback.replace(); playback.restore(); await microtasks(); await playback.play(12); - // The ring re-anchors on the new broadcast rather than discarding it as already played. + // The ring and the shared clock re-anchor on the new broadcast rather than discarding it as + // already played. + expect(reanchor).toHaveBeenCalledTimes(1); + reanchor.mockRestore(); expect(drain(ring)).toBeGreaterThan(0); expect(contexts).toHaveLength(1); expect(playback.rings()).toEqual([ring]); diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index e5aa18b8c4..83cc96b270 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -334,13 +334,15 @@ export class Decoder { const active = broadcast.relativeBroadcast(effect, identity.broadcast); if (!active) return; - // Another broadcast (a new name, or a republish) brings its own timeline, which a ring anchored - // on the previous one would discard as already played. A return to the same instance keeps the - // anchor, so the relay's redelivered last group is still dropped as old. Every handle to one - // instance shares `closed`, so it identifies the instance where the handle itself does not. + // Another broadcast (a new name, or a republish) brings its own timeline, which a ring and a + // clock anchored on the previous one would discard as already played. A return to the same + // instance keeps the anchor, so the relay's redelivered last group is still dropped as old. + // Every handle to one instance shares `closed`, so it identifies the instance where the + // handle itself does not. if (this.#instance !== undefined && this.#instance !== active.closed) { this.#ring.peek()?.reset(); this.#decodeBuffered.set([]); + this.sync.reset(); } this.#instance = active.closed; diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index 594ca34421..d5d3b7be68 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -313,7 +313,7 @@ describe("restart", () => { const old = source.out.active.peek(); expect(old).toBeDefined(); - // A newer publisher announces the exact path: a start, not a restart, for this name. + // A newer publisher announces the exact path, which now serves the name: a restart. const exact = owner.createBroadcast(name); exact.announce({ epoch: Moq.Epoch.mint() }); await settle(); @@ -333,9 +333,97 @@ describe("restart", () => { owner.close(); await settle(); }); + + it("resumes on a covering prefix of the same epoch after a gap", async () => { + const owner = new Origin.Producer(); + const name = Path.from("pool/job.hang"); + const epoch = Moq.Epoch.mint(); + const exact = owner.createBroadcast(name); + exact.announce({ epoch }); + const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); + await settle(); + const old = source.out.active.peek(); + expect(old).toBeDefined(); + + // The exact route goes, which ends the request on it, and the prefix arrives under the same + // epoch before the table is read again. The player still has to ask again. + exact.close(); + const pool = owner.dynamic(Path.from("pool"), { epoch }); + const requests = pool.requested(); + const upstream = new Moq.Broadcast.Producer(); + void (async () => { + for await (const request of requests) request.accept(upstream); + })(); + await settle(); + const resumed = source.out.active.peek(); + expect(resumed).toBeDefined(); + expect(resumed).not.toBe(old); + expect(source.out.error.peek()).toBeUndefined(); + + upstream.close(); + pool.close(); + source.close(); + owner.close(); + await settle(); + }); + + it("falls back to a covering prefix when the exact route ends", async () => { + const owner = new Origin.Producer(); + const name = Path.from("pool/job.hang"); + const pool = owner.dynamic(Path.from("pool")); + const requests = pool.requested(); + const upstream = new Moq.Broadcast.Producer(); + void (async () => { + for await (const request of requests) request.accept(upstream); + })(); + const exact = publish(owner, name); + const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); + await settle(); + const old = source.out.active.peek(); + expect(old).toBeDefined(); + + // The exact route goes while the prefix still covers the name: a restart onto the prefix. + exact.close(); + await settle(); + const fallback = source.out.active.peek(); + expect(fallback).toBeDefined(); + expect(fallback).not.toBe(old); + expect(source.out.error.peek()).toBeUndefined(); + + // Nothing covers the name any more: offline, not an error. + pool.close(); + await settle(); + expect(source.out.active.peek()).toBeUndefined(); + expect(source.out.error.peek()).toBeUndefined(); + + upstream.close(); + source.close(); + owner.close(); + await settle(); + }); }); describe("refusal", () => { + it("refuses a name outside the origin's scope", async () => { + const owner = new Origin.Producer(); + const scoped = owner.scope(Path.empty(), new Path.Patterns([Path.Pattern.subtree(Path.from("room"))])); + const source = new Broadcast({ + origin: scoped, + name: Path.from("lobby/live.hang"), + enabled: true, + announced: true, + catalogFormat: "manual", + }); + await settle(); + + expect(source.out.error.peek()?.message).toBe("path is outside the origin scope"); + expect(source.out.status.peek()).toBe("error"); + + source.close(); + owner.close(); + await settle(); + }); + for (const announced of [true, false]) { it(`reports a refusal until a fresh request (announced: ${announced})`, async () => { const owner = new Origin.Producer(); diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index b5e4993d0e..e82da627aa 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -260,9 +260,10 @@ export class Broadcast { // Subscribe to the broadcast, by default waiting for its announcement so we never race a // publisher that comes online after us. A request stays on the publisher instance it resolved - // and ends once that stops serving, so a republish or reconnect is followed by requesting - // again whenever the path is announced (a start or a restart); the request ending is never - // the trigger. Mirror its handle into `active`, and a refusal into `error`. + // and ends once that stops serving, so a republish, a reconnect, or a covering prefix taking + // over is followed by requesting again whenever the route serving the path starts or restarts; + // the request ending is never the trigger. Mirror its handle into `active`, and a refusal + // into `error`. #runBroadcast(effect: Effect): void { const enabled = effect.get(this.in.enabled); if (!enabled) return; @@ -273,21 +274,29 @@ export class Broadcast { const name = effect.get(this.in.name); const announced = effect.get(this.in.announced); + // Observed whether or not the first request waits for an announcement: announcements are + // the only restart signal. Each start or restart of the route serving the name requests + // afresh, and the fresh request replaces the current one unless both resolve the same + // broadcast. Following first means a name outside the origin's scope is refused here, + // before any request, and reported like any other refusal. + let stream: Moq.Announce.Consumer; + try { + stream = origin.follow(name); + } catch (err) { + effect.set(this.#out.error, err instanceof Error ? err : new Error(String(err)), undefined); + return; + } + effect.cleanup(() => stream.close()); + const current = new Signal(origin.request(name, { announced })); effect.cleanup(() => current.peek().close()); - // Observed whether or not the first request waits for an announcement: announcements are - // the only restart signal. Each start or restart covering the name requests afresh, and - // the fresh request replaces the current one unless both resolve the same broadcast. - const stream = origin.announced(Path.Pattern.literal(name), { hidden: true }); - effect.cleanup(() => stream.close()); effect.spawn(async () => { for (;;) { const entry = await effect.race(stream.next()); // An event that settled just before teardown still resumes here: open nothing then. if (!entry || effect.abort.aborted) break; if (entry.kind === "end" || entry.kind === "update") continue; - if (!Path.hasPrefix(entry.prefix, name)) continue; const previous = current.peek(); const fresh = origin.request(name, { announced }); const was = previous.active.peek(); diff --git a/js/watch/src/video/decoder.test.ts b/js/watch/src/video/decoder.test.ts index 2df963a73e..71ba45a02c 100644 --- a/js/watch/src/video/decoder.test.ts +++ b/js/watch/src/video/decoder.test.ts @@ -3,7 +3,7 @@ import * as Catalog from "@moq/hang/catalog"; import * as Container from "@moq/hang/container"; import * as Moq from "@moq/net"; import { Time } from "@moq/net"; -import { Signal } from "@moq/signals"; +import { type Effect, Signal } from "@moq/signals"; import type { Broadcast } from "../broadcast"; import { Sync } from "../sync"; import { Decoder } from "./decoder"; @@ -24,14 +24,17 @@ function config(fields: Record): Catalog.VideoConfig { } // The subscription is already closed, so the track never reaches VideoDecoder. -function closedConsumer(): Moq.Broadcast.Consumer { - return { closed: new Signal("ended") } as unknown as Moq.Broadcast.Consumer; +function closedConsumer(path: string): Moq.Broadcast.Consumer { + return { closed: new Signal("ended"), path } as unknown as Moq.Broadcast.Consumer; } +// Each source is another broadcast, at its own path. +let sources = 0; function watchBroadcast(catalog: Signal): Broadcast { + const media = closedConsumer(`source-${sources++}`); return { - out: { catalog }, - relativeBroadcast: () => closedConsumer(), + out: { catalog, active: new Signal(undefined) }, + relativeBroadcast: () => media, } as unknown as Broadcast; } @@ -245,13 +248,23 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[], hold: rea const video = config({ codedWidth: 16, codedHeight: 16, description: "01640028ffe100046764002801000268ee" }); if (kind === "cmaf") video.container = { kind, init: Buffer.from(Container.Cmaf.createVideoInitSegment(video)).toString("base64") }; - const producer = new Moq.Broadcast.Producer(); - const consumer = producer.consume(); + // The catalog's own broadcast, and the one the rendition reads. They differ once only one of + // them is republished, as with a rendition's `broadcast` override. + let producer = new Moq.Broadcast.Producer(); + let mediaProducer = producer; + const active = new Signal(producer.consume()); + const media = new Signal(producer.consume()); const catalog = new Signal({ video: { renditions: { video } } }); const broadcast = new Signal({ - out: { catalog }, - relativeBroadcast: () => consumer, + out: { catalog, active }, + relativeBroadcast: (effect: Effect) => effect.get(media), } as unknown as Broadcast); + const closeAll = () => { + active.peek().close(); + media.peek().close(); + producer.close(); + mediaProducer.close(); + }; const source = new Source({ broadcast, supported: async () => true }); const sync = new Sync({ delay: "instant" }); const enabled = new Signal(true); @@ -263,15 +276,35 @@ async function guardedPlayback(kind: "legacy" | "cmaf", reads: Read[], hold: rea catalog, submitted, outputs, + sync, get resets() { return resets; }, + // A restarted publisher takes the name: another instance of the same broadcast. + republish() { + closeAll(); + producer = new Moq.Broadcast.Producer(); + mediaProducer = producer; + active.set(producer.consume()); + media.set(producer.consume()); + }, + // Only the catalog's broadcast restarts; the media the rendition reads stays up. + republishCatalog() { + active.peek().close(); + producer = new Moq.Broadcast.Producer(); + active.set(producer.consume()); + }, + // Only the media the rendition reads restarts, under the same catalog. + republishMedia() { + media.peek().close(); + mediaProducer = new Moq.Broadcast.Producer(); + media.set(mediaProducer.consume()); + }, close() { decoder.close(); source.close(); sync.close(); - consumer.close(); - producer.close(); + closeAll(); next.mockRestore(); for (const [name, original] of [ ["VideoDecoder", originalDecoder], @@ -302,6 +335,59 @@ it("promoting from no active track holds its picture and timestamp until a frame } }); +it("switches to a republished broadcast at once, its timeline starting over", async () => { + const reads = [sample(10, 10_000_000)]; + const playback = await guardedPlayback("legacy", reads); + try { + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(10_000)); + const reset = spyOn(playback.sync, "reset"); + + reads.push(sample(0, 1000)); + playback.republish(); + await microtasks(); + + // Waiting to catch up to the old picture would hold it until the new timeline passed 10s. + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(1)); + expect(reset).toHaveBeenCalledTimes(1); + } finally { + playback.close(); + } +}); + +it("switches at once when only the media a rendition reads is republished", async () => { + const reads = [sample(10, 10_000_000)]; + const playback = await guardedPlayback("legacy", reads); + try { + const reset = spyOn(playback.sync, "reset"); + + reads.push(sample(0, 1000)); + playback.republishMedia(); + await microtasks(); + + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(1)); + expect(reset).toHaveBeenCalledTimes(1); + } finally { + playback.close(); + } +}); + +it("keeps the clock when only the catalog's broadcast is republished", async () => { + const reads = [sample(10, 10_000_000)]; + const playback = await guardedPlayback("legacy", reads); + try { + const reset = spyOn(playback.sync, "reset"); + + playback.republishCatalog(); + await microtasks(); + + // The media is the same instance, so its timeline did not start over. + expect(playback.decoder.out.timestamp.peek()).toBe(Time.Milli(10_000)); + expect(reset).not.toHaveBeenCalled(); + } finally { + playback.close(); + } +}); + it("clears the picture once no rendition is selectable", async () => { const playback = await guardedPlayback("legacy", [sample(10, 1000)]); try { diff --git a/js/watch/src/video/decoder.ts b/js/watch/src/video/decoder.ts index ecda6bb27e..9a4986e54d 100644 --- a/js/watch/src/video/decoder.ts +++ b/js/watch/src/video/decoder.ts @@ -98,6 +98,9 @@ export class Decoder { #pending = new Signal<{ track: string; catalog: Getter } | undefined>(undefined); // The active track's arrival estimate from its container consumer. #spread = new Signal(undefined); + // The path of the media the last rendition read, and its instance. Every handle to one instance + // shares `closed`, so it identifies the instance where the handle itself does not. + #instance?: { path: Moq.Path.Valid; closed: unknown }; readonly #identity: Computed; #signals = new Effect(); @@ -188,6 +191,15 @@ export class Decoder { return; } + // A republish of the media this rendition reads (the catalog's own broadcast, or the one a + // `broadcast` override names) is another instance, whose timeline starts over. Waiting for + // it to catch up to the old picture would freeze video until it passed the old timestamps, + // so switch at once and re-anchor the clock. A rendition switch within one instance, or a + // switch to another path, still waits. + const republished = this.#instance?.path === active.path && this.#instance.closed !== active.closed; + this.#instance = { path: active.path, closed: active.closed }; + if (republished) this.sync.reset(); + // Start a new pending effect. let pending: DecoderTrack | undefined = new DecoderTrack({ sync: this.sync, @@ -205,7 +217,7 @@ export class Decoder { if (!pending) return; const current = effect.get(this.#active); - if (current) { + if (current && !republished) { // A zero timestamp is a rendered frame, not a missing one. const pendingTimestamp = effect.get(pending.timestamp); if (pendingTimestamp === undefined) return; diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 5e401842cf..54ce126e8f 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -59,9 +59,9 @@ Decided: to m1. - A catalog `broadcast` reference by name follows the newest epoch, since a path cannot name one. -- Every first-party publisher that can restart mints its own: the apps and - moq-boy through [Apps](/quest/m0/broadcast-epoch/apps.md), the - ingest gateways, moqsink, and the bindings below. moq-stats mints one per +- Every first-party publisher that can restart mints its own: the apps, + moq-boy, the ingest gateways, moqsink, and the bindings below. Players + (`moq play`, `@moq/watch`, demo/web) follow the announce `Restart`. moq-stats mints one per [group announcement](/doc/concept/stats.md#broadcasts), which also gated the release (decided 2026-10-04). - The m1 quests gating this line moved under it in the 2026-10-05 audit, and @@ -97,7 +97,6 @@ then. MoQ Pro's VOD `storage.json` moves from the same seed to its own epoch. ## Required -- [Apps](/quest/m0/broadcast-epoch/apps.md) - moq-cli, the browser publish and watch components, and demo/web restart into a new epoch and reset on the switch - [Export ts](/quest/m0/broadcast-epoch/export-ts.md) - `export ts` and SRT egress linger only for the same epoch, and switch to a replacement only with `--stitch`, as a full program switch - [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 diff --git a/quest/m0/broadcast-epoch/apps.md b/quest/m0/broadcast-epoch/apps.md deleted file mode 100644 index 4d5ad02b4d..0000000000 --- a/quest/m0/broadcast-epoch/apps.md +++ /dev/null @@ -1,25 +0,0 @@ -# [M] Apps restart into a new epoch and reset on the switch - -## Goal - -`moq-cli` publish and play, `@moq/publish`, `@moq/watch`, `demo/web`, and -moq-boy announce a freshly minted epoch per run. Each publish -run is a new epoch, so a restart while the -old route lingers is a new broadcast rather than a resume into the old one, -which stalls viewers until the new run's group sequence catches up. A viewer -switches to a republish within an RTT. Logs show the epoch. - -## Plan - -- Publish sides mint per run as of #4942: moq-cli publish, HLS import, - archive replay, WHEP import, and transcode output; moq-boy; `@moq/publish`, - the clock, and moq-boy's viewer feedback. What remains is below. -- Watch sides handle "the broadcast changed" as a fresh catalog and decoder - reset. Players (`moq play`, `@moq/watch`, demo/web) start on `Start`, - restart on `Restart`, and stop on `End`, so a re-price or a same-epoch - failover (an `Update`) never restarts playback. #4970, which drove them - from announcements, closed unmerged. Test a republish mid-playback in the - browser and native players. -- moq-room is out (decided 2026-10-08): it only discovers participants, and - the caller that publishes into a room mints the epoch. - diff --git a/quest/m0/broadcast-epoch/moqsrc.md b/quest/m0/broadcast-epoch/moqsrc.md index 23304cd899..4a2759e99e 100644 --- a/quest/m0/broadcast-epoch/moqsrc.md +++ b/quest/m0/broadcast-epoch/moqsrc.md @@ -61,9 +61,8 @@ Decided in planning (2026-10-06, after #4955 found that no quest covered still drains to EOS. No timer and no property: announcements already say when the source is back, and a timer cannot tell a slow restart from a dead source. -- **Where.** The logic stays in `rs/moq-gst/src/source`. If - [Apps](/quest/m0/broadcast-epoch/apps.md) lands a shared helper for - following announcements first, use it instead. +- **Where.** The logic stays in `rs/moq-gst/src/source`, following the path + through moq-net's `origin::Consumer::follow`, as `moq play` does. - **Test.** An in-process test republishes the path under a newer epoch while the old publisher stays up: the new run's buffers render through a synced sink, a pad linked by name keeps flowing, and the bus carries no error. diff --git a/quest/m1/README.md b/quest/m1/README.md index 3049a6b040..1f37328816 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -194,7 +194,6 @@ ladders. - [Catalog colour](/quest/m1/color-catalog.md) - the catalog describes a rendition's colour and HDR properties, which the WebGPU HDR renderer reads - [WebGPU HDR](/quest/m1/webgpu-hdr.md) - HDR renditions play as HDR where the browser and display can show it, and tone-map to SDR elsewhere - [Request ID order](/quest/m1/request-id-order.md) - drafts 14 to 16 refuse a reused or lower Request ID in both languages -- [Watch follows a covering prefix](/quest/m1/watch-follow-prefix.md) - `@moq/watch` moves to a covering prefix when the exact route ends, as `moq play` does - [Import catalog drop](/quest/m1/import-catalog-drop.md) - moq-cli import never drops a catalog producer without finishing it - [JS restart keeps the request](/quest/m1/js-restart-keeps-request.md) - a resolved `@moq/net` request survives a Restart, as in Rust - [WebGPU on Safari](/quest/m1/webgpu-safari.md) - the WebGPU renderer is verified on Safari 26 for macOS and iOS diff --git a/quest/m1/watch-follow-prefix.md b/quest/m1/watch-follow-prefix.md deleted file mode 100644 index df7357422d..0000000000 --- a/quest/m1/watch-follow-prefix.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] @moq/watch follows the route that serves its path - -## Goal - -When the exact route for a path ends while a covering prefix (such as -`pool`) still serves it, `@moq/watch` moves to the prefix instead of going -offline, as Rust's `moq play` does through the shared follow helper. - -## Plan - -Found in #5154 (broadcast-epoch apps), decided 2026-10-09: -`Broadcast.#runBroadcast` in `js/watch` ignores `End`, so the player goes -offline even though a covering prefix still announces the path. Start after -#5154 lands, since it adds the helper this quest uses. - -- Decided: use the shared follow helper. #5154 moves it from - `moq_mux::Source::follow` into moq-net (`origin::Consumer::follow`) and - mirrors it in `@moq/net`, per the cross-package table; this quest wires - `@moq/watch`'s `Broadcast` onto the JS one. The serving route is the most - specific one covering the path; another route taking over is a `Restart`, - or an `Update` when both carry the same epoch; routes beneath the path are - ignored. Rejected: a private copy in `@moq/watch`, and keeping the Rust - helper in moq-mux (maintainer, 2026-10-09: a net-layer concern that should - mirror across languages). -- Test: a path served both exactly and by a `pool` prefix; ending the exact - route moves playback to the prefix as a restart, and ending both goes - offline. - -Public API: none. Wire: none. diff --git a/rs/moq-boy/src/main.rs b/rs/moq-boy/src/main.rs index 4d684bed90..549e0dac83 100644 --- a/rs/moq-boy/src/main.rs +++ b/rs/moq-boy/src/main.rs @@ -236,9 +236,11 @@ async fn run(config: &Config) -> Result<()> { let mut broadcast = publish_origin .create_broadcast(&broadcast_path) .context("failed to create broadcast")?; + let epoch = moq_net::Epoch::mint(); broadcast - .announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) .context("failed to announce broadcast")?; + tracing::info!(broadcast = %broadcast_path, %epoch, "announced"); // Consume origin: viewer broadcasts under the viewer prefix. // JS publishes viewer feedback at "{viewer_prefix}/{name}/{viewerId}" diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 68178ea6e3..03c45e6e15 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -529,6 +529,7 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res // One publisher instance per run, unless redundant publishers named a shared one. let epoch = moq.epoch.clone().unwrap_or_else(moq_net::Epoch::mint); + tracing::info!(%epoch, "publisher instance"); for stage in stages { let name = stage.broadcast(&moq); diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 62b70aa946..fe9d4f406b 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -8,6 +8,7 @@ use std::time::Duration; use anyhow::Context; use hang::moq_net; use moq_mux::catalog::{self, Stream}; +use moq_net::announce::Event as AnnounceEvent; // tokio's clock, which is the wall clock unless a test pauses it to drive the // playout clock itself. use tokio::time::Instant; @@ -15,7 +16,6 @@ use tokio::time::Instant; use super::args::Args; use super::output::{Output, Sink, Speaker}; use super::playback::{Kind, Playback, joined}; -use super::source::subscribe; use super::timeline::{AudioTimeline, Presentation, fit, timestamp}; use super::video::Video; use super::window::Event; @@ -63,8 +63,62 @@ impl Media { output.send(event); } + /// Follow the path's announcements for as long as the origin lasts: a start + /// plays the broadcast, a restart drops it for the instance that replaced it, + /// and an end lets what is playing finish while waiting for the next start. An + /// update is the same instance, which its subscriptions already ride out. async fn play(self) -> anyhow::Result<()> { - let source = subscribe(self.origin.clone(), &self.broadcast).await?; + let mut follow = self.origin.follow(&self.broadcast)?; + let source = moq_mux::Source::new(self.origin.clone(), &self.broadcast); + let mut playing = None; + + loop { + tokio::select! { + event = follow.next() => { + // The origin is gone, so nothing can announce the path again. + let Some(event) = event else { return Ok(()) }; + let announce = match event { + AnnounceEvent::Start(announce) | AnnounceEvent::Restart(announce) => announce, + // The same instance, which what is playing rides out. With nothing + // playing, it is the path served again after a gap the announcements + // folded away (a covering route of one epoch arriving as the exact one + // went), so play it. + AnnounceEvent::Update(announce) if playing.is_none() => announce, + AnnounceEvent::Update(_) => continue, + AnnounceEvent::End(_) => { + tracing::info!(broadcast = %self.broadcast, "offline, waiting for it to return"); + continue; + } + }; + tracing::info!( + broadcast = %self.broadcast, + epoch = announce.route.epoch.as_ref().map(tracing::field::display), + "online" + ); + + // Another instance's timeline is its own, so nothing of the old one carries over. + drop(playing.take()); + *self.presentation.lock().unwrap() = Presentation::new(self.args.video_delay()); + self.video.lock().unwrap().clear(); + self.drained.notify_one(); + self.output.send(Event::Wake); + playing = Some(Box::pin(self.play_broadcast(source.clone()))); + } + result = async { playing.as_mut().expect("guarded").await }, if playing.is_some() => { + playing = None; + match result { + Ok(()) => tracing::info!(broadcast = %self.broadcast, "broadcast ended"), + Err(err) if err.is::() => return Err(err), + Err(err) => tracing::warn!(broadcast = %self.broadcast, err = format!("{err:#}"), "broadcast ended"), + } + } + } + } + } + + /// Play the broadcast at the path until its catalog and every track it + /// started end. + async fn play_broadcast(&self, source: moq_mux::Source) -> anyhow::Result<()> { let broadcast = source .broadcast() .await @@ -126,7 +180,9 @@ impl Media { match snapshot.context("failed to read the catalog")? { Some(snapshot) => playback.received(snapshot), None => { - anyhow::ensure!(playback.played, "the catalog contains no playable audio or video renditions"); + if !playback.played { + return Err(Unplayable("the catalog contains no playable audio or video renditions".into()).into()); + } playback.catalog_ended = true; } } @@ -268,15 +324,28 @@ impl Media { // Renditions on offer and not one of them playable, with nothing // already running to fall back on. - anyhow::ensure!( - !tasks.is_empty() || rejected.is_empty(), - "no playable rendition in the catalog: {}", - rejected.join("; ") - ); + if tasks.is_empty() && !rejected.is_empty() { + return Err( + Unplayable(format!("no playable rendition in the catalog: {}", rejected.join("; "))).into(), + ); + } } } } +/// A broadcast this build cannot play, which ends the player. A broadcast that +/// ends or fails on the wire does not: the next announcement plays again. +#[derive(Debug)] +struct Unplayable(String); + +impl std::fmt::Display for Unplayable { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +impl std::error::Error for Unplayable {} + struct AudioPlayback { changed: Arc, speaker: O::Speaker, @@ -538,12 +607,7 @@ mod tests { drop(new); catalog.finish().unwrap(); - player.await.unwrap(); - match recorder.events().pop() { - Some(Event::Ended) => {} - Some(Event::Failed(err)) => panic!("playback failed: {err}"), - _ => panic!("playback never ended"), - } + settle(player, &recorder).await; let played = recorder.played(); let old_end = played.iter().filter(|p| p.sample == OLD).map(|p| p.to).max().unwrap(); @@ -555,6 +619,233 @@ mod tests { let old_duration: Duration = played.iter().filter(|p| p.sample == OLD).map(|p| p.to - p.from).sum(); assert_eq!(old_duration, Duration::from_secs(1)); } + + /// Publish an audio-only broadcast at `room` under `route`, returning what keeps + /// it up and its one rendition. + fn publish( + origin: &moq_net::origin::Producer, + route: moq_net::origin::Route, + ) -> ( + moq_net::broadcast::Producer, + catalog::Producer, + moq_mux::container::Producer, + ) { + let mut broadcast = origin.create_broadcast("room").unwrap(); + let catalog = catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let audio = rendition(&broadcast, &catalog, "audio"); + broadcast.announce(route).unwrap(); + (broadcast, catalog, audio) + } + + /// Write `count` packets of `sample` in real time, its timeline starting at 0. + async fn write(audio: &mut moq_mux::container::Producer, count: u64, sample: f32) { + for index in 0..count { + audio.write(packet(index, sample)).unwrap(); + tokio::time::sleep(PACKET_DURATION).await; + } + } + + /// How long `sample` played for, and through how many sinks. + fn played(recorder: &Recorder, sample: f32) -> (Duration, usize) { + let played: Vec<_> = recorder.played().into_iter().filter(|p| p.sample == sample).collect(); + let mut sinks: Vec<_> = played.iter().map(|p| p.sink).collect(); + sinks.dedup(); + (played.iter().map(|p| p.to - p.from).sum(), sinks.len()) + } + + /// The player follows the path for as long as the origin lasts, so it never + /// ends on its own: give it long enough to play out what it was handed, then + /// stop it. + async fn settle(player: tokio::task::JoinHandle<()>, recorder: &Recorder) { + tokio::time::sleep(Duration::from_secs(5)).await; + for event in recorder.events() { + match event { + Event::Failed(err) => panic!("playback failed: {err}"), + Event::Ended | Event::Finished => panic!("the player stopped following the path"), + Event::Wake => {} + } + } + assert!(!player.is_finished(), "the player stopped following the path"); + player.abort(); + } + + /// A restarted publisher announces a fresh epoch while its old run still + /// stands. The restart drops the old broadcast and plays the new one, whose + /// timeline starts over, instead of waiting for it to catch up. + #[tokio::test] + async fn a_republish_plays_the_new_broadcast() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + let epoch = || moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (_old_broadcast, _old_catalog, mut old) = publish(&origin, epoch()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + + let (_new_broadcast, _new_catalog, mut new) = publish(&origin, epoch()); + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert!( + played(&recorder, OLD).0 > Duration::ZERO, + "the old broadcast never played" + ); + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); + } + + /// Without epochs (moq-lite 06), a restart that overlaps the old route still + /// replaces it: the newest announcement wins the path. + #[tokio::test] + async fn an_epochless_republish_plays_the_new_broadcast() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + + let (_old_broadcast, _old_catalog, mut old) = publish(&origin, Default::default()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + + let (_new_broadcast, _new_catalog, mut new) = publish(&origin, Default::default()); + // A restart without an epoch waits out the origin's update hold, since a + // withdrawal wave's stale route looks just like one, and the player joins the + // new broadcast at its live edge. + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD).await; + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert!( + played(&recorder, OLD).0 > Duration::ZERO, + "the old broadcast never played" + ); + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); + } + + /// A publisher that stops cleanly ends the path, and its restart starts it + /// afresh. The player waits through the gap and plays the restarted broadcast. + #[tokio::test] + async fn a_broadcast_that_returns_plays_again() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + + let (old_broadcast, mut old_catalog, mut old) = publish(&origin, Default::default()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + old.finish().unwrap(); + old_catalog.finish().unwrap(); + drop((old_broadcast, old_catalog, old)); + + // Offline for longer than any timeout the player might give up on. + tokio::time::sleep(Duration::from_secs(60)).await; + assert!(!player.is_finished(), "the player stopped following the path"); + + let (_new_broadcast, _new_catalog, mut new) = publish(&origin, Default::default()); + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert_eq!( + played(&recorder, OLD).0, + PACKET_DURATION * 10, + "the old broadcast lost its tail" + ); + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the new broadcast did not play in full" + ); + } + + /// A covering route of the same epoch that arrives as the exact route goes, both + /// before the player looks, reaches it as an update: the announcements never show + /// the path unserved. A player with nothing playing still plays it. + #[tokio::test] + async fn an_update_while_idle_plays_the_route() { + tokio::time::pause(); + + const OLD: f32 = 0.25; + const NEW: f32 = 0.5; + let origin = moq_tokio::origin::spawn(); + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (exact, mut exact_catalog, mut old) = publish(&origin, route.clone()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + write(&mut old, 10, OLD).await; + old.finish().unwrap(); + exact_catalog.finish().unwrap(); + drop((exact_catalog, old)); + // The broadcast ends while its route stands, so the player goes idle. + tokio::time::sleep(Duration::from_secs(60)).await; + + let mut served = moq_net::broadcast::Info::new().produce(); + let served_catalog = catalog::Producer::new(&mut served, Default::default()).unwrap(); + let mut new = rendition(&served, &served_catalog, "audio"); + let pool = origin.dynamic("", route).unwrap(); + drop(exact); + let consumer = served.consume(); + tokio::spawn(async move { + while let Ok(request) = pool.requested_broadcast().await { + request.accept(consumer.clone()); + } + }); + write(&mut new, 10, NEW).await; + + settle(player, &recorder).await; + assert_eq!( + played(&recorder, NEW).0, + PACKET_DURATION * 10, + "the idle player never played the covering route" + ); + drop((served, served_catalog)); + } + + /// Re-pricing the same instance is an update, which playback rides through on + /// one sink rather than starting over. + #[tokio::test] + async fn a_reprice_keeps_playing() { + tokio::time::pause(); + + const SAMPLE: f32 = 0.25; + let origin = moq_tokio::origin::spawn(); + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (broadcast, _catalog, mut audio) = publish(&origin, route.clone()); + let recorder = Recorder::default(); + let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); + for index in 0..20 { + if index == 10 { + broadcast.announce(route.clone().with_cost(5)).unwrap(); + } + audio.write(packet(index, SAMPLE)).unwrap(); + tokio::time::sleep(PACKET_DURATION).await; + } + + settle(player, &recorder).await; + assert_eq!( + played(&recorder, SAMPLE), + (PACKET_DURATION * 20, 1), + "the update restarted playback" + ); + } + #[tokio::test] async fn a_finite_audio_track_plays_its_final_samples() { tokio::time::pause(); @@ -570,12 +861,7 @@ mod tests { audio.finish().unwrap(); drop(audio); catalog.finish().unwrap(); - player.await.unwrap(); - match recorder.events().pop() { - Some(Event::Ended) => {} - Some(Event::Failed(err)) => panic!("playback failed: {err}"), - _ => panic!("playback never ended"), - } + settle(player, &recorder).await; let played: Duration = recorder.played().iter().map(|p| p.to - p.from).sum(); assert_eq!(played, PACKET_DURATION, "the finished track lost its tail"); } diff --git a/rs/moq-cli/src/play/mod.rs b/rs/moq-cli/src/play/mod.rs index c455dd1e8d..8f2af57b9c 100644 --- a/rs/moq-cli/src/play/mod.rs +++ b/rs/moq-cli/src/play/mod.rs @@ -18,7 +18,6 @@ mod args; mod buffer; mod layout; mod playback; -mod source; mod timeline; #[cfg(all(test, feature = "play"))] diff --git a/rs/moq-cli/src/play/source.rs b/rs/moq-cli/src/play/source.rs deleted file mode 100644 index 5696665246..0000000000 --- a/rs/moq-cli/src/play/source.rs +++ /dev/null @@ -1,51 +0,0 @@ -//! Resolving the broadcast to play. - -use anyhow::Context; -use hang::moq_net; - -/// Wait for a route to cover `broadcast` on `origin`, then subscribe to it. -/// -/// The wait is the whole point. Subscribing goes through -/// `origin::Consumer::request_broadcast`, which resolves `Unroutable` on the -/// spot when no session has registered a handler yet rather than waiting for -/// one, and the media task starts well before the first handshake lands. The -/// window is already up, so this shows as a black frame rather than as a hang. -pub(super) async fn subscribe(origin: moq_net::origin::Consumer, broadcast: &str) -> anyhow::Result { - origin - .routed(broadcast) - .await - .with_context(|| format!("origin closed before broadcast `{broadcast}` was announced"))?; - - Ok(moq_mux::Source::new(origin, broadcast)) -} - -#[cfg(test)] -mod tests { - use super::*; - use std::time::Duration; - - /// Subscribing before the announcement lands doesn't wait, it fails: with no - /// session yet there is no handler registered on the origin, and - /// `request_broadcast` resolves `Unroutable` immediately. The media task is - /// spawned right after the reconnect loop starts, so it gets there first. - #[tokio::test] - async fn subscribe_waits_for_the_announcement() { - tokio::time::pause(); - - let origin = moq_tokio::origin::spawn(); - let consumer = origin.consume(); - - // Direct resolution has no route before the announcement. - let unannounced = moq_mux::Source::new(consumer.clone(), "room.hang").broadcast().await; - assert!(unannounced.is_err(), "expected an unroutable broadcast"); - - // Waiting first parks instead, for as long as it takes. - let mut waiting = std::pin::pin!(subscribe(consumer, "room.hang")); - let parked = tokio::time::timeout(Duration::from_secs(60), &mut waiting).await; - assert!(parked.is_err(), "expected to still be waiting on the announcement"); - - let _broadcast = origin.create_broadcast("room.hang").unwrap(); - _broadcast.announce(Default::default()).unwrap(); - waiting.await.unwrap(); - } -} diff --git a/rs/moq-cli/src/play/window.rs b/rs/moq-cli/src/play/window.rs index 850b3774f4..b516778123 100644 --- a/rs/moq-cli/src/play/window.rs +++ b/rs/moq-cli/src/play/window.rs @@ -37,7 +37,8 @@ pub(super) enum Event { Wake, /// Stop now: Ctrl-C, or the transport is gone and there is nothing more coming. Finished, - /// Every track reached its end. Present what is still queued, then stop. + /// The origin closed, so the broadcast can never come back. Present what is + /// still queued, then stop. Ended, Failed(String), } diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs new file mode 100644 index 0000000000..f8dd57ddc4 --- /dev/null +++ b/rs/moq-net/src/model/follow.rs @@ -0,0 +1,294 @@ +use std::task::{Poll, ready}; + +use super::origin_impl::{Announce, AnnounceConsumer, AnnounceEvent}; +use crate::PathOwned; + +/// The announcements covering one path, as one broadcast coming online, being replaced, and +/// going offline. Created by [`Consumer::follow`](crate::origin::Consumer::follow). +/// +/// Several routes can cover the path at once, such as the path itself and a prefix above it, +/// and a request resolves through the most specific one, so only that route's events come +/// through. Another route taking over is a [`Restart`](AnnounceEvent::Restart), unless both +/// carry the same epoch: those serve the same bytes, so it is an +/// [`Update`](AnnounceEvent::Update). Routes beneath the path serve other broadcasts and are +/// ignored. +pub struct Follow { + announced: AnnounceConsumer, + path: PathOwned, + /// Every route standing over the path. + covering: Vec, +} + +impl Follow { + pub(super) fn new(announced: AnnounceConsumer, path: PathOwned) -> Self { + Self { + announced, + path, + covering: Vec::new(), + } + } + + /// The next change to the route serving the path, or `None` once the origin closes. + pub async fn next(&mut self) -> Option { + kio::wait(|waiter| self.poll_next(waiter)).await + } + + /// Poll for the next change, registering `waiter` when there is none yet. + /// + /// While nothing serves the path, every announcement already on hand is folded into one + /// start, so the replay a late follower begins with (a covering prefix, then the exact path + /// beneath it) starts on the route serving the path rather than starting and restarting. + /// Once something serves it, each change comes through on its own: an end followed by a + /// start is a gap that ended any request on the old route, even when both routes carry one + /// epoch. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { + if self.serving().is_none() { + let ended = loop { + match self.announced.poll_next(waiter) { + Poll::Ready(Some(event)) => { + self.fold(event); + } + Poll::Ready(None) => break true, + Poll::Pending => break false, + } + }; + return match (self.serving().cloned(), ended) { + (Some(serving), _) => Poll::Ready(Some(AnnounceEvent::Start(serving))), + (None, true) => Poll::Ready(None), + (None, false) => Poll::Pending, + }; + } + + loop { + let Some(event) = ready!(self.announced.poll_next(waiter)) else { + return Poll::Ready(None); + }; + if let Some(event) = self.fold(event) { + return Poll::Ready(Some(event)); + } + } + } + + /// The most specific route covering the path, which is the one a request resolves. + fn serving(&self) -> Option<&Announce> { + self.covering + .iter() + .max_by_key(|announce| announce.prefix.as_str().len()) + } + + /// Apply one route's event, returning what it means for the path, if anything. + fn fold(&mut self, event: AnnounceEvent) -> Option { + let (AnnounceEvent::Start(announce) + | AnnounceEvent::Update(announce) + | AnnounceEvent::Restart(announce) + | AnnounceEvent::End(announce)) = &event; + if !self.path.has_prefix(&announce.prefix) { + return None; + } + let prefix = announce.prefix.clone(); + let before = self.serving().cloned(); + + self.covering.retain(|standing| standing.prefix != prefix); + let restart = match event { + AnnounceEvent::End(_) => false, + AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) => { + self.covering.push(announce); + false + } + AnnounceEvent::Restart(announce) => { + self.covering.push(announce); + true + } + }; + + let after = self.serving().cloned(); + match (before, after) { + (None, None) => None, + (None, Some(after)) => Some(AnnounceEvent::Start(after)), + (Some(before), None) => Some(AnnounceEvent::End(before)), + (Some(before), Some(after)) if before.prefix != after.prefix => { + let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; + Some(match same { + true => AnnounceEvent::Update(after), + false => AnnounceEvent::Restart(after), + }) + } + // A route less specific than the serving one changed, which no request sees. + (Some(_), Some(after)) if after.prefix != prefix => None, + (Some(_), Some(after)) if restart => Some(AnnounceEvent::Restart(after)), + (Some(_), Some(after)) => Some(AnnounceEvent::Update(after)), + } + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::super::ProduceTest; + use super::*; + use crate::{ + Epoch, Error, Hop, Pattern, Patterns, + origin::{Cost, Route}, + }; + + /// The follower's next event, which must arrive within the origin's update hold. + async fn followed(follow: &mut Follow) -> (&'static str, String) { + let event = moq_net_sim::timeout(Duration::from_secs(1), follow.next()) + .await + .expect("no event") + .expect("the origin closed"); + match event { + AnnounceEvent::Start(announce) => ("start", announce.prefix.to_string()), + AnnounceEvent::Update(announce) => ("update", announce.prefix.to_string()), + AnnounceEvent::Restart(announce) => ("restart", announce.prefix.to_string()), + AnnounceEvent::End(announce) => ("end", announce.prefix.to_string()), + } + } + + /// The follower stays quiet past the origin's update hold. + async fn quiet(follow: &mut Follow) { + if let Ok(event) = moq_net_sim::timeout(Duration::from_secs(1), follow.next()).await { + panic!("expected nothing, got {event:?}"); + } + } + + fn route(epoch: &Epoch) -> Route { + Route::default().with_epoch(epoch.clone()) + } + + #[moq_net_sim::test] + async fn follow_reports_the_route_serving_the_path() { + let origin = Hop::new(1).unwrap().produce(); + let mut follow = origin.consume().follow("pool/job").unwrap(); + let epoch = Epoch::mint(); + + // A prefix above the path covers it. + let pool = origin.dynamic("pool", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + + // A route beneath the path serves another broadcast. + let _beneath = origin.publish("pool/job/thumbnail", Route::default()).unwrap(); + quiet(&mut follow).await; + + // The path itself, from the same instance, takes over without a restart. + let exact = origin.publish("pool/job", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("update", "pool/job".into())); + + // The covering prefix no longer serves the path, so its changes are not seen. + pool.update(pool.route().with_cost(5)).unwrap(); + quiet(&mut follow).await; + + // Another instance at the path. + exact.announce(route(&Epoch::mint())).unwrap(); + assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); + + // It goes, so the prefix serves the path again: another instance. + drop(exact); + assert_eq!(followed(&mut follow).await, ("restart", "pool".into())); + + pool.update(pool.route().with_cost(9)).unwrap(); + assert_eq!(followed(&mut follow).await, ("update", "pool".into())); + + drop(pool); + assert_eq!(followed(&mut follow).await, ("end", "pool".into())); + } + + /// A follower that joins late starts on the route serving the path, not on whichever + /// covering route the replay happened to reach first. + #[moq_net_sim::test] + async fn follow_starts_on_the_serving_route_when_joining_late() { + let origin = Hop::new(1).unwrap().produce(); + let _pool = origin.dynamic("pool", Route::default()).unwrap(); + let _exact = origin.publish("pool/job", Route::default()).unwrap(); + + let mut follow = origin.consume().follow("pool/job").unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool/job".into())); + quiet(&mut follow).await; + } + + /// A gap with nothing serving the path ends any request on it, so a covering route that + /// arrives after it is a start of its own, even under the same epoch. + #[moq_net_sim::test] + async fn follow_keeps_a_gap_between_routes_of_one_epoch() { + let origin = Hop::new(1).unwrap().produce(); + let epoch = Epoch::mint(); + let exact = origin.publish("pool/job", route(&epoch)).unwrap(); + let mut follow = origin.consume().follow("pool/job").unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool/job".into())); + + drop(exact); + assert_eq!(followed(&mut follow).await, ("end", "pool/job".into())); + let _pool = origin.dynamic("pool", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + } + + /// Without an epoch nothing says two routes serve the same bytes. + #[moq_net_sim::test] + async fn follow_restarts_onto_a_more_specific_route_without_an_epoch() { + let origin = Hop::new(1).unwrap().produce(); + let mut follow = origin.consume().follow("pool/job").unwrap(); + + let _pool = origin.dynamic("pool", Route::default()).unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + + let _exact = origin.publish("pool/job", Route::default()).unwrap(); + assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); + } + + /// A route whose claim covers only paths beneath the followed one never serves it, so it + /// cannot mask the route that does, even when it would win the prefix on cost. + #[moq_net_sim::test] + async fn follow_ignores_a_route_scoped_beneath_the_path() { + let origin = Hop::new(1).unwrap().produce(); + let chat = Patterns::from("room/*/chat".parse::().unwrap()); + let _chat = origin + .scope("", &chat) + .unwrap() + .dynamic("room/alice", Route::default()) + .unwrap(); + let mut follow = origin.consume().follow("room/alice").unwrap(); + + let first = origin.publish("room/alice", Route::default().with_cost(5)).unwrap(); + let event = moq_net_sim::timeout(Duration::from_secs(1), follow.next()) + .await + .unwrap() + .unwrap(); + assert!( + matches!(&event, AnnounceEvent::Start(announce) if announce.route.cost == Cost::new(5)), + "{event:?}" + ); + + // A republish is another instance of the route that serves the path. + let _second = origin.publish("room/alice", Route::default().with_cost(5)).unwrap(); + drop(first); + let event = moq_net_sim::timeout(Duration::from_secs(1), follow.next()) + .await + .unwrap() + .unwrap(); + assert!( + matches!(&event, AnnounceEvent::Restart(announce) if announce.route.cost == Cost::new(5)), + "{event:?}" + ); + } + + /// A path no pattern can spell is refused rather than followed through the whole scope, + /// where a route that never serves it could win its prefix. + #[test] + fn follow_refuses_a_path_no_pattern_can_spell() { + let origin = Hop::new(1).unwrap().produce(); + assert!(matches!( + origin.consume().follow("camera*main"), + Err(Error::InvalidPath(_)) + )); + } + + /// A path the consumer's scope can never cover fails at once instead of waiting forever. + #[test] + fn follow_refuses_a_path_outside_the_scope() { + let origin = Hop::new(1).unwrap().produce(); + let scope = Patterns::from(Pattern::subtree("pool/job/cam").unwrap()); + let scoped = origin.consume().scope("", &scope).unwrap(); + assert!(matches!(scoped.follow("pool/job"), Err(Error::Unauthorized))); + } +} diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index eba650c029..6940c846c2 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -10,6 +10,8 @@ pub mod track; // modules so neither leaks the other's plumbing. #[path = "origin.rs"] mod origin_impl; +// Reduces the routes covering one path to the one serving it. +mod follow; // The failover state machine origin fronts run; pure, so its transitions are // tested without a runtime. mod front; @@ -48,6 +50,7 @@ pub mod origin { /// Subscribing to route (un)announcements from an origin. pub mod announce { + pub use super::follow::Follow; pub use super::origin_impl::{Announce, AnnounceConsumer as Consumer, AnnounceEvent as Event}; } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 659e73d57d..e349210637 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -4452,37 +4452,44 @@ impl Consumer { /// [`Self::routed_broadcast`]: pairing this with [`Self::request_broadcast`] /// leaves a gap where the covering route can retract. pub async fn routed(&self, path: impl AsPath) -> Option { + // A follower's first event is always a start. + match self.follow(path).ok()?.next().await? { + AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) | AnnounceEvent::Restart(announce) => { + Some(announce.route) + } + AnnounceEvent::End(_) => None, + } + } + + /// Follow the announcements of the routes covering `path`, reduced to the one serving it. + /// + /// This is how a player follows a broadcast across publisher restarts: play on a + /// [`Start`](AnnounceEvent::Start), drop everything and request the path afresh on a + /// [`Restart`](AnnounceEvent::Restart), and stop on an [`End`](AnnounceEvent::End). An + /// [`Update`](AnnounceEvent::Update) is the same publisher instance re-priced or failed + /// over, which subscriptions already ride out. Fails with [`Error::Unauthorized`] when + /// this consumer's scope can never cover the path, and with [`Error::InvalidPath`] when no + /// pattern can spell it (a segment containing `*`). + pub fn follow(&self, path: impl AsPath) -> Result { let path = path.as_path(); - // Scope a fresh consumer down to this path's subtree, so we only wake for - // announcements that overlap the requested path. - // A max-depth path cannot be spelled as `path/**` (`**` would be a 33rd - // segment), so watch the existing stream and match covering claims instead. - let consumer = match Pattern::subtree(path.as_str()) { - Ok(subtree) => self.scope("", &Patterns::from(subtree)).ok()?, - Err(InvalidPattern::TooManySegments) => self.clone(), - Err(_) => return None, - }; + // Scope down to the path itself, which still sees every route whose claim covers it: + // the rest of the origin never wakes it, and a route claiming only paths beneath it + // (a scoped dynamic at the same prefix) never masks the one that serves it, just as a + // request skips it. + let consumer = self.scope("", &Patterns::from(Pattern::literal(path.as_str())?))?; - // `scope` keeps narrower permissions intact: if we ask for `foo` on a - // consumer limited to `foo/specific`, `foo` itself is unauthorized. Bail - // rather than loop forever. + // `scope` keeps narrower permissions intact: on a consumer limited to + // `foo/specific`, no route can ever cover `foo`. if !consumer.allowed().matches(path.as_str()) { - return None; + return Err(Error::Unauthorized); } - // Use an untagged stream: this is a lookup, not egress announce - // forwarding, so it must not drive the announce guards. Hiding narrows - // discovery, not lookup, so a hidden path resolves like any other. - let mut announced = consumer.untagged().with_hidden(true).announced(); - loop { - if let AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) | AnnounceEvent::Restart(announce) = - announced.next().await? - && path.has_prefix(&announce.prefix) - { - return Some(announce.route); - } - } + // Untagged: this is a lookup, not egress announce forwarding, so it must not drive + // the announce guards. Hiding narrows discovery, not lookup, so a hidden path + // follows like any other. + let announced = consumer.untagged().with_hidden(true).announced(); + Ok(crate::announce::Follow::new(announced, path.to_owned())) } /// Block until `path` resolves to a broadcast: [`Self::request_broadcast`],