From 83d937ba4dcf975f196cbf841daa845bbb10ad93 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Wed, 7 Oct 2026 10:32:38 +0200 Subject: [PATCH 1/3] fix(watch): re-subscribe a media track whose setup timed out Co-Authored-By: Claude Opus 5.5 --- js/watch/src/broadcast.test.ts | 59 ++++++++++++++++++++++++++++++++- js/watch/src/media.test.ts | 60 ++++++++++++++++++++++++++++++++++ js/watch/src/media.ts | 18 ++++++++-- 3 files changed, 134 insertions(+), 3 deletions(-) diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index 6762fc8c28..3730370eca 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -1,9 +1,11 @@ import { describe, expect, it } from "bun:test"; import * as Catalog from "@moq/hang/catalog"; +import * as Container from "@moq/hang/container"; 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 +232,61 @@ 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("refusal", () => { for (const announced of [true, false]) { it(`reports a refusal until a fresh request (announced: ${announced})`, async () => { diff --git a/js/watch/src/media.test.ts b/js/watch/src/media.test.ts index 5f0f00ef09..c331dfb806 100644 --- a/js/watch/src/media.test.ts +++ b/js/watch/src/media.test.ts @@ -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,57 @@ test("media does not subscribe through a closed broadcast handle", () => { handle.close(); } }); + +// A subscription that timed out before its first response is opened again; any other reset ends it. +// Driven the way the decoders drive it: one effect that subscribes and reads until `nextMedia` runs out. +// ControlTimeout is what the subscriber rejects a track with once SUBSCRIBE goes unanswered for 10 s. +for (const [code, resumes] of [ + ["ControlTimeout", true], + ["Unroutable", false], + ["Cancel", false], + ["NotFound", false], +] as const) { + test(`media ${resumes ? "re-subscribes" : "stays ended"} after ${code}`, async () => { + 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(new Moq.Error.Stream(Moq.StreamCode[code])); + + // 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 ? [42] : []); + } finally { + effect.close(); + handle.close(); + broadcast.close(); + } + }); +} diff --git a/js/watch/src/media.ts b/js/watch/src/media.ts index baa0224bd9..ecfa801cd2 100644 --- a/js/watch/src/media.ts +++ b/js/watch/src/media.ts @@ -1,11 +1,13 @@ 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 times out before its first response. + * * @internal */ export function subscribeMedia( @@ -22,6 +24,18 @@ export function subscribeMedia( const subscriber = props.broadcast.track(props.track).subscribe(subscription()); effect.cleanup(() => subscriber.close()); + // A SUBSCRIBE that went unanswered says nothing about the track, and each attempt already waits + // out the setup deadline, so subscribe again at once. Every other reset ends the track; a + // withdrawn route (Unroutable) comes back as a new broadcast handle, which reruns this anyway. + const retry = new Signal(false); + effect.get(retry); + effect.run((inner) => { + const closed = inner.get(subscriber.closed); + if (!(closed instanceof NetError.Stream && closed.code === StreamCode.ControlTimeout)) return; + console.warn(`media subscription to ${props.track} timed out, subscribing again`, closed); + retry.set(true); + }); + effect.run((inner) => { subscriber.update({ priority: props.priority, maxAge: inner.get(props.maxAge) }); }); From 9696392f8652c74c4f1174033c9648356f9c2740 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Wed, 7 Oct 2026 12:25:11 +0200 Subject: [PATCH 2/3] fix(watch): retry only the local setup timeout, and the catalog too Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/subscriber.test.ts | 26 +++++++++++++++++++ js/watch/src/broadcast.test.ts | 33 ++++++++++++++++++++++++ js/watch/src/broadcast.ts | 2 ++ js/watch/src/media.test.ts | 25 +++++++++++------- js/watch/src/media.ts | 41 ++++++++++++++++++++++-------- 5 files changed, 108 insertions(+), 19 deletions(-) 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 3730370eca..f8f4d1d91a 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } 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, Time } from "@moq/net"; import { Effect, Signal } from "@moq/signals"; @@ -287,6 +288,38 @@ describe("republish", () => { }); }); +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(); + } + }); +}); + 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..17d99244be 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 { resubscribeOnTimeout } 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()); + resubscribeOnTimeout(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 c331dfb806..f4c630a17b 100644 --- a/js/watch/src/media.test.ts +++ b/js/watch/src/media.test.ts @@ -88,16 +88,23 @@ test("media does not subscribe through a closed broadcast handle", () => { } }); -// A subscription that timed out before its first response is opened again; any other reset ends it. +// 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; any other reset ends it. // Driven the way the decoders drive it: one effect that subscribes and reads until `nextMedia` runs out. -// ControlTimeout is what the subscriber rejects a track with once SUBSCRIBE goes unanswered for 10 s. -for (const [code, resumes] of [ - ["ControlTimeout", true], - ["Unroutable", false], - ["Cancel", false], - ["NotFound", false], +for (const [label, failure, resumes] of [ + ["a local setup timeout", timedOut(), true], + ["a peer's ControlTimeout reset", new Moq.Error.Stream(Moq.StreamCode.ControlTimeout), false], + ["Unroutable", new Moq.Error.Stream(Moq.StreamCode.Unroutable), false], + ["Cancel", new Moq.Error.Stream(Moq.StreamCode.Cancel), false], + ["NotFound", new Moq.Error.Stream(Moq.StreamCode.NotFound), false], ] as const) { - test(`media ${resumes ? "re-subscribes" : "stays ended"} after ${code}`, async () => { + test(`media ${resumes ? "re-subscribes" : "stays ended"} after ${label}`, async () => { const broadcast = new Moq.Broadcast.Producer(); const handle = broadcast.consume(); const format = new Container.Legacy.Format("data"); @@ -126,7 +133,7 @@ for (const [code, resumes] of [ // Let the effect subscribe before the subscription fails. await microtasks(); - failing.close(new Moq.Error.Stream(Moq.StreamCode[code])); + failing.close(failure); // The publisher is fine and serves the next subscription. const serving = broadcast.createTrack("video"); diff --git a/js/watch/src/media.ts b/js/watch/src/media.ts index ecfa801cd2..55e0f55663 100644 --- a/js/watch/src/media.ts +++ b/js/watch/src/media.ts @@ -23,24 +23,45 @@ 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()); + resubscribeOnTimeout(effect, subscriber, props.track); - // A SUBSCRIBE that went unanswered says nothing about the track, and each attempt already waits - // out the setup deadline, so subscribe again at once. Every other reset ends the track; a - // withdrawn route (Unroutable) comes back as a new broadcast handle, which reruns this anyway. + effect.run((inner) => { + subscriber.update({ priority: props.priority, maxAge: inner.get(props.maxAge) }); + }); + + return subscriber; +} + +/** + * Rerun `effect` once `subscriber` closes because its SUBSCRIBE went unanswered past this side's + * setup deadline. + * + * Such a failure says nothing about the track, and each attempt already waited out the deadline, + * so the rerun subscribes again at once. 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 resubscribeOnTimeout(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); - if (!(closed instanceof NetError.Stream && closed.code === StreamCode.ControlTimeout)) return; - console.warn(`media subscription to ${props.track} timed out, subscribing again`, closed); + if (!timedOut(closed)) return; + console.warn(`subscription to ${track} timed out, subscribing again`, closed); retry.set(true); }); +} - effect.run((inner) => { - subscriber.update({ priority: props.priority, maxAge: inner.get(props.maxAge) }); - }); - - return subscriber; +// Only the deadline raised here, which wraps a TimeoutError. A peer can reset with the same code at +// once (and relays forward it), so retrying that would spin. +function timedOut(err: Error | null | undefined): boolean { + return ( + err instanceof NetError.Stream && + err.code === StreamCode.ControlTimeout && + err.cause instanceof Error && + err.cause.name === "TimeoutError" + ); } /** Read the next media frame, ending playback when its subscription is reset. @internal */ From bd5ba99713048aca8ee46eccdc37efa7504f822b Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Wed, 7 Oct 2026 15:20:22 +0200 Subject: [PATCH 3/3] fix(watch): re-subscribe after a relay resets a track on an open broadcast A publisher that restarts before the relay drops its old session reroutes the same broadcast handle; the relay then resets every track the old session served with Internal (or SessionClosed). Nothing reruns the subscribing effect, so the catalog and media stay dead. Retry those two codes after a fixed 1 s delay through the same helper as the setup timeout. Co-Authored-By: Claude Opus 5.5 --- js/watch/src/broadcast.test.ts | 45 ++++++++++++++++++++++++++++++- js/watch/src/broadcast.ts | 4 +-- js/watch/src/media.test.ts | 30 ++++++++++++++------- js/watch/src/media.ts | 49 ++++++++++++++++++++-------------- 4 files changed, 96 insertions(+), 32 deletions(-) diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index f8f4d1d91a..6f314dd05c 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -1,4 +1,4 @@ -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"; @@ -318,6 +318,49 @@ describe("catalog", () => { 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", () => { diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 17d99244be..e40b023f7e 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -5,7 +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 { resubscribeOnTimeout } from "./media"; +import { resubscribe } from "./media"; import { toHang } from "./msf"; type ReferencedRendition = { @@ -288,7 +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()); - resubscribeOnTimeout(effect, track, trackName); + 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 f4c630a17b..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"; @@ -95,16 +95,22 @@ const timedOut = () => 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; any other reset ends it. +// 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(), true], - ["a peer's ControlTimeout reset", new Moq.Error.Stream(Moq.StreamCode.ControlTimeout), false], - ["Unroutable", new Moq.Error.Stream(Moq.StreamCode.Unroutable), false], - ["Cancel", new Moq.Error.Stream(Moq.StreamCode.Cancel), false], - ["NotFound", new Moq.Error.Stream(Moq.StreamCode.NotFound), false], + ["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 ${resumes ? "re-subscribes" : "stays ended"} after ${label}`, async () => { + 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"); @@ -140,11 +146,17 @@ for (const [label, failure, resumes] of [ new Container.Legacy.Producer(serving, format).encode(new Uint8Array([42]), Time.Micro(0), true); await microtasks(); - expect(frames).toEqual(resumes ? [42] : []); + 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 55e0f55663..1d596dde5d 100644 --- a/js/watch/src/media.ts +++ b/js/watch/src/media.ts @@ -6,7 +6,8 @@ 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 times out before its first response. + * Reruns `effect` to subscribe again when the subscription fails for a reason that says nothing about + * the track (see {@link resubscribe}). * * @internal */ @@ -23,7 +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()); - resubscribeOnTimeout(effect, subscriber, props.track); + resubscribe(effect, subscriber, props.track); effect.run((inner) => { subscriber.update({ priority: props.priority, maxAge: inner.get(props.maxAge) }); @@ -32,36 +33,44 @@ 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 because its SUBSCRIBE went unanswered past this side's - * setup deadline. + * 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}. * - * Such a failure says nothing about the track, and each attempt already waited out the deadline, - * so the rerun subscribes again at once. Every other reset ends the track: a withdrawn route - * (Unroutable) comes back as a new broadcast handle, which reruns the caller anyway. + * 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 resubscribeOnTimeout(effect: Effect, subscriber: Moq.Track.Subscriber, track: string): void { +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); - if (!timedOut(closed)) return; - console.warn(`subscription to ${track} timed out, subscribing again`, closed); - retry.set(true); + 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. A peer can reset with the same code at -// once (and relays forward it), so retrying that would spin. -function timedOut(err: Error | null | undefined): boolean { - return ( - err instanceof NetError.Stream && - err.code === StreamCode.ControlTimeout && - err.cause instanceof Error && - err.cause.name === "TimeoutError" - ); +// 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 */