From 623cc042d54ed4476d48d3f3df8a978fa8e1905c Mon Sep 17 00:00:00 2001 From: map1336 <270788582+fperex@users.noreply.github.com> Date: Sat, 11 Jul 2026 22:14:23 -0400 Subject: [PATCH 1/2] fix(net): classify routine stream teardown apart from real faults Every run loop logged any error from a stream as a warning, so a normal unsubscribe, a publisher handover or a peer closing the session produced scary warnings on every viewer that stopped watching. isStreamAbort() decides on the application code the relay encodes (WebTransportError.streamErrorCode natively, or the trailing number in qmux's RESET_STREAM/STOP_SENDING and Connection closed messages on the WebSocket fallback). Coded faults are a denylist, so an unknown code is treated as routine teardown while a client-actionable fault (auth, not-found, protocol, unroutable) still surfaces at warn. Call sites in the lite/ietf publishers, the subscriber and the connection run loop now pick their log level from it. The bandwidth probe streams no longer swallow genuine faults at debug. --- js/net/src/ietf/publisher.ts | 7 ++- js/net/src/index.ts | 2 + js/net/src/lite/connection.ts | 5 ++- js/net/src/lite/publisher.ts | 16 +++++-- js/net/src/lite/subscriber.ts | 20 ++++++--- js/net/src/util/error.test.ts | 83 +++++++++++++++++++++++++++++++++++ js/net/src/util/error.ts | 63 ++++++++++++++++++++++++++ 7 files changed, 184 insertions(+), 12 deletions(-) create mode 100644 js/net/src/util/error.test.ts diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index c4072535cd..18d4c16142 100644 --- a/js/net/src/ietf/publisher.ts +++ b/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(); diff --git a/js/net/src/index.ts b/js/net/src/index.ts index 3a4f29419b..62b0501d5b 100644 --- a/js/net/src/index.ts +++ b/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"; diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index d6dc916281..b7c394c53a 100644 --- a/js/net/src/lite/connection.ts +++ b/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(); } diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 7d420cb154..3c95c43e61 100644 --- a/js/net/src/lite/publisher.ts +++ b/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(); } } diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 4e337ffab9..572199a199 100644 --- a/js/net/src/lite/subscriber.ts +++ b/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); diff --git a/js/net/src/util/error.test.ts b/js/net/src/util/error.test.ts new file mode 100644 index 0000000000..cd751c446f --- /dev/null +++ b/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: " / "STOP_SENDING: "). + 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 +}); diff --git a/js/net/src/util/error.ts b/js/net/src/util/error.ts index c0016fa3a4..6054f3e628 100644 --- a/js/net/src/util/error.ts +++ b/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([ + 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: " / "STOP_SENDING: " + * 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}`); } From 32b92b079d652af27c7e81c3b76ca399c174aedc Mon Sep 17 00:00:00 2001 From: map1336 <270788582+fperex@users.noreply.github.com> Date: Sat, 11 Jul 2026 22:27:33 -0400 Subject: [PATCH 2/2] fix(watch): survive an audio decoder error, resample, and re-anchor on resubscribe Audio had the same fatal-decoder problem as video: an AudioDecoder that errored never decoded again, so the stream went silent until a reload. Restart it in place on a budget, with the same config-supersession check. Also fix three sources of crackle and drift: - Resample when the catalog rate and the AudioContext rate differ, instead of writing samples into a ring built at the wrong rate. - Snap timestamps onto the frame cadence so a jittery per-frame timestamp does not zero-fill or overwrite a sample every frame. - Re-anchor the discontinuity counter on resubscribe. Audio.Decoder is built once and its #discontinuity outlives the subscription, but the container Consumer's rewind counter is per-consumer and restarts at zero, so after any rewind the next resubscribe delivered 0, looked like a rewind, and reset the Sync reference shared with video. Resume the AudioContext on a user gesture rather than assuming it starts running. --- js/hang/package.json | 1 + js/hang/src/container/consumer.test.ts | 22 ++ js/hang/src/container/consumer.ts | 8 +- js/watch/src/audio/buffer.ts | 61 ++++- js/watch/src/audio/decoder.ts | 354 ++++++++++++++++++++++--- js/watch/src/audio/emitter.ts | 44 ++- js/watch/src/audio/ring-buffer.ts | 8 +- js/watch/src/audio/snap.test.ts | 34 +++ js/watch/src/audio/snap.ts | 19 ++ js/watch/src/audio/source.ts | 9 +- 10 files changed, 491 insertions(+), 69 deletions(-) create mode 100644 js/watch/src/audio/snap.test.ts create mode 100644 js/watch/src/audio/snap.ts diff --git a/js/hang/package.json b/js/hang/package.json index 0a9d151c79..f5dda0c3c9 100644 --- a/js/hang/package.json +++ b/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": { diff --git a/js/hang/src/container/consumer.test.ts b/js/hang/src/container/consumer.test.ts index 1d47a03317..02fa1b7044 100644 --- a/js/hang/src/container/consumer.test.ts +++ b/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(); +}); diff --git a/js/hang/src/container/consumer.ts b/js/hang/src/container/consumer.ts index 4895b65ff2..c8385dfb8c 100644 --- a/js/hang/src/container/consumer.ts +++ b/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. diff --git a/js/watch/src/audio/buffer.ts b/js/watch/src/audio/buffer.ts index 1e2c4fd1d3..f2e37efb4c 100644 --- a/js/watch/src/audio/buffer.ts +++ b/js/watch/src/audio/buffer.ts @@ -3,6 +3,9 @@ import { Effect, type Getter, Signal } from "@moq/signals"; import type { Data, InitPost, InitShared, Latency, Reset, State } from "./render"; import { allocSharedRingBuffer, SharedRingBuffer } from "./shared-ring-buffer"; +// A parked backpressure waiter; `cleanup` detaches its abort listener when it resolves. +type Waiter = { timestamp: Time.Micro; resolve: () => void; cleanup?: () => void }; + /** * Timestamp-based backpressure for buffered playback. The decoded PCM ring only holds the latency * floor; everything above it (the buffered lookahead, up to the ceiling) stays upstream as encoded @@ -14,7 +17,7 @@ import { allocSharedRingBuffer, SharedRingBuffer } from "./shared-ring-buffer"; class Backpressure { readonly #enabled: boolean; #headroom: Time.Micro; - #waiters: Array<{ timestamp: Time.Micro; resolve: () => void }> = []; + #waiters: Array = []; constructor(enabled: boolean, headroom: Time.Micro) { this.#enabled = enabled; @@ -26,26 +29,45 @@ class Backpressure { this.#headroom = headroom; } - wait(timestamp: Time.Micro, playhead: Time.Micro): Promise { + // `signal` (the decode effect's abort) unparks a waiter when the loop is torn down, so a frozen + // playhead (e.g. a mute that stopped the render graph) can never strand the spawn. + wait(timestamp: Time.Micro, playhead: Time.Micro, signal?: AbortSignal): Promise { if (!this.#enabled) return Promise.resolve(); + if (signal?.aborted) return Promise.resolve(); if (playhead >= ((timestamp - this.#headroom) | 0)) return Promise.resolve(); - return new Promise((resolve) => this.#waiters.push({ timestamp, resolve })); + return new Promise((resolve) => { + const waiter: Waiter = { timestamp, resolve }; + if (signal) { + const onAbort = () => { + const i = this.#waiters.indexOf(waiter); + if (i >= 0) this.#waiters.splice(i, 1); + resolve(); + }; + signal.addEventListener("abort", onAbort, { once: true }); + waiter.cleanup = () => signal.removeEventListener("abort", onAbort); + } + this.#waiters.push(waiter); + }); } // Resolve every waiter the playhead has reached. Thresholds are recomputed live so a changed // headroom takes effect on queued waiters too. advance(playhead: Time.Micro): void { if (this.#waiters.length === 0) return; - this.#waiters = this.#waiters.filter(({ timestamp, resolve }) => { - if (playhead < ((timestamp - this.#headroom) | 0)) return true; - resolve(); + this.#waiters = this.#waiters.filter((waiter) => { + if (playhead < ((waiter.timestamp - this.#headroom) | 0)) return true; + waiter.cleanup?.(); + waiter.resolve(); return false; }); } // Resolve everything unconditionally (reset/close): never strand a decode loop. flush(): void { - for (const { resolve } of this.#waiters) resolve(); + for (const waiter of this.#waiters) { + waiter.cleanup?.(); + waiter.resolve(); + } this.#waiters = []; } } @@ -82,8 +104,10 @@ export interface AudioBuffer { * applies backpressure: it stays pending while decoding `timestamp` would run more than the latency * floor ahead of the playhead, so the caller holds the (encoded) frame instead of decoding it too * far ahead of the floor-sized ring. Resolves immediately when not buffered (the ring bounds itself). + * `signal` cancels the wait: the promise resolves (never rejects) immediately on abort, so check + * `signal.aborted` after awaiting. */ - wait(timestamp: Time.Micro): Promise; + wait(timestamp: Time.Micro, signal?: AbortSignal): Promise; /** Current playback timestamp (derived from reader position). */ readonly timestamp: Getter; @@ -129,6 +153,7 @@ class SharedAudioBuffer implements AudioBuffer { readonly channels: number; #worklet: AudioWorkletNode; #ring: SharedRingBuffer; + readonly #buffered: boolean; readonly #timestamp = new Signal(0 as Time.Micro); readonly timestamp: Getter = this.#timestamp; @@ -145,6 +170,8 @@ class SharedAudioBuffer implements AudioBuffer { this.channels = channels; this.rate = rate; + this.#buffered = buffered; + // The ring holds the latency floor as decoded PCM (headroom above it for overflow). In // buffered mode the lookahead above the floor stays encoded upstream, held back by `wait()`. const capacity = Math.max(rate, latencySamples * 2); @@ -159,6 +186,13 @@ class SharedAudioBuffer implements AudioBuffer { // Poll the shared control array and reflect it into signals. this.#signals.interval(() => { + // A buffered ring that has fully drained while un-stalled is a deadlock: only the parked decode + // loop can refill it, but it is blocked on wait(). Re-stall so the loop resumes and re-anchors + // (the fall-through stalled branch flushes backpressure). Never in live mode (no lookahead). + // `.length` is normally not for control-flow (see its doc), but it's exact here: this poll is + // the sole writer of WRITE/STALLED, and the worklet only ever advances READ toward WRITE, so a + // zero observed on this thread can't be a stale/racy read. + if (this.#buffered && !this.#ring.stalled && this.#ring.length === 0) this.#ring.reset(); const stalled = this.#ring.stalled; this.#timestamp.set(this.#ring.timestamp); this.#stalled.set(stalled); @@ -194,10 +228,10 @@ class SharedAudioBuffer implements AudioBuffer { this.#backpressure.flush(); // the old timeline is gone; let the decode loop re-anchor } - wait(timestamp: Time.Micro): Promise { + wait(timestamp: Time.Micro, signal?: AbortSignal): Promise { // Stalled = still filling the floor (bootstrap or underflow): let frames through to refill. if (this.#ring.stalled) return Promise.resolve(); - return this.#backpressure.wait(timestamp, this.#ring.timestamp); + return this.#backpressure.wait(timestamp, this.#ring.timestamp, signal); } close(): void { @@ -271,15 +305,18 @@ class PostAudioBuffer implements AudioBuffer { reset(): void { const msg: Reset = { type: "reset" }; this.#worklet.port.postMessage(msg); + // Reflect stalled locally at once: the worklet's confirming state message round-trips a frame + // later, and until then the next wait() must not park on the stale (un-stalled) flag. + this.#stalled.set(true); this.#backpressure.flush(); // the old timeline is gone; let the decode loop re-anchor } - wait(timestamp: Time.Micro): Promise { + wait(timestamp: Time.Micro, signal?: AbortSignal): Promise { // Stalled = still filling the floor (bootstrap or underflow): let frames through to refill. if (this.#stalled.peek()) return Promise.resolve(); // Uses the worklet-reported playhead, which lags by a state-message interval; the floor's // headroom covers that. The worklet still drops the oldest if a frame slips through. - return this.#backpressure.wait(timestamp, this.#timestamp.peek()); + return this.#backpressure.wait(timestamp, this.#timestamp.peek(), signal); } close(): void { diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index 6033e08d71..602f18a72c 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -10,6 +10,7 @@ import type { Sync } from "../sync"; import { type AudioBuffer, createAudioBuffer } from "./buffer"; // Compiled and inlined as a blob URL via vite-plugin-worklet. import RenderWorklet from "./render-worklet.ts?worklet"; +import { snapTimestamp } from "./snap"; import type { Source } from "./source"; type DecoderInput = { @@ -37,6 +38,28 @@ type DecoderOutput = { buffered: Signal; }; +// Opus decoders always emit 48 kHz PCM (RFC 6716) no matter what rate the catalog advertises, so +// build the render pipeline at that rate up front instead of discovering it on the first frame. +const OPUS_OUTPUT_RATE = 48000; + +// Decoder restart policy (mirrors js/watch/src/video/decoder.ts). Cap in-place rebuilds so a permanently +// bad config can't loop forever (reset on a successful decode); rapid repeats back off ~one frame. +const MAX_AUDIO_RESTARTS = 5; +const RESTART_RAPID_MS = 300; +const RESTART_BACKOFF_MS = 500; + +// Snap a decoded frame's timestamp to the previous frame's exact end when they're within this window, so +// the timestamp-indexed ring writes back-to-back. Comfortably above the worst-case publisher timestamp +// quantization (~one capture quantum, ~2.9 ms) and well below any genuine gap (>= one 20 ms Opus frame: +// packet loss, DTX silence, publisher restart), which must still zero-fill. +const SNAP_US = 5000; + +// A wire timestamp jump larger than this is a publisher mute/pause, not jitter: re-anchor playback to +// live instead of playing out the silent gap. A constant (never maxBuffer-scaled) that clears Opus DTX +// comfort-noise spacing (~400 ms) and can never trip on a 20 ms cadence or on packet loss (loss does +// not gap PTS). +const REANCHOR_GAP_US = 1_000_000; + export interface AudioStats { /** Number of encoded bytes received. */ bytesReceived: number; @@ -63,6 +86,29 @@ export class Decoder { }; readonly output = readonlys(this.#output); + // The decoder's real output rate paired with the config it was measured against. #emit sets this + // on a mismatch; #runWorklet then rebuilds the AudioContext/worklet/ring at the true rate. The + // config tag means a rate measured for one stream is never applied to the next (which would build + // one wrong-rate pipeline before self-healing on the following frame). + #decodedRate = new Signal<{ config: Catalog.AudioConfig; rate: number } | undefined>(undefined); + + // The rate the decoder is expected to output (the pipeline's SOURCE rate), set when the ring is built. + // #emit resamples from this to the ring's actual context rate, and treats a sample at a DIFFERENT rate + // as a real decoder-rate change (-> #decodedRate rebuild) rather than something to resample. + #ringSourceRate: number | undefined; + + // Expected timestamp (us) of the next decoded frame. Every frame is snapped onto it when it lands + // within the window (see SNAP_US), which is an identity no-op on an already-contiguous stream. Reset + // at every re-anchor (discontinuity, reset, ring rebuild). + #expectedNext: Time.Micro | undefined; + + // Decoder restart bookkeeping (see #onDecoderError), mirroring the video decoder. + #restart = new Signal(0); + #restartCount = 0; + #lastRestart = 0; + // The config the current restart budget applies to; a rebuild for a different config resets the budget. + #budgetConfig: Catalog.AudioConfig | undefined; + // Decode buffer: audio sent to worklet but not yet played #decodeBuffered = new Signal([]); @@ -100,7 +146,6 @@ export class Decoder { }); this.#signals.run(this.#runWorklet.bind(this)); - this.#signals.run(this.#runEnabled.bind(this)); this.#signals.run(this.#runLatency.bind(this)); this.#signals.run(this.#runDecoder.bind(this)); } @@ -108,6 +153,7 @@ export class Decoder { #runWorklet(effect: Effect): void { // It takes a second or so to initialize the AudioContext/AudioWorklet, so do it even if disabled. // This is less efficient for video-only playback but makes muting/unmuting instant. + // NOTE: You should disconnect/reconnect the worklet to save power when disabled. //const enabled = effect.get(this.enabled); //if (!enabled) return; @@ -115,7 +161,13 @@ export class Decoder { const config = effect.get(this.source.output.config); if (!config) return; - const sampleRate = config.sampleRate; + // The pipeline must run at the rate the decoder actually OUTPUTS, which is not always the + // catalog rate: Opus always emits 48 kHz, and #emit corrects any other divergence after the + // first decoded frame (e.g. a catalog written from a 16 kHz capture context). A rate measured + // against a different config is ignored, so switching streams starts from the catalog rate. + const measured = effect.get(this.#decodedRate); + const decodedRate = measured?.config === config ? measured.rate : undefined; + const sampleRate = decodedRate ?? (config.codec === "opus" ? OPUS_OUTPUT_RATE : config.sampleRate); const channelCount = config.numberOfChannels; const context = new AudioContext({ @@ -126,11 +178,36 @@ export class Decoder { effect.cleanup(() => context.close()); + // Safari (and possibly others) ignore the requested sampleRate and run the context at the hardware + // rate. The worklet + ring run at the context's ACTUAL rate; #emit resamples the decoder's output + // (sampleRate) to it. `sampleRate` stays the SOURCE rate that #emit compares against to tell a real + // decoder-rate change apart from this deliberate resample. + const ringRate = context.sampleRate; + if (ringRate !== sampleRate) { + console.debug(`audio: requested ${sampleRate} Hz context, got ${ringRate} Hz; resampling to match`); + } + + // Safari (and Chrome's autoplay policy) create the AudioContext suspended and only resume it + // inside a user gesture; a reactive resume() is ignored. Resume on the first interaction, from + // context creation (not gated on playback). The Emitter (re)builds the graph once the context + // actually reaches "running" (see emitter.ts), which is what makes Safari start rendering. + const resume = () => { + if (context.state === "suspended") void context.resume().catch(() => {}); + }; + resume(); + // pointerdown/keydown cover desktop; iOS Safari has historically only unlocked audio on a + // click-class gesture (fired at touchend), so listen for that too. resume() is idempotent. + effect.event(document, "pointerdown", resume); + effect.event(document, "click", resume); + effect.event(document, "keydown", resume); + effect.spawn(async () => { // Register the AudioWorklet processor await context.audioWorklet.addModule(RenderWorklet); - // Ensure the context is running before creating the worklet + // The context may have been closed while addModule awaited (effect torn down); bail if so. + // A suspended context is expected here and fine: the worklet is created now and renders + // once a gesture resumes the context (see the resume handler above). if (context.state === "closed") return; // Create the worklet node. outputChannelCount must be set explicitly @@ -143,17 +220,21 @@ export class Decoder { }); effect.cleanup(() => worklet.disconnect()); - // Initial target latency in samples. + // Initial target latency in samples (at the ring's actual rate). const latency = this.sync.output.buffer.peek(); - const latencySamples = Math.ceil(sampleRate * Time.Second.fromMilli(latency)); + const latencySamples = Math.ceil(ringRate * Time.Second.fromMilli(latency)); const buffered = this.sync.output.buffered.peek(); // Let the factory pick the best transport (SharedArrayBuffer or postMessage). - const ring = createAudioBuffer(worklet, channelCount, sampleRate, latencySamples, buffered); + const ring = createAudioBuffer(worklet, channelCount, ringRate, latencySamples, buffered); this.#ring = ring; + this.#ringSourceRate = sampleRate; + this.#expectedNext = undefined; effect.cleanup(() => { ring.close(); this.#ring = undefined; + this.#ringSourceRate = undefined; + this.#expectedNext = undefined; }); // Mirror ring state (timestamp/stalled) onto our public signals. @@ -170,16 +251,6 @@ export class Decoder { }); } - #runEnabled(effect: Effect): void { - const values = effect.getAll([this.input.enabled, this.#output.context]); - if (!values) return; - const [_, context] = values; - - context.resume(); - - // NOTE: You should disconnect/reconnect the worklet to save power when disabled. - } - #runLatency(effect: Effect): void { // Gate on the worklet signal so this effect re-runs once the ring is created. const worklet = effect.get(this.#output.root); @@ -194,6 +265,9 @@ export class Decoder { } #runDecoder(effect: Effect): void { + // Re-run (rebuild subscription + decoder) when #onDecoderError bumps #restart. + effect.get(this.#restart); + const enabled = effect.get(this.input.enabled); if (!enabled) return; @@ -206,6 +280,15 @@ export class Decoder { const config = effect.get(this.source.output.config); if (!config) return; + // A rebuild for a NEW config (not a #restart bump, which keeps the same config object) starts a + // fresh restart budget. The video decoder gets this free via a per-track instance; here the counter + // lives on the long-lived Decoder, so reset it explicitly, or an exhausted budget from an old config + // would kill a healthy new stream on its first transient error. + if (config !== this.#budgetConfig) { + this.#budgetConfig = config; + this.#restartCount = 0; + } + // Honor a per-rendition `broadcast` override: subscribe on the resolved source // broadcast instead of the catalog's own broadcast. const active = broadcast.relativeBroadcast(effect, config.broadcast); @@ -214,6 +297,12 @@ export class Decoder { const sub = active.track(track).subscribe({ priority: Catalog.PRIORITY.audio }); effect.cleanup(() => sub.close()); + // Both branches below build a fresh container Consumer, whose rewind counter restarts at zero. + // This field outlives the subscription (unlike video, which keeps it on a per-track object), so a + // count left over from the previous stream would make the first frame look like a rewind and reset + // the Sync reference that video shares. Re-anchor the count with the consumer that reports it. + this.#discontinuity = 0; + if (config.container.kind === "cmaf") { this.#runCmafDecoder(effect, sub, config); } else { @@ -239,8 +328,9 @@ export class Decoder { }); effect.spawn(async () => { + const abort = effect.abort; // pin this run's signal; the getter is replaced on every re-run const loaded = await Util.Libav.polyfill(); - if (!loaded) return; // cancelled + if (!loaded || abort.aborted) return; let warmed = 0; @@ -254,7 +344,7 @@ export class Decoder { } this.#emit(data); }, - error: (error) => console.error(error), + error: (error) => this.#onDecoderError(error, effect), }); effect.cleanup(() => { if (decoder.state !== "closed") decoder.close(); @@ -272,9 +362,15 @@ export class Decoder { description, }); + let prevTs: number | undefined; for (;;) { const next = await consumer.next(); - if (!next) break; + if (!next) { + // Track ended cleanly. If our config is unchanged (a deep-equal republish) the effect + // won't re-run on its own, so resubscribe; a real local teardown aborted before here. + if (!abort.aborted) this.#onCleanEnd(effect); + break; + } // Publisher rewound the timeline: flush + re-anchor before decoding the new frame. this.#onDiscontinuity(next.discontinuity); @@ -282,6 +378,13 @@ export class Decoder { const { frame } = next; if (!frame) continue; + // A large forward wire gap is a publisher mute/pause: re-anchor to live and skip the wait so + // the frame re-seeds the (now stalled) ring immediately instead of playing out the silent gap. + const wireTs = frame.timestamp as number; + const gapped = prevTs !== undefined && wireTs - prevTs > REANCHOR_GAP_US; + if (gapped) this.#reanchor(); + prevTs = wireTs; + // Mark that we received this frame right now. const timestamp = Time.Milli.fromMicro(frame.timestamp as Time.Micro); this.sync.received(timestamp, "audio"); @@ -292,7 +395,10 @@ export class Decoder { // Backpressure: in buffered mode this holds the encoded frame until the playhead nears // it, keeping the lookahead above the floor as Opus instead of decoded PCM. No-op live. - await this.#ring?.wait(frame.timestamp as Time.Micro); + if (!gapped) { + await this.#ring?.wait(frame.timestamp as Time.Micro, abort); + if (abort.aborted) break; + } const chunk = new EncodedAudioChunk({ type: frame.keyframe ? "key" : "delta", @@ -300,7 +406,19 @@ export class Decoder { timestamp: frame.timestamp, }); - decoder.decode(chunk); + if (decoder.state === "closed") break; + try { + decoder.decode(chunk); + } catch (err) { + // A wrong-config chunk makes decode() throw synchronously. Audio frames are independent, + // so drop the bad one and continue; a closed decoder (from the async error callback) ends + // the loop and #onDecoderError rebuilds via #restart. + if (err instanceof DOMException && err.name === "DataError") { + console.debug("audio decode error; dropping frame", err); + continue; + } + break; + } } }); } @@ -333,12 +451,13 @@ export class Decoder { }); effect.spawn(async () => { + const abort = effect.abort; // pin this run's signal; the getter is replaced on every re-run const loaded = await Util.Libav.polyfill(); - if (!loaded) return; // cancelled + if (!loaded || abort.aborted) return; const decoder = new AudioDecoder({ output: (data) => this.#emit(data), - error: (error) => console.error(error), + error: (error) => this.#onDecoderError(error, effect), }); effect.cleanup(() => { if (decoder.state !== "closed") decoder.close(); @@ -352,9 +471,14 @@ export class Decoder { description, }); + let prevTs: number | undefined; for (;;) { const next = await consumer.next(); - if (!next) break; + if (!next) { + // See #runLegacyDecoder: resubscribe on a clean end unless we tore down locally. + if (!abort.aborted) this.#onCleanEnd(effect); + break; + } // Publisher rewound the timeline: flush + re-anchor before decoding the new frame. this.#onDiscontinuity(next.discontinuity); @@ -362,6 +486,12 @@ export class Decoder { const { frame } = next; if (!frame) continue; + // A large forward wire gap is a publisher mute/pause: re-anchor to live and skip the wait. + const wireTs = frame.timestamp as number; + const gapped = prevTs !== undefined && wireTs - prevTs > REANCHOR_GAP_US; + if (gapped) this.#reanchor(); + prevTs = wireTs; + const timestamp = Time.Milli.fromMicro(frame.timestamp); this.sync.received(timestamp, "audio"); @@ -371,23 +501,71 @@ export class Decoder { // Backpressure: in buffered mode this holds the encoded frame until the playhead nears // it, keeping the lookahead above the floor as Opus instead of decoded PCM. No-op live. - await this.#ring?.wait(frame.timestamp); + if (!gapped) { + await this.#ring?.wait(frame.timestamp, abort); + if (abort.aborted) break; + } if (decoder.state === "closed") break; - decoder.decode( - new EncodedAudioChunk({ - type: frame.keyframe ? "key" : "delta", - data: frame.data, - timestamp: frame.timestamp, - }), - ); + try { + decoder.decode( + new EncodedAudioChunk({ + type: frame.keyframe ? "key" : "delta", + data: frame.data, + timestamp: frame.timestamp, + }), + ); + } catch (err) { + // See #runLegacyDecoder: drop a bad chunk (DataError) and continue; else end the loop. + if (err instanceof DOMException && err.name === "DataError") { + console.debug("audio decode error; dropping frame", err); + continue; + } + break; + } } }); } + // Recover from a fatal AudioDecoder error by rebuilding in place (re-run #runDecoder) instead of + // leaving the loop abandoned. Capped (reset on a successful #emit); rapid repeats back off. Unlike the + // video decoder's per-track wrapper, #runDecoder is the persistent effect, so on exhaustion we stop + // retrying WITHOUT closing it - a later config change re-runs it and a working decoder resets the budget. + #onDecoderError(error: unknown, effect: Effect): void { + // If the catalog has already moved past the config this decoder was built for, the error is just + // wrong-config bytes during a codec/rate switch and #runDecoder is about to rebuild for the new + // config, so don't burn the restart budget re-decoding stale bytes. Compare fields, not identity: + // Signal.set stores a deep-equal object without notifying, so peek() can return a new-but-equal one. + const current = this.source.output.config.peek(); + const budget = this.#budgetConfig; + if (budget && current && (current.codec !== budget.codec || current.container.kind !== budget.container.kind)) { + console.debug("audio decoder error; config superseded, rebuild pending", error); + return; + } + + if (this.#restartCount >= MAX_AUDIO_RESTARTS) { + console.error("audio decoder error; giving up until the next config change", error); + return; + } + // Measure "rapid" against when the restart is DISPATCHED (see the video decoder), not the error + // time, so a backed-off restart doesn't reset the interval and oscillate immediate/backoff. + const rapid = performance.now() - this.#lastRestart < RESTART_RAPID_MS; + this.#restartCount++; + if (this.#restartCount === 1) console.warn("audio decoder error; restarting", error); + else console.debug("audio decoder error; restarting", error); + const restart = () => { + this.#lastRestart = performance.now(); + this.#restart.update((n) => n + 1); + }; + if (rapid) effect.timer(restart, RESTART_BACKOFF_MS); + else restart(); + } + #emit(sample: AudioData) { - const timestamp = sample.timestamp as Time.Micro; - const timestampMilli = Time.Milli.fromMicro(timestamp); + // A frame decoded successfully: reset the restart budget. + this.#restartCount = 0; + + let timestamp = sample.timestamp as Time.Micro; const ring = this.#ring; if (!ring) { @@ -396,8 +574,40 @@ export class Decoder { return; } + // If the decoder output an UNEXPECTED rate (not what the pipeline was built for, e.g. HE-AAC + // emitting 2x the advertised rate), rebuild the pipeline at the real rate and drop until it's up. + // Compared to the SOURCE rate (not ring.rate) so the deliberate resample-to-context-rate below is + // not mistaken for a decoder-rate change, which would thrash the rebuild and silence Safari. + const sourceRate = this.#ringSourceRate; + if (sourceRate !== undefined && sample.sampleRate !== sourceRate) { + const config = this.source.output.config.peek(); + const prev = this.#decodedRate.peek(); + // Skip the set when this config already has this rate recorded (the rebuild is just in + // flight), but always record it for a new config even at a rate a prior stream used. + if (config && (prev?.config !== config || prev.rate !== sample.sampleRate)) { + console.warn(`audio decoder outputs ${sample.sampleRate} Hz, expected ${sourceRate} Hz; rebuilding`); + this.#decodedRate.set({ config, rate: sample.sampleRate }); + } + sample.close(); + return; + } + // Calculate end time from sample duration const durationMicro = ((sample.numberOfFrames / sample.sampleRate) * 1_000_000) as Time.Micro; + + // Snap a near-contiguous frame to the previous frame's exact end so the timestamp-indexed ring writes + // back-to-back instead of zero-filling or overwriting a sample every frame (Safari-to-Safari crackle: + // Safari's decoder passes the publisher's quantized wire timestamps straight through, where Chrome's + // regenerates an exact cadence). Magnitude-keyed, not rate-keyed: an already-contiguous stream makes + // this an identity no-op, so Chrome/Firefox are unaffected. The publisher also snaps at the encoder + // (see encoder.ts); this covers unfixed publishers. The window is capped at half a frame so a genuine + // gap (>= one frame: loss, DTX silence, restart) always exceeds it and still zero-fills, whatever the + // frame duration. + const snapWindow = Math.min(SNAP_US, durationMicro / 2); + timestamp = snapTimestamp(this.#expectedNext, timestamp, snapWindow) as Time.Micro; + this.#expectedNext = (timestamp + durationMicro) as Time.Micro; + + const timestampMilli = Time.Milli.fromMicro(timestamp); const durationMilli = Time.Milli.fromMicro(durationMicro); const end = Time.Milli.add(timestampMilli, durationMilli); @@ -407,13 +617,24 @@ export class Decoder { // Firefox's Opus decoder sometimes outputs more channels than requested // (e.g. 6 for stereo). Clamp to the ring's channel count. const channels = Math.min(sample.numberOfChannels, ring.channels); - const channelData: Float32Array[] = []; + let channelData: Float32Array[] = []; for (let channel = 0; channel < channels; channel++) { const data = new Float32Array(sample.numberOfFrames); sample.copyTo(data, { format: "f32-planar", planeIndex: channel }); channelData.push(data); } + // Resample to the ring's rate when the context runs at a different rate than the decoder outputs + // (Safari pins ~44100 while Opus decodes 48000). outLen comes from the ring's own index rounding so + // consecutive frames stay exactly contiguous (no zero-fill gap / "floating point inaccuracy" warn). + // A no-op when the rates match (Chrome/Firefox honor the requested rate). + if (sample.sampleRate !== ring.rate) { + const startIndex = Math.round(Time.Second.fromMicro(timestamp) * ring.rate); + const endMicro = (timestamp + durationMicro) as Time.Micro; + const outLen = Math.max(0, Math.round(Time.Second.fromMicro(endMicro) * ring.rate) - startIndex); + channelData = channelData.map((data) => resampleLinear(data, outLen)); + } + // Hand off to the ring. Shared transport writes directly; post transport // transfers the ArrayBuffers. ring.insert(timestamp, channelData); @@ -451,23 +672,42 @@ export class Decoder { }); } - // Flush the audio buffer and re-stall, re-anchoring playback to the next frame. - // Use in buffered mode at an utterance boundary (see Sync.reset). - reset(): void { + // Flush the audio buffer and re-stall, re-anchoring playback to the next frame. Drops stale buffered + // PCM WITHOUT touching Sync: a forward gap (mute/pause) keeps the publisher's epoch, and video shares + // the Sync, so resetting it here would perturb video. + #reanchor(): void { this.#ring?.reset(); + this.#expectedNext = undefined; + } + + // Public utterance-boundary flush (buffered mode, see Sync.reset). + reset(): void { + this.#reanchor(); } - // React to the container consumer's discontinuity counter. When it changes the publisher - // has rewound the timeline, so flush the queued PCM and re-anchor the shared clock before - // the first frame of the new utterance is decoded. This makes the wire signal trigger the - // same flush as a manual `reset()`, with no app involvement. + // React to the container consumer's discontinuity counter. It changes only on a BACKWARD rewind + // (publisher timeline reset), so flush the queued PCM and re-anchor the shared clock before the new + // utterance. Forward gaps (mute/pause) are handled in the decode loop and must NOT reset Sync. #onDiscontinuity(count: number): void { if (count === this.#discontinuity) return; this.#discontinuity = count; - this.#ring?.reset(); + this.#reanchor(); this.sync.reset(); } + // The publisher closed the audio track but our catalog config is unchanged (a deep-equal republish, + // e.g. a Sample-rate override that re-pins 48 kHz), so #runDecoder won't re-run on its own and we'd + // go silent. Resubscribe via the restart budget (a successful #emit resets it); a genuine local + // teardown aborts the loop before this is reached. + #onCleanEnd(effect: Effect): void { + if (this.#restartCount >= MAX_AUDIO_RESTARTS) return; + this.#restartCount++; + effect.timer(() => { + this.#lastRestart = performance.now(); + this.#restart.update((n) => n + 1); + }, RESTART_BACKOFF_MS); + } + close() { this.#signals.close(); } @@ -476,7 +716,35 @@ export class Decoder { static supported = supported; } +// Linear-resample one channel of PCM to `outLen` samples. The caller derives `outLen` from the ring's +// index rounding so consecutive frames stay contiguous; this just maps the input samples across it. +// Bypassed (returns the input) when no rate change is needed. +function resampleLinear(input: Float32Array, outLen: number): Float32Array { + const inLen = input.length; + if (outLen === inLen) return input; + const out = new Float32Array(outLen); + if (outLen === 0 || inLen === 0) return out; + if (inLen === 1) { + out.fill(input[0]); + return out; + } + const step = inLen / outLen; + for (let j = 0; j < outLen; j++) { + const pos = j * step; + const i0 = Math.floor(pos); + const i1 = Math.min(i0 + 1, inLen - 1); + const frac = pos - i0; + out[j] = input[i0] * (1 - frac) + input[i1] * frac; + } + return out; +} + async function supported(config: Catalog.AudioConfig): Promise { + // Load the Opus polyfill first. Safari 16.4-18.7 ships no WebCodecs audio API at all, so a bare + // AudioDecoder.isConfigSupported would throw ReferenceError and drop every rendition. polyfill() + // returns immediately when a native AudioDecoder already exists (Chrome, Firefox, Safari 26+). + await Util.Libav.polyfill(); + // Opus in CMAF uses raw packets; dOps is not a valid OGG Identification Header. let description: Uint8Array | undefined; if (config.codec !== "opus") { diff --git a/js/watch/src/audio/emitter.ts b/js/watch/src/audio/emitter.ts index 45fd1bce97..551526f2f6 100644 --- a/js/watch/src/audio/emitter.ts +++ b/js/watch/src/audio/emitter.ts @@ -53,20 +53,42 @@ export class Emitter { this.#signals.run((effect) => { const root = effect.get(this.source.output.root); if (!root) return; + const context = root.context; - const gain = new GainNode(root.context, { gain: effect.get(this.input.volume) }); - root.connect(gain); - - effect.set(this.#gain, gain); + // Safari starts the AudioContext suspended and will NOT render a source->destination edge + // that was wired while suspended, even after a later resume(). So build the graph only once + // the context is actually running: the first gesture-driven resume (see decoder.ts) flips + // this, and the edge is then wired live, exactly like the working mute->unmute path. + const running = new Signal(context.state === "running"); + effect.event(context, "statechange", () => running.set(context.state === "running")); effect.run((inner) => { - // We only connect/disconnect when enabled to save power. - // Otherwise the worklet keeps running in the background returning 0s. - const enabled = inner.get(this.#output.enabled); - if (!enabled) return; - - gain.connect(root.context.destination); // speakers - inner.cleanup(() => gain.disconnect()); + if (!inner.get(running)) return; + + // peek (not get) the volume: the fade effect below owns volume changes. Subscribing here + // would rebuild the whole graph on every change and cut the fade short with a click. + const gain = new GainNode(context, { gain: this.input.volume.peek() }); + root.connect(gain); + inner.cleanup(() => { + // The decoder can tear down its worklet first, dropping the root->gain edge; a + // disconnect of an already-disconnected node throws InvalidAccessError. Swallow it: + // this cleanup runs inside the signals dispose loop, where a throw would wedge the + // effect and leave audio permanently silent. + try { + root.disconnect(gain); + } catch {} + }); + + inner.set(this.#gain, gain); + + inner.run((leaf) => { + // We only connect/disconnect when enabled to save power. + // Otherwise the worklet keeps running in the background returning 0s. + if (!leaf.get(this.#output.enabled)) return; + + gain.connect(context.destination); // speakers + leaf.cleanup(() => gain.disconnect()); + }); }); }); diff --git a/js/watch/src/audio/ring-buffer.ts b/js/watch/src/audio/ring-buffer.ts index 83adc03309..ee5ae63220 100644 --- a/js/watch/src/audio/ring-buffer.ts +++ b/js/watch/src/audio/ring-buffer.ts @@ -190,7 +190,13 @@ export class AudioRingBuffer { if (this.#stalled) return 0; const samples = Math.min(this.#writeIndex - this.#readIndex, output[0].length); - if (samples === 0) return 0; + if (samples === 0) { + // Buffered ring fully drained while playing: re-stall so the parked decode loop is released to + // refill (a drained un-stalled ring is a deadlock; only that loop can refill it). Live mode holds + // no lookahead, so a momentary underflow there is normal and must not re-stall. + if (this.#buffered) this.#stalled = true; + return 0; + } for (let channel = 0; channel < this.channels; channel++) { const dst = output[channel]; diff --git a/js/watch/src/audio/snap.test.ts b/js/watch/src/audio/snap.test.ts new file mode 100644 index 0000000000..684e000871 --- /dev/null +++ b/js/watch/src/audio/snap.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "bun:test"; +import { snapTimestamp } from "./snap"; + +const THRESHOLD = 5000; // us, matches SNAP_US in decoder.ts + +describe("snapTimestamp", () => { + it("snaps a near-contiguous timestamp to the expected value", () => { + expect(snapTimestamp(20000, 20002, THRESHOLD)).toBe(20000); + expect(snapTimestamp(20000, 19998, THRESHOLD)).toBe(20000); + expect(snapTimestamp(20000, 20000, THRESHOLD)).toBe(20000); + // Exactly at the threshold still snaps. + expect(snapTimestamp(20000, 25000, THRESHOLD)).toBe(20000); + }); + + it("passes through a genuine gap (beyond the threshold)", () => { + expect(snapTimestamp(20000, 40000, THRESHOLD)).toBe(40000); + expect(snapTimestamp(20000, 25001, THRESHOLD)).toBe(25001); + }); + + it("passes through when there is no expectation yet", () => { + expect(snapTimestamp(undefined, 12345, THRESHOLD)).toBe(12345); + }); + + it("caps the window at half a frame so a real gap is never snapped (adaptive window in #emit)", () => { + // #emit uses Math.min(SNAP_US, durationMicro / 2). For an exotic 2.5 ms frame that window is 1250 us, + // smaller than SNAP_US, so a one-frame gap (2500 us) must pass through instead of being snapped shut. + const durationMicro = 2500; + const window = Math.min(THRESHOLD, durationMicro / 2); + expect(window).toBe(1250); + // Sub-window jitter still snaps; a full-frame gap does not. + expect(snapTimestamp(2500, 3000, window)).toBe(2500); + expect(snapTimestamp(2500, 5000, window)).toBe(5000); + }); +}); diff --git a/js/watch/src/audio/snap.ts b/js/watch/src/audio/snap.ts new file mode 100644 index 0000000000..b7046f149d --- /dev/null +++ b/js/watch/src/audio/snap.ts @@ -0,0 +1,19 @@ +/** + * Timestamp snapping for the audio render pipeline. Kept in its own module (free of the worklet blob + * import in decoder.ts) so it can be unit tested directly. + * + * @module + */ + +/** + * Snap `actual` to `expected` when they are within `thresholdMicro`, else return `actual` unchanged. + * + * Used to make near-contiguous decoded frames write back-to-back in the watcher's timestamp-indexed ring + * buffer, avoiding a zero-fill or overwrite of a sample every frame (Safari-to-Safari crackle). Genuine + * gaps (at least one frame apart: packet loss, DTX silence, publisher restart) exceed the threshold and + * pass through so the ring still zero-fills them. + */ +export function snapTimestamp(expected: number | undefined, actual: number, thresholdMicro: number): number { + if (expected !== undefined && Math.abs(actual - expected) <= thresholdMicro) return expected; + return actual; +} diff --git a/js/watch/src/audio/source.ts b/js/watch/src/audio/source.ts index 366e323a40..25cc943ff8 100644 --- a/js/watch/src/audio/source.ts +++ b/js/watch/src/audio/source.ts @@ -87,7 +87,14 @@ export class Source { const available: Record = {}; for (const [name, config] of Object.entries(renditions)) { - const isSupported = await supported(config); + // A throwing probe (bad description hex / rejecting isConfigSupported) must not abort the loop + // and drop every audio rendition; treat this one as unsupported and continue. + let isSupported = false; + try { + isSupported = await supported(config); + } catch (err) { + console.warn(`audio rendition ${name} (${config.codec}) probe threw; treating as unsupported`, err); + } if (isSupported) available[name] = config; }