fix(core): coalesce in-flight audio decodes; cap decodable source size

Two independent OOM paths in the WebAudio decode pipeline, both hit by a
composition whose audio rides inside large camera-video containers:

- No in-flight dedup: decoding a long source takes seconds, while the warm
  pass / play() / rate changes may re-request the same src many times a
  second. Each re-request launched ANOTHER whole-file fetch — dozens of
  concurrent copies of the same source could stack up before the first
  decode ever reached the buffer cache. decodeAudioElement now coalesces
  concurrent requests per src onto one fetch+decode promise.
- No size bound: WebAudio decode requires the ENTIRE file in an ArrayBuffer.
  Sources beyond 96MB are now permanently skipped for the session
  (content-length short-circuit before any bytes buffer, plus a capped
  stream read when length is undeclared) and keep playing through the
  streamed HTMLMediaElement path instead.

Author: I-bangash

packages/core/src/runtime/webAudioTransport.test.ts

@@ -471,6 +471,73 @@ describe("WebAudioTransport", () => {
       vi.unstubAllGlobals();
     });
 
+    it("coalesces concurrent decodes of the same src onto ONE fetch", async () => {
+      const transport = transportWithDecode(async () => ({}) as AudioBuffer);
+      let release: (() => void) | null = null;
+      const gate = new Promise<void>((r) => {
+        release = r;
+      });
+      const fetchMock = vi.fn(async () => {
+        await gate;
+        return { ok: true, arrayBuffer: async () => new ArrayBuffer(8) };
+      });
+      vi.stubGlobal("fetch", fetchMock);
+
+      // The periodic warm pass / play / rate changes can all request the same
+      // src while a long fetch+decode is still in flight — without coalescing,
+      // each launched ANOTHER whole-file fetch (the OOM amplifier).
+      const first = transport.decodeAudioElement(el("long.mp4"));
+      const second = transport.decodeAudioElement(el("long.mp4"));
+      release!();
+      const [a, b] = await Promise.all([first, second]);
+      expect(a).not.toBeNull();
+      expect(b).toBe(a);
+      expect(fetchMock).toHaveBeenCalledTimes(1);
+      vi.unstubAllGlobals();
+    });
+
+    it("skips oversized sources via content-length and never re-fetches them", async () => {
+      const decode = vi.fn(async () => ({}) as AudioBuffer);
+      const transport = transportWithDecode(decode);
+      const fetchMock = vi.fn(async () => ({
+        ok: true,
+        headers: { get: () => String(200 * 1024 * 1024) },
+        body: { cancel: async () => undefined },
+        arrayBuffer: async () => new ArrayBuffer(8),
+      }));
+      vi.stubGlobal("fetch", fetchMock);
+
+      expect(await transport.decodeAudioElement(el("huge-camera.mp4"))).toBeNull();
+      expect(decode).not.toHaveBeenCalled();
+      // Permanent per-session skip — the element keeps its streamed fallback.
+      expect(await transport.decodeAudioElement(el("huge-camera.mp4"))).toBeNull();
+      expect(fetchMock).toHaveBeenCalledTimes(1);
+      vi.unstubAllGlobals();
+    });
+
+    it("caps an unbounded streamed body instead of buffering it all", async () => {
+      const transport = transportWithDecode(async () => ({}) as AudioBuffer);
+      // Chunks report huge byteLengths without allocating them — the cap logic
+      // only inspects sizes until it aborts.
+      const chunk = { byteLength: 60 * 1024 * 1024 } as Uint8Array;
+      let reads = 0;
+      const fetchMock = vi.fn(async () => ({
+        ok: true,
+        headers: { get: () => null },
+        body: {
+          getReader: () => ({
+            read: async () => (reads++ < 3 ? { done: false, value: chunk } : { done: true }),
+            cancel: async () => undefined,
+          }),
+        },
+      }));
+      vi.stubGlobal("fetch", fetchMock);
+
+      expect(await transport.decodeAudioElement(el("no-length.mp4"))).toBeNull();
+      expect(reads).toBeLessThanOrEqual(2); // aborted once past the cap
+      vi.unstubAllGlobals();
+    });
+
     it("first fetch per src uses the HTTP cache; only a RETRY bypasses it with no-store", async () => {
       const transport = transportWithDecode(async () => ({}) as AudioBuffer);
       const fetchMock = vi

packages/core/src/runtime/webAudioTransport.ts

@@ -80,6 +80,41 @@ const EDGE_RAMP_SECONDS = 0.008;
 const STOP_FADE_TIME_CONSTANT = 0.004;
 const STOP_FADE_TAIL_SECONDS = 0.02;
 
+/**
+ * Decoding a source through WebAudio requires fetching the ENTIRE file into an
+ * ArrayBuffer first — for audio that rides inside a large camera/screen-capture
+ * video container, that is hundreds of MB per source and can OOM the tab
+ * before a single decoded sample exists. Sources whose bytes exceed this cap
+ * are permanently skipped for the session (they keep playing through the
+ * streamed HTMLMediaElement path instead of the sample-accurate transport).
+ */
+const MAX_DECODE_SOURCE_BYTES = 96 * 1024 * 1024;
+
+/** Read a response body while enforcing MAX_DECODE_SOURCE_BYTES; null = over cap. */
+async function readBodyCapped(body: ReadableStream<Uint8Array>): Promise<ArrayBuffer | null> {
+  const reader = body.getReader();
+  const chunks: Uint8Array[] = [];
+  let total = 0;
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    if (!value) continue;
+    total += value.byteLength;
+    if (total > MAX_DECODE_SOURCE_BYTES) {
+      void reader.cancel().catch(() => undefined);
+      return null;
+    }
+    chunks.push(value);
+  }
+  const out = new Uint8Array(total);
+  let offset = 0;
+  for (const chunk of chunks) {
+    out.set(chunk, offset);
+    offset += chunk.byteLength;
+  }
+  return out.buffer;
+}
+
 /**
  * Anti-pop gain automation: a hard cut starts/stops mid-waveform, which is an
  * audible click. Ramp gain over EDGE_RAMP_SECONDS at the source's effective
@@ -122,6 +157,13 @@ export class WebAudioTransport {
   private _bufferCache = new Map<string, AudioBuffer>();
   private _failedSrcs = new Set<string>();
   private _fetchAttemptedSrcs = new Set<string>();
+  // In-flight decode per src. Decoding a long source takes SECONDS (full-file
+  // fetch + decodeAudioData) while callers — the periodic warm pass, play(),
+  // rate changes — may re-request the same src many times a second. Without
+  // this map each re-request launched ANOTHER whole-file fetch: dozens of
+  // concurrent copies of the same source stacked up in memory and could OOM
+  // the tab before the first decode ever landed in `_bufferCache`.
+  private _pendingDecodes = new Map<string, Promise<AudioBuffer | null>>();
   private _activeSources: ScheduledSource[] = [];
   private _masterGain: GainNode | null = null;
   // Composition-time reference frame: at AudioContext time `_rateAnchorCtx`,
@@ -160,8 +202,21 @@ export class WebAudioTransport {
     if (this._failedSrcs.has(src)) return null;
     if (!this._ctx) return null;
 
+    // Coalesce concurrent requests for the same src onto one fetch+decode.
+    const pending = this._pendingDecodes.get(src);
+    if (pending) return pending;
+    const task = this._decodeSrc(src);
+    this._pendingDecodes.set(src, task);
+    try {
+      return await task;
+    } finally {
+      this._pendingDecodes.delete(src);
+    }
+  }
+
+  private async _decodeSrc(src: string): Promise<AudioBuffer | null> {
     const arrayBuffer = await this._fetchAudioBytes(src);
-    if (!arrayBuffer) return null;
+    if (!arrayBuffer || !this._ctx) return null;
 
     // A decode failure means the bytes themselves are unusable (corrupt or an
     // unsupported codec) — that IS permanent, so blacklist to avoid re-decoding
@@ -199,13 +254,44 @@ export class WebAudioTransport {
         swallow("webAudioTransport.fetch", new Error(`${response.status} ${src}`));
         return null;
       }
-      return await response.arrayBuffer();
+      return await this._readResponseCapped(response, src);
     } catch (err) {
       swallow("webAudioTransport.fetch", err);
       return null;
     }
   }
 
+  /** Permanent per-session skip for a source too large to decode; the element
+   *  keeps playing through the streamed HTMLMediaElement path. */
+  private _markOversized(src: string, size: string): null {
+    this._failedSrcs.add(src);
+    swallow("webAudioTransport.oversize", new Error(`${size} ${src}`));
+    return null;
+  }
+
+  /**
+   * Enforce MAX_DECODE_SOURCE_BYTES while reading a response. Declared length
+   * short-circuits before any bytes buffer; otherwise the capped stream read
+   * bounds the damage. Optional-chained so environments whose fetch shim lacks
+   * headers/body (tests, exotic embedders) use the plain arrayBuffer path.
+   */
+  private async _readResponseCapped(response: Response, src: string): Promise<ArrayBuffer | null> {
+    const declared = Number(response.headers?.get?.("content-length") ?? "");
+    if (Number.isFinite(declared) && declared > MAX_DECODE_SOURCE_BYTES) {
+      void response.body?.cancel?.().catch(() => undefined);
+      return this._markOversized(src, `${declared}B`);
+    }
+    if (response.body?.getReader) {
+      const capped = await readBodyCapped(response.body);
+      return capped ?? this._markOversized(src, `>${MAX_DECODE_SOURCE_BYTES}B`);
+    }
+    const buf = await response.arrayBuffer();
+    if (buf.byteLength > MAX_DECODE_SOURCE_BYTES) {
+      return this._markOversized(src, `${buf.byteLength}B`);
+    }
+    return buf;
+  }
+
   startGeneration(): number {
     this._playGeneration += 1;
     return this._playGeneration;
@@ -407,6 +493,7 @@ export class WebAudioTransport {
     this.stopAll();
     this._bufferCache.clear();
     this._failedSrcs.clear();
+    this._pendingDecodes.clear();
     if (this._ctx) {
       try {
         void this._ctx.close();