From 6fcb8eb989efc049af8a7567fce8e35b10f24168 Mon Sep 17 00:00:00 2001 From: Christian Pillsbury Date: Wed, 11 Mar 2026 12:00:04 -0700 Subject: [PATCH] feat(spf): stream segment fetches via ReadableStream body (#890) --- .../templates/firefox-mse-repro/index.html | 107 ++++ .../templates/firefox-mse-repro/main.ts | 531 ++++++++++++++++++ .../spf/src/dom/features/end-of-stream.ts | 4 +- .../spf/src/dom/features/load-segments.ts | 67 ++- .../src/dom/features/segment-loader-actor.ts | 31 +- .../src/dom/features/setup-sourcebuffer.ts | 128 ++--- .../dom/features/tests/end-of-stream.test.ts | 35 ++ .../dom/features/tests/load-segments.test.ts | 152 +++++ .../features/tests/setup-sourcebuffer.test.ts | 409 ++++++-------- packages/spf/src/dom/media/append-segment.ts | 45 +- .../spf/src/dom/media/source-buffer-actor.ts | 83 ++- .../dom/media/tests/append-segment.test.ts | 162 ++++++ .../media/tests/source-buffer-actor.test.ts | 152 +++++ .../dom/network/chunked-stream-iterable.ts | 57 ++ packages/spf/src/dom/network/fetch.ts | 22 + .../tests/chunked-stream-iterable.test.ts | 154 +++++ .../spf/src/dom/network/tests/fetch.test.ts | 70 ++- .../spf/src/dom/playback-engine/engine.ts | 11 +- 18 files changed, 1847 insertions(+), 373 deletions(-) create mode 100644 packages/sandbox/templates/firefox-mse-repro/index.html create mode 100644 packages/sandbox/templates/firefox-mse-repro/main.ts create mode 100644 packages/spf/src/dom/media/tests/append-segment.test.ts create mode 100644 packages/spf/src/dom/network/chunked-stream-iterable.ts create mode 100644 packages/spf/src/dom/network/tests/chunked-stream-iterable.test.ts diff --git a/packages/sandbox/templates/firefox-mse-repro/index.html b/packages/sandbox/templates/firefox-mse-repro/index.html new file mode 100644 index 00000000..d02b09a1 --- /dev/null +++ b/packages/sandbox/templates/firefox-mse-repro/index.html @@ -0,0 +1,107 @@ + + + + + + Firefox MSE Init Order Repro + + + +

Firefox MSE Init Segment Order Reproduction

+

+ Tests whether appending a video media segment before audio init causes mozHasAudio = false in Firefox. +

+ +
+
+ + +
+ +

1 · Load HLS Playlist

+ +
+ + +
+
+ +

2 · Manual Setup

+
+ + + +
+ +

3 · Manual Steps

+
+ + +
+
+ + + +
+ +

4 · Test Scenarios

+ +
+

✓ Correct order: Both inits committed, then media segments.

+ +
+ +
+

+ ✗ Bug trigger: Video init → video segment 1 → + then audio init → audio segment 1.
+ In Firefox this should leave mozHasAudio = false. +

+ +
+
+ +
+

Log

+
+
+
+ + + + diff --git a/packages/sandbox/templates/firefox-mse-repro/main.ts b/packages/sandbox/templates/firefox-mse-repro/main.ts new file mode 100644 index 00000000..511a212c --- /dev/null +++ b/packages/sandbox/templates/firefox-mse-repro/main.ts @@ -0,0 +1,531 @@ +/** + * Firefox MSE Init Segment Order Reproduction Harness + * + * Tests the Firefox bug where appending a video media segment before the + * audio SourceBuffer has received its initialization segment causes + * mozHasAudio to be permanently false. + */ + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +interface PlaylistInfo { + videoCodec: string; + audioCodec: string; + videoInitUrl: string; + audioInitUrl: string; + videoSegmentUrls: string[]; + audioSegmentUrls: string[]; +} + +// --------------------------------------------------------------------------- +// DOM refs +// --------------------------------------------------------------------------- + +const video = document.getElementById('video') as HTMLVideoElement; +const logEl = document.getElementById('log')!; +const stateEl = document.getElementById('state')!; +const codecInfoEl = document.getElementById('codec-info')!; +const urlInput = document.getElementById('url') as HTMLInputElement; + +const btnParse = document.getElementById('btn-parse') as HTMLButtonElement; +const btnReset = document.getElementById('btn-reset') as HTMLButtonElement; +const btnMs = document.getElementById('btn-ms') as HTMLButtonElement; +const btnVsb = document.getElementById('btn-vsb') as HTMLButtonElement; +const btnAsb = document.getElementById('btn-asb') as HTMLButtonElement; +const btnVinit = document.getElementById('btn-vinit') as HTMLButtonElement; +const btnAinit = document.getElementById('btn-ainit') as HTMLButtonElement; +const btnVseg = document.getElementById('btn-vseg') as HTMLButtonElement; +const btnAseg = document.getElementById('btn-aseg') as HTMLButtonElement; +const btnPlay = document.getElementById('btn-play') as HTMLButtonElement; +const btnTestOk = document.getElementById('btn-test-ok') as HTMLButtonElement; +const btnTestBad = document.getElementById('btn-test-bad') as HTMLButtonElement; + +// --------------------------------------------------------------------------- +// State +// --------------------------------------------------------------------------- + +let info: PlaylistInfo | null = null; +let mediaSource: MediaSource | null = null; +let videoSB: SourceBuffer | null = null; +let audioSB: SourceBuffer | null = null; +let videoSegIdx = 0; +let audioSegIdx = 0; + +// --------------------------------------------------------------------------- +// Logging +// --------------------------------------------------------------------------- + +function log(msg: string, type: 'info' | 'ok' | 'err' | 'warn' | 'sep' = 'info') { + const el = document.createElement('div'); + el.className = type; + const ts = new Date().toLocaleTimeString('en', { + hour12: false, + hour: '2-digit', + minute: '2-digit', + second: '2-digit', + }); + el.textContent = `[${ts}] ${msg}`; + logEl.appendChild(el); + logEl.scrollTop = logEl.scrollHeight; +} + +// --------------------------------------------------------------------------- +// State display +// --------------------------------------------------------------------------- + +function updateState() { + const rs = ['HAVE_NOTHING', 'HAVE_METADATA', 'HAVE_CURRENT_DATA', 'HAVE_FUTURE_DATA', 'HAVE_ENOUGH_DATA']; + const mozHasAudio = (video as unknown as Record).mozHasAudio; + const mozHasVideo = (video as unknown as Record).mozHasVideo; + + function row(key: string, val: string, className: string) { + return `
${key}${val}
`; + } + + function boolRow(key: string, val: unknown) { + if (val === undefined) return row(key, 'n/a', 'val-none'); + return row(key, String(val), val ? 'val-true' : 'val-false'); + } + + function rangeStr(r: TimeRanges | undefined) { + if (!r || r.length === 0) return 'empty'; + return Array.from({ length: r.length }, (_, i) => `[${r.start(i).toFixed(2)},${r.end(i).toFixed(2)}]`).join(' '); + } + + stateEl.innerHTML = [ + row('readyState', rs[video.readyState] ?? String(video.readyState), 'val'), + boolRow('mozHasAudio', mozHasAudio), + boolRow('mozHasVideo', mozHasVideo), + row('ms.readyState', mediaSource?.readyState ?? 'none', mediaSource ? 'val' : 'val-none'), + row('videoSB', videoSB ? rangeStr(videoSB.buffered) : 'none', videoSB ? 'val' : 'val-none'), + row('audioSB', audioSB ? rangeStr(audioSB.buffered) : 'none', audioSB ? 'val' : 'val-none'), + ].join(''); +} + +setInterval(updateState, 250); + +for (const evt of ['loadedmetadata', 'loadeddata', 'canplay', 'playing', 'waiting', 'stalled', 'error'] as const) { + video.addEventListener(evt, () => { + log(`video: ${evt}`, evt === 'error' ? 'err' : 'info'); + updateState(); + }); +} + +// --------------------------------------------------------------------------- +// m3u8 parsing +// --------------------------------------------------------------------------- + +function resolveUrl(url: string, base: string): string { + if (url.startsWith('http')) return url; + try { + return new URL(url, base).href; + } catch { + return url; + } +} + +function parseMaster(text: string, baseUrl: string) { + const lines = text.split('\n').map((l) => l.trim()); + let videoUrl = ''; + let audioUrl = ''; + let codecsAttr = ''; + + for (let i = 0; i < lines.length; i++) { + const line = lines[i]!; + + if (line.startsWith('#EXT-X-MEDIA:') && /TYPE=AUDIO/.test(line) && !audioUrl) { + const m = line.match(/URI="([^"]+)"/); + if (m) audioUrl = resolveUrl(m[1]!, baseUrl); + } + + if (line.startsWith('#EXT-X-STREAM-INF:') && !videoUrl) { + const cm = line.match(/CODECS="([^"]+)"/); + if (cm) codecsAttr = cm[1]!; + const next = lines[i + 1]; + if (next && !next.startsWith('#')) videoUrl = resolveUrl(next, baseUrl); + } + } + + return { videoUrl, audioUrl, codecsAttr }; +} + +function parseMedia(text: string, baseUrl: string) { + const lines = text.split('\n').map((l) => l.trim()); + let initUrl = ''; + const segmentUrls: string[] = []; + + for (const line of lines) { + if (line.startsWith('#EXT-X-MAP:')) { + const m = line.match(/URI="([^"]+)"/); + if (m) initUrl = resolveUrl(m[1]!, baseUrl); + } + if (line && !line.startsWith('#')) segmentUrls.push(resolveUrl(line, baseUrl)); + } + + return { initUrl, segmentUrls }; +} + +function splitCodecs(codecsAttr: string) { + const parts = codecsAttr.split(',').map((c) => c.trim()); + const video = parts.filter((c) => /^(avc|hvc|vp0|av0)/i.test(c)).join(','); + const audio = parts.filter((c) => /^(mp4a|ac-3|ec-3|opus)/i.test(c)).join(','); + return { video: video || 'avc1.64001f', audio: audio || 'mp4a.40.2' }; +} + +// --------------------------------------------------------------------------- +// Parse button +// --------------------------------------------------------------------------- + +btnParse.addEventListener('click', async () => { + btnParse.disabled = true; + log('--- Parsing HLS playlist ---', 'sep'); + const url = urlInput.value.trim(); + log(`GET ${url}`); + + try { + const masterText = await fetch(url).then((r) => r.text()); + const { videoUrl, audioUrl, codecsAttr } = parseMaster(masterText, url); + + if (!videoUrl) throw new Error('No video rendition found in master playlist'); + log(`Video playlist: ${videoUrl}`, 'ok'); + log(`Audio playlist: ${audioUrl || '(none — muxed?)'}`, audioUrl ? 'ok' : 'warn'); + + const [videoText, audioText] = await Promise.all([ + fetch(videoUrl).then((r) => r.text()), + audioUrl ? fetch(audioUrl).then((r) => r.text()) : Promise.resolve(''), + ]); + + const videoMedia = parseMedia(videoText, videoUrl); + const audioMedia = audioUrl ? parseMedia(audioText, audioUrl) : { initUrl: '', segmentUrls: [] }; + + const codecs = splitCodecs(codecsAttr); + log(`Video codec: ${codecs.video}`, 'ok'); + log(`Audio codec: ${codecs.audio}`, 'ok'); + log(`Video init: ${videoMedia.initUrl}`, 'ok'); + log(`Audio init: ${audioMedia.initUrl || '(none)'}`, audioMedia.initUrl ? 'ok' : 'warn'); + log(`Video segments: ${videoMedia.segmentUrls.length}, Audio segments: ${audioMedia.segmentUrls.length}`, 'ok'); + + if (!audioUrl || !audioMedia.initUrl) { + log('⚠ No separate audio playlist — test requires demuxed audio/video HLS', 'warn'); + } + + info = { + videoCodec: codecs.video, + audioCodec: codecs.audio, + videoInitUrl: videoMedia.initUrl, + audioInitUrl: audioMedia.initUrl, + videoSegmentUrls: videoMedia.segmentUrls.slice(0, 6), + audioSegmentUrls: audioMedia.segmentUrls.slice(0, 6), + }; + + codecInfoEl.textContent = `video/mp4; codecs="${info.videoCodec}" | audio/mp4; codecs="${info.audioCodec}"`; + codecInfoEl.classList.add('visible'); + + btnMs.disabled = false; + btnTestOk.disabled = !audioMedia.initUrl; + btnTestBad.disabled = !audioMedia.initUrl; + } catch (e) { + log(`Error: ${e}`, 'err'); + btnParse.disabled = false; + } +}); + +// --------------------------------------------------------------------------- +// Reset +// --------------------------------------------------------------------------- + +btnReset.addEventListener('click', () => { + if (mediaSource && mediaSource.readyState === 'open') { + try { + mediaSource.endOfStream(); + } catch { + /* ignore */ + } + } + video.src = ''; + mediaSource = null; + videoSB = null; + audioSB = null; + videoSegIdx = 0; + audioSegIdx = 0; + btnVseg.textContent = 'Append Video Seg 1'; + btnAseg.textContent = 'Append Audio Seg 1'; + for (const b of [btnMs, btnVsb, btnAsb, btnVinit, btnAinit, btnVseg, btnAseg, btnPlay]) b.disabled = true; + btnParse.disabled = false; + log('--- Reset ---', 'sep'); + updateState(); +}); + +// --------------------------------------------------------------------------- +// Manual setup +// --------------------------------------------------------------------------- + +btnMs.addEventListener('click', () => { + mediaSource = new MediaSource(); + mediaSource.addEventListener( + 'sourceopen', + () => { + log('MediaSource: sourceopen', 'ok'); + btnVsb.disabled = false; + btnAsb.disabled = false; + updateState(); + }, + { once: true } + ); + mediaSource.addEventListener('sourceended', () => { + log('MediaSource: sourceended', 'info'); + updateState(); + }); + mediaSource.addEventListener('sourceclose', () => { + log('MediaSource: sourceclose', 'warn'); + updateState(); + }); + video.src = URL.createObjectURL(mediaSource); + btnMs.disabled = true; + log('MediaSource created, attaching to video…'); +}); + +function addSB(type: 'video' | 'audio') { + if (!mediaSource || !info) return; + const mime = type === 'video' ? `video/mp4; codecs="${info.videoCodec}"` : `audio/mp4; codecs="${info.audioCodec}"`; + try { + const sb = mediaSource.addSourceBuffer(mime); + sb.addEventListener('updateend', updateState); + sb.addEventListener('error', (e) => { + log(`${type} SB error: ${e}`, 'err'); + updateState(); + }); + log(`${type} SourceBuffer added: ${mime}`, 'ok'); + if (type === 'video') { + videoSB = sb; + btnVsb.disabled = true; + btnVinit.disabled = false; + } else { + audioSB = sb; + btnAsb.disabled = true; + btnAinit.disabled = false; + } + updateState(); + } catch (e) { + log(`Failed addSourceBuffer(${mime}): ${e}`, 'err'); + } +} + +btnVsb.addEventListener('click', () => addSB('video')); +btnAsb.addEventListener('click', () => addSB('audio')); + +// --------------------------------------------------------------------------- +// Append helper +// --------------------------------------------------------------------------- + +async function append(sb: SourceBuffer, url: string, label: string): Promise { + log(`Fetching ${label}…`); + const data = await fetch(url).then((r) => r.arrayBuffer()); + log(`${label}: ${data.byteLength} bytes`); + + if (sb.updating) + await new Promise((resolve) => { + sb.addEventListener('updateend', () => resolve(), { once: true }); + }); + + return new Promise((resolve, reject) => { + const onEnd = () => { + cleanup(); + log(`${label}: updateend ✓`, 'ok'); + resolve(); + }; + const onErr = () => { + cleanup(); + reject(new Error(`${label}: SourceBuffer error`)); + }; + const cleanup = () => { + sb.removeEventListener('updateend', onEnd); + sb.removeEventListener('error', onErr); + }; + sb.addEventListener('updateend', onEnd); + sb.addEventListener('error', onErr); + try { + sb.appendBuffer(data); + } catch (e) { + cleanup(); + reject(e); + } + }); +} + +// --------------------------------------------------------------------------- +// Manual step buttons +// --------------------------------------------------------------------------- + +btnVinit.addEventListener('click', async () => { + if (!videoSB || !info?.videoInitUrl) return; + btnVinit.disabled = true; + try { + await append(videoSB, info.videoInitUrl, 'video init'); + btnVseg.disabled = false; + } catch (e) { + log(`${e}`, 'err'); + } + updateState(); +}); + +btnAinit.addEventListener('click', async () => { + if (!audioSB || !info?.audioInitUrl) return; + btnAinit.disabled = true; + try { + await append(audioSB, info.audioInitUrl, 'audio init'); + btnAseg.disabled = false; + } catch (e) { + log(`${e}`, 'err'); + } + updateState(); +}); + +btnVseg.addEventListener('click', async () => { + if (!videoSB || !info) return; + const url = info.videoSegmentUrls[videoSegIdx]; + if (!url) { + log('No more video segments', 'warn'); + return; + } + try { + await append(videoSB, url, `video seg ${videoSegIdx + 1}`); + videoSegIdx++; + btnVseg.textContent = `Append Video Seg ${videoSegIdx + 1}`; + btnPlay.disabled = false; + } catch (e) { + log(`${e}`, 'err'); + } + updateState(); +}); + +btnAseg.addEventListener('click', async () => { + if (!audioSB || !info) return; + const url = info.audioSegmentUrls[audioSegIdx]; + if (!url) { + log('No more audio segments', 'warn'); + return; + } + try { + await append(audioSB, url, `audio seg ${audioSegIdx + 1}`); + audioSegIdx++; + btnAseg.textContent = `Append Audio Seg ${audioSegIdx + 1}`; + btnPlay.disabled = false; + } catch (e) { + log(`${e}`, 'err'); + } + updateState(); +}); + +btnPlay.addEventListener('click', () => { + video.play().catch((e) => log(`play() rejected: ${e}`, 'warn')); +}); + +// --------------------------------------------------------------------------- +// Test scenarios +// --------------------------------------------------------------------------- + +async function freshSetup(): Promise { + if (mediaSource && mediaSource.readyState === 'open') { + try { + mediaSource.endOfStream(); + } catch { + /* ignore */ + } + } + video.src = ''; + videoSB = null; + audioSB = null; + videoSegIdx = 0; + audioSegIdx = 0; + + mediaSource = new MediaSource(); + + await new Promise((resolve) => { + mediaSource!.addEventListener( + 'sourceopen', + () => { + log('MediaSource: sourceopen', 'ok'); + + videoSB = mediaSource!.addSourceBuffer(`video/mp4; codecs="${info!.videoCodec}"`); + videoSB.addEventListener('updateend', updateState); + log(`Video SB added`, 'ok'); + + audioSB = mediaSource!.addSourceBuffer(`audio/mp4; codecs="${info!.audioCodec}"`); + audioSB.addEventListener('updateend', updateState); + log(`Audio SB added`, 'ok'); + + resolve(); + }, + { once: true } + ); + + video.src = URL.createObjectURL(mediaSource!); + }); +} + +async function runTest(label: string, steps: Array<() => Promise>) { + for (const b of [btnTestOk, btnTestBad, btnParse, btnReset]) b.disabled = true; + log(`--- ${label} ---`, 'sep'); + + try { + await freshSetup(); + for (const step of steps) await step(); + updateState(); + log(`Test complete — check mozHasAudio above`, 'ok'); + video.play().catch(() => { + /* autoplay may be blocked */ + }); + } catch (e) { + log(`Test error: ${e}`, 'err'); + } + + for (const b of [btnTestOk, btnTestBad, btnParse, btnReset]) b.disabled = false; + btnPlay.disabled = false; +} + +btnTestOk.addEventListener('click', () => { + if (!info) return; + runTest('CORRECT ORDER: both inits → both segments', [ + () => { + log('Step 1: append video init'); + return append(videoSB!, info!.videoInitUrl, 'video init'); + }, + () => { + log('Step 2: append audio init'); + return append(audioSB!, info!.audioInitUrl, 'audio init'); + }, + () => { + log('Step 3: append video seg 1'); + return append(videoSB!, info!.videoSegmentUrls[0]!, 'video seg 1'); + }, + () => { + log('Step 4: append audio seg 1'); + return append(audioSB!, info!.audioSegmentUrls[0]!, 'audio seg 1'); + }, + ]); +}); + +btnTestBad.addEventListener('click', () => { + if (!info) return; + runTest('BUG ORDER: video init → video seg → audio init → audio seg', [ + () => { + log('Step 1: append video init'); + return append(videoSB!, info!.videoInitUrl, 'video init'); + }, + () => { + log('Step 2: append video seg 1 ← BEFORE audio init'); + return append(videoSB!, info!.videoSegmentUrls[0]!, 'video seg 1'); + }, + () => { + log('Step 3: append audio init (too late?)'); + return append(audioSB!, info!.audioInitUrl, 'audio init'); + }, + () => { + log('Step 4: append audio seg 1'); + return append(audioSB!, info!.audioSegmentUrls[0]!, 'audio seg 1'); + }, + ]); +}); diff --git a/packages/spf/src/dom/features/end-of-stream.ts b/packages/spf/src/dom/features/end-of-stream.ts index f066010e..d8a002ef 100644 --- a/packages/spf/src/dom/features/end-of-stream.ts +++ b/packages/spf/src/dom/features/end-of-stream.ts @@ -61,7 +61,9 @@ function isLastSegmentAppended(segments: readonly { id: string }[], actor: Sourc if (segments.length === 0) return true; const lastSeg = segments[segments.length - 1]; if (!lastSeg) return false; - return actor?.snapshot.context.segments.some((s) => s.id === lastSeg.id) ?? false; + // A partial segment is still streaming — the last segment is not ready until + // its entry is present and not marked partial. + return actor?.snapshot.context.segments.some((s) => s.id === lastSeg.id && !s.partial) ?? false; } /** diff --git a/packages/spf/src/dom/features/load-segments.ts b/packages/spf/src/dom/features/load-segments.ts index 9e7e4c4a..ddb417e9 100644 --- a/packages/spf/src/dom/features/load-segments.ts +++ b/packages/spf/src/dom/features/load-segments.ts @@ -6,7 +6,8 @@ import type { AddressableObject, Presentation, ResolvedTrack } from '../../core/ import { isResolvedTrack } from '../../core/types'; import { getSelectedTrack, type TrackSelectionState } from '../../core/utils/track-selection'; import type { SourceBufferActor } from '../media/source-buffer-actor'; -import { fetchResolvableBytes } from '../network/fetch'; +import { ChunkedStreamIterable, type ChunkedStreamIterableOptions } from '../network/chunked-stream-iterable'; +import { fetchResolvable } from '../network/fetch'; import { type BufferState, createSegmentLoaderActor, @@ -29,31 +30,61 @@ const ActorKeyByType = { // ============================================================================ /** - * Creates a fetch function that transparently samples bandwidth after each - * completed request. Callers receive bytes; throughput tracking is invisible. + * Creates a fetch function that eagerly starts the HTTP request (TTFB is + * awaited), then returns a lazy iterable over the response body that + * transparently samples bandwidth per chunk. * - * `onSample` is an optional callback invoked after each sample is recorded, - * used for bridging throughput state outward (e.g. migration bridge to global - * state). A callback is used rather than a subscription so that no immediate - * fire occurs at setup time — subscriptions fire on registration and would - * trigger spurious state changes before any work has started. + * Separating connection start from body reading ensures `fetch()` is called + * as soon as the task begins — not deferred until the actor's append loop + * first pulls a chunk. This makes fetch timing predictable and observable + * (e.g. in tests that record fetched URLs) regardless of downstream consumers. + * + * `onSample` bridges throughput state outward; see Phase 2 comment for detail. */ +type FetchOptions = RequestInit & ChunkedStreamIterableOptions; + function createTrackedFetch( throughput: WritableState, onSample?: (next: BandwidthState) => void -): (addressable: AddressableObject, options?: RequestInit) => Promise { +): (addressable: AddressableObject, options?: FetchOptions) => Promise> { return async (addressable, options) => { - const start = performance.now(); - const data = await fetchResolvableBytes(addressable, options); - const elapsed = performance.now() - start; - const next = sampleBandwidth(throughput.current, elapsed, data.byteLength); - throughput.patch(next); - throughput.flush(); - onSample?.(next); - return data; + const { minChunkSize, ...fetchOptions } = options ?? {}; + const response = await fetchResolvable(addressable, fetchOptions); + if (!response.body) throw new Error('Response has no body'); + const body = response.body; + return { + [Symbol.asyncIterator]: async function* () { + let chunkStart = performance.now(); + for await (const chunk of new ChunkedStreamIterable( + body, + ...(minChunkSize !== undefined ? [{ minChunkSize }] : []) + )) { + const elapsed = performance.now() - chunkStart; + const next = sampleBandwidth(throughput.current, elapsed, chunk.byteLength); + throughput.patch(next); + throughput.flush(); + onSample?.(next); + yield chunk; + chunkStart = performance.now(); + } + }, + }; }; } +/** + * Non-tracking fetch: eagerly starts the request and returns the response body + * as a lazy chunk iterable. Used for audio tracks which don't sample bandwidth. + * Pass `minChunkSize: Infinity` to accumulate the full body as a single chunk + * (equivalent to arrayBuffer() but through the same streaming path). + */ +async function fetchStream(addressable: AddressableObject, options?: FetchOptions): Promise> { + const { minChunkSize, ...fetchOptions } = options ?? {}; + const response = await fetchResolvable(addressable, fetchOptions); + if (!response.body) throw new Error('Response has no body'); + return new ChunkedStreamIterable(response.body, ...(minChunkSize !== undefined ? [{ minChunkSize }] : [])); +} + // ============================================================================ // STATE & OWNERS // ============================================================================ @@ -249,7 +280,7 @@ export function loadSegments( } : undefined ) - : fetchResolvableBytes; + : fetchStream; const segmentLoader = createState(undefined); diff --git a/packages/spf/src/dom/features/segment-loader-actor.ts b/packages/spf/src/dom/features/segment-loader-actor.ts index b3ee154d..339e7ba7 100644 --- a/packages/spf/src/dom/features/segment-loader-actor.ts +++ b/packages/spf/src/dom/features/segment-loader-actor.ts @@ -101,11 +101,16 @@ export interface SegmentLoaderActor { * operation (if still needed) or preempts it. * * @param sourceBufferActor - Shared SourceBufferActor reference (not owned) - * @param fetchBytes - Tracked fetch closure (owns throughput sampling) + * @param fetchBytes - Tracked fetch closure (owns throughput sampling for segments). + * Accepts an optional `minChunkSize` in options; init segments pass `Infinity` + * so the entire body accumulates as one chunk before appending. */ export function createSegmentLoaderActor( sourceBufferActor: SourceBufferActor, - fetchBytes: (addressable: AddressableObject, options?: RequestInit) => Promise + fetchBytes: ( + addressable: AddressableObject, + options?: RequestInit & { minChunkSize?: number } + ) => Promise> ): SegmentLoaderActor { let pendingTasks: LoadTask[] | null = null; let inFlightInitTrackId: string | null = null; @@ -115,7 +120,9 @@ export function createSegmentLoaderActor( let destroyed = false; const getBufferedSegments = (allSegments: readonly Segment[]): Segment[] => { - const bufferedIds = new Set(sourceBufferActor.snapshot.context.segments.map((s) => s.id)); + // Exclude partial segments — they are still being streamed and must not be + // treated as fully buffered for load planning or buffer window calculations. + const bufferedIds = new Set(sourceBufferActor.snapshot.context.segments.filter((s) => !s.partial).map((s) => s.id)); return allSegments.filter((s) => bufferedIds.has(s.id)); }; @@ -166,6 +173,9 @@ export function createSegmentLoaderActor( // content in the actor context. Preserves buffered high-quality content during // ABR downgrades; loads during upgrades and for uncovered positions. const existing = actorCtx.segments.find((s) => Math.abs(s.startTime - seg.startTime) < EPSILON); + // Partial segments are still streaming — treat as not buffered so they + // are always re-planned (avoids relying on incomplete data). + if (existing?.partial) return true; if (!existing?.trackBandwidth || !track.bandwidth) return true; return track.bandwidth > existing.trackBandwidth; }); @@ -205,7 +215,12 @@ export function createSegmentLoaderActor( if (task.type === 'append-init') { inFlightInitTrackId = task.meta.trackId; if (!signal.aborted) { - const data = await fetchBytes(task, { signal }); + // Init segments are small and need the full body before the + // same-track-seek vs track-switch commit decision can be made. + // minChunkSize: Infinity causes ChunkedStreamIterable to accumulate + // all chunks and yield exactly one — equivalent to arrayBuffer() but + // through the same streaming path as media segments. + const data = await fetchBytes(task, { signal, minChunkSize: Infinity }); // For seeks on the same track: commit even if aborted — avoids re-fetching the // same init next time. For track switches: don't commit the old track's init; // the new track's init follows in pendingTasks. @@ -220,12 +235,14 @@ export function createSegmentLoaderActor( return; } - // append-segment + // append-segment: await headers eagerly (starts the HTTP connection and + // records the fetch in observers like tests), then pass the body stream + // directly to the actor so chunks are appended as they arrive. inFlightSegmentId = task.meta.id; if (!signal.aborted) { - const data = await fetchBytes(task, { signal }); + const stream = await fetchBytes(task, { signal }); if (!signal.aborted) { - await sourceBufferActor.send({ type: 'append-segment', data, meta: task.meta }, signal); + await sourceBufferActor.send({ type: 'append-segment', data: stream, meta: task.meta }, signal); } } } finally { diff --git a/packages/spf/src/dom/features/setup-sourcebuffer.ts b/packages/spf/src/dom/features/setup-sourcebuffer.ts index 9fe770dd..344b6356 100644 --- a/packages/spf/src/dom/features/setup-sourcebuffer.ts +++ b/packages/spf/src/dom/features/setup-sourcebuffer.ts @@ -7,46 +7,12 @@ import { BufferKeyByType, getSelectedTrack, type TrackSelectionState } from '../ import { createSourceBuffer } from '../media/mediasource-setup'; import { createSourceBufferActor, type SourceBufferActor } from '../media/source-buffer-actor'; -/** Map track type to SourceBufferActor owner property key. */ -const ActorKeyByType = { - video: 'videoBufferActor', - audio: 'audioBufferActor', -} as const; - /** * Media track type for SourceBuffer setup. * Text tracks are excluded as they don't use MSE SourceBuffers. */ export type MediaTrackType = 'video' | 'audio'; -/** - * Setup SourceBuffer task (module-level, pure). - * Creates SourceBuffer for resolved track and waits a frame before completing. - */ -const setupSourceBufferTask = async ( - { currentState, currentOwners }: { currentState: SourceBufferState; currentOwners: SourceBufferOwners }, - context: { owners: WritableState; config: SourceBufferConfig } -): Promise => { - // Wait for track to be resolved with codecs before creating SourceBuffer - const track = getSelectedTrack(currentState, context.config.type); - if (!track || !isResolvedTrack(track)) return; - if (!track.codecs || track.codecs.length === 0) return; - - const mimeCodec = buildMimeCodec(track); - - // Create SourceBuffer and its actor together — single owners.patch so - // subscribers see both arrive at the same time with no intermediate state. - const buffer = createSourceBuffer(currentOwners.mediaSource!, mimeCodec); - const actor = createSourceBufferActor(buffer); - - const bufferKey = BufferKeyByType[context.config.type]; - const actorKey = ActorKeyByType[context.config.type]; - context.owners.patch({ [bufferKey]: buffer, [actorKey]: actor }); - - // Wait a frame to allow async state updates to flush - await new Promise((resolve) => requestAnimationFrame(resolve)); -}; - /** * State shape for SourceBuffer setup. */ @@ -121,57 +87,71 @@ export function shouldSetupBuffer(owners: SourceBufferOwners, type: MediaTrackTy } /** - * Configuration for SourceBuffer setup. - */ -export interface SourceBufferConfig { - type: T; -} - -/** - * Setup SourceBuffer orchestration. + * Setup all needed SourceBuffers as a single coordinated operation. * - * Triggers when: - * - MediaSource exists and is in 'open' state - * - Track is selected (same condition as resolveTrack) + * Waits until ALL selected tracks (video and/or audio) are resolved with + * codecs, then creates every SourceBuffer in one synchronous block before + * patching owners. This guarantees that downstream consumers (e.g. + * loadSegments) never see a partial set of SourceBuffers — preventing the + * Firefox bug where appending to a video SourceBuffer before the audio + * SourceBuffer exists causes mozHasAudio to be permanently false. * - * Creates SourceBuffer when track becomes resolved with codecs. - * This allows setupSourceBuffer to run in parallel with resolveTrack. + * Handles video-only, audio-only, and combined presentations correctly: + * only the tracks that are actually selected are waited on and created. * - * Note: Text tracks don't use SourceBuffers and should be handled separately. - * - * Generic over track type - create one orchestration per track type: * @example - * const videoCleanup = setupSourceBuffer({ state, owners }, { type: 'video' }); - * const audioCleanup = setupSourceBuffer({ state, owners }, { type: 'audio' }); + * const cleanup = setupSourceBuffers({ state, owners }); */ -export function setupSourceBuffer( - { - state, - owners, - }: { - state: WritableState; - owners: WritableState; - }, - config: SourceBufferConfig -): () => void { - let currentTask: Promise | null = null; +export function setupSourceBuffers({ + state, + owners, +}: { + state: WritableState; + owners: WritableState; +}): () => void { + let setupDone = false; const cleanup = combineLatest([state, owners]).subscribe( async ([currentState, currentOwners]: [SourceBufferState, SourceBufferOwners]) => { - // Check orchestration conditions (track selected, MediaSource open) - if (currentTask) return; // Task already in progress - if (!canSetupBuffer(currentState, currentOwners, config.type) || !shouldSetupBuffer(currentOwners, config.type)) - return; + if (setupDone) return; + if (!currentOwners.mediaSource) return; - // Invoke task (no abort needed - synchronous SourceBuffer creation + frame wait) - currentTask = setupSourceBufferTask({ currentState, currentOwners }, { owners, config }); + const videoSelected = !!currentState.selectedVideoTrackId; + const audioSelected = !!currentState.selectedAudioTrackId; - try { - await currentTask; - } finally { - // Cleanup orchestration state - currentTask = null; + if (!videoSelected && !audioSelected) return; + + const videoTrack = videoSelected ? getSelectedTrack(currentState, 'video') : null; + const audioTrack = audioSelected ? getSelectedTrack(currentState, 'audio') : null; + + // Wait until every selected track is resolved with codecs before + // creating any SourceBuffer. This is the coordination guarantee. + if (videoSelected && (!videoTrack || !isResolvedTrack(videoTrack) || !videoTrack.codecs?.length)) return; + if (audioSelected && (!audioTrack || !isResolvedTrack(audioTrack) || !audioTrack.codecs?.length)) return; + + setupDone = true; + + // Create all SourceBuffers synchronously — no await between addSourceBuffer + // calls — then patch owners once so subscribers see all buffers simultaneously. + const patch: Partial = {}; + + if (videoSelected && videoTrack && isResolvedTrack(videoTrack)) { + const buffer = createSourceBuffer(currentOwners.mediaSource!, buildMimeCodec(videoTrack)); + patch.videoBuffer = buffer; + patch.videoBufferActor = createSourceBufferActor(buffer); } + + if (audioSelected && audioTrack && isResolvedTrack(audioTrack)) { + const buffer = createSourceBuffer(currentOwners.mediaSource!, buildMimeCodec(audioTrack)); + patch.audioBuffer = buffer; + patch.audioBufferActor = createSourceBufferActor(buffer); + } + + owners.patch(patch); + + // Wait a frame to allow async state updates to flush before downstream + // orchestrations (loadSegments) begin reacting to the new owners. + await new Promise((resolve) => requestAnimationFrame(resolve)); } ); diff --git a/packages/spf/src/dom/features/tests/end-of-stream.test.ts b/packages/spf/src/dom/features/tests/end-of-stream.test.ts index 2537292c..bd34a012 100644 --- a/packages/spf/src/dom/features/tests/end-of-stream.test.ts +++ b/packages/spf/src/dom/features/tests/end-of-stream.test.ts @@ -189,6 +189,41 @@ describe('hasLastSegmentLoaded', () => { expect(hasLastSegmentLoaded(state, owners)).toBe(true); }); + it('returns false when last segment is present but marked partial', () => { + const track = makeResolvedVideoTrack(4); + const state: EndOfStreamState = { + selectedVideoTrackId: 'video-1', + presentation: makePresentation(track), + }; + // Last segment (seg-3) is in context but still streaming + const actor = createSourceBufferActor(makeSourceBuffer(), { + initTrackId: 'video-1', + segments: [ + { id: 'seg-0', startTime: 0, duration: 2.5, trackId: 'video-1' }, + { id: 'seg-1', startTime: 2.5, duration: 2.5, trackId: 'video-1' }, + { id: 'seg-2', startTime: 5, duration: 2.5, trackId: 'video-1' }, + { id: 'seg-3', startTime: 7.5, duration: 2.5, trackId: 'video-1', partial: true }, + ], + }); + expect(hasLastSegmentLoaded(state, { videoBufferActor: actor })).toBe(false); + }); + + it('returns true when last segment is present and partial is cleared', () => { + const track = makeResolvedVideoTrack(4); + const state: EndOfStreamState = { + selectedVideoTrackId: 'video-1', + presentation: makePresentation(track), + }; + const actor = createSourceBufferActor(makeSourceBuffer(), { + initTrackId: 'video-1', + segments: [ + { id: 'seg-2', startTime: 5, duration: 2.5, trackId: 'video-1' }, + { id: 'seg-3', startTime: 7.5, duration: 2.5, trackId: 'video-1' }, // partial: undefined = complete + ], + }); + expect(hasLastSegmentLoaded(state, { videoBufferActor: actor })).toBe(true); + }); + it('returns false when video last segment is loaded but audio last segment is not', () => { const videoTrack = makeResolvedVideoTrack(4); const presentation = { diff --git a/packages/spf/src/dom/features/tests/load-segments.test.ts b/packages/spf/src/dom/features/tests/load-segments.test.ts index 64f109de..0946a1ae 100644 --- a/packages/spf/src/dom/features/tests/load-segments.test.ts +++ b/packages/spf/src/dom/features/tests/load-segments.test.ts @@ -901,3 +901,155 @@ describe('loadSegments forward buffer flushing', () => { cleanup(); }); }); + +// --------------------------------------------------------------------------- +// Streaming bandwidth tracking +// --------------------------------------------------------------------------- + +describe('loadSegments bandwidth tracking', () => { + function makeStreamingFetch(chunks: Uint8Array[]) { + return vi.fn().mockImplementation(() => { + const body = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + }, + }); + return Promise.resolve(new Response(body)); + }); + } + + it('samples bandwidth per chunk and updates state.bandwidthState', async () => { + const chunkSize = 50_000; // 50 KB — below 128 KB default, so whole segment = one flush + const numChunks = 3; + const chunks = Array.from({ length: numChunks }, () => new Uint8Array(chunkSize).fill(1)); + + globalThis.fetch = makeStreamingFetch(chunks); + + const { loadSegments } = await import('../load-segments'); + const { createState: cs } = await import('../../../core/state/create-state'); + + const segment = { id: 's1', url: 'http://example.com/s1.m4s', startTime: 0, duration: 10 }; + const track = { + type: 'video' as const, + id: 'track-1', + url: 'http://example.com/video.m3u8', + mimeType: 'video/mp4', + codecs: ['avc1.42E01E'], + bandwidth: 1_000_000, + initialization: { url: 'http://example.com/init.mp4' }, + segments: [segment], + startTime: 0, + duration: 10, + }; + + // Seeding bandwidthState activates the onSample bridge → state.bandwidthState updates + const initialBandwidth = { + fastEstimate: 0, + fastTotalWeight: 0, + slowEstimate: 0, + slowTotalWeight: 0, + bytesSampled: 0, + }; + + const state = cs({ + preload: 'auto', + selectedVideoTrackId: 'track-1', + currentTime: 0, + bandwidthState: initialBandwidth, + presentation: { + id: 'p1', + url: 'http://example.com/playlist.m3u8', + startTime: 0, + duration: 10, + selectionSets: [{ id: 'ss1', type: 'video', switchingSets: [{ id: 'sw1', type: 'video', tracks: [track] }] }], + }, + }); + + const { sourceBuffer, actor } = makeSourceBufferWithActor(); + const owners = cs({ videoBuffer: sourceBuffer, videoBufferActor: actor }); + + const cleanup = loadSegments({ state, owners }, { type: 'video' }); + + // Wait for both init and segment to be appended (actor context will have 1 segment) + await vi.waitFor( + () => { + expect(owners.current.videoBufferActor?.snapshot.context.segments).toHaveLength(1); + }, + { timeout: 3000 } + ); + + // All bytes (init + segment) should be counted in bytesSampled + const totalExpected = chunkSize * numChunks * 2; // init fetch + segment fetch, each 3×50KB + expect(state.current.bandwidthState?.bytesSampled).toBeGreaterThan(0); + expect(state.current.bandwidthState?.bytesSampled).toBeLessThanOrEqual(totalExpected); + + cleanup(); + }); + + it('appended data matches the concatenated streaming chunks', async () => { + const part1 = new Uint8Array([1, 2, 3, 4]); + const part2 = new Uint8Array([5, 6, 7, 8]); + + globalThis.fetch = vi.fn().mockImplementation(() => { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(part1); + controller.enqueue(part2); + controller.close(); + }, + }); + return Promise.resolve(new Response(body)); + }); + + const { loadSegments } = await import('../load-segments'); + const { createState: cs } = await import('../../../core/state/create-state'); + + const segment = { id: 's1', url: 'http://example.com/s1.m4s', startTime: 0, duration: 10 }; + const track = { + type: 'video' as const, + id: 'track-1', + url: 'http://example.com/video.m3u8', + mimeType: 'video/mp4', + codecs: ['avc1.42E01E'], + bandwidth: 500_000, + initialization: { url: 'http://example.com/init.mp4' }, + segments: [segment], + startTime: 0, + duration: 10, + }; + + const state = cs({ + preload: 'auto', + selectedVideoTrackId: 'track-1', + currentTime: 0, + presentation: { + id: 'p1', + url: 'http://example.com/playlist.m3u8', + startTime: 0, + duration: 10, + selectionSets: [{ id: 'ss1', type: 'video', switchingSets: [{ id: 'sw1', type: 'video', tracks: [track] }] }], + }, + }); + + const { sourceBuffer, actor } = makeSourceBufferWithActor(); + const owners = cs({ videoBuffer: sourceBuffer, videoBufferActor: actor }); + + const cleanup = loadSegments({ state, owners }, { type: 'video' }); + + await vi.waitFor( + () => { + expect(owners.current.videoBufferActor?.snapshot.context.segments).toHaveLength(1); + }, + { timeout: 3000 } + ); + + // Each response body yields [1,2,3,4,5,6,7,8] — both init and segment appends should match + const calls = (sourceBuffer.appendBuffer as ReturnType).mock.calls; + for (const [data] of calls) { + expect(Array.from(new Uint8Array(data as ArrayBuffer))).toEqual([1, 2, 3, 4, 5, 6, 7, 8]); + } + + cleanup(); + }); +}); diff --git a/packages/spf/src/dom/features/tests/setup-sourcebuffer.test.ts b/packages/spf/src/dom/features/tests/setup-sourcebuffer.test.ts index 316e25e5..aa0d661a 100644 --- a/packages/spf/src/dom/features/tests/setup-sourcebuffer.test.ts +++ b/packages/spf/src/dom/features/tests/setup-sourcebuffer.test.ts @@ -6,7 +6,7 @@ import { canSetupBuffer, type SourceBufferOwners, type SourceBufferState, - setupSourceBuffer, + setupSourceBuffers, shouldSetupBuffer, } from '../setup-sourcebuffer'; @@ -238,268 +238,193 @@ describe('shouldSetupBuffer', () => { }); }); -describe('setupSourceBuffer', () => { +describe('setupSourceBuffers', () => { beforeEach(() => { vi.clearAllMocks(); }); - describe('video track', () => { - it('creates SourceBuffer for resolved video track', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); + it('creates video SourceBuffer for video-only source', async () => { + const { createSourceBuffer } = await import('../../media/mediasource-setup'); - const videoTrack = createResolvedVideoTrack(); - const state = createState({}); - const owners = createState({}); + const videoTrack = createResolvedVideoTrack(); + const state = createState({}); + const owners = createState({}); + const cleanup = setupSourceBuffers({ state, owners }); - const cleanup = setupSourceBuffer({ state, owners }, { type: 'video' }); - - // Set up conditions - const mediaSource = {} as MediaSource; - owners.patch({ mediaSource }); - state.patch({ - presentation: createPresentationWithTracks({ video: videoTrack }), - selectedVideoTrackId: 'video-1', - }); - - // Wait for async operation - await vi.waitFor(() => { - expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'video/mp4; codecs="avc1.42E01E"'); - }); - - cleanup(); + const mediaSource = {} as MediaSource; + owners.patch({ mediaSource }); + state.patch({ + presentation: createPresentationWithTracks({ video: videoTrack }), + selectedVideoTrackId: 'video-1', }); - it('updates owners with videoBuffer reference', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); - - const mockBuffer = { - mimeCodec: 'video/mp4; codecs="avc1.42E01E"', - mode: 'segments', - updating: false, - }; - vi.mocked(createSourceBuffer).mockReturnValue(mockBuffer as unknown as SourceBuffer); - - const videoTrack = createResolvedVideoTrack(); - const state = createState({}); - const owners = createState({}); - - const cleanup = setupSourceBuffer({ state, owners }, { type: 'video' }); - - owners.patch({ mediaSource: {} as MediaSource }); - state.patch({ - presentation: createPresentationWithTracks({ video: videoTrack }), - selectedVideoTrackId: 'video-1', - }); - - await vi.waitFor(() => { - const currentOwners = owners.current; - expect(currentOwners.videoBuffer).toBe(mockBuffer); - }); - - cleanup(); + await vi.waitFor(() => { + expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'video/mp4; codecs="avc1.42E01E"'); + expect(owners.current.videoBuffer).toBeDefined(); + expect(owners.current.audioBuffer).toBeUndefined(); }); - it('does not create if track not resolved', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); - - const unresolvedTrack = { - type: 'video' as const, - id: 'video-1', - url: 'http://example.com/video.m3u8', - bandwidth: 1000000, - mimeType: 'video/mp4', - codecs: ['avc1.42E01E'], - }; - const presentation: Presentation = { - id: 'pres-1', - url: 'http://example.com/playlist.m3u8', - selectionSets: [ - { - id: 'video-set', - type: 'video', - switchingSets: [ - { - id: 'video-switching', - type: 'video', - tracks: [unresolvedTrack], - }, - ], - }, - ], - startTime: 0, - }; - - const state = createState({}); - const owners = createState({}); - - const cleanup = setupSourceBuffer({ state, owners }, { type: 'video' }); - - owners.patch({ mediaSource: {} as MediaSource }); - state.patch({ - presentation, - selectedVideoTrackId: 'video-1', - }); - - await new Promise((resolve) => setTimeout(resolve, 50)); - - expect(createSourceBuffer).not.toHaveBeenCalled(); - - cleanup(); - }); - - it('does not create if track missing codecs', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); - - const videoTrack: VideoTrack = { - ...createResolvedVideoTrack(), - codecs: [], - }; - - const state = createState({}); - const owners = createState({}); - - const cleanup = setupSourceBuffer({ state, owners }, { type: 'video' }); - - owners.patch({ mediaSource: {} as MediaSource }); - state.patch({ - presentation: createPresentationWithTracks({ video: videoTrack }), - selectedVideoTrackId: 'video-1', - }); - - await new Promise((resolve) => setTimeout(resolve, 50)); - - expect(createSourceBuffer).not.toHaveBeenCalled(); - - cleanup(); - }); - - it('does not create multiple buffers (deduplication)', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); - - const videoTrack = createResolvedVideoTrack(); - const state = createState({}); - const owners = createState({}); - - const cleanup = setupSourceBuffer({ state, owners }, { type: 'video' }); - - owners.patch({ mediaSource: {} as MediaSource }); - state.patch({ - presentation: createPresentationWithTracks({ video: videoTrack }), - selectedVideoTrackId: 'video-1', - }); - - await vi.waitFor(() => { - expect(createSourceBuffer).toHaveBeenCalledTimes(1); - }); - - // Trigger another update - state.patch({ - presentation: createPresentationWithTracks({ video: videoTrack }), - }); - - await new Promise((resolve) => setTimeout(resolve, 50)); - - expect(createSourceBuffer).toHaveBeenCalledTimes(1); - - cleanup(); - }); + cleanup(); }); - describe('audio track', () => { - it('creates SourceBuffer for resolved audio track', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); + it('creates audio SourceBuffer for audio-only source', async () => { + const { createSourceBuffer } = await import('../../media/mediasource-setup'); - const audioTrack = createResolvedAudioTrack(); - const state = createState({}); - const owners = createState({}); + const audioTrack = createResolvedAudioTrack(); + const state = createState({}); + const owners = createState({}); + const cleanup = setupSourceBuffers({ state, owners }); - const cleanup = setupSourceBuffer({ state, owners }, { type: 'audio' }); - - const mediaSource = {} as MediaSource; - owners.patch({ mediaSource }); - state.patch({ - presentation: createPresentationWithTracks({ audio: audioTrack }), - selectedAudioTrackId: 'audio-1', - }); - - await vi.waitFor(() => { - expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'audio/mp4; codecs="mp4a.40.2"'); - }); - - cleanup(); + const mediaSource = {} as MediaSource; + owners.patch({ mediaSource }); + state.patch({ + presentation: createPresentationWithTracks({ audio: audioTrack }), + selectedAudioTrackId: 'audio-1', }); - it('updates owners with audioBuffer reference', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); - - const mockBuffer = { - mimeCodec: 'audio/mp4; codecs="mp4a.40.2"', - mode: 'segments', - updating: false, - }; - vi.mocked(createSourceBuffer).mockReturnValue(mockBuffer as unknown as SourceBuffer); - - const audioTrack = createResolvedAudioTrack(); - const state = createState({}); - const owners = createState({}); - - const cleanup = setupSourceBuffer({ state, owners }, { type: 'audio' }); - - owners.patch({ mediaSource: {} as MediaSource }); - state.patch({ - presentation: createPresentationWithTracks({ audio: audioTrack }), - selectedAudioTrackId: 'audio-1', - }); - - await vi.waitFor(() => { - const currentOwners = owners.current; - expect(currentOwners.audioBuffer).toBe(mockBuffer); - }); - - cleanup(); + await vi.waitFor(() => { + expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'audio/mp4; codecs="mp4a.40.2"'); + expect(owners.current.audioBuffer).toBeDefined(); + expect(owners.current.videoBuffer).toBeUndefined(); }); + + cleanup(); }); - describe('multi-track orchestration', () => { - it('creates video and audio track types in parallel', async () => { - const { createSourceBuffer } = await import('../../media/mediasource-setup'); + it('creates both SourceBuffers together when both tracks are selected', async () => { + const { createSourceBuffer } = await import('../../media/mediasource-setup'); - const videoTrack = createResolvedVideoTrack(); - const audioTrack = createResolvedAudioTrack(); + const videoTrack = createResolvedVideoTrack(); + const audioTrack = createResolvedAudioTrack(); + const state = createState({}); + const owners = createState({}); + const cleanup = setupSourceBuffers({ state, owners }); - const state = createState({}); - const owners = createState({}); - - // Set up both orchestrations - const videoCleanup = setupSourceBuffer({ state, owners }, { type: 'video' }); - const audioCleanup = setupSourceBuffer({ state, owners }, { type: 'audio' }); - - // Set up conditions - const mediaSource = {} as MediaSource; - owners.patch({ mediaSource }); - state.patch({ - presentation: createPresentationWithTracks({ - video: videoTrack, - audio: audioTrack, - }), - selectedVideoTrackId: 'video-1', - selectedAudioTrackId: 'audio-1', - }); - - // Wait for both to be created - await vi.waitFor(() => { - expect(createSourceBuffer).toHaveBeenCalledTimes(2); - expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'video/mp4; codecs="avc1.42E01E"'); - expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'audio/mp4; codecs="mp4a.40.2"'); - }); - - // Verify both buffers are set - const currentOwners = owners.current; - expect(currentOwners.videoBuffer).toBeDefined(); - expect(currentOwners.audioBuffer).toBeDefined(); - - videoCleanup(); - audioCleanup(); + const mediaSource = {} as MediaSource; + owners.patch({ mediaSource }); + state.patch({ + presentation: createPresentationWithTracks({ video: videoTrack, audio: audioTrack }), + selectedVideoTrackId: 'video-1', + selectedAudioTrackId: 'audio-1', }); + + await vi.waitFor(() => { + expect(createSourceBuffer).toHaveBeenCalledTimes(2); + expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'video/mp4; codecs="avc1.42E01E"'); + expect(createSourceBuffer).toHaveBeenCalledWith(mediaSource, 'audio/mp4; codecs="mp4a.40.2"'); + expect(owners.current.videoBuffer).toBeDefined(); + expect(owners.current.audioBuffer).toBeDefined(); + }); + + cleanup(); + }); + + it('waits for audio to resolve before creating video SourceBuffer when both are selected', async () => { + const { createSourceBuffer } = await import('../../media/mediasource-setup'); + + const videoTrack = createResolvedVideoTrack(); + const unresolvedAudio = createResolvedAudioTrack(); + // Simulate unresolved audio track (no segments/initialization — not a ResolvedTrack) + const { + segments: _s, + initialization: _i, + startTime: _st, + duration: _d, + ...unresolvedAudioPartial + } = unresolvedAudio; + + const state = createState({}); + const owners = createState({}); + const cleanup = setupSourceBuffers({ state, owners }); + + owners.patch({ mediaSource: {} as MediaSource }); + // Both track IDs selected, but audio track is not yet resolved + state.patch({ + presentation: createPresentationWithTracks({ + video: videoTrack, + audio: unresolvedAudioPartial as AudioTrack, + }), + selectedVideoTrackId: 'video-1', + selectedAudioTrackId: 'audio-1', + }); + + // Video is resolved but audio is not — neither SourceBuffer should be created yet + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(createSourceBuffer).not.toHaveBeenCalled(); + + // Now audio resolves + state.patch({ + presentation: createPresentationWithTracks({ video: videoTrack, audio: unresolvedAudio }), + }); + + await vi.waitFor(() => { + expect(createSourceBuffer).toHaveBeenCalledTimes(2); + expect(owners.current.videoBuffer).toBeDefined(); + expect(owners.current.audioBuffer).toBeDefined(); + }); + + cleanup(); + }); + + it('does not create SourceBuffer when track has no codecs', async () => { + const { createSourceBuffer } = await import('../../media/mediasource-setup'); + + const videoTrack: VideoTrack = { ...createResolvedVideoTrack(), codecs: [] }; + const state = createState({}); + const owners = createState({}); + const cleanup = setupSourceBuffers({ state, owners }); + + owners.patch({ mediaSource: {} as MediaSource }); + state.patch({ + presentation: createPresentationWithTracks({ video: videoTrack }), + selectedVideoTrackId: 'video-1', + }); + + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(createSourceBuffer).not.toHaveBeenCalled(); + + cleanup(); + }); + + it('does not create SourceBuffers more than once', async () => { + const { createSourceBuffer } = await import('../../media/mediasource-setup'); + + const videoTrack = createResolvedVideoTrack(); + const state = createState({}); + const owners = createState({}); + const cleanup = setupSourceBuffers({ state, owners }); + + owners.patch({ mediaSource: {} as MediaSource }); + state.patch({ + presentation: createPresentationWithTracks({ video: videoTrack }), + selectedVideoTrackId: 'video-1', + }); + + await vi.waitFor(() => expect(createSourceBuffer).toHaveBeenCalledTimes(1)); + + state.patch({ presentation: createPresentationWithTracks({ video: videoTrack }) }); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(createSourceBuffer).toHaveBeenCalledTimes(1); + + cleanup(); + }); + + it('does not create without a MediaSource', async () => { + const { createSourceBuffer } = await import('../../media/mediasource-setup'); + + const videoTrack = createResolvedVideoTrack(); + const state = createState({}); + const owners = createState({}); + const cleanup = setupSourceBuffers({ state, owners }); + + state.patch({ + presentation: createPresentationWithTracks({ video: videoTrack }), + selectedVideoTrackId: 'video-1', + }); + + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(createSourceBuffer).not.toHaveBeenCalled(); + + cleanup(); }); }); diff --git a/packages/spf/src/dom/media/append-segment.ts b/packages/spf/src/dom/media/append-segment.ts index e1380a5d..0955b33d 100644 --- a/packages/spf/src/dom/media/append-segment.ts +++ b/packages/spf/src/dom/media/append-segment.ts @@ -1,25 +1,39 @@ /** - * Segment appender helper (P11) + * Segment appender helper. * - * Appends media segments (ArrayBuffer) to SourceBuffer. + * Appends media data (ArrayBuffer or AsyncIterable stream) to a + * SourceBuffer, waiting for `updateend` between calls so the browser can + * process each append before the next one arrives. */ +/** Data accepted by appendSegment — a full buffer or an async chunk stream. */ +export type AppendData = ArrayBuffer | AsyncIterable; + /** - * Append a media segment to a SourceBuffer. + * Append media data to a SourceBuffer. * - * Waits for the SourceBuffer to be ready (not updating), then appends - * the segment data. Returns a promise that resolves when append completes. + * Accepts either a full ArrayBuffer (single append) or an AsyncIterable of + * Uint8Array chunks (one append per chunk, in order). Waits for `updateend` + * between each call so appends are serialized correctly. * - * @param sourceBuffer - The SourceBuffer to append to - * @param segmentData - The segment data as ArrayBuffer - * @returns Promise that resolves when append completes - * - * @example - * const data = await fetch(segmentUrl).then(r => r.arrayBuffer()); - * await appendSegment(videoSourceBuffer, data); + * Errors from the SourceBuffer (`error` event) or from the iterable are + * propagated as rejections. */ -export async function appendSegment(sourceBuffer: SourceBuffer, segmentData: ArrayBuffer): Promise { - // Wait for SourceBuffer to be ready (not currently updating) +export async function appendSegment(sourceBuffer: SourceBuffer, data: AppendData, signal?: AbortSignal): Promise { + if (data instanceof ArrayBuffer) { + await appendChunk(sourceBuffer, data); + } else { + for await (const chunk of data) { + // Check between chunks so an abort can stop streaming before the next + // appendBuffer call. The current chunk (if any) has already landed in the + // SourceBuffer; the partial: true flag in the actor model reflects this. + if (signal?.aborted) throw signal.reason ?? new DOMException('Aborted', 'AbortError'); + await appendChunk(sourceBuffer, chunk); + } + } +} + +async function appendChunk(sourceBuffer: SourceBuffer, data: ArrayBuffer | Uint8Array): Promise { if (sourceBuffer.updating) { await new Promise((resolve) => { const onUpdateEnd = () => { @@ -30,7 +44,6 @@ export async function appendSegment(sourceBuffer: SourceBuffer, segmentData: Arr }); } - // Append the segment data return new Promise((resolve, reject) => { const onUpdateEnd = () => { cleanup(); @@ -51,7 +64,7 @@ export async function appendSegment(sourceBuffer: SourceBuffer, segmentData: Arr sourceBuffer.addEventListener('error', onError); try { - sourceBuffer.appendBuffer(segmentData); + sourceBuffer.appendBuffer(data as ArrayBuffer); } catch (error) { cleanup(); reject(error); diff --git a/packages/spf/src/dom/media/source-buffer-actor.ts b/packages/spf/src/dom/media/source-buffer-actor.ts index 8ed1ed3c..65ffdb93 100644 --- a/packages/spf/src/dom/media/source-buffer-actor.ts +++ b/packages/spf/src/dom/media/source-buffer-actor.ts @@ -2,7 +2,7 @@ import type { Actor, ActorSnapshot } from '../../core/actor'; import { createState } from '../../core/state/create-state'; import { SerialRunner, Task } from '../../core/task'; import type { Segment, Track } from '../../core/types'; -import { appendSegment } from './append-segment'; +import { type AppendData, appendSegment } from './append-segment'; import { flushBuffer } from './buffer-flusher'; // ============================================================================= @@ -20,8 +20,10 @@ export type AppendSegmentMeta = Pick & trackBandwidth?: number; }; -export type AppendInitMessage = { type: 'append-init'; data: ArrayBuffer; meta: { trackId: Track['id'] } }; -export type AppendSegmentMessage = { type: 'append-segment'; data: ArrayBuffer; meta: AppendSegmentMeta }; +export type { AppendData }; + +export type AppendInitMessage = { type: 'append-init'; data: AppendData; meta: { trackId: Track['id'] } }; +export type AppendSegmentMessage = { type: 'append-segment'; data: AppendData; meta: AppendSegmentMeta }; export type RemoveMessage = { type: 'remove'; start: number; end: number }; export type SourceBufferMessage = AppendInitMessage | AppendSegmentMessage | RemoveMessage; @@ -31,7 +33,18 @@ export type SourceBufferActorStatus = 'idle' | 'updating' | 'destroyed'; /** Non-finite (extended) data managed by the actor — the XState "context". */ export interface SourceBufferActorContext { initTrackId?: string | undefined; - segments: Array & { trackId: Track['id']; trackBandwidth?: number }>; + segments: Array< + Pick & { + trackId: Track['id']; + trackBandwidth?: number; + /** + * True while a streaming append is in progress for this segment. + * The segment's data is partially present in the SourceBuffer. + * Downstream code must not treat a partial segment as fully buffered. + */ + partial?: boolean; + } + >; bufferedRanges: BufferedRange[]; } @@ -79,6 +92,13 @@ interface MessageTaskOptions { signal: AbortSignal; getCtx: () => SourceBufferActorContext; sourceBuffer: SourceBuffer; + /** + * Called when a streaming append transitions to a partial state — i.e. + * the first chunk of an AsyncIterable has been committed and the segment + * now has data in the SourceBuffer but is not yet complete. Not called for + * full ArrayBuffer appends (which are atomic). + */ + onPartialContext: (ctx: SourceBufferActorContext) => void; } function appendInitTask( @@ -100,15 +120,13 @@ function appendInitTask( function appendSegmentTask( message: AppendSegmentMessage, - { signal, getCtx, sourceBuffer }: MessageTaskOptions + { signal, getCtx, sourceBuffer, onPartialContext }: MessageTaskOptions ): Task { return new Task( async (taskSignal) => { const ctx = getCtx(); if (taskSignal.aborted) return ctx; - await appendSegment(sourceBuffer, message.data); - // No abort check here: the physical SourceBuffer has been modified, so - // the model must be updated to match regardless of signal state. + const { meta } = message; // Remove any existing entry at the same start time (same "slot" in the // timeline), then record the new segment. Assumes time-aligned segments @@ -116,6 +134,32 @@ function appendSegmentTask( // parsed timestamps. const EPSILON = 0.0001; const filtered = ctx.segments.filter((s) => Math.abs(s.startTime - meta.startTime) >= EPSILON); + + // For streaming data: emit partial state before the first chunk so + // downstream code can see the in-progress segment and treat it as + // incomplete. ArrayBuffer appends are atomic so no partial state is + // needed — context is updated once at task completion. + if (!(message.data instanceof ArrayBuffer)) { + onPartialContext({ + ...ctx, + segments: [ + ...filtered, + { + id: meta.id, + startTime: meta.startTime, + duration: meta.duration, + trackId: meta.trackId, + ...(meta.trackBandwidth !== undefined && { trackBandwidth: meta.trackBandwidth }), + partial: true, + }, + ], + bufferedRanges: ctx.bufferedRanges, + }); + } + + await appendSegment(sourceBuffer, message.data, taskSignal); + // No abort check here: the physical SourceBuffer has been modified, so + // the model must be updated to match regardless of signal state. return { ...ctx, segments: [ @@ -234,7 +278,17 @@ export function createSourceBufferActor( // tick is rejected — the actor is now committed to this operation. state.patch({ status: 'updating' }); - const task = messageToTask(message, { signal, getCtx: () => state.current.context, sourceBuffer }); + const onPartialContext = (ctx: SourceBufferActorContext) => { + state.patch({ status: 'updating', context: ctx }); + state.flush(); + }; + + const task = messageToTask(message, { + signal, + getCtx: () => state.current.context, + sourceBuffer, + onPartialContext, + }); return runner.schedule(task).then(applyResult).catch(handleError); }, @@ -265,8 +319,16 @@ export function createSourceBufferActor( // appends do not fail — but worth revisiting if MSE error recovery lands. let workingCtx = state.current.context; + // Partial context updates from streaming appends patch state directly so + // external subscribers see in-progress state, but workingCtx is only + // advanced on task completion to preserve batch context threading. + const onPartialContext = (ctx: SourceBufferActorContext) => { + state.patch({ status: 'updating', context: ctx }); + state.flush(); + }; + for (const message of messages.slice(0, -1)) { - const task = messageToTask(message, { signal, getCtx: () => workingCtx, sourceBuffer }); + const task = messageToTask(message, { signal, getCtx: () => workingCtx, sourceBuffer, onPartialContext }); const result = runner.schedule(task); result.then((newCtx) => { workingCtx = newCtx; @@ -277,6 +339,7 @@ export function createSourceBufferActor( signal, getCtx: () => workingCtx, sourceBuffer, + onPartialContext, }); return runner.schedule(lastTask).then(applyResult).catch(handleError); }, diff --git a/packages/spf/src/dom/media/tests/append-segment.test.ts b/packages/spf/src/dom/media/tests/append-segment.test.ts new file mode 100644 index 00000000..8dc5449a --- /dev/null +++ b/packages/spf/src/dom/media/tests/append-segment.test.ts @@ -0,0 +1,162 @@ +import { describe, expect, it, vi } from 'vitest'; +import { appendSegment } from '../append-segment'; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeSourceBuffer(): SourceBuffer { + const listeners: Record = {}; + + return { + updating: false, + appendBuffer: vi.fn(() => { + setTimeout(() => { + for (const listener of listeners.updateend ?? []) listener(new Event('updateend')); + }, 0); + }), + addEventListener: vi.fn((type: string, listener: EventListener) => { + listeners[type] ??= []; + listeners[type].push(listener); + }), + removeEventListener: vi.fn((type: string, listener: EventListener) => { + listeners[type] = (listeners[type] ?? []).filter((l) => l !== listener); + }), + } as unknown as SourceBuffer; +} + +async function* chunks(...buffers: ArrayBuffer[]): AsyncGenerator { + for (const buf of buffers) yield new Uint8Array(buf); +} + +// --------------------------------------------------------------------------- +// ArrayBuffer path +// --------------------------------------------------------------------------- + +describe('appendSegment', () => { + it('calls appendBuffer once for an ArrayBuffer', async () => { + const sb = makeSourceBuffer(); + await appendSegment(sb, new ArrayBuffer(8)); + expect(sb.appendBuffer).toHaveBeenCalledTimes(1); + }); + + it('resolves after updateend for ArrayBuffer', async () => { + const sb = makeSourceBuffer(); + await expect(appendSegment(sb, new ArrayBuffer(4))).resolves.toBeUndefined(); + }); + + it('waits for updating=false before appending', async () => { + const listeners: Record = {}; + let updating = true; + + const sb = { + get updating() { + return updating; + }, + appendBuffer: vi.fn(() => { + setTimeout(() => { + updating = false; + for (const l of listeners.updateend ?? []) l(new Event('updateend')); + }, 0); + }), + addEventListener: vi.fn((type: string, listener: EventListener) => { + listeners[type] ??= []; + listeners[type].push(listener); + }), + removeEventListener: vi.fn((type: string, listener: EventListener) => { + listeners[type] = (listeners[type] ?? []).filter((l) => l !== listener); + }), + } as unknown as SourceBuffer; + + // Simulate an external updateend that clears updating + setTimeout(() => { + updating = false; + for (const l of listeners.updateend ?? []) l(new Event('updateend')); + }, 10); + + await appendSegment(sb, new ArrayBuffer(4)); + expect(sb.appendBuffer).toHaveBeenCalledOnce(); + }); + + // --------------------------------------------------------------------------- + // AsyncIterable path + // --------------------------------------------------------------------------- + + it('calls appendBuffer once per chunk for AsyncIterable', async () => { + const sb = makeSourceBuffer(); + await appendSegment(sb, chunks(new ArrayBuffer(4), new ArrayBuffer(4), new ArrayBuffer(4))); + expect(sb.appendBuffer).toHaveBeenCalledTimes(3); + }); + + it('resolves after all chunks are appended', async () => { + const sb = makeSourceBuffer(); + await expect(appendSegment(sb, chunks(new ArrayBuffer(4), new ArrayBuffer(4)))).resolves.toBeUndefined(); + }); + + it('propagates errors thrown from the AsyncIterable', async () => { + const sb = makeSourceBuffer(); + + async function* errorStream(): AsyncGenerator { + yield new Uint8Array(4); + throw new Error('stream failed'); + } + + await expect(appendSegment(sb, errorStream())).rejects.toThrow('stream failed'); + }); + + it('passes chunk bytes through to appendBuffer unchanged', async () => { + const sb = makeSourceBuffer(); + const data = new Uint8Array([1, 2, 3, 4]); + + await appendSegment( + sb, + (async function* () { + yield data; + })() + ); + + const appended = (sb.appendBuffer as ReturnType).mock.calls[0]?.[0]; + expect(Array.from(new Uint8Array(appended as ArrayBuffer))).toEqual([1, 2, 3, 4]); + }); + + it('appends chunks in order', async () => { + const appended: number[][] = []; + + const listeners: Record = {}; + const sb = { + updating: false, + appendBuffer: vi.fn((data: ArrayBuffer) => { + appended.push(Array.from(new Uint8Array(data))); + setTimeout(() => { + for (const l of listeners.updateend ?? []) l(new Event('updateend')); + }, 0); + }), + addEventListener: vi.fn((type: string, listener: EventListener) => { + listeners[type] ??= []; + listeners[type].push(listener); + }), + removeEventListener: vi.fn((type: string, listener: EventListener) => { + listeners[type] = (listeners[type] ?? []).filter((l) => l !== listener); + }), + } as unknown as SourceBuffer; + + const chunk1 = new Uint8Array([1, 2]); + const chunk2 = new Uint8Array([3, 4]); + const chunk3 = new Uint8Array([5, 6]); + + await appendSegment( + sb, + (async function* () { + yield chunk1; + yield chunk2; + yield chunk3; + })() + ); + + expect(appended).toEqual([ + [1, 2], + [3, 4], + [5, 6], + ]); + }); +}); diff --git a/packages/spf/src/dom/media/tests/source-buffer-actor.test.ts b/packages/spf/src/dom/media/tests/source-buffer-actor.test.ts index ef851656..217a095e 100644 --- a/packages/spf/src/dom/media/tests/source-buffer-actor.test.ts +++ b/packages/spf/src/dom/media/tests/source-buffer-actor.test.ts @@ -437,6 +437,158 @@ describe('createSourceBufferActor', () => { // destroy() // --------------------------------------------------------------------------- + // --------------------------------------------------------------------------- + // Partial segment state — streaming AsyncIterable appends + // --------------------------------------------------------------------------- + + it('does not emit a partial snapshot for ArrayBuffer appends', async () => { + const sourceBuffer = makeSourceBuffer([[0, 10]]); + const actor = createSourceBufferActor(sourceBuffer); + + const snapshots: (typeof actor.snapshot)[] = []; + const unsub = actor.subscribe((s) => snapshots.push(s)); + + await actor.send( + { + type: 'append-segment', + data: new ArrayBuffer(8), + meta: { id: 's1', startTime: 0, duration: 10, trackId: 'track-1' }, + }, + neverAborted + ); + unsub(); + + const hadPartial = snapshots.some((s) => s.context.segments.some((seg) => seg.partial)); + expect(hadPartial).toBe(false); + + actor.destroy(); + }); + + it('emits a partial:true snapshot before completing a streaming append', async () => { + const sourceBuffer = makeSourceBuffer([[0, 10]]); + const actor = createSourceBufferActor(sourceBuffer); + + const snapshots: (typeof actor.snapshot)[] = []; + const unsub = actor.subscribe((s) => + snapshots.push({ ...s, context: { ...s.context, segments: [...s.context.segments] } }) + ); + + async function* twoChunks() { + yield new Uint8Array(4); + yield new Uint8Array(4); + } + + await actor.send( + { type: 'append-segment', data: twoChunks(), meta: { id: 's1', startTime: 0, duration: 10, trackId: 'track-1' } }, + neverAborted + ); + unsub(); + + const partialSnapshot = snapshots.find((s) => + s.context.segments.some((seg) => seg.id === 's1' && seg.partial === true) + ); + expect(partialSnapshot).toBeDefined(); + + actor.destroy(); + }); + + it('clears partial flag on segment after streaming append completes', async () => { + const sourceBuffer = makeSourceBuffer([[0, 10]]); + const actor = createSourceBufferActor(sourceBuffer); + + async function* oneChunk() { + yield new Uint8Array(8); + } + + await actor.send( + { type: 'append-segment', data: oneChunk(), meta: { id: 's1', startTime: 0, duration: 10, trackId: 'track-1' } }, + neverAborted + ); + + const seg = actor.snapshot.context.segments.find((s) => s.id === 's1'); + expect(seg).toBeDefined(); + expect(seg?.partial).toBeUndefined(); + + actor.destroy(); + }); + + it('leaves partial:true entry in context when streaming append is aborted', async () => { + // Use a controllable iterable that pauses, allowing abort mid-stream + let resolveFirst: () => void; + const firstChunkReady = new Promise((r) => { + resolveFirst = r; + }); + + async function* pausingStream() { + yield new Uint8Array(4); + // Pause here — abort will fire before the second chunk + await firstChunkReady; + yield new Uint8Array(4); + } + + const sourceBuffer = makeSourceBuffer([ + [0, 5], + [5, 10], + ]); + const actor = createSourceBufferActor(sourceBuffer); + const ac = new AbortController(); + + const pending = actor.send( + { + type: 'append-segment', + data: pausingStream(), + meta: { id: 's1', startTime: 0, duration: 10, trackId: 'track-1' }, + }, + ac.signal + ); + + // Wait until partial state is emitted (first chunk queued) + await vi.waitFor(() => { + expect(actor.snapshot.context.segments.some((s) => s.id === 's1' && s.partial === true)).toBe(true); + }); + + // Abort — the stream is paused waiting for resolveFirst + ac.abort(); + resolveFirst!(); + + // Let the task settle (it will reject with AbortError — swallow it) + await pending.catch(() => {}); + + // partial: true entry should remain — accurately reflects data in SourceBuffer + const seg = actor.snapshot.context.segments.find((s) => s.id === 's1'); + expect(seg).toBeDefined(); + expect(seg?.partial).toBe(true); + + actor.destroy(); + }); + + it('replaces a partial:true entry when the same segment is fully re-appended', async () => { + const sourceBuffer = makeSourceBuffer([[0, 10]]); + const actor = createSourceBufferActor(sourceBuffer); + + // First: put a partial segment in context via initialContext shortcut + const actorWithPartial = createSourceBufferActor(sourceBuffer, { + segments: [{ id: 's1', startTime: 0, duration: 10, trackId: 'track-1', partial: true }], + }); + + // Now fully append the same segment (ArrayBuffer path — atomic, no partial) + await actorWithPartial.send( + { + type: 'append-segment', + data: new ArrayBuffer(8), + meta: { id: 's1', startTime: 0, duration: 10, trackId: 'track-1' }, + }, + neverAborted + ); + + const seg = actorWithPartial.snapshot.context.segments.find((s) => s.id === 's1'); + expect(seg).toBeDefined(); + expect(seg?.partial).toBeUndefined(); + + actorWithPartial.destroy(); + actor.destroy(); + }); + it('destroy() aborts the in-progress operation', async () => { const sourceBuffer = makeSourceBuffer(); const actor = createSourceBufferActor(sourceBuffer); diff --git a/packages/spf/src/dom/network/chunked-stream-iterable.ts b/packages/spf/src/dom/network/chunked-stream-iterable.ts new file mode 100644 index 00000000..fd1d67dc --- /dev/null +++ b/packages/spf/src/dom/network/chunked-stream-iterable.ts @@ -0,0 +1,57 @@ +const DEFAULT_MIN_CHUNK_SIZE = 2 ** 17; // 128 KB + +export interface ChunkedStreamIterableOptions { + minChunkSize?: number; +} + +/** + * Adapts a `ReadableStream` (e.g. `response.body`) into an + * `AsyncIterable` that yields chunks no smaller than + * `minChunkSize` bytes. Smaller network chunks are accumulated and yielded + * together once the threshold is met. Any remainder is flushed on stream end. + * + * Errors from the underlying stream propagate naturally — the reader lock is + * always released via `finally`. + */ +export class ChunkedStreamIterable implements AsyncIterable { + readonly minChunkSize: number; + #readableStream: ReadableStream; + + constructor( + readableStream: ReadableStream, + { minChunkSize = DEFAULT_MIN_CHUNK_SIZE }: ChunkedStreamIterableOptions = {} + ) { + this.#readableStream = readableStream; + this.minChunkSize = minChunkSize; + } + + async *[Symbol.asyncIterator](): AsyncGenerator { + let pending: Uint8Array | undefined; + const reader = this.#readableStream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + if (pending) yield pending; + break; + } + + pending = pending ? concat(pending, value) : value; + + if (pending.length >= this.minChunkSize) { + yield pending; + pending = undefined; + } + } + } finally { + reader.releaseLock(); + } + } +} + +function concat(a: Uint8Array, b: Uint8Array): Uint8Array { + const result = new Uint8Array(a.length + b.length); + result.set(a); + result.set(b, a.length); + return result; +} diff --git a/packages/spf/src/dom/network/fetch.ts b/packages/spf/src/dom/network/fetch.ts index fb913058..c76d5094 100644 --- a/packages/spf/src/dom/network/fetch.ts +++ b/packages/spf/src/dom/network/fetch.ts @@ -4,9 +4,11 @@ * Two-function approach for composability: * 1. fetchResolvable() - Fetch AddressableObject (handles byte ranges) * 2. getResponseText() - Extract text from Response + * 3. fetchResolvableStream() - Stream body as Uint8Array chunks */ import type { AddressableObject } from '../../core/types'; +import { ChunkedStreamIterable, type ChunkedStreamIterableOptions } from './chunked-stream-iterable'; /** * Minimal Response-like interface for text extraction. @@ -69,6 +71,26 @@ export async function fetchResolvableBytes( return response.arrayBuffer(); } +/** + * Fetch resolvable as a stream of Uint8Array chunks. + * + * Convenience wrapper around fetchResolvable that yields the body as chunks + * via ChunkedStreamIterable. Headers are awaited before the first chunk is + * yielded (TTFB is accounted for before iteration begins). + * + * Throws if the response body is null (e.g. non-body HTTP status). + * Errors from the underlying stream propagate naturally as thrown errors. + */ +export async function* fetchResolvableStream( + addressable: AddressableObject, + options?: RequestInit & ChunkedStreamIterableOptions +): AsyncGenerator { + const { minChunkSize, ...fetchOptions } = options ?? {}; + const response = await fetchResolvable(addressable, fetchOptions); + if (!response.body) throw new Error('Response has no body'); + yield* new ChunkedStreamIterable(response.body, ...(minChunkSize !== undefined ? [{ minChunkSize }] : [])); +} + /** * Extract text from Response. * diff --git a/packages/spf/src/dom/network/tests/chunked-stream-iterable.test.ts b/packages/spf/src/dom/network/tests/chunked-stream-iterable.test.ts new file mode 100644 index 00000000..1ad1a516 --- /dev/null +++ b/packages/spf/src/dom/network/tests/chunked-stream-iterable.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it } from 'vitest'; +import { ChunkedStreamIterable } from '../chunked-stream-iterable'; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeStream(...chunks: Uint8Array[]): ReadableStream { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) { + controller.enqueue(chunks[i++]); + } else { + controller.close(); + } + }, + }); +} + +function bytes(size: number, fill = 1): Uint8Array { + return new Uint8Array(size).fill(fill); +} + +async function collect(iterable: AsyncIterable): Promise { + const result: Uint8Array[] = []; + for await (const chunk of iterable) { + result.push(chunk); + } + return result; +} + +function totalBytes(chunks: Uint8Array[]): number { + return chunks.reduce((sum, c) => sum + c.length, 0); +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe('ChunkedStreamIterable', () => { + it('exposes minChunkSize', () => { + const stream = makeStream(); + const iterable = new ChunkedStreamIterable(stream, { minChunkSize: 1024 }); + expect(iterable.minChunkSize).toBe(1024); + }); + + it('defaults minChunkSize to 128 KB', () => { + const stream = makeStream(); + const iterable = new ChunkedStreamIterable(stream); + expect(iterable.minChunkSize).toBe(2 ** 17); + }); + + it('yields a single chunk when it meets minChunkSize exactly', async () => { + const minChunkSize = 64; + const stream = makeStream(bytes(64)); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize })); + expect(chunks).toHaveLength(1); + expect(chunks[0]!.length).toBe(64); + }); + + it('yields a single chunk when it exceeds minChunkSize', async () => { + const minChunkSize = 64; + const stream = makeStream(bytes(100)); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize })); + expect(chunks).toHaveLength(1); + expect(chunks[0]!.length).toBe(100); + }); + + it('accumulates small chunks until minChunkSize is met', async () => { + const minChunkSize = 64; + // 3 × 30-byte chunks — first two should accumulate, third triggers flush at 90 bytes + const stream = makeStream(bytes(30, 1), bytes(30, 2), bytes(30, 3)); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize })); + expect(chunks).toHaveLength(1); + expect(chunks[0]!.length).toBe(90); + }); + + it('flushes remaining bytes on stream end even if below minChunkSize', async () => { + const minChunkSize = 128; + const stream = makeStream(bytes(50)); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize })); + expect(chunks).toHaveLength(1); + expect(chunks[0]!.length).toBe(50); + }); + + it('preserves all bytes across multiple yielded chunks', async () => { + const minChunkSize = 50; + // 3 × 40-byte chunks → first two accumulate to 80 (≥50, yield), third is remainder + const stream = makeStream(bytes(40, 1), bytes(40, 2), bytes(40, 3)); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize })); + expect(totalBytes(chunks)).toBe(120); + }); + + it('concatenates chunk bytes correctly', async () => { + const minChunkSize = 4; + const a = new Uint8Array([1, 2]); + const b = new Uint8Array([3, 4]); + const stream = makeStream(a, b); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize })); + expect(chunks).toHaveLength(1); + expect(Array.from(chunks[0]!)).toEqual([1, 2, 3, 4]); + }); + + it('yields nothing for an empty stream', async () => { + const stream = makeStream(); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize: 64 })); + expect(chunks).toHaveLength(0); + }); + + it('propagates errors from the underlying stream', async () => { + const errorStream = new ReadableStream({ + start(controller) { + controller.error(new Error('network failure')); + }, + }); + + await expect(collect(new ChunkedStreamIterable(errorStream, { minChunkSize: 64 }))).rejects.toThrow( + 'network failure' + ); + }); + + it('releases the reader lock after normal completion', async () => { + const stream = makeStream(bytes(10)); + const iterable = new ChunkedStreamIterable(stream, { minChunkSize: 64 }); + await collect(iterable); + // If lock was not released, getReader() would throw + expect(() => stream.getReader()).not.toThrow(); + }); + + it('releases the reader lock after an error', async () => { + const errorStream = new ReadableStream({ + start(controller) { + controller.error(new Error('fail')); + }, + }); + + const iterable = new ChunkedStreamIterable(errorStream, { minChunkSize: 64 }); + await expect(collect(iterable)).rejects.toThrow(); + // Lock should be released even though we errored + expect(errorStream.locked).toBe(false); + }); + + it('handles multiple large chunks correctly', async () => { + const minChunkSize = 50; + // Each chunk already meets minChunkSize → each yielded individually + const stream = makeStream(bytes(60, 1), bytes(70, 2), bytes(80, 3)); + const chunks = await collect(new ChunkedStreamIterable(stream, { minChunkSize })); + expect(chunks).toHaveLength(3); + expect(chunks[0]!.length).toBe(60); + expect(chunks[1]!.length).toBe(70); + expect(chunks[2]!.length).toBe(80); + }); +}); diff --git a/packages/spf/src/dom/network/tests/fetch.test.ts b/packages/spf/src/dom/network/tests/fetch.test.ts index 5c8e2266..6b526474 100644 --- a/packages/spf/src/dom/network/tests/fetch.test.ts +++ b/packages/spf/src/dom/network/tests/fetch.test.ts @@ -1,7 +1,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; import type { AddressableObject } from '../../../core/types'; import type { ResponseLike } from '../fetch'; -import { fetchResolvable, getResponseText } from '../fetch'; +import { fetchResolvable, fetchResolvableStream, getResponseText } from '../fetch'; describe('fetchResolvable', () => { beforeEach(() => { @@ -63,6 +63,74 @@ describe('fetchResolvable', () => { }); }); +describe('fetchResolvableStream', () => { + beforeEach(() => { + vi.restoreAllMocks(); + }); + + function makeBodyStream(...chunks: Uint8Array[]): ReadableStream { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) { + controller.enqueue(chunks[i++]); + } else { + controller.close(); + } + }, + }); + } + + async function collect(gen: AsyncGenerator): Promise { + const result: Uint8Array[] = []; + for await (const chunk of gen) result.push(chunk); + return result; + } + + it('yields chunks from the response body', async () => { + const data = new Uint8Array(256).fill(0xff); + const body = makeBodyStream(data); + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(body)); + + const chunks = await collect(fetchResolvableStream({ url: 'https://example.com/seg.m4s' }, { minChunkSize: 128 })); + const total = chunks.reduce((sum, c) => sum + c.length, 0); + expect(total).toBe(256); + }); + + it('passes the URL and byte-range header through to fetch', async () => { + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(makeBodyStream())); + + await collect( + fetchResolvableStream( + { url: 'https://example.com/seg.m4s', byteRange: { start: 0, end: 99 } }, + { minChunkSize: 64 } + ) + ); + + const req: Request = fetchSpy.mock.calls[0]![0] as Request; + expect(req.headers.get('Range')).toBe('bytes=0-99'); + }); + + it('throws when the response has no body', async () => { + const nullBodyResponse = new Response(null, { status: 204 }); + vi.spyOn(globalThis, 'fetch').mockResolvedValue(nullBodyResponse); + + await expect(collect(fetchResolvableStream({ url: 'https://example.com/seg.m4s' }))).rejects.toThrow( + 'Response has no body' + ); + }); + + it('does not pass minChunkSize as a fetch RequestInit option', async () => { + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(makeBodyStream())); + + await collect(fetchResolvableStream({ url: 'https://example.com/seg.m4s' }, { minChunkSize: 512 })); + + // fetch should have been called with a Request, not an object with minChunkSize + const req: Request = fetchSpy.mock.calls[0]![0] as Request; + expect(req).toBeInstanceOf(Request); + }); +}); + describe('getResponseText', () => { it('extracts text from ResponseLike', async () => { const response: ResponseLike = { diff --git a/packages/spf/src/dom/playback-engine/engine.ts b/packages/spf/src/dom/playback-engine/engine.ts index 6f1ccf64..9b2b45bc 100644 --- a/packages/spf/src/dom/playback-engine/engine.ts +++ b/packages/spf/src/dom/playback-engine/engine.ts @@ -20,7 +20,7 @@ import { loadSegments } from '../features/load-segments'; import type { TextTrackBufferState } from '../features/load-text-track-cues'; import { loadTextTrackCues } from '../features/load-text-track-cues'; import { setupMediaSource } from '../features/setup-mediasource'; -import { setupSourceBuffer } from '../features/setup-sourcebuffer'; +import { setupSourceBuffers } from '../features/setup-sourcebuffer'; import { setupTextTracks } from '../features/setup-text-tracks'; import { syncSelectedTextTrackFromDom } from '../features/sync-selected-text-track-from-dom'; import { syncTextTrackModes } from '../features/sync-text-track-modes'; @@ -271,9 +271,12 @@ export function createPlaybackEngine(config: PlaybackEngineConfig = {}): Playbac // 4.5. Update MediaSource duration (when presentation duration available) updateDuration({ state, owners }), - // 5. Setup SourceBuffers (when MediaSource ready and tracks resolved) - setupSourceBuffer({ state, owners }, { type: 'video' }), - setupSourceBuffer({ state, owners }, { type: 'audio' }), + // 5. Setup SourceBuffers (when MediaSource ready and all selected tracks resolved) + // Both SourceBuffers are created in a single synchronous operation to guarantee + // neither is visible to loadSegments before the other exists — preventing the + // Firefox bug where appending video data before audio SB is created causes + // mozHasAudio to be permanently false. + setupSourceBuffers({ state, owners }), // 5.5. Track currentTime from mediaElement (feeds forward buffer management) //