diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 1df08f571b..bf915879c5 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -7,6 +7,7 @@ import { HopSchema, isAnonymous, MAX_HOPS, Route, UNKNOWN_HOP } from "../hop.ts" import * as Path from "../path.ts"; import { type Reader, Writer } from "../stream.ts"; import * as Time from "../time.ts"; +import { TimeoutError } from "../util/timeout.ts"; import { type AnnounceBroadcast, AnnounceInit, AnnounceOk, encodeAnnounceBroadcast } from "./announce.ts"; import { Group } from "./group.ts"; import { Probe } from "./probe.ts"; @@ -969,6 +970,31 @@ test("a lite subscribe that times out waiting on a stream slot for the SUBSCRIBE } }); +// @moq/watch subscribes again after this deadline but never after a peer's reset with the same code, +// so the local timeout has to keep the timer's TimeoutError as its cause. +test("a subscribe whose TRACK_INFO never arrives closes with a control timeout caused by the timer", async () => { + jest.useFakeTimers(); + const warn = spyOn(console, "warn").mockImplementation(() => {}); + const { quic, streams } = fakeSession(); + const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); + try { + const track = subscriber.consume(Path.from("room")).track("video").subscribe(); + await drainUntil(() => streams.length === 1); + await streams[0].reading; + + jest.advanceTimersByTime(SUBSCRIBE_SETUP_TIMEOUT_MS); + await drainUntil(() => track.closed.peek() !== undefined); + const closed = track.closed.peek(); + expect(closed).toBeInstanceOf(StreamError); + expect((closed as StreamError).code).toBe(StreamCode.ControlTimeout); + expect((closed as StreamError).cause).toBeInstanceOf(TimeoutError); + } finally { + subscriber.close(); + warn.mockRestore(); + jest.useRealTimers(); + } +}); + test("a fetch started after the subscriber closes rejects without opening a stream", async () => { const { quic, streams } = fakeSession(); const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index 6762fc8c28..6f314dd05c 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -1,9 +1,12 @@ -import { describe, expect, it } from "bun:test"; +import { describe, expect, it, jest } from "bun:test"; import * as Catalog from "@moq/hang/catalog"; +import * as Container from "@moq/hang/container"; +import * as Json from "@moq/json"; import * as Moq from "@moq/net"; -import { Origin, Path } from "@moq/net"; +import { Origin, Path, Time } from "@moq/net"; import { Effect, Signal } from "@moq/signals"; import { Broadcast } from "./broadcast"; +import { nextMedia, subscribeMedia } from "./media"; function publish(origin: Origin.Producer, path: Path.Valid) { const broadcast = origin.createBroadcast(path); @@ -230,6 +233,136 @@ describe("blind resolution", () => { }); }); +describe("republish", () => { + it("hands media a new broadcast to subscribe to after an unannounce and re-announce", async () => { + // A media subscription ended by its route going away is not retried on the old handle: + // the re-announce has to bring a new one, which reruns the subscribing effect. + const owner = new Origin.Producer(); + const path = Path.from("room/cam.hang"); + const first = publish(owner, path); + first.createTrack("video"); + const source = new Broadcast({ origin: owner, name: path, enabled: true, catalogFormat: "manual" }); + const format = new Container.Legacy.Format("data"); + const effect = new Effect(); + const frames: number[] = []; + effect.run((inner) => { + const active = source.relativeBroadcast(inner, undefined); + if (!active) return; + const sub = subscribeMedia(inner, { + broadcast: active, + track: "video", + priority: 0, + maxAge: new Signal(Time.Milli(10_000)), + }); + if (!sub) return; + const consumer = new Container.Consumer(sub, { format }); + inner.cleanup(() => consumer.close()); + inner.spawn(async () => { + for (;;) { + const next = await nextMedia(consumer); + if (!next) break; + if (next.frame) frames.push(next.frame.payload[0]); + } + }); + }); + try { + await settle(); + const old = source.out.active.peek(); + expect(old).toBeDefined(); + + first.close(); + await settle(); + const second = publish(owner, path); + const track = second.createTrack("video"); + await settle(); + expect(source.out.active.peek()).not.toBe(old); + + new Container.Legacy.Producer(track, format).encode(new Uint8Array([42]), Time.Micro(0), true); + await settle(); + expect(frames).toEqual([42]); + } finally { + effect.close(); + source.close(); + owner.close(); + } + }); +}); + +describe("catalog", () => { + it("subscribes again after the catalog's setup times out", async () => { + const owner = new Origin.Producer(); + const path = Path.from("room/cam.hang"); + const producer = publish(owner, path); + const failing = producer.createTrack(Catalog.TRACK); + const source = new Broadcast({ origin: owner, name: path, enabled: true, catalogFormat: "hang" }); + try { + await settle(); + expect(source.out.status.peek()).toBe("loading"); + + // What the subscriber rejects a track with once its own setup deadline passes. + const timeout = Object.assign(new Error("subscribe timed out after 10000ms"), { name: "TimeoutError" }); + failing.close(new Moq.Error.Stream(Moq.StreamCode.ControlTimeout, { cause: timeout })); + const serving = producer.createTrack(Catalog.TRACK); + const catalog = new Json.Snapshot.Producer({ + track: serving, + compression: "none", + deltaRatio: 0, + }); + catalog.update({ video: { renditions: { hd: video("avc1.42001f") } } } as Catalog.Root); + await settle(); + + expect(source.out.status.peek()).toBe("live"); + expect(videoRenditions(source)).toEqual(["hd"]); + } finally { + source.close(); + owner.close(); + } + }); + + it("subscribes again after a relay resets the catalog of a broadcast that stays open", async () => { + // A publisher that restarts before the relay drops its old session reroutes the same broadcast + // handle, then the relay resets the catalog the old session served. + const owner = new Origin.Producer(); + const path = Path.from("room/cam.hang"); + const producer = publish(owner, path); + const serve = () => { + const track = producer.createTrack(Catalog.TRACK); + new Json.Snapshot.Producer({ track, compression: "none", deltaRatio: 0 }).update({ + video: { renditions: { hd: video("avc1.42001f") } }, + } as Catalog.Root); + return track; + }; + const first = serve(); + const source = new Broadcast({ origin: owner, name: path, enabled: true, catalogFormat: "hang" }); + const warn = console.warn; + console.warn = () => {}; + try { + await settle(); + expect(source.out.status.peek()).toBe("live"); + const handle = source.out.active.peek(); + + jest.useFakeTimers(); + first.close(new Moq.Error.Stream(Moq.StreamCode.Internal)); + serve(); + for (let i = 0; i < 20; i++) await Promise.resolve(); + expect(source.out.status.peek()).toBe("offline"); + + jest.advanceTimersByTime(1000); + jest.useRealTimers(); + await settle(); + + expect(source.out.active.peek()).toBe(handle); + expect(source.out.status.peek()).toBe("live"); + expect(videoRenditions(source)).toEqual(["hd"]); + } finally { + jest.useRealTimers(); + console.warn = warn; + source.close(); + owner.close(); + } + }); +}); + describe("refusal", () => { for (const announced of [true, false]) { it(`reports a refusal until a fresh request (announced: ${announced})`, async () => { diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 157138e9e2..e40b023f7e 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -5,6 +5,7 @@ import type * as Moq from "@moq/net"; import { Error as NetError, Path } from "@moq/net"; import { Effect, type Getter, getter, type Inputs, type Readonlys, readonlys, Signal } from "@moq/signals"; +import { resubscribe } from "./media"; import { toHang } from "./msf"; type ReferencedRendition = { @@ -287,6 +288,7 @@ export class Broadcast { const trackName = format === "hang" ? Catalog.TRACK : format === "hangz" ? Catalog.TRACK_COMPRESSED : "catalog"; const track = broadcast.track(trackName).subscribe({ priority: Catalog.PRIORITY.catalog }); effect.cleanup(() => track.close()); + resubscribe(effect, track, trackName); // The hang catalog is reconstructed from snapshots (and future deltas) via @moq/json, with // "hangz" decompressing the `.z` track; MSF stays on its own one-blob-per-group fetch. diff --git a/js/watch/src/media.test.ts b/js/watch/src/media.test.ts index 5f0f00ef09..c31107b2bc 100644 --- a/js/watch/src/media.test.ts +++ b/js/watch/src/media.test.ts @@ -1,4 +1,4 @@ -import { expect, test } from "bun:test"; +import { expect, jest, test } from "bun:test"; import { Container } from "@moq/hang"; import * as Moq from "@moq/net"; import { Time } from "@moq/net"; @@ -7,10 +7,16 @@ import { nextMedia, subscribeMedia } from "./media"; const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); +// Drain reactive work without waiting on a timer. +async function microtasks() { + for (let i = 0; i < 20; i++) await Promise.resolve(); +} + test("media max age is present on the initial subscription and later updates", async () => { let initial: Moq.Track.Subscription | undefined; const updates: Moq.Track.Subscription[] = []; const subscriber = { + closed: new Once(), close: () => undefined, update: (subscription: Moq.Track.Subscription) => updates.push(subscription), } as unknown as Moq.Track.Subscriber; @@ -81,3 +87,76 @@ test("media does not subscribe through a closed broadcast handle", () => { handle.close(); } }); + +// What the subscriber rejects a track with once its own setup deadline passes: the code, wrapping the +// TimeoutError its timer raised. A peer's reset with the same code wraps the transport error instead. +const timedOut = () => + new Moq.Error.Stream(Moq.StreamCode.ControlTimeout, { + cause: Object.assign(new Error("subscribe timed out after 10000ms"), { name: "TimeoutError" }), + }); + +// A subscription that timed out here before its first response is opened again at once, and one a relay +// reset because its upstream died is opened again after a delay; any other reset ends it. +// Driven the way the decoders drive it: one effect that subscribes and reads until `nextMedia` runs out. +for (const [label, failure, resumes] of [ + ["a local setup timeout", timedOut(), "at once"], + ["Internal", new Moq.Error.Stream(Moq.StreamCode.Internal), "after a delay"], + ["SessionClosed", new Moq.Error.Stream(Moq.StreamCode.SessionClosed), "after a delay"], + ["a peer's ControlTimeout reset", new Moq.Error.Stream(Moq.StreamCode.ControlTimeout), "never"], + ["Unroutable", new Moq.Error.Stream(Moq.StreamCode.Unroutable), "never"], + ["Cancel", new Moq.Error.Stream(Moq.StreamCode.Cancel), "never"], + ["NotFound", new Moq.Error.Stream(Moq.StreamCode.NotFound), "never"], +] as const) { + test(`media re-subscribes ${resumes} on ${label}`, async () => { + jest.useFakeTimers(); + const warn = console.warn; + console.warn = () => {}; + const broadcast = new Moq.Broadcast.Producer(); + const handle = broadcast.consume(); + const format = new Container.Legacy.Format("data"); + const failing = broadcast.createTrack("video"); + const effect = new Effect(); + const frames: number[] = []; + try { + effect.run((inner) => { + const sub = subscribeMedia(inner, { + broadcast: handle, + track: "video", + priority: 0, + maxAge: new Signal(Time.Milli(10_000)), + }); + if (!sub) return; + const consumer = new Container.Consumer(sub, { format }); + inner.cleanup(() => consumer.close()); + inner.spawn(async () => { + for (;;) { + const next = await nextMedia(consumer); + if (!next) break; + if (next.frame) frames.push(next.frame.payload[0]); + } + }); + }); + + // Let the effect subscribe before the subscription fails. + await microtasks(); + failing.close(failure); + + // The publisher is fine and serves the next subscription. + const serving = broadcast.createTrack("video"); + new Container.Legacy.Producer(serving, format).encode(new Uint8Array([42]), Time.Micro(0), true); + + await microtasks(); + expect(frames).toEqual(resumes === "at once" ? [42] : []); + + jest.advanceTimersByTime(1000); + await microtasks(); + expect(frames).toEqual(resumes === "never" ? [] : [42]); + } finally { + effect.close(); + handle.close(); + broadcast.close(); + console.warn = warn; + jest.useRealTimers(); + } + }); +} diff --git a/js/watch/src/media.ts b/js/watch/src/media.ts index baa0224bd9..1d596dde5d 100644 --- a/js/watch/src/media.ts +++ b/js/watch/src/media.ts @@ -1,11 +1,14 @@ import type { Container } from "@moq/hang"; import type * as Moq from "@moq/net"; -import { Error as NetError, type Time } from "@moq/net"; -import type { Effect, Getter } from "@moq/signals"; +import { Error as NetError, StreamCode, type Time } from "@moq/net"; +import { type Effect, type Getter, Signal } from "@moq/signals"; /** * Open a media subscription with its max age on the initial request and every update. * + * Reruns `effect` to subscribe again when the subscription fails for a reason that says nothing about + * the track (see {@link resubscribe}). + * * @internal */ export function subscribeMedia( @@ -21,6 +24,7 @@ export function subscribeMedia( const subscription = () => ({ priority: props.priority, maxAge: props.maxAge.peek() }); const subscriber = props.broadcast.track(props.track).subscribe(subscription()); effect.cleanup(() => subscriber.close()); + resubscribe(effect, subscriber, props.track); effect.run((inner) => { subscriber.update({ priority: props.priority, maxAge: inner.get(props.maxAge) }); @@ -29,6 +33,46 @@ export function subscribeMedia( return subscriber; } +// Spaces the attempts when a relay keeps resetting a track it cannot serve. +const RESET_RETRY_MS = 1000; + +/** + * Rerun `effect` once `subscriber` closes for a reason that says nothing about the track. + * + * - This side's setup deadline passed: each attempt already waited it out, so subscribe again at once. + * - The peer reset with Internal or SessionClosed: a relay sends these when the upstream session + * serving the track dies. The broadcast handle can stay open across that, when the publisher + * restarted on a new session the relay routes to, so nothing else would rerun the caller. + * Subscribe again after {@link RESET_RETRY_MS}. + * + * Every other reset ends the track: a withdrawn route (Unroutable) comes back as a new broadcast + * handle, which reruns the caller anyway. + * + * @internal + */ +export function resubscribe(effect: Effect, subscriber: Moq.Track.Subscriber, track: string): void { + const retry = new Signal(false); + effect.get(retry); + effect.run((inner) => { + const closed = inner.get(subscriber.closed); + const delay = retryDelay(closed); + if (delay === undefined) return; + console.warn(`subscription to ${track} ended, subscribing again`, closed); + if (delay === 0) retry.set(true); + else inner.timer(() => retry.set(true), delay); + }); +} + +// Only the deadline raised here, which wraps a TimeoutError, retries at once. A peer can reset with +// the same code at once (and relays forward it), so retrying that would spin. +function retryDelay(err: Error | null | undefined): number | undefined { + if (!(err instanceof NetError.Stream)) return undefined; + if (err.code === StreamCode.ControlTimeout && err.cause instanceof Error && err.cause.name === "TimeoutError") + return 0; + if (err.code === StreamCode.Internal || err.code === StreamCode.SessionClosed) return RESET_RETRY_MS; + return undefined; +} + /** Read the next media frame, ending playback when its subscription is reset. @internal */ export async function nextMedia(consumer: Container.Consumer) { try {