fix(watch): survive an audio decoder error, resample, and re-anchor on resubscribe

js/hang/package.json

@@ -15,6 +15,7 @@
 	"scripts": {
 		"build": "rimraf dist && tsc -b && bun ../common/package.ts",
 		"check": "tsc --noEmit",
+		"test": "bun test --only-failures",
 		"release": "bun ../common/release.ts"
 	},
 	"dependencies": {

js/hang/src/container/consumer.test.ts

@@ -585,3 +585,25 @@ test("Consumer does not duration-skip when the gap is not covered", async () =>
 	expect(frames.map((f) => f.group)).toEqual([0, 0, 1]);
 	consumer.close();
 });
+
+// A fresh Consumer always starts at zero, on both a fresh track and a resubscribe of a track that has
+// already been read. Downstream decoders that outlive their subscription rely on this to re-anchor their
+// own copy of the count; a nonzero start would make the first frame of a new stream look like a rewind.
+test("Consumer starts at discontinuity 0, including on resubscribe", async () => {
+	const track = new Track.Producer("test");
+
+	const first = new Consumer(track.subscribe(), { format: new LegacyFormat(), latency: 500 as Time.Milli });
+	expect(first.discontinuity).toBe(0);
+
+	writeGroupWithLegacyFrames(track, 0, [0 as Time.Micro, 33_000 as Time.Micro]);
+	await drainFrames(first, 200);
+	expect(first.discontinuity).toBe(0);
+	first.close();
+
+	// Resubscribing the same track builds a new Consumer, whose counter restarts rather than resuming.
+	const second = new Consumer(track.subscribe(), { format: new LegacyFormat(), latency: 500 as Time.Milli });
+	expect(second.discontinuity).toBe(0);
+	second.close();
+
+	track.close();
+});

js/hang/src/container/consumer.ts

@@ -111,7 +111,13 @@ export class Consumer {
 	async #run() {
 		// Start fetching groups in the background
 		for (;;) {
-			const consumer = await this.#track.recvGroup();
+			// A reset/closed track (publisher handover, unsubscribe, session close) rejects recvGroup;
+			// classify it (routine teardown -> debug, real fault -> warn) instead of letting it escape
+			// uncaught to the anonymous spawn handler as a contextless "spawn error".
+			const consumer = await this.#track.recvGroup().catch((err: unknown) => {
+				console[Moq.isStreamAbort(err) ? "debug" : "warn"]("track consumer stopped", this.#track.name, err);
+				return undefined;
+			});
 			if (!consumer) break;
 
 			// To improve TTV, we always start with the first group.

js/net/src/ietf/publisher.ts

@@ -3,7 +3,7 @@ import type * as broadcast from "../broadcast.ts";
 import type * as group from "../group.ts";
 import * as Path from "../path.ts";
 import { type Stream, Writer } from "../stream.ts";
-import { error } from "../util/error.ts";
+import { error, isStreamAbort } from "../util/error.ts";
 import type { Session } from "./adapter.ts";
 import { Frame, Group as GroupMessage } from "./object.ts";
 import { PublishDone } from "./publish.ts";
@@ -196,7 +196,10 @@ export class Publisher {
 			stream.close();
 		} catch (err: unknown) {
 			const e = error(err);
-			console.warn(`publish error: broadcast=${name} track=${track.name} error=${e.message}`);
+			// A downstream unsubscribe/handover aborts the stream; that is expected teardown, not a fault.
+			console[isStreamAbort(e) ? "debug" : "warn"](
+				`publish error: broadcast=${name} track=${track.name} error=${e.message}`,
+			);
 			stream.abort(e);
 		} finally {
 			track.close();

js/net/src/index.ts

@@ -23,5 +23,7 @@ export * as Path from "./path.ts";
 export * as Time from "./time.ts";
 /** Track role handles. */
 export * as Track from "./track.ts";
+/** Classify an error as a routine stream teardown vs a client-actionable fault (for log-level decisions). */
+export { isStreamAbort } from "./util/error.ts";
 /** QUIC variable-length integer encoding and decoding. */
 export * as Varint from "./varint.ts";

js/net/src/lite/connection.ts

@@ -6,6 +6,7 @@ import type { Established } from "../connection/established.ts";
 import * as Path from "../path.ts";
 import { type Reader, Readers, Stream, Writer } from "../stream.ts";
 import type * as Time from "../time.ts";
+import { isStreamAbort } from "../util/error.ts";
 import { AnnounceRequest } from "./announce.ts";
 import { Fetch } from "./fetch.ts";
 import { Goaway } from "./goaway.ts";
@@ -150,7 +151,9 @@ export class Connection implements Established {
 		try {
 			await Promise.all(tasks);
 		} catch (err) {
-			console.error("fatal error running connection", err);
+			// Routine teardown (unsubscribe, handover, peer close) rejects the same way a fault does;
+			// isStreamAbort tells them apart from the error content, same as every other run-loop catch.
+			console[isStreamAbort(err) ? "debug" : "error"]("connection run loop stopped", err);
 		} finally {
 			this.close();
 		}

js/net/src/lite/publisher.ts

@@ -5,7 +5,7 @@ import * as Path from "../path.ts";
 import { type Stream, Writer } from "../stream.ts";
 import { Timescale } from "../time.ts";
 import type * as track from "../track.ts";
-import { error } from "../util/error.ts";
+import { error, isStreamAbort } from "../util/error.ts";
 import { AnnounceInit, AnnounceOk, type AnnounceRequest, encodeAnnounceBroadcast } from "./announce.ts";
 import { Datagram as DatagramMessage } from "./datagram.ts";
 import type { Fetch } from "./fetch.ts";
@@ -307,7 +307,10 @@ export class Publisher {
 			await datagrams;
 		} catch (err: unknown) {
 			const e = error(err);
-			console.warn(`publish error: broadcast=${msg.broadcast} track=${track.name} error=${e.message}`);
+			// A downstream unsubscribe/handover aborts the stream; that is expected teardown, not a fault.
+			console[isStreamAbort(e) ? "debug" : "warn"](
+				`publish error: broadcast=${msg.broadcast} track=${track.name} error=${e.message}`,
+			);
 			track.close(e);
 			stream.abort(e);
 			await datagrams;
@@ -394,7 +397,10 @@ export class Publisher {
 			stream.close();
 		} catch (err: unknown) {
 			const e = error(err);
-			console.warn(`publish error: broadcast=${broadcast} track=${track.name} error=${e.message}`);
+			// A downstream unsubscribe/handover resets the stream; that is expected teardown, not a fault.
+			console[isStreamAbort(e) ? "debug" : "warn"](
+				`publish error: broadcast=${broadcast} track=${track.name} error=${e.message}`,
+			);
 			track.close(e);
 			stream.reset(e);
 		}
@@ -603,7 +609,9 @@ export class Publisher {
 				}
 			}
 		} catch (err: unknown) {
-			console.warn("probe stream error", err);
+			// Best-effort bandwidth side channel: a reset on reconnect/teardown is routine, but a real
+			// fault (auth, protocol, unroutable) must still surface.
+			console[isStreamAbort(err) ? "debug" : "warn"]("probe stream error", err);
 			stream.close();
 		}
 	}

js/net/src/lite/subscriber.ts

@@ -8,7 +8,7 @@ import * as Path from "../path.ts";
 import { type Reader, Stream } from "../stream.ts";
 import * as Time from "../time.ts";
 import type * as track from "../track.ts";
-import { error } from "../util/error.ts";
+import { error, isStreamAbort } from "../util/error.ts";
 import { withTimeout } from "../util/timeout.ts";
 import { AnnounceInit, AnnounceOk, AnnounceRequest, decodeAnnounceBroadcastMaybe } from "./announce.ts";
 import { Datagram as DatagramMessage } from "./datagram.ts";
@@ -326,7 +326,11 @@ export class Subscriber {
 			const e = error(err);
 			request.reject(e);
 			this.#subscribes.delete(id);
-			console.warn(`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${e.message}`);
+			// A publisher handover/unsubscribe (or session close) resets the stream during setup; that is
+			// expected teardown, not a fault. A timeout is not a stream abort, so it still logs at warn.
+			console[isStreamAbort(e) ? "debug" : "warn"](
+				`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${e.message}`,
+			);
 			// If the stream eventually opens after the timeout, abort it so we
 			// don't leak it. Cover both branches: setup may resolve late, or it
 			// may reject (e.g. encode/decode failure) after the stream is open.
@@ -365,7 +369,10 @@ export class Subscriber {
 		} catch (err) {
 			const e = error(err);
 			producer.close(e);
-			console.warn(`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${e.message}`);
+			// A publisher handover/unsubscribe resets the stream; that is expected teardown, not a fault.
+			console[isStreamAbort(e) ? "debug" : "warn"](
+				`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${e.message}`,
+			);
 			stream.abort(e);
 		} finally {
 			this.#subscribes.delete(id);
@@ -708,7 +715,8 @@ export class Subscriber {
 				reader.releaseLock();
 			}
 		} catch (err: unknown) {
-			console.warn("datagram stream error", err);
+			// Best-effort loop: a deliberate session close resets this stream on every teardown.
+			console[isStreamAbort(err) ? "debug" : "warn"]("datagram stream error", err);
 		}
 	}
 
@@ -778,7 +786,9 @@ export class Subscriber {
 				}
 			}
 		} catch (err: unknown) {
-			console.warn("probe stream error", err);
+			// Best-effort bandwidth/RTT side channel: a reset on reconnect/teardown just means no estimate
+			// right now, but a real fault (auth, protocol, unroutable) must still surface.
+			console[isStreamAbort(err) ? "debug" : "warn"]("probe stream error", err);
 		} finally {
 			this.#recvBandwidth.set(undefined);
 			this.#rtt?.set(undefined);

js/net/src/util/error.test.ts

@@ -0,0 +1,83 @@
+import { expect, test } from "bun:test";
+import { isStreamAbort } from "./error.ts";
+
+// A stand-in for WebTransportError, which the test runner does not define. isStreamAbort keys on
+// `err.name` + duck-typed `source`/`streamErrorCode`, so this reproduces what the browser passes.
+function wtError(source: string, streamErrorCode?: number | null): Error {
+	const err = new Error("The stream was aborted by the remote server.");
+	err.name = "WebTransportError";
+	return Object.assign(err, { source, streamErrorCode });
+}
+
+// A stand-in for the DOMException Safari throws when a stream is used after it ended.
+function invalidState(message: string): Error {
+	const err = new Error(message);
+	err.name = "InvalidStateError";
+	return err;
+}
+
+test("routine stream resets (Cancel/Dropped/Closed/no-code) are teardown", () => {
+	// Native WebTransport (streamErrorCode carries the relay's rs/moq-net error.rs code).
+	expect(isStreamAbort(wtError("stream", 0))).toBe(true); // Cancel: normal unsubscribe
+	expect(isStreamAbort(wtError("stream", 24))).toBe(true); // Dropped
+	expect(isStreamAbort(wtError("stream", 25))).toBe(true); // Closed
+	expect(isStreamAbort(wtError("stream", null))).toBe(true); // reset with no app code -> routine
+	// WebSocket/qmux fallback ("RESET_STREAM: <code>" / "STOP_SENDING: <code>").
+	expect(isStreamAbort(new Error("RESET_STREAM: 0"))).toBe(true);
+	expect(isStreamAbort(new Error("STOP_SENDING: 0"))).toBe(true);
+});
+
+test("client-actionable fault codes surface (warn), not teardown", () => {
+	expect(isStreamAbort(wtError("stream", 6))).toBe(false); // Unauthorized
+	expect(isStreamAbort(wtError("stream", 13))).toBe(false); // NotFound (wrong path)
+	expect(isStreamAbort(wtError("stream", 15))).toBe(false); // ProtocolViolation
+	expect(isStreamAbort(wtError("stream", 30))).toBe(false); // Unroutable
+	expect(isStreamAbort(new Error("RESET_STREAM: 13"))).toBe(false); // NotFound over qmux
+	expect(isStreamAbort(new Error("STOP_SENDING: 6"))).toBe(false); // Unauthorized over qmux
+});
+
+test("write-after-close over the WebSocket/qmux fallback is teardown", () => {
+	// Generic Streams-API errors from writing after the stream ended (a peer reset racing an in-flight
+	// write). Chromium/Firefox and Safari word it differently; both are routine teardown, not faults.
+	expect(isStreamAbort(new Error("The stream is closed or closing"))).toBe(true);
+	expect(isStreamAbort(invalidState("The object is in an invalid state."))).toBe(true);
+});
+
+test("a session close is teardown, unless the peer signalled a fault", () => {
+	// The exact strings qmux constructs when the session ends (see @moq/qmux session.js).
+	expect(isStreamAbort(new Error("Connection closed"))).toBe(true); // local close()
+	expect(isStreamAbort(new Error("Connection closed: 0: "))).toBe(true); // peer CONNECTION_CLOSE, no error
+	expect(isStreamAbort(new Error("Connection closed: 1006 "))).toBe(true); // abnormal WebSocket closure
+
+	// A client-actionable code on the CONNECTION_CLOSE frame still surfaces.
+	expect(isStreamAbort(new Error("Connection closed: 6: unauthorized"))).toBe(false); // Unauthorized
+	expect(isStreamAbort(new Error("Connection closed: 15: bad frame"))).toBe(false); // ProtocolViolation
+});
+
+test("a coded fault wins over the teardown message heuristics", () => {
+	// A fault code must not be reclassified as teardown just because the browser's prose happens to
+	// mention a closing stream.
+	const err = wtError("stream", 6); // Unauthorized
+	err.message = "The stream is closed or closing";
+	expect(isStreamAbort(err)).toBe(false);
+});
+
+test("an unrelated error mentioning an invalid state still surfaces", () => {
+	// Keyed on `name`, so one of our own ordering bugs is not silently downgraded to debug.
+	expect(isStreamAbort(new Error("decoder is in an invalid state"))).toBe(false);
+});
+
+test("non-WebTransport errors are surfaced", () => {
+	expect(isStreamAbort(new Error("first subscribe response must be SUBSCRIBE_OK"))).toBe(false);
+	expect(isStreamAbort("not an error")).toBe(false);
+	expect(isStreamAbort(undefined)).toBe(false);
+});
+
+test("WebTransport errors are classified by code regardless of source", () => {
+	// Chrome surfaces a write-side abort (a downstream unsubscribe seen by the publisher) as a
+	// WebTransportError with source "session" and the relay's Cancel(0) code: routine, must not warn.
+	expect(isStreamAbort(wtError("session", 0))).toBe(true);
+	expect(isStreamAbort(wtError("session", null))).toBe(true);
+	// A genuine session-level fault code still surfaces.
+	expect(isStreamAbort(wtError("session", 6))).toBe(false); // Unauthorized
+});

js/net/src/util/error.ts

@@ -3,6 +3,69 @@ export function error(err: unknown): Error {
 	return err instanceof Error ? err : new Error(String(err));
 }
 
+// Stream-reset application codes that indicate a client-actionable FAULT (bad auth, wrong path, protocol
+// violation, unroutable, ...) rather than a routine teardown. Mirrors rs/moq-net/src/error.rs
+// `Error::to_code()`. Kept as a DENYLIST: any code NOT listed (Cancel=0, Timeout=3, Transport=4,
+// Dropped=24, Closed=25, ...) is treated as an expected teardown, so a normal unsubscribe/handover reset
+// stays quiet whatever code it carries, while a genuine failure surfaces.
+const STREAM_FAULT_CODES = new Set<number>([
+	6, // Unauthorized
+	9, // Version
+	11, // BoundsExceeded
+	12, // Duplicate
+	13, // NotFound
+	14, // WrongSize
+	15, // ProtocolViolation
+	16, // UnexpectedMessage
+	17, // Unsupported
+	27, // FrameTooLarge
+	30, // Unroutable
+]);
+
+/**
+ * True when `err` is a stream-lifecycle event a caller can log at debug rather than warn: false for a
+ * client-actionable fault (see STREAM_FAULT_CODES) or a non-stream error.
+ *
+ * Reads the application code the relay encodes (rs/moq-net `error.rs`): `WebTransportError.streamErrorCode`
+ * on native transports, or the trailing number in qmux's "RESET_STREAM: <code>" / "STOP_SENDING: <code>"
+ * message on the WebSocket fallback. Keyed on `err.name` rather than `instanceof WebTransportError` so it
+ * is safe where that global is undefined (e.g. the test runner).
+ */
+export function isStreamAbort(err: unknown): boolean {
+	if (!(err instanceof Error)) return false;
+
+	// Coded stream resets decide first: a client-actionable fault code always wins over the message
+	// heuristics below.
+	if (err.name === "WebTransportError") {
+		// Trust the relay's numeric app code regardless of `source`: Chrome reports a WRITE-side abort
+		// (a downstream unsubscribe seen by the publisher) as source "session", not "stream", so gating on
+		// source would leave that routine teardown warning on every closed viewer.
+		const c = (err as { streamErrorCode?: number | null }).streamErrorCode;
+		const code = typeof c === "number" ? c : undefined;
+		return code === undefined || !STREAM_FAULT_CODES.has(code);
+	}
+
+	const match = /^(?:RESET_STREAM|STOP_SENDING): (\d+)/.exec(err.message);
+	if (match) return !STREAM_FAULT_CODES.has(Number(match[1]));
+
+	// The session itself ended, which fails every stream still open on it. qmux words this
+	// "Connection closed", optionally carrying the peer's CONNECTION_CLOSE code. (Native WebTransport
+	// reports the equivalent as a WebTransportError with no `streamErrorCode`, handled above.)
+	const closed = /^Connection closed(?::\s*(\d+))?/.exec(err.message);
+	if (closed) {
+		const code = closed[1] === undefined ? undefined : Number(closed[1]);
+		return code === undefined || !STREAM_FAULT_CODES.has(code);
+	}
+
+	// A write/close after the stream already ended (a peer reset racing an in-flight write) surfaces as a
+	// generic Streams-API error rather than a coded RESET_STREAM/STOP_SENDING, common over the WebSocket/qmux
+	// fallback. Engines word it differently: Chromium/Firefox say the stream is "closed or closing"; Safari
+	// throws InvalidStateError. Key Safari on `name` rather than the prose, so an unrelated error that merely
+	// mentions an invalid state still surfaces.
+	if (err.name === "InvalidStateError") return true;
+	return /closed or closing/i.test(err.message);
+}
+
 export function unreachable(value: never): never {
 	throw new Error(`unreachable: ${value}`);
 }