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/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}`); } 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; }