From f3854e5c02e2f1ca6f437e52b8eef626d0fb3a66 Mon Sep 17 00:00:00 2001 From: byrongamatos Date: Wed, 8 Jul 2026 15:49:46 +0200 Subject: [PATCH 1/2] refactor(stems): extract the streaming path to src/streaming.js MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the bounded-memory iOS WAV streaming layer out of main.js into its own module (~440 lines): the reader/WAV helpers (trackRead/ensureBytes/dropBytes/ readWavHeader/dequeueTrackFrames), the worklet-feed pump (appendRound/waitPos/ runPump) + seek path (cancelStreamReaders/openTrackStreams/repositionStream), setupStreaming, resetStreamState, and streamOffsetBuffered/streamingSupported/ isWavResponse. main.js 2281 -> 1840 lines. The layer would form a static import cycle (transport imports setupStreaming/ repositionStream; the pump calls transportPlay, setupStreaming needs persistedSongGain + the shared onWorkletMessage — all in main.js). Broken with an injected seam: configureStreaming({ startPendingPlay, songGain, onWorkletMessage }) wired once at boot. Everything else the layer calls is already an extracted module (audio-ctx, mix, mix-gains, util, wav-pcm, prefs, state), imported by the same names, so the bodies moved verbatim. Removed now-dead main.js imports (parseWavHeader/pcm16ToFloat32, ensureCtxAtRate). New tests/streaming.test.mjs covers the pure exports (isWavResponse, streamingSupported, streamOffsetBuffered window math). The pump/seek internals stay private, so the seek-token race guard remains on-device-verified. CHANGELOG: backfilled steps 5-9.1 (missed) + added step 10. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 41 ++++ src/main.js | 487 ++------------------------------------- src/streaming.js | 480 ++++++++++++++++++++++++++++++++++++++ tests/streaming.test.mjs | 59 +++++ 4 files changed, 593 insertions(+), 474 deletions(-) create mode 100644 src/streaming.js create mode 100644 tests/streaming.test.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index d10ca16..04bcb7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,47 @@ All notable changes to the Stems Toggle plugin are documented here. ### Changed +- **ES-module migration, step 10 — extract the streaming path to + `src/streaming.js`.** The bounded-memory iOS WAV streaming layer (the + reader/WAV helpers, the worklet-feed pump + its seek/reposition, `setupStreaming`, + and `streamOffsetBuffered`/`streamingSupported`/`isWavResponse`) moves out of + `main.js` into its own module (~440 lines). The one static cycle it would create + — the pump resuming a deferred play via `transportPlay`, plus `persistedSongGain` + and the shared `onWorkletMessage`, all in `main.js` — is broken with an injected + `configureStreaming({ startPendingPlay, songGain, onWorkletMessage })` seam + (transport already imports back into streaming). New `tests/streaming.test.mjs` + covers the pure exports. Move-only, no behaviour change; on-device seek/playback + smoke passed. +- **ES-module migration, step 9.1 — guard the streaming pump against seek races.** + A pre-existing race (surfaced by review of the step-9 lift): `repositionStream` + cancels + reopens track readers on the same objects and clears `pumpStop` before + older pump/`appendRound` continuations exit, so a stale append could post + misaligned data or steal PCM from the new pump. Fix: a per-run seek token, + re-checked before each track read and before posting; a stranded `posWaiter` + resolved on stop; and `repositionStream` restarts as `runPump(!S.buffersReady)` + so a seek during the initial prefill still establishes readiness. On-device + seek-stress smoke passed. +- **ES-module migration, step 9 — streaming scalars → `ST` object in + `src/state.js`.** The 9 reassigned streaming-path scalars (`streaming`, + `streamTracks`, `streamSampleRate`/`streamTotalSamples`, `jsWriteFrontier`, + `pumpStop`, `lastWorkletPos`, `posWaiter`, `streamSeekToken`) move into an `ST` + container (same read-only-binding reason as `S`), read across transport + + `onWorkletMessage` + `onSongReady`. Move-only, no behaviour change. +- **ES-module migration, step 8 — shim scalars → `SH` object in `src/state.js`.** + The 7 `#audio`-shim scalars (`shimsInstalled`/`shimsUsable` + the captured core + descriptors and native play/pause) move into an `SH` container; `SH.shimsUsable` + gates the sloppak takeover from transport + `onSongReady`. Move-only. +- **ES-module migration, step 7 — extract mix routing to `src/mix.js`.** + `applyMixRouting` (posts worklet gains) + `makeStemGainHandle` (the per-stem + GainNode stand-in) move out of `main.js`; pairs with the pure `mix-gains.js`. + New `tests/mix.test.mjs` covers the previously-untested gain-handle write path. +- **ES-module migration, step 6 — extract the AudioContext/worklet lifecycle to + `src/audio-ctx.js`.** `ensureCtx`/`resumeCtx`/`ensureWorklet`/`ensureCtxAtRate`/ + `updateLatencyOffset` (+ the `WORKLET_URL` const) — the leaf tier that touches + only shared state + native Web Audio — move out of `main.js`. +- **ES-module migration, step 5 — extract pure helpers to `src/util.js`.** + `clampVolume`, `coerceBool`, `hashString` (FNV-1a), `isGuitarStemId` move to + their own module with real-import tests. - **ES-module migration, step 4b — the reassigned scalars → `S` object in `src/state.js` (R1 pilot).** The ~28 reassigned module scalars (the audio graph `audioCtx`/`masterGain`/`analyserNode`, `stemState`, the worklet + decode flags, diff --git a/src/main.js b/src/main.js index b32734a..7db063e 100644 --- a/src/main.js +++ b/src/main.js @@ -1,10 +1,13 @@ import { computeMixGains } from './mix-gains.js'; -import { parseWavHeader, pcm16ToFloat32 } from './wav-pcm.js'; import { clampVolume, coerceBool, hashString, isGuitarStemId } from './util.js'; import { - ensureCtx, resumeCtx, ensureWorklet, ensureCtxAtRate, updateLatencyOffset, + ensureCtx, resumeCtx, ensureWorklet, updateLatencyOffset, } from './audio-ctx.js'; import { makeStemGainHandle } from './mix.js'; +import { + setupStreaming, repositionStream, resetStreamState, streamingSupported, + streamOffsetBuffered, isWavResponse, configureStreaming, +} from './streaming.js'; import { karaokeDefault, setKaraokeDefault, loadDefaultMuted, saveDefaultMuted, @@ -48,6 +51,14 @@ import { // the small lead guarantees `when` is still in the future for all of them. const START_LEAD = 0.03; + // Wire the streaming layer's seams back to this orchestration layer (it can't + // import main.js). These are hoisted function declarations, so callable now. + configureStreaming({ + startPendingPlay: transportPlay, // resume a deferred play() when buffers arrive + songGain: persistedSongGain, // seed master gain from the mixer / persisted vol + onWorkletMessage, // shared worklet 'ready'/'ended'/'pos' handler + }); + // ── Plugin state ── // The Song-fader bridge (window.slopsmith.stems.setMasterVolume) is a // shared global. Track whether WE installed it and snapshot whatever was @@ -315,17 +326,6 @@ import { } } - // True when `offset` (s) falls inside the PCM the streaming worklet currently - // holds around its read frontier. Outside it — e.g. replaying from 0 after the - // song ended and the pump drained the window to the tail — the worklet would - // read silence, so the caller must refetch (repositionStream) first. - function streamOffsetBuffered(offset) { - if (!ST.streaming || ST.streamSampleRate <= 0) return true; - const posSamp = Math.round(offset * ST.streamSampleRate); - const behind = Math.ceil(0.1 * ST.streamSampleRate); - return posSamp >= ST.lastWorkletPos - behind && posSamp <= ST.jsWriteFrontier; - } - function transportPlay() { resumeCtx(); if (!S.buffersReady) { S.pendingPlay = true; return; } @@ -1137,467 +1137,6 @@ import { return true; } - // ══════════════════════════════════════════════════════════════════════ - // Streaming (bounded-memory) playback — the iOS WAV path - // - // The iOS client proxy transcodes each OGG stem to RIFF/WAV (16-bit PCM) - // and streams it. Decoding whole stems to AudioBuffers (buildGraphFromBuffers - // above) jettisons the WKWebView content process at ~6 stems (~500 MB). Here - // we read the PCM incrementally off fetch().body and feed the worklet's - // bounded ring via 'append', dropping consumed PCM, so peak memory is a - // few-second window per track — independent of song length or stem count. - // Raw PCM is sliceable at any sample, so NO decoder/WebCodecs/demuxer is - // needed. Desktop (audio/ogg, not sliceable) keeps the full-decode path. - // ══════════════════════════════════════════════════════════════════════ - const STREAM_AHEAD_SEC = 2.0; // keep ~this far buffered ahead of pos - const STREAM_PREFILL_SEC = 0.5; // buffer this much before starting - const STREAM_CAP_SEC = 3.5; // worklet per-track window capacity - const STREAM_CHUNK_FRAMES = 8192; // max frames appended per pump round - const EMPTY_BYTES = new Uint8Array(0); - - // Streaming-path state (this-song flag, track readers, sample-rate/length, - // pump write-frontier + stop flag, worklet read-frontier + pos-waiter, seek - // token) → ST in state.js. Reassigned scalars need a container; read across - // the transport layer + onWorkletMessage + onSongReady, so shared, not local. - - function streamingSupported() { - return typeof ReadableStream !== 'undefined' - && typeof fetch === 'function' - && typeof AudioWorkletNode !== 'undefined'; - } - function isWavResponse(resp) { - try { - const ct = ((resp && resp.headers && resp.headers.get('content-type')) || '').toLowerCase(); - return ct.indexOf('wav') !== -1; - } catch (_) { return false; } - } - - // --- parseWavHeader (pure; node-testable, see tests/wav-pcm.test.mjs) --- - // Parse a RIFF/WAV header from the leading bytes; returns - // { nch, sampleRate, bitsPerSample, dataOffset, dataSize } or null. Only - // 16-bit PCM is accepted (what the proxy emits). Chunk-walks fmt/data so a - // non-canonical header (extra chunks before `data`) still parses. - // parseWavHeader + pcm16ToFloat32 → src/wav-pcm.js (imported above). - - // Pull one chunk from a track's reader into its leftover byte buffer. - async function trackRead(t) { - if (t.done) return false; - const { done, value } = await t.reader.read(); - if (done) { t.done = true; return false; } - if (value && value.length) { - if (t.leftover.length === 0) { - t.leftover = value; - } else { - const merged = new Uint8Array(t.leftover.length + value.length); - merged.set(t.leftover, 0); - merged.set(value, t.leftover.length); - t.leftover = merged; - } - } - return true; - } - async function ensureBytes(t, n) { - while (t.leftover.length < n && !t.done) await trackRead(t); - return t.leftover.length >= n; - } - async function dropBytes(t, n) { - while (n > 0) { - if (t.leftover.length === 0) { if (!(await trackRead(t))) break; } - if (t.leftover.length === 0) break; - const take = Math.min(n, t.leftover.length); - t.leftover = t.leftover.subarray(take); - n -= take; - } - } - - // Read the WAV header off a freshly opened track reader, parse it, and trim - // `leftover` to the start of the PCM body. Returns the header or null. - async function readWavHeader(t) { - let header = parseWavHeader(t.leftover); - while (!header && !t.done && t.leftover.length < (1 << 16)) { - await trackRead(t); - header = parseWavHeader(t.leftover); - } - if (!header) return null; - t.leftover = t.leftover.subarray(header.dataOffset); - return header; - } - - // Dequeue `frames` per-channel samples of real PCM from a track (reading - // from its reader as needed), returning one Float32Array(frames) per channel. - async function dequeueTrackFrames(t, frames) { - if (t.skipBytes > 0) { await dropBytes(t, t.skipBytes); t.skipBytes = 0; } - const wantBytes = frames * t.byteAlign; - await ensureBytes(t, wantBytes); - const availBytes = Math.min(wantBytes, t.leftover.length); - const availFrames = Math.floor(availBytes / t.byteAlign); - const chunk = t.leftover.subarray(0, availFrames * t.byteAlign); - t.leftover = t.leftover.subarray(availFrames * t.byteAlign); - return pcm16ToFloat32(chunk, 0, frames, t.nch); // pads past availFrames with 0 - } - - // Append one aligned block for every track at the current write frontier, - // reading PCM as needed and zero-padding tracks past their own end. - async function appendRound(token = ST.streamSeekToken) { - const remaining = ST.streamTotalSamples - ST.jsWriteFrontier; - if (remaining <= 0) return false; - const aheadTarget = Math.min(ST.streamTotalSamples, - ST.lastWorkletPos + Math.ceil(STREAM_AHEAD_SEC * ST.streamSampleRate)); - const frames = Math.min(STREAM_CHUNK_FRAMES, remaining, Math.max(0, aheadTarget - ST.jsWriteFrontier)); - if (frames <= 0) return false; - - const blocks = []; - const transfer = []; - for (const t of ST.streamTracks) { - // Bail before touching each track once a seek supersedes us: dequeue - // reads t.reader fresh, and repositionStream reopens readers on these - // same track objects — consuming a later track here would steal PCM - // from the new pump. The in-flight track (if any) resolves off its own - // now-cancelled reader, so it can't reach the fresh ones. - if (token !== ST.streamSeekToken) return false; - const realWanted = Math.max(0, Math.min(frames, t.totalFrames - ST.jsWriteFrontier)); - let chans; - if (realWanted <= 0) { - chans = []; - for (let ch = 0; ch < t.nch; ch++) chans.push(new Float32Array(frames)); - } else { - // dequeueTrackFrames pads to `realWanted`; pad the rest to `frames`. - chans = await dequeueTrackFrames(t, realWanted); - if (realWanted < frames) { - for (let ch = 0; ch < t.nch; ch++) { - const full = new Float32Array(frames); - full.set(chans[ch], 0); - chans[ch] = full; - } - } - } - for (const c of chans) transfer.push(c.buffer); - blocks.push({ channels: chans }); - } - // A seek (repositionStream) bumps streamSeekToken and resets jsWriteFrontier. - // If it landed while we were awaiting dequeueTrackFrames above, these blocks - // are misaligned to the old frontier — drop them. Checking pumpStop alone is - // not enough: repositionStream clears it again before this continuation runs. - if (ST.pumpStop || token !== ST.streamSeekToken || !S.workletNode) return false; - try { - S.workletNode.port.postMessage({ type: 'append', base: ST.jsWriteFrontier, frames, tracks: blocks }, transfer); - } catch (_) { return false; } - ST.jsWriteFrontier += frames; - return true; - } - - // Await the worklet's next backpressure ('pos') message, or a short timeout. - function waitPos() { - return new Promise((resolve) => { - ST.posWaiter = resolve; - setTimeout(() => { if (ST.posWaiter === resolve) { ST.posWaiter = null; resolve(); } }, 100); - }); - } - - // The pump: keep the worklet window ~STREAM_AHEAD_SEC ahead of its read - // frontier. On the initial run, prefill then start (honouring pending play). - async function runPump(isInitial) { - // Capture the seek token for this pump run: a seek supersedes us by bumping - // it, so every loop guard + append re-checks it after awaits (see appendRound). - const token = ST.streamSeekToken; - try { - const prefillTo = Math.min(ST.streamTotalSamples, - ST.jsWriteFrontier + Math.ceil(STREAM_PREFILL_SEC * ST.streamSampleRate)); - while (!ST.pumpStop && token === ST.streamSeekToken && ST.jsWriteFrontier < prefillTo) { - if (!(await appendRound(token))) break; - } - if (ST.pumpStop || token !== ST.streamSeekToken) return; - if (isInitial) { - S.buffersReady = true; - if (S.pendingPlay) { S.pendingPlay = false; transportPlay(); } - } - while (!ST.pumpStop && token === ST.streamSeekToken && ST.jsWriteFrontier < ST.streamTotalSamples) { - const target = Math.min(ST.streamTotalSamples, - ST.lastWorkletPos + Math.ceil(STREAM_AHEAD_SEC * ST.streamSampleRate)); - if (ST.jsWriteFrontier >= target) { await waitPos(); continue; } - if (!(await appendRound(token))) break; - } - } catch (e) { - if (!ST.pumpStop && (!e || e.name !== 'AbortError')) console.warn('[stems] stream pump error:', e); - } - } - - function cancelStreamReaders() { - for (const t of ST.streamTracks) { - try { t.reader && t.reader.cancel(); } catch (_) {} - t.reader = null; - t.leftover = EMPTY_BYTES; - t.done = true; - } - } - - // (Re)open every track's byte stream starting at `fromSample`. Uses a Range - // request: a 206-capable proxy serves pure PCM from there (efficient seek); - // a proxy that ignores Range returns 200 from 0, so we skip the header + - // preceding samples ourselves. - async function openTrackStreams(fromSample, gen, token) { - await Promise.all(ST.streamTracks.map(async (t) => { - // Byte offset of sample `fromSample` in this track's WAV (header + - // linear PCM). dataOffset is the parsed header size (44 for the - // proxy's canonical WAV, but honour a non-canonical one too). - const byteOffset = t.dataOffset + fromSample * t.byteAlign; - const headers = { Range: 'bytes=' + byteOffset + '-' }; - const resp = await fetch(t.url, { signal: S.abortController.signal, headers }); - if (gen !== S.loadGeneration || token !== ST.streamSeekToken) { - try { resp.body && resp.body.cancel(); } catch (_) {} - return; - } - t.reader = resp.body.getReader(); - t.leftover = EMPTY_BYTES; - t.done = false; - // 206 → body already starts at fromSample (no header). 200 → whole - // file from 0; skip the header + the preceding PCM ourselves. - t.skipBytes = (resp.status === 206) ? 0 : byteOffset; - })); - } - - // Streaming seek: flush the worklet window, refetch every track from the - // target, and resume the pump. Sample-accurate; O(1) network with a - // 206-capable proxy, O(offset) discard otherwise. - async function repositionStream(targetSec) { - const token = ++ST.streamSeekToken; - const gen = S.loadGeneration; - ST.pumpStop = true; - cancelStreamReaders(); - // Unblock an old pump parked in waitPos() so it re-checks its (now stale) - // token and exits promptly, instead of hanging until the 100ms timeout. - if (ST.posWaiter) { const r = ST.posWaiter; ST.posWaiter = null; try { r(); } catch (_) {} } - const Tsamp = Math.max(0, Math.round(targetSec * ST.streamSampleRate)); - if (S.workletNode) { - try { S.workletNode.port.postMessage({ type: 'seek', offset: Tsamp / ST.streamSampleRate }); } catch (_) {} - } - ST.jsWriteFrontier = Tsamp; - ST.lastWorkletPos = Tsamp; - try { - await openTrackStreams(Tsamp, gen, token); - } catch (e) { - if (token === ST.streamSeekToken && (!e || e.name !== 'AbortError')) { - console.warn('[stems] seek refetch failed:', e); - } - return; - } - if (token !== ST.streamSeekToken || gen !== S.loadGeneration) return; - ST.pumpStop = false; - // If a seek supersedes the initial prefill, runPump(true) returns before - // flipping S.buffersReady / honouring pendingPlay. Carry the initial role - // forward until readiness is actually established, else the stream would - // stay silent (play() only sets pendingPlay) until a reload. - runPump(!S.buffersReady); - } - - // Build the streaming graph + pump from the fetched WAV streams. `probeResp` - // is stem[0]'s already-open response. Returns true once set up (the pump - // runs asynchronously), false on failure. On failure it does NOT teardown() - // (that would bump S.loadGeneration and hide the failure from onSongReady's - // supersession check) — the caller tears down + falls back. A `false` return - // with `gen === S.loadGeneration` is a real failure; a stale gen is supersession. - async function setupStreaming(stems, probeResp, fullUrl, gen) { - // 1. Open readers + parse headers for every stem (and the full mix). - let restResps = []; - try { - restResps = await Promise.all(stems.slice(1).map((s) => - fetch(s.url, { signal: S.abortController.signal, headers: { Range: 'bytes=0-' } }))); - } catch (e) { - if (gen !== S.loadGeneration) return false; - console.warn('[stems] stem fetch failed; cannot stream:', e); - return false; // caller (onSongReady) tears down + falls back - } - if (gen !== S.loadGeneration) return false; - - let fullResp = null; - if (fullUrl) { - try { - const fr = await fetch(fullUrl, { signal: S.abortController.signal, headers: { Range: 'bytes=0-' } }); - if (gen !== S.loadGeneration) { try { fr.body && fr.body.cancel(); } catch (_) {} return false; } - if (isWavResponse(fr)) fullResp = fr; else { try { fr.body && fr.body.cancel(); } catch (_) {} } - } catch (_) { /* no full mix → separated stems only */ } - } - - const responses = [probeResp].concat(restResps); - const built = []; - for (let i = 0; i < stems.length; i++) { - const resp = responses[i]; - if (!resp || !resp.body) return false; - const t = { - id: stems[i].id, url: stems[i].url, default: !!stems[i].default, - reader: resp.body.getReader(), leftover: EMPTY_BYTES, done: false, skipBytes: 0, - nch: 2, byteAlign: 4, totalFrames: 0, dataOffset: 44, - }; - const hdr = await readWavHeader(t); - if (gen !== S.loadGeneration) { try { t.reader.cancel(); } catch (_) {} return false; } - if (!hdr) { console.warn('[stems] stem WAV header parse failed; cannot stream'); return false; } - t.nch = hdr.nch; t.byteAlign = hdr.nch * 2; t.dataOffset = hdr.dataOffset; - t.totalFrames = Math.floor(hdr.dataSize / t.byteAlign); - // All stems must share one sample rate — the context is pinned to it - // and the worklet indexes every track by the same sample clock. A - // mixed-rate pack would play some stems at the wrong speed, so refuse - // to stream (the caller falls back). Demucs output is homogeneous; - // this is a defensive guard. - if (!ST.streamSampleRate) { - ST.streamSampleRate = hdr.sampleRate; - } else if (hdr.sampleRate !== ST.streamSampleRate) { - console.warn('[stems] stem "' + t.id + '" rate ' + hdr.sampleRate - + ' != ' + ST.streamSampleRate + '; cannot stream a mixed-rate pack'); - return false; - } - built.push(t); - } - - let fullTrack = null; - if (fullResp && fullResp.body) { - const t = { - id: '__full', url: fullUrl, - reader: fullResp.body.getReader(), leftover: EMPTY_BYTES, done: false, skipBytes: 0, - nch: 2, byteAlign: 4, totalFrames: 0, dataOffset: 44, - }; - const hdr = await readWavHeader(t); - if (gen !== S.loadGeneration) { try { t.reader.cancel(); } catch (_) {} return false; } - // The full mix rides the same sample clock as the stems; the pristine - // mixdown can be encoded at the source rate (≠ the demucs stem rate), - // in which case it'd play at the wrong speed. Only keep it when its - // rate matches; otherwise drop it and play the separated stems. - if (hdr && hdr.sampleRate === ST.streamSampleRate) { - t.nch = hdr.nch; t.byteAlign = hdr.nch * 2; t.dataOffset = hdr.dataOffset; - t.totalFrames = Math.floor(hdr.dataSize / t.byteAlign); - fullTrack = t; - } else { - if (hdr) console.warn('[stems] original_audio rate ' - + hdr.sampleRate + ' != ' + ST.streamSampleRate + '; using separated stems only'); - try { t.reader.cancel(); } catch (_) {} - } - } - - const maxStemFrames = built.reduce((m, t) => Math.max(m, t.totalFrames), 0); - if (maxStemFrames <= 0) return false; - ST.streamTotalSamples = maxStemFrames; - - // 2. Pin the AudioContext to the source rate + (re)load the worklet. - const okCtx = await ensureCtxAtRate(ST.streamSampleRate); - if (gen !== S.loadGeneration) return false; - if (!okCtx || !S.audioCtx) { - console.warn('[stems] could not run the AudioContext at the stem sample rate; not streaming'); - return false; - } - S.useWorklet = true; - - // 3. Full-mix tolerance (same rule as buildGraphFromBuffers): only keep - // it when its length matches the stems, clamped so it can't extend the - // song past where the stems / highway end. - S.fullTrackIndex = -1; - ST.streamTracks = built.slice(); - if (fullTrack) { - const tol = Math.max(2048, Math.round(0.05 * ST.streamSampleRate)); - if (Math.abs(fullTrack.totalFrames - maxStemFrames) > tol) { - console.warn('[stems] original_audio length off by ' - + (fullTrack.totalFrames - maxStemFrames) + ' frames; using separated stems only.'); - try { fullTrack.reader.cancel(); } catch (_) {} - } else { - fullTrack.totalFrames = Math.min(fullTrack.totalFrames, maxStemFrames); - ST.streamTracks.push(fullTrack); - S.fullTrackIndex = built.length; - } - } - - // 4. Mix graph + stem UI state (mirrors buildGraphFromBuffers, no buffers). - S.masterGain = S.audioCtx.createGain(); - S.masterGain.gain.value = persistedSongGain(); - S.masterGain.connect(S.audioCtx.destination); - S.analyserNode = S.audioCtx.createAnalyser(); - S.analyserNode.fftSize = 256; - S.masterGain.connect(S.analyserNode); - S.workletPostReady = false; - try { - S.workletNode = new AudioWorkletNode(S.audioCtx, 'stem-mixer', { - numberOfInputs: 0, numberOfOutputs: 1, outputChannelCount: [2], - }); - S.workletNode.port.onmessage = onWorkletMessage; - S.workletNode.connect(S.masterGain); - } catch (e) { - console.warn('[stems] stem-mixer node failed; not streaming:', e); - return false; - } - - const karaoke = karaokeDefault(); - const defaultMuted = loadDefaultMuted(); - const savedMuted = loadMuted(S.currentFilename); - const savedVols = loadVolumes(S.currentFilename); - S.stemState = built.map((t, i) => { - const gain = makeStemGainHandle(i); - let on; - if (savedMuted) { - on = !savedMuted.has(t.id); - } else { - on = !!t.default; - if (defaultMuted.has(t.id)) on = false; - if (karaoke && /vocal/i.test(t.id)) on = false; - } - const vol = clampVolume(savedVols[t.id]); - const initialVol = vol == null ? 1 : vol; - gain.gain.value = on ? initialVol : 0; - return { id: t.id, url: t.url, default: t.default, buffer: null, source: null, gain, on, vol: initialVol }; - }); - - // 5. Transport + initial gains, then open the worklet window. - // Preserve any playhead seeded before/while setting up — the core-playback - // takeover (core was already playing stem[0]) or a seek during setup left - // it on transport.baseOffset. Start the window there instead of 0; the - // tracks' readers begin at PCM sample 0, so skip forward to the offset. - transport.duration = ST.streamTotalSamples / ST.streamSampleRate; - const initialOffsetSamples = Math.max(0, Math.min( - Math.round((transport.baseOffset || 0) * ST.streamSampleRate), ST.streamTotalSamples)); - transport.baseOffset = initialOffsetSamples / ST.streamSampleRate; - transport.baseCtxTime = 0; - transport.playing = false; - const core = document.getElementById('audio'); - const coreRate = core ? Number(core.playbackRate) : 1; - transport.rate = (Number.isFinite(coreRate) && coreRate > 0) ? coreRate : 1; - ST.jsWriteFrontier = initialOffsetSamples; - ST.lastWorkletPos = initialOffsetSamples; - if (initialOffsetSamples > 0) { - for (const t of ST.streamTracks) t.skipBytes = initialOffsetSamples * t.byteAlign; - } - - const { stemGains, fullGain } = computeMixGains(S.stemState, S.fullTrackIndex >= 0); - const gains = stemGains.slice(); - if (S.fullTrackIndex >= 0) gains.push(fullGain); - const openTracks = ST.streamTracks.map((t) => ({ nch: t.nch, length: t.totalFrames })); - try { - S.workletNode.port.postMessage({ - type: 'open', tracks: openTracks, gains, - sampleRate: ST.streamSampleRate, cap: Math.ceil(STREAM_CAP_SEC * ST.streamSampleRate), - startSample: initialOffsetSamples, - }); - } catch (e) { - console.warn('[stems] worklet open failed; not streaming:', e); - return false; - } - S.workletPostReady = true; - updateLatencyOffset(); - - ST.pumpStop = false; // teardown() set this true; clear it before pumping - ST.streaming = true; - runPump(true); // async — prefills, then starts on pending play - return true; - } - - function resetStreamState() { - ST.pumpStop = true; - ST.streamSeekToken++; - cancelStreamReaders(); - ST.streamTracks = []; - ST.streaming = false; - ST.streamSampleRate = 0; - ST.streamTotalSamples = 0; - ST.jsWriteFrontier = 0; - ST.lastWorkletPos = 0; - if (ST.posWaiter) { const r = ST.posWaiter; ST.posWaiter = null; try { r(); } catch (_) {} } - } - // ── Song-fader bridge ── // audio-mixer.js's "Song" fader probes window.slopsmith.stems.setMasterVolume // and routes itself there when present, so the fader can drive every stem diff --git a/src/streaming.js b/src/streaming.js new file mode 100644 index 0000000..1d1afda --- /dev/null +++ b/src/streaming.js @@ -0,0 +1,480 @@ +// ══════════════════════════════════════════════════════════════════════ +// Streaming (bounded-memory) playback — the iOS WAV path +// +// The iOS client proxy transcodes each OGG stem to RIFF/WAV (16-bit PCM) +// and streams it. Decoding whole stems to AudioBuffers (buildGraphFromBuffers +// in main.js) jettisons the WKWebView content process at ~6 stems (~500 MB). +// Here we read the PCM incrementally off fetch().body and feed the worklet's +// bounded ring via 'append', dropping consumed PCM, so peak memory is a +// few-second window per track — independent of song length or stem count. +// Raw PCM is sliceable at any sample, so NO decoder/WebCodecs/demuxer is +// needed. Desktop (audio/ogg, not sliceable) keeps the full-decode path. +// ══════════════════════════════════════════════════════════════════════ +import { S, ST, transport } from './state.js'; +import { ensureCtxAtRate, updateLatencyOffset } from './audio-ctx.js'; +import { makeStemGainHandle } from './mix.js'; +import { computeMixGains } from './mix-gains.js'; +import { clampVolume } from './util.js'; +import { parseWavHeader, pcm16ToFloat32 } from './wav-pcm.js'; +import { karaokeDefault, loadDefaultMuted, loadMuted, loadVolumes } from './prefs.js'; + +// Injected seams to main.js's transport/orchestration layer. Injected (not +// imported) to avoid a static import cycle: transport already imports +// repositionStream/setupStreaming from here, so this layer can't import back. +let startPendingPlay = () => {}; // resume a deferred play() once buffers are ready +let songGain = () => 0.8; // seed the master gain from the mixer / persisted volume +let onWorkletMessage = () => {}; // the shared worklet 'ready'/'ended'/'pos' handler +export function configureStreaming(hooks) { + if (hooks.startPendingPlay) startPendingPlay = hooks.startPendingPlay; + if (hooks.songGain) songGain = hooks.songGain; + if (hooks.onWorkletMessage) onWorkletMessage = hooks.onWorkletMessage; +} + +const STREAM_AHEAD_SEC = 2.0; // keep ~this far buffered ahead of pos +const STREAM_PREFILL_SEC = 0.5; // buffer this much before starting +const STREAM_CAP_SEC = 3.5; // worklet per-track window capacity +const STREAM_CHUNK_FRAMES = 8192; // max frames appended per pump round +const EMPTY_BYTES = new Uint8Array(0); + +export function streamingSupported() { + return typeof ReadableStream !== 'undefined' + && typeof fetch === 'function' + && typeof AudioWorkletNode !== 'undefined'; +} +export function isWavResponse(resp) { + try { + const ct = ((resp && resp.headers && resp.headers.get('content-type')) || '').toLowerCase(); + return ct.indexOf('wav') !== -1; + } catch (_) { return false; } +} + +// True when `offset` (s) falls inside the PCM the streaming worklet currently +// holds around its read frontier. Outside it — e.g. replaying from 0 after the +// song ended and the pump drained the window to the tail — the worklet would +// read silence, so the caller must refetch (repositionStream) first. +export function streamOffsetBuffered(offset) { + if (!ST.streaming || ST.streamSampleRate <= 0) return true; + const posSamp = Math.round(offset * ST.streamSampleRate); + const behind = Math.ceil(0.1 * ST.streamSampleRate); + return posSamp >= ST.lastWorkletPos - behind && posSamp <= ST.jsWriteFrontier; +} + +// Pull one chunk from a track's reader into its leftover byte buffer. +async function trackRead(t) { + if (t.done) return false; + const { done, value } = await t.reader.read(); + if (done) { t.done = true; return false; } + if (value && value.length) { + if (t.leftover.length === 0) { + t.leftover = value; + } else { + const merged = new Uint8Array(t.leftover.length + value.length); + merged.set(t.leftover, 0); + merged.set(value, t.leftover.length); + t.leftover = merged; + } + } + return true; +} +async function ensureBytes(t, n) { + while (t.leftover.length < n && !t.done) await trackRead(t); + return t.leftover.length >= n; +} +async function dropBytes(t, n) { + while (n > 0) { + if (t.leftover.length === 0) { if (!(await trackRead(t))) break; } + if (t.leftover.length === 0) break; + const take = Math.min(n, t.leftover.length); + t.leftover = t.leftover.subarray(take); + n -= take; + } +} + +// Read the WAV header off a freshly opened track reader, parse it, and trim +// `leftover` to the start of the PCM body. Returns the header or null. +// (parseWavHeader + pcm16ToFloat32 are the pure decoders in wav-pcm.js.) +async function readWavHeader(t) { + let header = parseWavHeader(t.leftover); + while (!header && !t.done && t.leftover.length < (1 << 16)) { + await trackRead(t); + header = parseWavHeader(t.leftover); + } + if (!header) return null; + t.leftover = t.leftover.subarray(header.dataOffset); + return header; +} + +// Dequeue `frames` per-channel samples of real PCM from a track (reading +// from its reader as needed), returning one Float32Array(frames) per channel. +async function dequeueTrackFrames(t, frames) { + if (t.skipBytes > 0) { await dropBytes(t, t.skipBytes); t.skipBytes = 0; } + const wantBytes = frames * t.byteAlign; + await ensureBytes(t, wantBytes); + const availBytes = Math.min(wantBytes, t.leftover.length); + const availFrames = Math.floor(availBytes / t.byteAlign); + const chunk = t.leftover.subarray(0, availFrames * t.byteAlign); + t.leftover = t.leftover.subarray(availFrames * t.byteAlign); + return pcm16ToFloat32(chunk, 0, frames, t.nch); // pads past availFrames with 0 +} + +// Append one aligned block for every track at the current write frontier, +// reading PCM as needed and zero-padding tracks past their own end. +async function appendRound(token = ST.streamSeekToken) { + const remaining = ST.streamTotalSamples - ST.jsWriteFrontier; + if (remaining <= 0) return false; + const aheadTarget = Math.min(ST.streamTotalSamples, + ST.lastWorkletPos + Math.ceil(STREAM_AHEAD_SEC * ST.streamSampleRate)); + const frames = Math.min(STREAM_CHUNK_FRAMES, remaining, Math.max(0, aheadTarget - ST.jsWriteFrontier)); + if (frames <= 0) return false; + + const blocks = []; + const transfer = []; + for (const t of ST.streamTracks) { + // Bail before touching each track once a seek supersedes us: dequeue + // reads t.reader fresh, and repositionStream reopens readers on these + // same track objects — consuming a later track here would steal PCM + // from the new pump. The in-flight track (if any) resolves off its own + // now-cancelled reader, so it can't reach the fresh ones. + if (token !== ST.streamSeekToken) return false; + const realWanted = Math.max(0, Math.min(frames, t.totalFrames - ST.jsWriteFrontier)); + let chans; + if (realWanted <= 0) { + chans = []; + for (let ch = 0; ch < t.nch; ch++) chans.push(new Float32Array(frames)); + } else { + // dequeueTrackFrames pads to `realWanted`; pad the rest to `frames`. + chans = await dequeueTrackFrames(t, realWanted); + if (realWanted < frames) { + for (let ch = 0; ch < t.nch; ch++) { + const full = new Float32Array(frames); + full.set(chans[ch], 0); + chans[ch] = full; + } + } + } + for (const c of chans) transfer.push(c.buffer); + blocks.push({ channels: chans }); + } + // A seek (repositionStream) bumps streamSeekToken and resets jsWriteFrontier. + // If it landed while we were awaiting dequeueTrackFrames above, these blocks + // are misaligned to the old frontier — drop them. Checking pumpStop alone is + // not enough: repositionStream clears it again before this continuation runs. + if (ST.pumpStop || token !== ST.streamSeekToken || !S.workletNode) return false; + try { + S.workletNode.port.postMessage({ type: 'append', base: ST.jsWriteFrontier, frames, tracks: blocks }, transfer); + } catch (_) { return false; } + ST.jsWriteFrontier += frames; + return true; +} + +// Await the worklet's next backpressure ('pos') message, or a short timeout. +function waitPos() { + return new Promise((resolve) => { + ST.posWaiter = resolve; + setTimeout(() => { if (ST.posWaiter === resolve) { ST.posWaiter = null; resolve(); } }, 100); + }); +} + +// The pump: keep the worklet window ~STREAM_AHEAD_SEC ahead of its read +// frontier. On the initial run, prefill then start (honouring pending play). +async function runPump(isInitial) { + // Capture the seek token for this pump run: a seek supersedes us by bumping + // it, so every loop guard + append re-checks it after awaits (see appendRound). + const token = ST.streamSeekToken; + try { + const prefillTo = Math.min(ST.streamTotalSamples, + ST.jsWriteFrontier + Math.ceil(STREAM_PREFILL_SEC * ST.streamSampleRate)); + while (!ST.pumpStop && token === ST.streamSeekToken && ST.jsWriteFrontier < prefillTo) { + if (!(await appendRound(token))) break; + } + if (ST.pumpStop || token !== ST.streamSeekToken) return; + if (isInitial) { + S.buffersReady = true; + if (S.pendingPlay) { S.pendingPlay = false; startPendingPlay(); } + } + while (!ST.pumpStop && token === ST.streamSeekToken && ST.jsWriteFrontier < ST.streamTotalSamples) { + const target = Math.min(ST.streamTotalSamples, + ST.lastWorkletPos + Math.ceil(STREAM_AHEAD_SEC * ST.streamSampleRate)); + if (ST.jsWriteFrontier >= target) { await waitPos(); continue; } + if (!(await appendRound(token))) break; + } + } catch (e) { + if (!ST.pumpStop && (!e || e.name !== 'AbortError')) console.warn('[stems] stream pump error:', e); + } +} + +function cancelStreamReaders() { + for (const t of ST.streamTracks) { + try { t.reader && t.reader.cancel(); } catch (_) {} + t.reader = null; + t.leftover = EMPTY_BYTES; + t.done = true; + } +} + +// (Re)open every track's byte stream starting at `fromSample`. Uses a Range +// request: a 206-capable proxy serves pure PCM from there (efficient seek); +// a proxy that ignores Range returns 200 from 0, so we skip the header + +// preceding samples ourselves. +async function openTrackStreams(fromSample, gen, token) { + await Promise.all(ST.streamTracks.map(async (t) => { + // Byte offset of sample `fromSample` in this track's WAV (header + + // linear PCM). dataOffset is the parsed header size (44 for the + // proxy's canonical WAV, but honour a non-canonical one too). + const byteOffset = t.dataOffset + fromSample * t.byteAlign; + const headers = { Range: 'bytes=' + byteOffset + '-' }; + const resp = await fetch(t.url, { signal: S.abortController.signal, headers }); + if (gen !== S.loadGeneration || token !== ST.streamSeekToken) { + try { resp.body && resp.body.cancel(); } catch (_) {} + return; + } + t.reader = resp.body.getReader(); + t.leftover = EMPTY_BYTES; + t.done = false; + // 206 → body already starts at fromSample (no header). 200 → whole + // file from 0; skip the header + the preceding PCM ourselves. + t.skipBytes = (resp.status === 206) ? 0 : byteOffset; + })); +} + +// Streaming seek: flush the worklet window, refetch every track from the +// target, and resume the pump. Sample-accurate; O(1) network with a +// 206-capable proxy, O(offset) discard otherwise. +export async function repositionStream(targetSec) { + const token = ++ST.streamSeekToken; + const gen = S.loadGeneration; + ST.pumpStop = true; + cancelStreamReaders(); + // Unblock an old pump parked in waitPos() so it re-checks its (now stale) + // token and exits promptly, instead of hanging until the 100ms timeout. + if (ST.posWaiter) { const r = ST.posWaiter; ST.posWaiter = null; try { r(); } catch (_) {} } + const Tsamp = Math.max(0, Math.round(targetSec * ST.streamSampleRate)); + if (S.workletNode) { + try { S.workletNode.port.postMessage({ type: 'seek', offset: Tsamp / ST.streamSampleRate }); } catch (_) {} + } + ST.jsWriteFrontier = Tsamp; + ST.lastWorkletPos = Tsamp; + try { + await openTrackStreams(Tsamp, gen, token); + } catch (e) { + if (token === ST.streamSeekToken && (!e || e.name !== 'AbortError')) { + console.warn('[stems] seek refetch failed:', e); + } + return; + } + if (token !== ST.streamSeekToken || gen !== S.loadGeneration) return; + ST.pumpStop = false; + // If a seek supersedes the initial prefill, runPump(true) returns before + // flipping S.buffersReady / honouring pendingPlay. Carry the initial role + // forward until readiness is actually established, else the stream would + // stay silent (play() only sets pendingPlay) until a reload. + runPump(!S.buffersReady); +} + +// Build the streaming graph + pump from the fetched WAV streams. `probeResp` +// is stem[0]'s already-open response. Returns true once set up (the pump +// runs asynchronously), false on failure. On failure it does NOT teardown() +// (that would bump S.loadGeneration and hide the failure from onSongReady's +// supersession check) — the caller tears down + falls back. A `false` return +// with `gen === S.loadGeneration` is a real failure; a stale gen is supersession. +export async function setupStreaming(stems, probeResp, fullUrl, gen) { + // 1. Open readers + parse headers for every stem (and the full mix). + let restResps = []; + try { + restResps = await Promise.all(stems.slice(1).map((s) => + fetch(s.url, { signal: S.abortController.signal, headers: { Range: 'bytes=0-' } }))); + } catch (e) { + if (gen !== S.loadGeneration) return false; + console.warn('[stems] stem fetch failed; cannot stream:', e); + return false; // caller (onSongReady) tears down + falls back + } + if (gen !== S.loadGeneration) return false; + + let fullResp = null; + if (fullUrl) { + try { + const fr = await fetch(fullUrl, { signal: S.abortController.signal, headers: { Range: 'bytes=0-' } }); + if (gen !== S.loadGeneration) { try { fr.body && fr.body.cancel(); } catch (_) {} return false; } + if (isWavResponse(fr)) fullResp = fr; else { try { fr.body && fr.body.cancel(); } catch (_) {} } + } catch (_) { /* no full mix → separated stems only */ } + } + + const responses = [probeResp].concat(restResps); + const built = []; + for (let i = 0; i < stems.length; i++) { + const resp = responses[i]; + if (!resp || !resp.body) return false; + const t = { + id: stems[i].id, url: stems[i].url, default: !!stems[i].default, + reader: resp.body.getReader(), leftover: EMPTY_BYTES, done: false, skipBytes: 0, + nch: 2, byteAlign: 4, totalFrames: 0, dataOffset: 44, + }; + const hdr = await readWavHeader(t); + if (gen !== S.loadGeneration) { try { t.reader.cancel(); } catch (_) {} return false; } + if (!hdr) { console.warn('[stems] stem WAV header parse failed; cannot stream'); return false; } + t.nch = hdr.nch; t.byteAlign = hdr.nch * 2; t.dataOffset = hdr.dataOffset; + t.totalFrames = Math.floor(hdr.dataSize / t.byteAlign); + // All stems must share one sample rate — the context is pinned to it + // and the worklet indexes every track by the same sample clock. A + // mixed-rate pack would play some stems at the wrong speed, so refuse + // to stream (the caller falls back). Demucs output is homogeneous; + // this is a defensive guard. + if (!ST.streamSampleRate) { + ST.streamSampleRate = hdr.sampleRate; + } else if (hdr.sampleRate !== ST.streamSampleRate) { + console.warn('[stems] stem "' + t.id + '" rate ' + hdr.sampleRate + + ' != ' + ST.streamSampleRate + '; cannot stream a mixed-rate pack'); + return false; + } + built.push(t); + } + + let fullTrack = null; + if (fullResp && fullResp.body) { + const t = { + id: '__full', url: fullUrl, + reader: fullResp.body.getReader(), leftover: EMPTY_BYTES, done: false, skipBytes: 0, + nch: 2, byteAlign: 4, totalFrames: 0, dataOffset: 44, + }; + const hdr = await readWavHeader(t); + if (gen !== S.loadGeneration) { try { t.reader.cancel(); } catch (_) {} return false; } + // The full mix rides the same sample clock as the stems; the pristine + // mixdown can be encoded at the source rate (≠ the demucs stem rate), + // in which case it'd play at the wrong speed. Only keep it when its + // rate matches; otherwise drop it and play the separated stems. + if (hdr && hdr.sampleRate === ST.streamSampleRate) { + t.nch = hdr.nch; t.byteAlign = hdr.nch * 2; t.dataOffset = hdr.dataOffset; + t.totalFrames = Math.floor(hdr.dataSize / t.byteAlign); + fullTrack = t; + } else { + if (hdr) console.warn('[stems] original_audio rate ' + + hdr.sampleRate + ' != ' + ST.streamSampleRate + '; using separated stems only'); + try { t.reader.cancel(); } catch (_) {} + } + } + + const maxStemFrames = built.reduce((m, t) => Math.max(m, t.totalFrames), 0); + if (maxStemFrames <= 0) return false; + ST.streamTotalSamples = maxStemFrames; + + // 2. Pin the AudioContext to the source rate + (re)load the worklet. + const okCtx = await ensureCtxAtRate(ST.streamSampleRate); + if (gen !== S.loadGeneration) return false; + if (!okCtx || !S.audioCtx) { + console.warn('[stems] could not run the AudioContext at the stem sample rate; not streaming'); + return false; + } + S.useWorklet = true; + + // 3. Full-mix tolerance (same rule as buildGraphFromBuffers): only keep + // it when its length matches the stems, clamped so it can't extend the + // song past where the stems / highway end. + S.fullTrackIndex = -1; + ST.streamTracks = built.slice(); + if (fullTrack) { + const tol = Math.max(2048, Math.round(0.05 * ST.streamSampleRate)); + if (Math.abs(fullTrack.totalFrames - maxStemFrames) > tol) { + console.warn('[stems] original_audio length off by ' + + (fullTrack.totalFrames - maxStemFrames) + ' frames; using separated stems only.'); + try { fullTrack.reader.cancel(); } catch (_) {} + } else { + fullTrack.totalFrames = Math.min(fullTrack.totalFrames, maxStemFrames); + ST.streamTracks.push(fullTrack); + S.fullTrackIndex = built.length; + } + } + + // 4. Mix graph + stem UI state (mirrors buildGraphFromBuffers, no buffers). + S.masterGain = S.audioCtx.createGain(); + S.masterGain.gain.value = songGain(); + S.masterGain.connect(S.audioCtx.destination); + S.analyserNode = S.audioCtx.createAnalyser(); + S.analyserNode.fftSize = 256; + S.masterGain.connect(S.analyserNode); + S.workletPostReady = false; + try { + S.workletNode = new AudioWorkletNode(S.audioCtx, 'stem-mixer', { + numberOfInputs: 0, numberOfOutputs: 1, outputChannelCount: [2], + }); + S.workletNode.port.onmessage = onWorkletMessage; + S.workletNode.connect(S.masterGain); + } catch (e) { + console.warn('[stems] stem-mixer node failed; not streaming:', e); + return false; + } + + const karaoke = karaokeDefault(); + const defaultMuted = loadDefaultMuted(); + const savedMuted = loadMuted(S.currentFilename); + const savedVols = loadVolumes(S.currentFilename); + S.stemState = built.map((t, i) => { + const gain = makeStemGainHandle(i); + let on; + if (savedMuted) { + on = !savedMuted.has(t.id); + } else { + on = !!t.default; + if (defaultMuted.has(t.id)) on = false; + if (karaoke && /vocal/i.test(t.id)) on = false; + } + const vol = clampVolume(savedVols[t.id]); + const initialVol = vol == null ? 1 : vol; + gain.gain.value = on ? initialVol : 0; + return { id: t.id, url: t.url, default: t.default, buffer: null, source: null, gain, on, vol: initialVol }; + }); + + // 5. Transport + initial gains, then open the worklet window. + // Preserve any playhead seeded before/while setting up — the core-playback + // takeover (core was already playing stem[0]) or a seek during setup left + // it on transport.baseOffset. Start the window there instead of 0; the + // tracks' readers begin at PCM sample 0, so skip forward to the offset. + transport.duration = ST.streamTotalSamples / ST.streamSampleRate; + const initialOffsetSamples = Math.max(0, Math.min( + Math.round((transport.baseOffset || 0) * ST.streamSampleRate), ST.streamTotalSamples)); + transport.baseOffset = initialOffsetSamples / ST.streamSampleRate; + transport.baseCtxTime = 0; + transport.playing = false; + const core = document.getElementById('audio'); + const coreRate = core ? Number(core.playbackRate) : 1; + transport.rate = (Number.isFinite(coreRate) && coreRate > 0) ? coreRate : 1; + ST.jsWriteFrontier = initialOffsetSamples; + ST.lastWorkletPos = initialOffsetSamples; + if (initialOffsetSamples > 0) { + for (const t of ST.streamTracks) t.skipBytes = initialOffsetSamples * t.byteAlign; + } + + const { stemGains, fullGain } = computeMixGains(S.stemState, S.fullTrackIndex >= 0); + const gains = stemGains.slice(); + if (S.fullTrackIndex >= 0) gains.push(fullGain); + const openTracks = ST.streamTracks.map((t) => ({ nch: t.nch, length: t.totalFrames })); + try { + S.workletNode.port.postMessage({ + type: 'open', tracks: openTracks, gains, + sampleRate: ST.streamSampleRate, cap: Math.ceil(STREAM_CAP_SEC * ST.streamSampleRate), + startSample: initialOffsetSamples, + }); + } catch (e) { + console.warn('[stems] worklet open failed; not streaming:', e); + return false; + } + S.workletPostReady = true; + updateLatencyOffset(); + + ST.pumpStop = false; // teardown() set this true; clear it before pumping + ST.streaming = true; + runPump(true); // async — prefills, then starts on pending play + return true; +} + +export function resetStreamState() { + ST.pumpStop = true; + ST.streamSeekToken++; + cancelStreamReaders(); + ST.streamTracks = []; + ST.streaming = false; + ST.streamSampleRate = 0; + ST.streamTotalSamples = 0; + ST.jsWriteFrontier = 0; + ST.lastWorkletPos = 0; + if (ST.posWaiter) { const r = ST.posWaiter; ST.posWaiter = null; try { r(); } catch (_) {} } +} diff --git a/tests/streaming.test.mjs b/tests/streaming.test.mjs new file mode 100644 index 0000000..bc4e13b --- /dev/null +++ b/tests/streaming.test.mjs @@ -0,0 +1,59 @@ +// Unit tests for src/streaming.js's pure-ish exports. Real ES-module import. +// The pump/seek internals (appendRound/runPump/repositionStream) drive the Web +// Audio worklet + fetch and aren't exported, so their seek-token race guard is +// covered by the on-device seek-stress smoke, not here. +import test, { beforeEach } from 'node:test'; +import assert from 'node:assert/strict'; + +import { ST } from '../src/state.js'; +import { streamingSupported, isWavResponse, streamOffsetBuffered } from '../src/streaming.js'; + +test('isWavResponse: true only when content-type contains "wav"', () => { + const mk = (ct) => ({ headers: { get: () => ct } }); + assert.equal(isWavResponse(mk('audio/wav')), true); + assert.equal(isWavResponse(mk('audio/x-wav')), true); + assert.equal(isWavResponse(mk('AUDIO/WAV')), true); // case-insensitive + assert.equal(isWavResponse(mk('audio/ogg')), false); + assert.equal(isWavResponse(mk(null)), false); + assert.equal(isWavResponse(null), false); // no resp + assert.equal(isWavResponse({ headers: { get() { throw new Error('x'); } } }), false); +}); + +test('streamingSupported: true only when all three platform APIs exist', () => { + const orig = { R: globalThis.ReadableStream, f: globalThis.fetch, A: globalThis.AudioWorkletNode }; + try { + globalThis.ReadableStream = function () {}; + globalThis.fetch = function () {}; + globalThis.AudioWorkletNode = function () {}; + assert.equal(streamingSupported(), true); + delete globalThis.AudioWorkletNode; // no worklet → unsupported + assert.equal(streamingSupported(), false); + } finally { + globalThis.ReadableStream = orig.R; globalThis.fetch = orig.f; + if (orig.A === undefined) delete globalThis.AudioWorkletNode; else globalThis.AudioWorkletNode = orig.A; + } +}); + +beforeEach(() => { + ST.streaming = true; + ST.streamSampleRate = 48000; + ST.lastWorkletPos = 48000; // read frontier at 1.0s + ST.jsWriteFrontier = 96000; // written up to 2.0s +}); + +test('streamOffsetBuffered: true only inside [pos - 0.1s, writeFrontier]', () => { + // window is [48000 - 4800, 96000] = [43200, 96000] samples + assert.equal(streamOffsetBuffered(1.0), true); // 48000 — inside + assert.equal(streamOffsetBuffered(2.0), true); // 96000 — at the write frontier + assert.equal(streamOffsetBuffered(0.9), true); // 43200 — exactly pos-0.1s + assert.equal(streamOffsetBuffered(0.0), false); // before the window (replay-from-0) + assert.equal(streamOffsetBuffered(0.85), false); // 40800 — just behind the window + assert.equal(streamOffsetBuffered(2.1), false); // 100800 — past the write frontier +}); + +test('streamOffsetBuffered: always true when not streaming or rate unknown', () => { + ST.streaming = false; + assert.equal(streamOffsetBuffered(0.0), true); + ST.streaming = true; ST.streamSampleRate = 0; + assert.equal(streamOffsetBuffered(999), true); +}); From b9ab8568accc55a9c9e509623f55e011e258a953 Mon Sep 17 00:00:00 2001 From: byrongamatos Date: Wed, 8 Jul 2026 16:03:26 +0200 Subject: [PATCH 2/2] test(stems): cover all three streamingSupported() failure paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address CodeRabbit nitpick on #34 — pin the full && contract (missing ReadableStream / fetch / AudioWorkletNode), not just the worklet branch. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/streaming.test.mjs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/streaming.test.mjs b/tests/streaming.test.mjs index bc4e13b..f983a6d 100644 --- a/tests/streaming.test.mjs +++ b/tests/streaming.test.mjs @@ -26,6 +26,12 @@ test('streamingSupported: true only when all three platform APIs exist', () => { globalThis.fetch = function () {}; globalThis.AudioWorkletNode = function () {}; assert.equal(streamingSupported(), true); + delete globalThis.ReadableStream; // no stream → unsupported + assert.equal(streamingSupported(), false); + globalThis.ReadableStream = function () {}; + delete globalThis.fetch; // no fetch → unsupported + assert.equal(streamingSupported(), false); + globalThis.fetch = function () {}; delete globalThis.AudioWorkletNode; // no worklet → unsupported assert.equal(streamingSupported(), false); } finally {