fix(watch): re-subscribe a track after a setup timeout or an upstream reset

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));

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<Catalog.Root>({
+				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<Catalog.Root>({ 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 () => {

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.

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<Error | null>(),
 		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();
+		}
+	});
+}

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 {