From 557d8bd72f85a71df35327defafeeb839ee43fba Mon Sep 17 00:00:00 2001 From: Christian Pillsbury Date: Thu, 18 Jun 2026 14:01:25 -0700 Subject: [PATCH] refactor(spf): drive live reload via a RecurringRunner instead of an epoch signal (WIP) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the signal-as-event live-reload scheduler with a runner-driven model. The `resolveTrack` loader schedules its resolve work on a new `RecurringRunner` that re-runs the task on an injected `reschedule` policy; the separate `scheduleTrackReload` behavior and its per-type reload-epoch signals are deleted. Core (`core/tasks`): - `Task.run()` is now memoized (runs once, shares the result across calls) and gains `clone()` (fresh, pending, structurally identical) — added to `TaskLike`. - `RecurringRunner`: single-slot, id-keyed (dedup same id / abort-and-replace on new id), time-free. Each cycle runs `Promise.all([task.run(), reschedule(task, previous, signal)])` and re-runs a `clone()` while reschedule resolves `true`. - `Reschedule = (task, previous, signal) => PromiseLike` — invoked concurrently with the run, observes it via the memoized `run()`, owns its delay. - `delayedReschedule(cadence)` builds a Reschedule from a pure ms-cadence fn, start-anchored (subtracts the run's elapsed) so reloads are measured from load-start per RFC 8216 §6.3.4, preserving half-on-unchanged. Supporting: - `@videojs/utils/time`: add cancellable `sleep(ms, signal)`. - `media/hls/reload-policy`: `mediaPlaylistReloadDelay` (pure cadence; relocated scheduler logic — target-duration, half-on-unchanged, stop-on-ENDLIST, retry). - `resolve-track`: baked universal completeness gate + injected `reschedule`; engine composes `delayedReschedule(mediaPlaylistReloadDelay)`. WIP: not fully validated end-to-end against a live stream through this path. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../spf/src/core/tasks/delayed-reschedule.ts | 35 ++ packages/spf/src/core/tasks/task.ts | 187 ++++++++++- .../tasks/tests/delayed-reschedule.test.ts | 95 ++++++ .../spf/src/core/tasks/tests/task.test.ts | 245 +++++++++++++- packages/spf/src/media/hls/reload-policy.ts | 44 +++ .../src/media/hls/tests/reload-policy.test.ts | 54 +++ .../src/playback/behaviors/resolve-track.ts | 313 +++++++----------- .../behaviors/schedule-track-reload.ts | 198 ----------- .../behaviors/tests/resolve-track.test.ts | 85 ++--- .../tests/schedule-track-reload.test.ts | 137 -------- .../spf/src/playback/engines/hls/engine.ts | 51 +-- .../src/playback/engines/live-hls/engine.ts | 16 +- .../engines/live-playlist-spike/engine.ts | 22 +- packages/utils/src/time/index.ts | 1 + packages/utils/src/time/sleep.ts | 25 ++ packages/utils/src/time/tests/sleep.test.ts | 42 +++ 16 files changed, 926 insertions(+), 624 deletions(-) create mode 100644 packages/spf/src/core/tasks/delayed-reschedule.ts create mode 100644 packages/spf/src/core/tasks/tests/delayed-reschedule.test.ts create mode 100644 packages/spf/src/media/hls/reload-policy.ts create mode 100644 packages/spf/src/media/hls/tests/reload-policy.test.ts delete mode 100644 packages/spf/src/playback/behaviors/schedule-track-reload.ts delete mode 100644 packages/spf/src/playback/behaviors/tests/schedule-track-reload.test.ts create mode 100644 packages/utils/src/time/sleep.ts create mode 100644 packages/utils/src/time/tests/sleep.test.ts diff --git a/packages/spf/src/core/tasks/delayed-reschedule.ts b/packages/spf/src/core/tasks/delayed-reschedule.ts new file mode 100644 index 00000000..175ebf73 --- /dev/null +++ b/packages/spf/src/core/tasks/delayed-reschedule.ts @@ -0,0 +1,35 @@ +import { sleep } from '@videojs/utils/time'; +import type { Reschedule } from './task'; + +/** + * Build a {@link Reschedule} from a pure cadence function — the common + * timer-based, *start-anchored* implementation. + * + * Invoked concurrently with the run, it observes the result, then waits + * `cadence(current, previous)` milliseconds **measured from when it was invoked** + * (≈ the run's start): it subtracts the run's own elapsed time, so consecutive + * runs begin one cadence apart regardless of how long each run takes (per + * RFC 8216 §6.3.4's "measured from the last time the client began loading"). If + * the run takes longer than the cadence, the next run starts immediately. + * + * A `null` cadence stops the recurrence. An errored run passes `current` as + * `undefined`, so the cadence function can choose to retry (return a delay) or + * stop (return `null`). + */ +export function delayedReschedule( + cadence: (current: TValue | undefined, previous: TValue | undefined) => number | null +): Reschedule { + return async (task, previous, signal) => { + const startedAt = Date.now(); + let current: TValue | undefined; + try { + current = await task.run(); + } catch { + current = undefined; + } + const ms = cadence(current, previous); + if (ms === null) return false; + await sleep(Math.max(0, ms - (Date.now() - startedAt)), signal); + return true; + }; +} diff --git a/packages/spf/src/core/tasks/task.ts b/packages/spf/src/core/tasks/task.ts index e66d602a..6fd3e8d5 100644 --- a/packages/spf/src/core/tasks/task.ts +++ b/packages/spf/src/core/tasks/task.ts @@ -46,8 +46,11 @@ export interface TaskLike { readonly status: TaskStatus; readonly value: DeepReadonly | undefined; readonly error: DeepReadonly | undefined; + /** Run the work, memoized: repeated calls share one execution + result. */ run(): Promise; abort(): void; + /** A fresh, structurally identical task (same work + id) in a pending state — for re-running. */ + clone(): TaskLike; } /** @@ -58,6 +61,11 @@ export interface TaskLike { * propagates into the task's work without requiring the caller to track the * task separately. * + * `run()` is memoized: the work runs at most once per instance, and every call + * returns the same promise (so observers can `await run()` to read the result + * without re-triggering the work). To re-run the *same* work, take a `clone()` — + * a fresh instance with its own AbortController and a pending state. + * * Ordering guarantee: `value` is written before `status` transitions to `'done'`; * `error` is written before `status` transitions to `'error'`. Any reader * observing `status === 'done'` is guaranteed `value` is already present. @@ -65,17 +73,20 @@ export interface TaskLike { export class Task implements TaskLike { readonly id: string; readonly #runFn: (signal: AbortSignal) => Promise; + readonly #externalSignal: AbortSignal | undefined; readonly #abortController = new AbortController(); readonly #signal: AbortSignal; #status: TaskStatus = 'pending'; #value: TValue | undefined = undefined; #error: TError | undefined = undefined; + #promise: Promise | undefined = undefined; constructor(runFn: (signal: AbortSignal) => Promise, config?: TaskConfig) { this.#runFn = runFn; const rawId = config?.id; this.id = typeof rawId === 'function' ? rawId() : (rawId ?? generateId()); + this.#externalSignal = config?.signal; this.#signal = config?.signal ? anyAbortSignal([this.#abortController.signal, config.signal]) : this.#abortController.signal; @@ -93,23 +104,37 @@ export class Task implements TaskLike | undefined; } - async run(): Promise { - this.#status = 'running'; - try { - const result = await this.#runFn(this.#signal); - this.#value = result; // value before status — ordering guarantee - this.#status = 'done'; - return result; - } catch (e) { - this.#error = e as TError; // error before status — ordering guarantee - this.#status = 'error'; - throw e; - } + run(): Promise { + // Memoized: run the work once; repeated calls share the same promise (which + // resolves/rejects immediately once settled). Re-running needs a `clone()`. + this.#promise ??= (async () => { + this.#status = 'running'; + try { + const result = await this.#runFn(this.#signal); + this.#value = result; // value before status — ordering guarantee + this.#status = 'done'; + return result; + } catch (e) { + this.#error = e as TError; // error before status — ordering guarantee + this.#status = 'error'; + throw e; + } + })(); + return this.#promise; } abort(): void { this.#abortController.abort(); } + + /** + * A fresh task with the same work, id, and external signal, in a pending state + * (its own AbortController, no memoized result) — so it can be run again. Used + * to re-run structurally identical work (e.g. `RecurringRunner` reloads). + */ + clone(): Task { + return new Task(this.#runFn, { id: this.id, signal: this.#externalSignal }); + } } // ============================================================================= @@ -285,3 +310,141 @@ export class SerialRunner { this.abortAll(); } } + +// ============================================================================= +// RecurringRunner +// ============================================================================= + +/** + * Decides whether — and *when* — a {@link RecurringRunner} re-runs its task. + * Invoked **concurrently with the run** (so the inter-run interval can be + * measured from when the run *started*, not when it finished), with: + * - `task` — the in-flight run, observable via the memoized `task.run()` (does + * not re-trigger work), e.g. to read its result for a cadence/stop decision. + * - `previous` — the prior successful run's value (`undefined` on the first), + * for decisions that compare consecutive results. + * - `signal` — aborts the wait (and the recurrence). + * + * Resolves `true` to re-run (after whatever delay it owns) or `false` to stop. + * The runner deals only in this awaitable verdict — *how* the delay is produced + * (a timer, a frame, an event) and *when* it's measured from live entirely in + * the reschedule function, so the runner itself knows nothing about time. See + * `delayedReschedule` for the common timer-based, start-anchored implementation. + */ +export type Reschedule = ( + task: TaskLike, + previous: TValue | undefined, + signal: AbortSignal +) => PromiseLike; + +/** + * Runs a task, then re-runs it whenever a {@link Reschedule} function says to, + * until it says stop (or it's aborted) — the recurring sibling of + * {@link ConcurrentRunner} / {@link SerialRunner}, and like them it's handed a + * {@link TaskLike} to run. + * + * The runner has no notion of time: it just awaits whatever `reschedule` + * returns (a promise → re-run when it resolves; `null` → stop). With no + * reschedule it runs the task exactly once (the non-recurring default). + * + * Single-slot, keyed by task **id**: there is always at most one identified + * active task for re-running. Scheduling a task whose id matches the active one + * is a no-op — the existing recurrence keeps running (dedup by id). Scheduling a + * task with a *different* id aborts the prior task's in-flight run and pending + * reschedule, then takes over the slot (abort-and-replace) — the right shape + * when there's one logical unit of recurring work (e.g. reloading the *selected* + * track's media playlist). + * + * Each re-run is a fresh `clone()` of the task (since `Task.run()` is memoized — + * the same instance won't re-execute), carrying the same id so the slot's + * identity is stable across cycles. The run function should read any inputs that + * change between cycles at call time rather than capturing them once. + * `abortAll()` aborts the in-flight task and the pending reschedule; an aborted + * (or stopped) recurrence frees the slot, so a later schedule of the same id + * starts fresh. + */ +export class RecurringRunner { + readonly #reschedule: Reschedule | undefined; + // The identified active task — the one being (re)run. Held across cycles + // (including the inter-cycle wait); null once the recurrence stops/aborts. + #active: TaskLike | null = null; + // Aborts the active recurrence: its in-flight run (via the task) and the + // pending reschedule await (via the signal passed to `reschedule`). + #abort: AbortController | null = null; + #destroyed = false; + + constructor(reschedule?: Reschedule) { + this.#reschedule = reschedule; + } + + schedule(task: TaskLike): void { + if (this.#destroyed) return; + // Dedup by id: this id is already the active re-run target, so the existing + // recurrence continues uninterrupted (don't restart it). + if (this.#active?.id === task.id) return; + // Different id supersedes: abort the prior recurrence, then take over as the + // one identified active task. + this.#cancel(); + this.#active = task; + const ac = new AbortController(); + this.#abort = ac; + void this.#loop(task, ac); + } + + async #loop(task: TaskLike, ac: AbortController): Promise { + const signal = ac.signal; + let current = task; + let previous: TValue | undefined; + while (!signal.aborted) { + let result: TValue | undefined; + let again = false; + try { + // Start the run and the reschedule together: reschedule observes the run + // (via the memoized `task.run()`) and owns the inter-run delay, so the + // interval can be measured from the run's *start*. `Promise.all` waits + // for both — the result (the next cycle's `previous`) and the verdict + + // delay. An errored run resolves to `undefined`, leaving the retry/stop + // choice to reschedule. + [result, again] = await Promise.all([ + current.run().then( + (value) => value, + () => undefined + ), + this.#reschedule ? this.#reschedule(current, previous, signal) : Promise.resolve(false), + ]); + } catch { + return; // reschedule rejected (aborted during its delay) — stop without freeing + } + if (signal.aborted || !again) break; + // Only a successful run advances the comparison baseline. + if (result !== undefined) previous = result; + // Re-run the same work as a fresh task — `run()` is memoized, so re-running + // `current` wouldn't re-execute. The clone keeps the id (stable slot + // identity); track it as active so an abort hits the in-flight instance. + current = current.clone(); + this.#active = current; + } + // Loop ended on its own terms (stop / aborted-but-not-cancelled): free the + // slot iff this loop still owns it (a supersede/abortAll swapped #abort). + if (this.#abort === ac) { + this.#active = null; + this.#abort = null; + } + } + + #cancel(): void { + this.#abort?.abort(); + this.#abort = null; + this.#active?.abort(); + this.#active = null; + } + + abortAll(): void { + this.#cancel(); + } + + destroy(): void { + this.#destroyed = true; + this.abortAll(); + } +} diff --git a/packages/spf/src/core/tasks/tests/delayed-reschedule.test.ts b/packages/spf/src/core/tasks/tests/delayed-reschedule.test.ts new file mode 100644 index 00000000..86bb8223 --- /dev/null +++ b/packages/spf/src/core/tasks/tests/delayed-reschedule.test.ts @@ -0,0 +1,95 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { delayedReschedule } from '../delayed-reschedule'; +import { Task } from '../task'; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('delayedReschedule', () => { + it('observes the run result + previous, then waits the cadence before resolving true', async () => { + vi.useFakeTimers(); + const task = new Task(async () => 5); + const cadence = vi.fn(() => 100); + const reschedule = delayedReschedule(cadence); + + let resolved: boolean | undefined; + const done = reschedule(task, 4, new AbortController().signal).then((v) => { + resolved = v; + }); + + await vi.advanceTimersByTimeAsync(0); // run settles → cadence consulted + expect(cadence).toHaveBeenCalledWith(5, 4); + expect(resolved).toBeUndefined(); // still waiting out the cadence + + await vi.advanceTimersByTimeAsync(100); + await done; + expect(resolved).toBe(true); + }); + + it('start-anchors: subtracts the run elapsed from the cadence', async () => { + vi.useFakeTimers(); + // A run that takes 40ms; cadence 100 → next run ~60ms after the run settles + // (so the interval is 100ms measured from the run's start). + const task = new Task(async () => { + await new Promise((resolve) => setTimeout(resolve, 40)); + return 1; + }); + const reschedule = delayedReschedule(() => 100); + + let resolved = false; + const done = reschedule(task, undefined, new AbortController().signal).then(() => { + resolved = true; + }); + + await vi.advanceTimersByTimeAsync(40); // run completes; 40ms elapsed + await vi.advanceTimersByTimeAsync(59); + expect(resolved).toBe(false); // 99ms from start — not yet + + await vi.advanceTimersByTimeAsync(1); + await done; + expect(resolved).toBe(true); // 100ms from start + }); + + it('resolves false (stop) without waiting when the cadence returns null', async () => { + vi.useFakeTimers(); + const task = new Task(async () => 1); + const reschedule = delayedReschedule(() => null); + + await expect(reschedule(task, undefined, new AbortController().signal)).resolves.toBe(false); + }); + + it('passes undefined to the cadence when the run errors (so it can retry)', async () => { + vi.useFakeTimers(); + const task = new Task(async () => { + throw new Error('boom'); + }); + const cadence = vi.fn(() => 50); // retry on error + const reschedule = delayedReschedule(cadence); + + let resolved: boolean | undefined; + const done = reschedule(task, undefined, new AbortController().signal).then((v) => { + resolved = v; + }); + + await vi.advanceTimersByTimeAsync(0); + expect(cadence).toHaveBeenCalledWith(undefined, undefined); + + await vi.advanceTimersByTimeAsync(50); + await done; + expect(resolved).toBe(true); + }); + + it('rejects when aborted during the wait', async () => { + vi.useFakeTimers(); + const task = new Task(async () => 1); + const reschedule = delayedReschedule(() => 100); + const ac = new AbortController(); + + const done = reschedule(task, undefined, ac.signal); + await vi.advanceTimersByTimeAsync(0); // run settles → into the wait + ac.abort(new DOMException('Aborted', 'AbortError')); + + await expect(done).rejects.toBeInstanceOf(DOMException); + }); +}); diff --git a/packages/spf/src/core/tasks/tests/task.test.ts b/packages/spf/src/core/tasks/tests/task.test.ts index 48460ac5..f53c87f4 100644 --- a/packages/spf/src/core/tasks/tests/task.test.ts +++ b/packages/spf/src/core/tasks/tests/task.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, vi } from 'vitest'; -import { ConcurrentRunner, SerialRunner, Task } from '../task'; +import { ConcurrentRunner, RecurringRunner, type Reschedule, SerialRunner, Task } from '../task'; // ============================================================================= // Task @@ -170,6 +170,66 @@ describe('Task', () => { expect(t1.id).not.toBe(t2.id); }); }); + + describe('memoization', () => { + it('runs the work at most once and shares the result across run() calls', async () => { + const work = vi.fn(async () => 42); + const task = new Task(work); + + const [a, b] = await Promise.all([task.run(), task.run()]); + const c = await task.run(); // after settle + + expect(work).toHaveBeenCalledTimes(1); + expect([a, b, c]).toEqual([42, 42, 42]); + }); + + it('shares the rejection across run() calls', async () => { + const err = new Error('boom'); + const work = vi.fn(async () => { + throw err; + }); + const task = new Task(work); + + await expect(task.run()).rejects.toBe(err); + await expect(task.run()).rejects.toBe(err); + expect(work).toHaveBeenCalledTimes(1); + }); + }); + + describe('clone', () => { + it('produces a fresh, pending task with the same id and work', async () => { + let runs = 0; + const original = new Task(async () => ++runs, { id: 'x' }); + await original.run(); + + const cloned = original.clone(); + expect(cloned).not.toBe(original); + expect(cloned.id).toBe('x'); + expect(cloned.status).toBe('pending'); + + // The clone re-executes the same work (a fresh memoization). + await expect(cloned.run()).resolves.toBe(2); + expect(runs).toBe(2); + }); + + it('gives the clone an independent abort scope', async () => { + const signals: AbortSignal[] = []; + const original = new Task(async (signal) => { + signals.push(signal); + await new Promise((resolve) => setTimeout(resolve, 10)); + }); + + const cloned = original.clone(); + const run = cloned.run(); + cloned.abort(); + await run; + + // Aborting the clone aborts only the clone's signal, not the original's. + original.abort(); + expect(signals).toHaveLength(1); + expect(signals[0]?.aborted).toBe(true); + }); + }); }); // ============================================================================= @@ -573,3 +633,186 @@ describe('SerialRunner', () => { expect(taskSignal?.aborted).toBe(true); }); }); + +describe('RecurringRunner', () => { + /** A reschedule that parks forever, rejecting only when its signal aborts — so a + * recurrence stays "live" (awaiting) until superseded or aborted. */ + const parkUntilAborted: Reschedule = (_task, _previous, signal) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + }); + + /** Flush pending macrotasks so "did NOT happen" assertions are meaningful. */ + const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); + + it('runs the task once when no reschedule is supplied', async () => { + let runs = 0; + const task = new Task(async () => ++runs, { id: 'x' }); + const runner = new RecurringRunner(); + + runner.schedule(task); + await vi.waitFor(() => expect(runs).toBe(1)); + + await flush(); + expect(runs).toBe(1); + }); + + it('re-runs (a clone of) the task while reschedule resolves true, stops on false', async () => { + let runs = 0; + // The runner clones per cycle; the clones share this run fn's `runs` counter. + const task = new Task(async () => ++runs, { id: 'x' }); + const runner = new RecurringRunner(async (t) => (await t.run()) < 3); // continue while < 3 + + runner.schedule(task); + await vi.waitFor(() => expect(runs).toBe(3)); + + await flush(); + expect(runs).toBe(3); + }); + + it('observes the run and receives the previous successful value', async () => { + let n = 0; + const task = new Task(async () => ++n, { id: 'x' }); + const seen: Array<[number, number | undefined]> = []; + const runner = new RecurringRunner(async (t, previous) => { + const current = await t.run(); // observe via the memoized run + seen.push([current, previous]); + return current < 2; + }); + + runner.schedule(task); + await vi.waitFor(() => expect(seen.length).toBe(2)); + + expect(seen).toEqual([ + [1, undefined], + [2, 1], + ]); + }); + + it('keeps the loop alive across a transient error (observed value is undefined)', async () => { + let n = 0; + const task = new Task(async () => { + n += 1; + if (n === 1) throw new Error('boom'); + return n; + }); + // Retry on error (observed value undefined); keep going until a value reaches 3. + const runner = new RecurringRunner(async (t) => { + const current = await t.run().catch(() => undefined); + return current === undefined || current < 3; + }); + + runner.schedule(task); + await vi.waitFor(() => expect(n).toBe(3)); + + runner.destroy(); + }); + + it('aborts an in-flight run when superseded by a new id', async () => { + let aborted = false; + const slow = new Task( + (signal) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { + aborted = true; + reject(new DOMException('Aborted', 'AbortError')); + }); + }), + { id: 'a' } + ); + let ranB = false; + const taskB = new Task( + async () => { + ranB = true; + return 1; + }, + { id: 'b' } + ); + const runner = new RecurringRunner(parkUntilAborted); + + runner.schedule(slow); // parks mid-run, listening for abort + runner.schedule(taskB); // new id → abort slow's in-flight run, run B + expect(aborted).toBe(true); + + await vi.waitFor(() => expect(ranB).toBe(true)); + runner.destroy(); + }); + + it('ignores a schedule with the same id — the existing recurrence keeps running', async () => { + let runsA = 0; + const taskA = new Task(async () => ++runsA, { id: 'x' }); + let runsB = 0; + const taskB = new Task(async () => ++runsB, { id: 'x' }); // same id + const runner = new RecurringRunner(parkUntilAborted); + + runner.schedule(taskA); + await vi.waitFor(() => expect(runsA).toBe(1)); // A ran once, parked + + runner.schedule(taskB); // same id while A is live → ignored + await flush(); + expect(runsB).toBe(0); + expect(runsA).toBe(1); + + runner.destroy(); + }); + + it('a new id aborts the prior recurrence and takes over the slot', async () => { + let runsA = 0; + const taskA = new Task(async () => ++runsA, { id: 'a' }); + let runsB = 0; + const taskB = new Task(async () => ++runsB, { id: 'b' }); + const runner = new RecurringRunner(parkUntilAborted); + + runner.schedule(taskA); + await vi.waitFor(() => expect(runsA).toBe(1)); // A parked + + runner.schedule(taskB); // new id → abort A, run B + await vi.waitFor(() => expect(runsB).toBe(1)); + + await flush(); + expect(runsA).toBe(1); // A did not re-run + + runner.destroy(); + }); + + it('frees the slot when a recurrence stops, so the same id can start fresh', async () => { + let runs = 0; + // Fresh instances per schedule (the real pattern — callers build a new task + // each time); a shared counter observes runs across both. + const make = () => new Task(async () => ++runs, { id: 'x' }); + const runner = new RecurringRunner(async () => false); // stop after first run + + runner.schedule(make()); + await vi.waitFor(() => expect(runs).toBe(1)); + await flush(); // let the loop run reschedule → false → free the slot + + runner.schedule(make()); // not deduped (recurrence ended) → runs fresh + await vi.waitFor(() => expect(runs).toBe(2)); + + runner.destroy(); + }); + + it('abortAll stops the recurrence; no re-run', async () => { + let runs = 0; + const task = new Task(async () => ++runs, { id: 'x' }); + const runner = new RecurringRunner(parkUntilAborted); + + runner.schedule(task); + await vi.waitFor(() => expect(runs).toBe(1)); + + runner.abortAll(); + await flush(); + expect(runs).toBe(1); + }); + + it('does not run after destroy', async () => { + let runs = 0; + const task = new Task(async () => ++runs, { id: 'x' }); + const runner = new RecurringRunner(async () => true); + + runner.destroy(); + runner.schedule(task); + await flush(); + expect(runs).toBe(0); + }); +}); diff --git a/packages/spf/src/media/hls/reload-policy.ts b/packages/spf/src/media/hls/reload-policy.ts new file mode 100644 index 00000000..b5bb32ce --- /dev/null +++ b/packages/spf/src/media/hls/reload-policy.ts @@ -0,0 +1,44 @@ +import { getMediaPlaylistMetadata, type ResolvedTrack } from '../types'; + +/** Reload cadence when a playlist carries no usable target duration. */ +const FALLBACK_TARGET_DURATION = 6; + +/** Identity of a reload snapshot — window position + length. Changes when the window slid or grew. */ +function snapshotSignature(track: ResolvedTrack): string { + return `${getMediaPlaylistMetadata(track)?.mediaSequence ?? 0}:${track.segments.length}`; +} + +function targetDurationOf(track: ResolvedTrack | undefined): number { + return (track && getMediaPlaylistMetadata(track)?.targetDuration) || FALLBACK_TARGET_DURATION; +} + +/** + * Live media-playlist reload cadence, per RFC 8216bis §6.3.4 — a + * {@link RecurrencePolicy} for a `RecurringRunner` re-resolving the selected + * track. Structurally matches `RecurrencePolicy` without + * importing it (media stays core-free): `current` is the freshly resolved track + * (`undefined` if the reload errored), `previous` the prior resolved snapshot. + * + * - Complete playlist (VoD, or live that hit `#EXT-X-ENDLIST`) → `null`: stop. + * Keys off `Track.duration` (finite once complete), the single completeness + * source of truth. + * - Errored reload (`current` undefined) → retry at the last-known target- + * duration cadence (fallback when unknown), keeping the loop alive across + * transient fetch failures. + * - Unchanged window (same media sequence + segment count as `previous`) → poll + * at half the target duration; a moved/grown window (or the first reload) → + * full target duration. + * + * Returned delays are milliseconds. + */ +export function mediaPlaylistReloadDelay( + current: ResolvedTrack | undefined, + previous: ResolvedTrack | undefined +): number | null { + if (!current) return targetDurationOf(previous) * 1000; + if (Number.isFinite(current.duration)) return null; + + const target = targetDurationOf(current); + const changed = !previous || snapshotSignature(current) !== snapshotSignature(previous); + return (changed ? target : target / 2) * 1000; +} diff --git a/packages/spf/src/media/hls/tests/reload-policy.test.ts b/packages/spf/src/media/hls/tests/reload-policy.test.ts new file mode 100644 index 00000000..d87dca87 --- /dev/null +++ b/packages/spf/src/media/hls/tests/reload-policy.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, it } from 'vitest'; +import { MEDIA_PLAYLIST_METADATA_KEY, type ResolvedTrack } from '../../types'; +import { mediaPlaylistReloadDelay } from '../reload-policy'; + +/** Minimal resolved-track stand-in carrying only what the policy reads. */ +function track(opts: { + duration?: number; + targetDuration?: number; + mediaSequence?: number; + segments?: number; +}): ResolvedTrack { + return { + duration: opts.duration ?? Number.POSITIVE_INFINITY, + segments: Array.from({ length: opts.segments ?? 1 }), + metadata: { + [MEDIA_PLAYLIST_METADATA_KEY]: { + targetDuration: opts.targetDuration ?? 4, + mediaSequence: opts.mediaSequence ?? 0, + endList: Number.isFinite(opts.duration ?? Number.POSITIVE_INFINITY), + }, + }, + } as unknown as ResolvedTrack; +} + +describe('mediaPlaylistReloadDelay', () => { + it('stops (null) once the playlist is complete (finite duration)', () => { + expect(mediaPlaylistReloadDelay(track({ duration: 30 }), undefined)).toBeNull(); + }); + + it('polls at full target duration on the first reload of a live window', () => { + expect(mediaPlaylistReloadDelay(track({ targetDuration: 4 }), undefined)).toBe(4000); + }); + + it('polls at half target duration when the window is unchanged', () => { + const prev = track({ targetDuration: 4, mediaSequence: 10, segments: 3 }); + const same = track({ targetDuration: 4, mediaSequence: 10, segments: 3 }); + expect(mediaPlaylistReloadDelay(same, prev)).toBe(2000); + }); + + it('polls at full target duration when the window slid or grew', () => { + const prev = track({ targetDuration: 4, mediaSequence: 10, segments: 3 }); + const slid = track({ targetDuration: 4, mediaSequence: 11, segments: 3 }); + expect(mediaPlaylistReloadDelay(slid, prev)).toBe(4000); + }); + + it('retries at the last-known cadence when a reload errors (current undefined)', () => { + const prev = track({ targetDuration: 8 }); + expect(mediaPlaylistReloadDelay(undefined, prev)).toBe(8000); + }); + + it('falls back to 6s when no target duration / prior snapshot is available', () => { + expect(mediaPlaylistReloadDelay(undefined, undefined)).toBe(6000); + }); +}); diff --git a/packages/spf/src/playback/behaviors/resolve-track.ts b/packages/spf/src/playback/behaviors/resolve-track.ts index f8ae30ae..068aa020 100644 --- a/packages/spf/src/playback/behaviors/resolve-track.ts +++ b/packages/spf/src/playback/behaviors/resolve-track.ts @@ -1,7 +1,7 @@ -import { type AnySlotMap, defineBehavior } from '../../core/composition/create-composition'; +import { defineBehavior } from '../../core/composition/create-composition'; import { createMachineReactor } from '../../core/reactors/create-machine-reactor'; import { computed, peek, type ReadonlySignal, type Signal, update } from '../../core/signals/primitives'; -import { ConcurrentRunner, Task } from '../../core/tasks/task'; +import { RecurringRunner, type Reschedule, Task } from '../../core/tasks/task'; import { NON_FMP4_CONTAINER_MIMES, parseMediaPlaylist } from '../../media/hls/parse-media-playlist'; import type { MaybeResolvedPresentation, PartiallyResolvedTrack, ResolvedTrack } from '../../media/types'; import { deriveStreamType, getMediaPlaylistMetadata, isResolvedPresentation, isResolvedTrack } from '../../media/types'; @@ -17,27 +17,27 @@ import { AUDIO_TYPE_CONFIG, TEXT_TYPE_CONFIG, VIDEO_TYPE_CONFIG } from '../primi // `setupTrackResolution` has the same shape as a Behavior `setup` function: // `({ state, config }) => cleanup`. Each `resolveXTrack` export below calls it // from inside its own `defineBehavior` setup, supplying its per-type config -// inline. The orchestration — gate on a selection, decide whether a (re)load -// is due, schedule the fetch+parse, and patch the resolved track back into -// `state.presentation` (carrying the prior snapshot's timeline forward) — is -// shared. +// inline. The orchestration — gate on a selection, schedule the fetch+parse, +// and patch the resolved track back into `state.presentation` (carrying the +// prior snapshot's timeline forward) — is shared. // // This behavior is category [1] "content snapshot" from // [live-presentation-modeling.md](../../../internal/design/spf/live-presentation-modeling.md): -// it produces the windowed segment list. *Whether* a (re)load is due is decided -// by an injected `shouldLoadTrack(track, params)` gate, so the core stays -// live-agnostic — it knows nothing about reload epochs or completeness. The -// default (`loadIfUnresolved`) resolves only an unresolved track (the pre-live -// one-shot). Live-capable variants inject `shouldLoadLiveTrack`, which also -// reloads a resolved-but-incomplete window and subscribes the effect to the -// scheduler's per-type reload-epoch *ping* (read for subscription only — see -// `scheduleTrackReload`, category [3] "refetch policy"). The ping slot is read -// defensively inside that gate, off an optional state view, so the core never -// names or assumes it (the `bandwidthState?` pattern from `track-switching`). +// it produces the windowed segment list. *Whether* to (re)load is the `shouldLoadTrack` +// gate (resolve an unresolved track; reload a resolved-but-incomplete one — a live +// window may have slid past the playhead; reuse a complete one). *When* to reload +// (category [3] "refetch policy") is owned not by this behavior but by the +// `RecurringRunner` it schedules on: given a `reschedule` function (injected by +// the engine, which composes the playlist's target-duration cadence with a +// sleep), the runner re-runs the resolve task whenever `reschedule` resolves, +// until it returns `null` (the playlist completed). With no `reschedule` it runs +// once (VoD / non-live). The recurrence loop lives in the runner, which knows +// nothing about time — so the behavior stays free of timers and reload signals. // -// A same-id task already in flight is deduped by `ConcurrentRunner` -// (drop-if-busy), and the post-resolve presentation write is read with `peek`, -// so it never re-fires the effect. +// The runner is single-slot: a selection change re-schedules (abort-and-replace), +// and the reactor's `presentation-resolved` exit aborts it on source change. The +// post-resolve presentation write is read with `peek`, so it never re-fires the +// effect. // ============================================================================ /** @@ -53,7 +53,6 @@ export interface ResolveTrackState { } type SelectedTrackKey = 'selectedVideoTrackId' | 'selectedAudioTrackId' | 'selectedTextTrackId'; -type ReloadEpochKey = 'videoReloadEpoch' | 'audioReloadEpoch' | 'textReloadEpoch'; type ResolveTrackStateMap = { presentation: Signal; @@ -68,115 +67,109 @@ interface TrackResolutionConfig { /** Fetch a track's media-playlist text — already failover-decorated by the behavior. */ fetchResolvableText?: FetchText; /** - * Load gate — decides whether to (re)resolve the selected track. Defaults to - * `loadIfUnresolved` (initial resolve only). Live-capable variants inject - * `shouldLoadLiveTrack`, which also reloads incomplete windows and subscribes - * to the scheduler's reload-epoch ping. Receives the behavior's `params` - * (`{ state, context, config }`) untouched. + * Re-run policy handed to the `RecurringRunner`: returns a promise that resolves + * when the live track's playlist should reload, or `null` to stop. Absent → + * resolve once (VoD / non-live). Injected by the engine (which composes the + * target-duration cadence with a sleep); the behavior stays free of timers and + * reload signals. */ - shouldLoadTrack?: ShouldLoadTrack, AnySlotMap, TrackResolutionConfig>; + reschedule?: Reschedule; } /** - * Engine-config slice each `resolve*` behavior reads to build its failover- - * decorated playlist fetch. + * Engine-config slice each `resolve*` behavior reads. */ interface ResolveTrackConfig { /** CDN-id derivation for the failover trip; defaults to origin-based `getCdnId`. */ getCdnId?: GetCdnId; + /** Live media-playlist re-run policy; absent → resolve once. */ + reschedule?: Reschedule; } /** - * The behavior's setup deps, threaded straight through to `shouldLoadTrack` so a - * gate reads from the same surfaces the behavior does. `context` is optional — - * present at runtime, absent on direct setup calls and unread by today's gates, - * so the whole object passes through unchanged. + * Whether the loader should (re)load this track now: yes if it's unresolved (the + * initial resolve, or a retry of a failed one), or resolved-but-incomplete (a + * live window that may have slid past the playhead — reuse risks a stall); no if + * resolved + complete (VoD, or live that hit `#EXT-X-ENDLIST` — a complete + * playlist can never go stale). Completeness keys off `Track.duration`, the + * single completeness source of truth. This gate governs only whether to + * (re)*start* loading; the live reload *cadence* is the `RecurringRunner`'s job. */ -export interface ShouldLoadTrackParams { - state: State; - context?: Context; - config: Config; -} - -/** - * Decides whether the loader should (re)resolve the selected track now, reading - * whatever signals it needs off `params` at call time (so its `.get()`s - * subscribe the loader's effect to exactly what it consulted). Returning `true` - * schedules a fetch+parse; an in-flight same-id task is deduped by the runner. - */ -export type ShouldLoadTrack = ( - track: PartiallyResolvedTrack | ResolvedTrack, - params: ShouldLoadTrackParams -) => boolean; - -/** - * Default load gate: resolve only an unresolved track (the initial resolve, or a - * retry of a failed one). The pre-live one-shot behavior — it names no - * reload-epoch slot, so the core assumes no live ping. Live-capable variants - * inject `shouldLoadLiveTrack` instead. - */ -function loadIfUnresolved(track: PartiallyResolvedTrack | ResolvedTrack): boolean { - return !isResolvedTrack(track); -} - -/** - * Per-type reload-epoch ping slots, all *optional* — the live gate reads its - * slot defensively, so the core's state type (which omits them) stays - * assignable and the core never assumes the ping exists. Materialized by - * `scheduleTrackReload` when it's composed; absent otherwise. - */ -type ReloadEpochStateView = { [P in ReloadEpochKey]?: ReadonlySignal }; - -/** - * Live load gate — the `shouldLoadTrack` alternative the live-capable variants - * inject. Beyond the default's initial resolve, it reloads a resolved-but- - * incomplete window (Infinity `Track.duration`, the single completeness source - * of truth): a live window may have slid past the playhead, so reuse risks a - * stall; a complete playlist (VoD, or live that hit `#EXT-X-ENDLIST`) can never - * go stale, so it's reused. Reading the per-type reload-epoch slot subscribes - * the loader's effect to the scheduler's cadence ping (each bump re-fires it); - * the value is unused — a signal used as an event channel. The slot is read - * defensively, off the optional `ReloadEpochStateView`, with its per-type key - * closured in by the variant — so only this gate names it, never the core. - */ -function shouldLoadLiveTrack( - track: PartiallyResolvedTrack | ResolvedTrack, - params: ShouldLoadTrackParams & ReloadEpochStateView>, - reloadEpochKey: ReloadEpochKey -): boolean { - params.state[reloadEpochKey]?.get(); +function shouldLoadTrack(track: PartiallyResolvedTrack | ResolvedTrack): boolean { return !isResolvedTrack(track) || !Number.isFinite(track.duration); } -function setupTrackResolution(params: { +function setupTrackResolution({ + state, + config: { selectedKey, findTrackToResolve, fetchResolvableText = defaultFetchResolvableText, reschedule }, +}: { state: ResolveTrackStateMap; - context?: AnySlotMap; config: TrackResolutionConfig; }) { - // Destructure in the body (not the signature) so the whole `params` object — - // `{ state, context, config }` — passes straight through to `shouldLoadTrack`, - // letting a gate reach `context` in future without changing this seam. - const { state, config } = params; - const { - selectedKey, - findTrackToResolve, - fetchResolvableText = defaultFetchResolvableText, - shouldLoadTrack = loadIfUnresolved, - } = config; - // NOTE: This can/maybe will be pulled into a per-use case factory (e.g. something like createTaskRunner() with args TBD), - // likely eventually passed down via config or a new "definitions" argument. This will allow us to decide if we want our task runner/scheduler - // to e.g. run concurrently (like we currently are), serially with a queue, or abort the previous task and replace it with the newly scheduled one. (CJP). - const runner = new ConcurrentRunner(); + // The runner owns recurrence: with a `reschedule` (live) it re-runs the resolve + // task whenever reschedule resolves, until it returns null; with none (VoD) it + // runs once. Single-slot — a selection change re-schedules (abort-and-replace), + // and the reactor's state-exit aborts it on source change. + const runner = new RecurringRunner(reschedule); + + // The resolve task for `trackId`. The `RecurringRunner` re-runs this same task + // each reload cycle, so its run fn re-reads the current snapshot each time — + // carrying the prior window's timeline forward — then fetches+parses, patches + // `state.presentation`, and returns the resolved track (the runner's + // `reschedule` reads its metadata to decide the next cadence). + const createResolveTask = (trackId: string): Task => + new Task( + async (signal) => { + const presentation = peek(state.presentation); + const track = presentation ? findTrackToResolve(presentation, trackId) : undefined; + // The recurrence is aborted on source change, so a missing track here is + // a transient race; surfacing it as an error lets the policy retry. + if (!track) throw new Error('resolve-track: selected track not found'); + + // `fetchResolvableText` is the behavior's failover-decorated fetch: it + // trips the CDN on a failed fetch (network error or non-OK status). A + // parse failure is a content issue, not a CDN-availability one, so it + // doesn't trip. `track` is the prior snapshot (the unresolved shell on the + // first pass, the last resolved window on a live reload); the parser + // carries its timeline forward. + const text = await fetchResolvableText(track, { signal }); + const mediaTrack = parseMediaPlaylist(text, track); + + // Updater handles undefined inputs by returning current unchanged; + // isResolvedPresentation narrows for the patch. State-exit on + // resolving→unresolved fires runner.abortAll before any URL change + // settles, and per the Fetch spec the signal abort cancels in-flight body + // reads — so by the time we reach this point the presentation we resolved + // against is the live one. + update(state.presentation, (current) => { + if (!isResolvedPresentation(current)) return current; + const patched = updateTrackInPresentation(current, mediaTrack); + // Container is uniform within a type (an ABR ladder shares its + // container), so a detected non-fMP4 rendition (TS, raw AAC) implies + // every rendition of *this* type matches — relabel them all from one + // resolved playlist instead of fetching each. Scoped to this track's own + // type: never cross audio↔video (mixed-container sources exist, e.g. + // muxed-TS video + raw-.aac audio), which also keeps per-type + // resolutions' writes disjoint (no race). + const relabeled = NON_FMP4_CONTAINER_MIMES.has(mediaTrack.mimeType) + ? applyContainerMimeType(patched, mediaTrack.type, mediaTrack.mimeType) + : patched; + // Stream nature (category [2a]) — stable once a media playlist is + // parsed; recomputing each reload is harmless. + return { ...relabeled, streamType: deriveStreamType(getMediaPlaylistMetadata(mediaTrack)) }; + }); + return mediaTrack; + }, + { id: trackId } + ); // Reactor states model the FSM the previous effect-based body was // hand-rolling. 'presentation-resolved' is entered when the // presentation is fully parsed (has a Ham id + selectionSets); leaving - // it (presentation cleared or reset to an unresolved value) aborts all - // in-flight tasks via the entry-cleanup. Most URL changes go through - // 'presentation-unresolved' naturally (set undefined → set new partial - // → re-parse), so the common case is covered by state-exit alone; the - // task body's commit-time id check covers the pathological - // resolved→resolved-without-unresolved transition. + // it (presentation cleared or reset to an unresolved value) aborts the + // in-flight + scheduled reload via the entry-cleanup. Most URL changes go + // through 'presentation-unresolved' naturally (set undefined → set new + // partial → re-parse), so the common case is covered by state-exit alone. const derivedStateSignal = computed(() => isResolvedPresentation(state.presentation.get()) ? ('presentation-resolved' as const) @@ -191,75 +184,28 @@ function setupTrackResolution(params: { 'presentation-resolved': { // `entry` runs on state entry; the function it returns is the // state-exit cleanup. Returning `() => runner.abortAll()` binds - // abort-of-in-flight-resolutions to leaving 'presentation-resolved' - // (presentation cleared/reset, or behavior destroyed) — - // source-change cancellation expressed structurally through the - // state machine. + // abort-of-in-flight-resolution (and any pending reload) to leaving + // 'presentation-resolved' (presentation cleared/reset, or behavior + // destroyed) — source-change cancellation expressed structurally + // through the state machine. entry: () => () => runner.abortAll(), effects: [ () => { - // The reactor's state transitions handle relevant presentation - // changes (presentation-resolved ↔ presentation-unresolved); - // within 'presentation-resolved' we peek (untracked read) so - // internal updates (segments added by sibling tasks) don't re-fire - // the effect. `selectedKey` is read tracked up front so a selection - // change re-fires; the injected `shouldLoadTrack` gate takes any - // further tracked reads it needs (e.g. the live reload-epoch ping), - // subscribing the effect to exactly what it consulted. + // `selectedKey` is read tracked so a selection change re-fires this + // and re-schedules (the single-slot runner aborts+replaces the prior + // recurrence). Presentation is peeked (untracked) so the post-resolve + // write — and the runner's own reload cycles — don't re-fire it. const trackId = state[selectedKey].get(); const presentation = peek(state.presentation); if (!presentation || !trackId) return; const track = findTrackToResolve(presentation, trackId); - // The gate decides whether a (re)load is due (default: unresolved - // only; live: also incomplete-window reload). A same-id task already - // in flight is deduped by the runner (drop-if-busy). - if (!track || !shouldLoadTrack(track, params)) return; + // Gate: schedule only when there's loading to do (unresolved, or a + // resolved-but-incomplete live window). Recurrence past the first run + // is the runner's job, driven by `reschedule`. + if (!track || !shouldLoadTrack(track)) return; - runner.schedule( - // NOTE: This can/maybe will be pulled into a per-use case factory (e.g. something like createResolveTrackTask(track, context, config)), - // likely eventually passed down via config or a new "definitions" argument (CJP). - new Task( - async (signal) => { - // `fetchResolvableText` is the behavior's failover-decorated - // fetch: it trips the CDN on a failed fetch (network error or - // non-OK status). A parse failure is a content issue, not a - // CDN-availability one, so it doesn't trip. - // `track` is the prior snapshot (the unresolved shell on the - // first pass, the last resolved window on a live reload); the - // parser carries its timeline forward. - const text = await fetchResolvableText(track, { signal }); - const mediaTrack = parseMediaPlaylist(text, track); - - // Updater handles undefined inputs by returning current - // unchanged; isResolvedPresentation narrows for the patch. - // State-exit on resolving→unresolved fires runner.abortAll - // before any URL change settles, and per the Fetch spec the - // signal abort cancels in-flight body reads — so by the - // time we reach this point the presentation we resolved - // against is the live one. - update(state.presentation, (current) => { - if (!isResolvedPresentation(current)) return current; - const patched = updateTrackInPresentation(current, mediaTrack); - // Container is uniform within a type (an ABR ladder shares - // its container), so a detected non-fMP4 rendition (TS, - // raw AAC) implies every rendition of *this* type matches — - // relabel them all from one resolved playlist instead of - // fetching each. Scoped to this track's own type: never cross - // audio↔video (mixed-container sources exist, e.g. muxed-TS - // video + raw-.aac audio), which also keeps per-type - // resolutions' writes disjoint (no race). - const relabeled = NON_FMP4_CONTAINER_MIMES.has(mediaTrack.mimeType) - ? applyContainerMimeType(patched, mediaTrack.type, mediaTrack.mimeType) - : patched; - // Stream nature (category [2a]) — stable once a media - // playlist is parsed; recomputing each reload is harmless. - return { ...relabeled, streamType: deriveStreamType(getMediaPlaylistMetadata(mediaTrack)) }; - }); - }, - { id: track.id } - ) - ); + runner.schedule(createResolveTask(trackId)); }, ], }, @@ -304,27 +250,20 @@ export const resolveVideoTrack = defineBehavior({ setup: ({ state, config = {}, - ...otherProps }: { state: ResolveTrackStateMap<'selectedVideoTrackId'>; config?: ResolveTrackConfig; }) => { // Engine `config` layers over the per-type defaults (mirrors the other // per-type variants, see track-types.ts); `failoverFetch` reads its - // `selectedKey` + `getCdnId` from the merged result. `fetchResolvableText` - // is then placed AFTER the spread so the failover-decorated fetch wins — - // unlike segments, playlists expose no overridable per-type fetch. + // `selectedKey` + `getCdnId` from the merged result, and `reschedule` + // rides through to the runner. `fetchResolvableText` is placed AFTER the + // spread so the failover-decorated fetch wins — unlike segments, playlists + // expose no overridable per-type fetch. const trackConfig = { ...VIDEO_TRACK_RESOLUTION_CONFIG, ...config }; return setupTrackResolution({ - ...otherProps, state, - config: { - ...trackConfig, - fetchResolvableText: failoverFetch(defaultFetchResolvableText, state, trackConfig), - // Live-capable gate: closure in this type's reload-epoch slot so the - // core never names it. - shouldLoadTrack: (track, params) => shouldLoadLiveTrack(track, params, 'videoReloadEpoch'), - }, + config: { ...trackConfig, fetchResolvableText: failoverFetch(defaultFetchResolvableText, state, trackConfig) }, }); }, }); @@ -339,7 +278,6 @@ export const resolveAudioTrack = defineBehavior({ setup: ({ state, config = {}, - ...otherProps }: { state: ResolveTrackStateMap<'selectedAudioTrackId'>; config?: ResolveTrackConfig; @@ -347,13 +285,8 @@ export const resolveAudioTrack = defineBehavior({ // Key order is load-bearing — see resolveVideoTrack. const trackConfig = { ...AUDIO_TRACK_RESOLUTION_CONFIG, ...config }; return setupTrackResolution({ - ...otherProps, state, - config: { - ...trackConfig, - fetchResolvableText: failoverFetch(defaultFetchResolvableText, state, trackConfig), - shouldLoadTrack: (track, params) => shouldLoadLiveTrack(track, params, 'audioReloadEpoch'), - }, + config: { ...trackConfig, fetchResolvableText: failoverFetch(defaultFetchResolvableText, state, trackConfig) }, }); }, }); @@ -368,7 +301,6 @@ export const resolveTextTrack = defineBehavior({ setup: ({ state, config = {}, - ...otherProps }: { state: ResolveTrackStateMap<'selectedTextTrackId'>; config?: ResolveTrackConfig; @@ -376,13 +308,8 @@ export const resolveTextTrack = defineBehavior({ // Key order is load-bearing — see resolveVideoTrack. const trackConfig = { ...TEXT_TRACK_RESOLUTION_CONFIG, ...config }; return setupTrackResolution({ - ...otherProps, state, - config: { - ...trackConfig, - fetchResolvableText: failoverFetch(defaultFetchResolvableText, state, trackConfig), - shouldLoadTrack: (track, params) => shouldLoadLiveTrack(track, params, 'textReloadEpoch'), - }, + config: { ...trackConfig, fetchResolvableText: failoverFetch(defaultFetchResolvableText, state, trackConfig) }, }); }, }); diff --git a/packages/spf/src/playback/behaviors/schedule-track-reload.ts b/packages/spf/src/playback/behaviors/schedule-track-reload.ts deleted file mode 100644 index 03ce0056..00000000 --- a/packages/spf/src/playback/behaviors/schedule-track-reload.ts +++ /dev/null @@ -1,198 +0,0 @@ -/** - * Live media-playlist reload *scheduling*, per track type. - * - * Category [3] "refetch policy" from - * [live-presentation-modeling.md](../../../internal/design/spf/live-presentation-modeling.md): - * decides *when* a track's media playlist should be re-fetched, without doing - * any fetching itself. Once the presentation is resolved and a track of this - * type is selected, it bumps a per-type reload-epoch slot on a target-duration - * cadence (half that when the last reload was unchanged, per RFC 8216bis - * §6.3.4); the sibling `resolveTrack` loader (category [1]) watches that slot - * and performs the actual fetch+parse+merge. - * - * The split keeps "when to refetch" and "what segments are in the playlist" - * — categories that change at different rates — in separate behaviors. - * - * Inert for VoD: it stays `idle` once the resolved track reports - * `#EXT-X-ENDLIST` (a complete playlist never reloads), so no scheduler-aware - * engine config is needed. It enters on track *selection* (not resolution) so - * its bumps also drive retries of a failed/slow first resolve until the loader - * succeeds. - * - * Specialized per type via the shared `setupTrackReloadSchedule` (mirrors - * `resolve-track`): `scheduleVideoTrackReload` / `scheduleAudioTrackReload` / - * `scheduleTextTrackReload`, each gating on its own `selected*TrackId` + track - * type, so demuxed audio and video reload independently. - * - * Limitations (intentional, for now): selection is read once at loop start - * (mid-stream track switching isn't encoded in the reactor's two states). - */ -import { defineBehavior } from '../../core/composition/create-composition'; -import { createMachineReactor } from '../../core/reactors/create-machine-reactor'; -import { computed, peek, type ReadonlySignal, type Signal, update } from '../../core/signals/primitives'; -import { - getMediaPlaylistMetadata, - isResolvedPresentation, - isResolvedTrack, - type MaybeResolvedPresentation, - type ResolvedTrack, - type TrackType, -} from '../../media/types'; -import { findTrack } from '../../media/utils/tracks'; -import { AUDIO_TYPE_CONFIG, TEXT_TYPE_CONFIG, VIDEO_TYPE_CONFIG } from '../primitives/track-types'; - -export interface ScheduleTrackReloadState { - presentation?: MaybeResolvedPresentation; - selectedVideoTrackId?: string; - selectedAudioTrackId?: string; - selectedTextTrackId?: string; - videoReloadEpoch?: number; - audioReloadEpoch?: number; - textReloadEpoch?: number; -} - -type SelectedTrackKey = 'selectedVideoTrackId' | 'selectedAudioTrackId' | 'selectedTextTrackId'; -type ReloadEpochKey = 'videoReloadEpoch' | 'audioReloadEpoch' | 'textReloadEpoch'; -type ScheduleStateName = 'idle' | 'scheduling'; - -type ScheduleTrackReloadStateMap = { - presentation: ReadonlySignal; -} & { [P in K]: ReadonlySignal } & { [P in E]: Signal }; - -interface TrackReloadScheduleConfig { - type: TrackType; - selectedKey: K; - reloadEpochKey: E; -} - -/** Fallback reload cadence when the playlist carries no usable target duration. */ -const FALLBACK_TARGET_DURATION = 6; - -function sleep(ms: number, signal: AbortSignal): Promise { - return new Promise((resolve, reject) => { - if (signal.aborted) { - reject(new DOMException('Aborted', 'AbortError')); - return; - } - const timer = setTimeout(resolve, ms); - signal.addEventListener( - 'abort', - () => { - clearTimeout(timer); - reject(new DOMException('Aborted', 'AbortError')); - }, - { once: true } - ); - }); -} - -/** Identity of a reload snapshot — window position + length. Changes when the window slid or grew. */ -function snapshotSignature(track: ResolvedTrack): string { - return `${getMediaPlaylistMetadata(track)?.mediaSequence ?? 0}:${track.segments.length}`; -} - -function setupTrackReloadSchedule({ - state, - config: { type, selectedKey, reloadEpochKey }, -}: { - state: ScheduleTrackReloadStateMap; - config: TrackReloadScheduleConfig; -}) { - const derivedStateSignal = computed(() => { - const presentation = state.presentation.get(); - const trackId = state[selectedKey].get(); - if (!isResolvedPresentation(presentation) || !trackId) return 'idle'; - const track = findTrack(presentation, type, trackId); - if (!track) return 'idle'; - // A complete playlist (VoD, or live that has ended) never reloads. An - // unresolved track keeps us scheduling so the loader's first resolve is - // retried via the epoch bumps. - if (isResolvedTrack(track) && getMediaPlaylistMetadata(track)?.endList) return 'idle'; - return 'scheduling'; - }); - - return createMachineReactor({ - initial: 'idle', - monitor: () => derivedStateSignal.get(), - states: { - idle: {}, - scheduling: { - entry: () => { - const ac = new AbortController(); - const trackId = state[selectedKey].get()!; - - void (async () => { - // Signature of the snapshot seen at the previous iteration, to - // detect whether the last reload changed the window. - let lastSignature: string | null = null; - - while (!ac.signal.aborted) { - const presentation = peek(state.presentation); - const track = isResolvedPresentation(presentation) ? findTrack(presentation, type, trackId) : undefined; - const meta = track && isResolvedTrack(track) ? getMediaPlaylistMetadata(track) : undefined; - if (meta?.endList) break; - - const signature = track && isResolvedTrack(track) ? snapshotSignature(track) : null; - // First pass (no baseline) or a moved window counts as changed → - // full cadence; an unchanged window polls at half cadence. - const changed = signature === null || signature !== lastSignature; - lastSignature = signature; - - const target = meta?.targetDuration || FALLBACK_TARGET_DURATION; - try { - await sleep((changed ? target : target / 2) * 1000, ac.signal); - } catch { - return; // aborted during the wait - } - - update(state[reloadEpochKey], (epoch) => (epoch ?? 0) + 1); - } - })(); - - // State-exit (source change / destroy / endList) aborts the loop. - return () => ac.abort(); - }, - }, - }, - }); -} - -const VIDEO_RELOAD_SCHEDULE_CONFIG = { - type: VIDEO_TYPE_CONFIG.type, - selectedKey: VIDEO_TYPE_CONFIG.selectedKey, - reloadEpochKey: 'videoReloadEpoch', -} as const; -const AUDIO_RELOAD_SCHEDULE_CONFIG = { - type: AUDIO_TYPE_CONFIG.type, - selectedKey: AUDIO_TYPE_CONFIG.selectedKey, - reloadEpochKey: 'audioReloadEpoch', -} as const; -const TEXT_RELOAD_SCHEDULE_CONFIG = { - type: TEXT_TYPE_CONFIG.type, - selectedKey: TEXT_TYPE_CONFIG.selectedKey, - reloadEpochKey: 'textReloadEpoch', -} as const; - -/** Schedule live reloads of the selected video track's media playlist. */ -export const scheduleVideoTrackReload = defineBehavior({ - stateKeys: ['presentation', 'selectedVideoTrackId', 'videoReloadEpoch'], - contextKeys: [], - setup: ({ state }: { state: ScheduleTrackReloadStateMap<'selectedVideoTrackId', 'videoReloadEpoch'> }) => - setupTrackReloadSchedule({ state, config: VIDEO_RELOAD_SCHEDULE_CONFIG }), -}); - -/** Schedule live reloads of the selected audio track's media playlist (demuxed audio). */ -export const scheduleAudioTrackReload = defineBehavior({ - stateKeys: ['presentation', 'selectedAudioTrackId', 'audioReloadEpoch'], - contextKeys: [], - setup: ({ state }: { state: ScheduleTrackReloadStateMap<'selectedAudioTrackId', 'audioReloadEpoch'> }) => - setupTrackReloadSchedule({ state, config: AUDIO_RELOAD_SCHEDULE_CONFIG }), -}); - -/** Schedule live reloads of the selected text track's media playlist (live captions). */ -export const scheduleTextTrackReload = defineBehavior({ - stateKeys: ['presentation', 'selectedTextTrackId', 'textReloadEpoch'], - contextKeys: [], - setup: ({ state }: { state: ScheduleTrackReloadStateMap<'selectedTextTrackId', 'textReloadEpoch'> }) => - setupTrackReloadSchedule({ state, config: TEXT_RELOAD_SCHEDULE_CONFIG }), -}); diff --git a/packages/spf/src/playback/behaviors/tests/resolve-track.test.ts b/packages/spf/src/playback/behaviors/tests/resolve-track.test.ts index d4e44391..d23ea34c 100644 --- a/packages/spf/src/playback/behaviors/tests/resolve-track.test.ts +++ b/packages/spf/src/playback/behaviors/tests/resolve-track.test.ts @@ -1,12 +1,14 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type { StateSignals } from '../../../core/composition/create-composition'; import { signal } from '../../../core/signals/primitives'; +import type { TaskLike } from '../../../core/tasks/task'; import type { MaybeResolvedPresentation, PartiallyResolvedAudioTrack, PartiallyResolvedTextTrack, PartiallyResolvedVideoTrack, Presentation, + ResolvedTrack, } from '../../../media/types'; import { isResolvedTrack } from '../../../media/types'; import { type ResolveTrackState, resolveAudioTrack, resolveTextTrack, resolveVideoTrack } from '../resolve-track'; @@ -15,25 +17,13 @@ afterEach(() => { vi.restoreAllMocks(); }); -// The reload-epoch slots live outside `ResolveTrackState` (the loader's gate -// reads them defensively; `scheduleTrackReload` materializes them in the real -// composition). The test plays that role, so it extends the state shape here. -type TestResolveTrackState = ResolveTrackState & { - videoReloadEpoch?: number; - audioReloadEpoch?: number; - textReloadEpoch?: number; -}; - -function makeState(initial: TestResolveTrackState = {}): StateSignals { +function makeState(initial: ResolveTrackState = {}): StateSignals { return { presentation: signal(initial.presentation), selectedVideoTrackId: signal(initial.selectedVideoTrackId), selectedAudioTrackId: signal(initial.selectedAudioTrackId), selectedTextTrackId: signal(initial.selectedTextTrackId), failedCdns: signal(initial.failedCdns), - videoReloadEpoch: signal(initial.videoReloadEpoch), - audioReloadEpoch: signal(initial.audioReloadEpoch), - textReloadEpoch: signal(initial.textReloadEpoch), }; } @@ -451,49 +441,64 @@ http://example.com/seg0.m4s`; }; } - it('re-fetches when the reload epoch is bumped', async () => { + // Flush pending macrotasks so "did NOT happen" assertions are meaningful. + const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); + + it('re-resolves while the window is incomplete (reschedule resolves)', async () => { + const state = makeState({ presentation: liveVideoPresentation(), selectedVideoTrackId: 'track-1' }); + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => new Response(LIVE_PLAYLIST)); + + // Re-run twice, then stop — bounds the loop without timers. + let cycles = 0; + const reschedule = async () => cycles++ < 2; + const reactor = resolveVideoTrack.setup({ state, config: { reschedule } }); + + await vi.waitFor(() => expect(fetchSpy).toHaveBeenCalledTimes(3)); // initial + 2 reloads + expect(isResolvedTrack(findTrackById(state.presentation.get()!, 'track-1')!)).toBe(true); + + await flush(); + expect(fetchSpy).toHaveBeenCalledTimes(3); + + reactor.destroy(); + }); + + it('resolves once and never reloads when no reschedule is configured (VoD/non-live)', async () => { const state = makeState({ presentation: liveVideoPresentation(), selectedVideoTrackId: 'track-1' }); const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => new Response(LIVE_PLAYLIST)); const reactor = resolveVideoTrack.setup({ state }); - await vi.waitFor(() => expect(isResolvedTrack(findTrackById(state.presentation.get()!, 'track-1')!)).toBe(true)); + await vi.waitFor(() => expect(fetchSpy).toHaveBeenCalledTimes(1)); + await flush(); expect(fetchSpy).toHaveBeenCalledTimes(1); - // A scheduler bump re-fetches the (already-resolved, incomplete) track. - state.videoReloadEpoch.set(1); - await vi.waitFor(() => expect(fetchSpy).toHaveBeenCalledTimes(2)); - - // Re-setting the same epoch value is a signal no-op (no re-fire), so the - // scheduler's monotonic bumps drive reloads one-for-one without the loader - // tracking a last-serviced epoch. - state.videoReloadEpoch.set(1); - await new Promise((resolve) => setTimeout(resolve, 20)); - expect(fetchSpy).toHaveBeenCalledTimes(2); - reactor.destroy(); }); - it('does not reload a complete (VoD/ENDLIST) track on an epoch bump', async () => { + it('stops reloading once the playlist completes (reschedule returns null)', async () => { const state = makeState({ presentation: liveVideoPresentation(), selectedVideoTrackId: 'track-1' }); - // ENDLIST → complete → finite duration → never reloads, even if a bump arrives. + // Complete (ENDLIST) → finite duration → reschedule stops after the first resolve. const fetchSpy = vi .spyOn(globalThis, 'fetch') .mockImplementation(async () => new Response(`${LIVE_PLAYLIST}\n#EXT-X-ENDLIST`)); - const reactor = resolveVideoTrack.setup({ state }); + // Observe the resolved track; stop (false) once it's complete. + const reschedule = async (task: TaskLike) => { + const current = await task.run().catch(() => undefined); + return !(current && Number.isFinite(current.duration)); + }; + const reactor = resolveVideoTrack.setup({ state, config: { reschedule } }); await vi.waitFor(() => expect(isResolvedTrack(findTrackById(state.presentation.get()!, 'track-1')!)).toBe(true)); expect(fetchSpy).toHaveBeenCalledTimes(1); - state.videoReloadEpoch.set(1); - await new Promise((resolve) => setTimeout(resolve, 20)); + await flush(); expect(fetchSpy).toHaveBeenCalledTimes(1); reactor.destroy(); }); - it('retries an unresolved track on epoch bump after a transient failure', async () => { + it('retries an unresolved track after a transient failure (errored run → reschedule retry)', async () => { const state = makeState({ presentation: liveVideoPresentation(), selectedVideoTrackId: 'track-1' }); let calls = 0; vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { @@ -502,17 +507,13 @@ http://example.com/seg0.m4s`; return new Response(LIVE_PLAYLIST); }); - const reactor = resolveVideoTrack.setup({ state }); + // Retry while the observed run errored (undefined); stop once a result lands. + const reschedule = async (task: TaskLike) => { + const current = await task.run().catch(() => undefined); + return current === undefined; + }; + const reactor = resolveVideoTrack.setup({ state, config: { reschedule } }); - // First attempt fails and settles: track stays unresolved. (Waiting for - // the failed fetch to settle mirrors the scheduler's cadence delay — a bump - // arriving while the task is still in flight would be id-deduped.) - await vi.waitFor(() => expect(calls).toBe(1)); - await new Promise((resolve) => setTimeout(resolve, 20)); - expect(isResolvedTrack(findTrackById(state.presentation.get()!, 'track-1')!)).toBe(false); - - // The scheduler's next bump retries the unresolved track → resolves. - state.videoReloadEpoch.set(1); await vi.waitFor(() => expect(isResolvedTrack(findTrackById(state.presentation.get()!, 'track-1')!)).toBe(true)); expect(calls).toBe(2); diff --git a/packages/spf/src/playback/behaviors/tests/schedule-track-reload.test.ts b/packages/spf/src/playback/behaviors/tests/schedule-track-reload.test.ts deleted file mode 100644 index 51c7ec1c..00000000 --- a/packages/spf/src/playback/behaviors/tests/schedule-track-reload.test.ts +++ /dev/null @@ -1,137 +0,0 @@ -import { afterEach, describe, expect, it, vi } from 'vitest'; -import { signal } from '../../../core/signals/primitives'; -import { - type MaybeResolvedPresentation, - MEDIA_PLAYLIST_METADATA_KEY, - type PartiallyResolvedVideoTrack, - type Presentation, - type VideoTrack, -} from '../../../media/types'; -import { scheduleVideoTrackReload } from '../schedule-track-reload'; - -afterEach(() => { - vi.useRealTimers(); - vi.restoreAllMocks(); -}); - -const UNRESOLVED_VIDEO: PartiallyResolvedVideoTrack = { - type: 'video', - id: 'v-1', - url: 'https://example.com/video.m3u8', - bandwidth: 1_000_000, - mimeType: 'video/mp4', - codecs: [], -}; - -function resolvedVideo(opts: { endList?: boolean; segmentCount?: number; mediaSequence?: number } = {}): VideoTrack { - const { endList = false, segmentCount = 3, mediaSequence = 0 } = opts; - return { - type: 'video', - id: 'v-1', - url: 'https://example.com/video.m3u8', - mimeType: 'video/mp4', - codecs: ['avc1.640020'], - bandwidth: 1_000_000, - initialization: { url: 'https://example.com/init.mp4' }, - duration: endList ? 12 : Number.POSITIVE_INFINITY, - startTime: 0, - segments: Array.from({ length: segmentCount }, (_, i) => ({ - id: `segment-${mediaSequence + i}`, - url: `${mediaSequence + i}.m4s`, - duration: 4, - startTime: i * 4, - })), - metadata: { [MEDIA_PLAYLIST_METADATA_KEY]: { mediaSequence, targetDuration: 4, endList } }, - }; -} - -function presentationWith(track: VideoTrack | PartiallyResolvedVideoTrack): Presentation { - return { - id: 'pres-1', - url: 'https://example.com/master.m3u8', - startTime: 0, - selectionSets: [{ id: 'video-set', type: 'video', switchingSets: [{ id: 'sw', type: 'video', tracks: [track] }] }], - }; -} - -function makeState(presentation: MaybeResolvedPresentation, trackId: string | undefined = 'v-1') { - return { - presentation: signal(presentation), - selectedVideoTrackId: signal(trackId), - videoReloadEpoch: signal(undefined), - }; -} - -describe('scheduleVideoTrackReload', () => { - it('bumps the reload epoch on the target-duration cadence', async () => { - vi.useFakeTimers(); - const state = makeState(presentationWith(resolvedVideo())); - - const reactor = scheduleVideoTrackReload.setup({ state }); - - expect(state.videoReloadEpoch.get()).toBeUndefined(); - await vi.advanceTimersByTimeAsync(4000); // one TARGETDURATION - expect(state.videoReloadEpoch.get()).toBe(1); - - reactor.destroy(); - }); - - it('polls at half cadence when the window is unchanged', async () => { - vi.useFakeTimers(); - const state = makeState(presentationWith(resolvedVideo())); - - const reactor = scheduleVideoTrackReload.setup({ state }); - - // First reload after a full TARGETDURATION (4s). - await vi.advanceTimersByTimeAsync(4000); - expect(state.videoReloadEpoch.get()).toBe(1); - - // Snapshot unchanged (presentation not updated) → next poll at half (2s). - await vi.advanceTimersByTimeAsync(1999); - expect(state.videoReloadEpoch.get()).toBe(1); - await vi.advanceTimersByTimeAsync(1); - expect(state.videoReloadEpoch.get()).toBe(2); - - reactor.destroy(); - }); - - it('stays idle for a complete (endList) playlist', async () => { - vi.useFakeTimers(); - const state = makeState(presentationWith(resolvedVideo({ endList: true }))); - - const reactor = scheduleVideoTrackReload.setup({ state }); - - await vi.advanceTimersByTimeAsync(60_000); - expect(state.videoReloadEpoch.get()).toBeUndefined(); - - reactor.destroy(); - }); - - it('keeps bumping while the track is unresolved (drives first-resolve retries)', async () => { - vi.useFakeTimers(); - const state = makeState(presentationWith(UNRESOLVED_VIDEO)); - - const reactor = scheduleVideoTrackReload.setup({ state }); - - // No targetDuration available yet → fallback cadence (6s). - await vi.advanceTimersByTimeAsync(6000); - expect(state.videoReloadEpoch.get()).toBe(1); - await vi.advanceTimersByTimeAsync(6000); - expect(state.videoReloadEpoch.get()).toBe(2); - - reactor.destroy(); - }); - - it('stays idle with no selected track', async () => { - vi.useFakeTimers(); - const state = makeState(presentationWith(resolvedVideo())); - state.selectedVideoTrackId.set(undefined); - - const reactor = scheduleVideoTrackReload.setup({ state }); - - await vi.advanceTimersByTimeAsync(60_000); - expect(state.videoReloadEpoch.get()).toBeUndefined(); - - reactor.destroy(); - }); -}); diff --git a/packages/spf/src/playback/engines/hls/engine.ts b/packages/spf/src/playback/engines/hls/engine.ts index 6546248e..648628bb 100644 --- a/packages/spf/src/playback/engines/hls/engine.ts +++ b/packages/spf/src/playback/engines/hls/engine.ts @@ -5,6 +5,8 @@ import { type StateSignals, } from '../../../core/composition/create-composition'; import { makeShareSignals, type ShareSignalsConfig } from '../../../core/composition/share-signals'; +import { delayedReschedule } from '../../../core/tasks/delayed-reschedule'; +import type { Reschedule } from '../../../core/tasks/task'; import type { QualityConfig } from '../../../media/abr/quality-selection'; import type { BackBufferConfig } from '../../../media/buffer/back-buffer'; import type { ForwardBufferConfig } from '../../../media/buffer/forward-buffer'; @@ -16,7 +18,15 @@ import { removeAllSubtitlesTracksFromMedia, } from '../../../media/dom/text/text-track-slots'; import { parseMultivariantPlaylist } from '../../../media/hls/parse-multivariant'; -import type { AudioTrack, CanPlayTrack, MaybeResolvedPresentation, TextTrack, VideoTrack } from '../../../media/types'; +import { mediaPlaylistReloadDelay } from '../../../media/hls/reload-policy'; +import type { + AudioTrack, + CanPlayTrack, + MaybeResolvedPresentation, + ResolvedTrack, + TextTrack, + VideoTrack, +} from '../../../media/types'; import type { GetCdnId } from '../../../media/utils/cdn'; import { getResolvedSelectedTrackDuration } from '../../../media/utils/track-selection'; import type { BandwidthConfig, BandwidthState } from '../../../network/bandwidth-estimator'; @@ -42,11 +52,6 @@ import { trackLoadTriggers } from '../../behaviors/dom/track-load-triggers'; import { updateMediaSourceDuration } from '../../behaviors/dom/update-mediasource-duration'; import { type ParsePresentation, resolvePresentation } from '../../behaviors/resolve-presentation'; import { resolveAudioTrack, resolveTextTrack, resolveVideoTrack } from '../../behaviors/resolve-track'; -import { - scheduleAudioTrackReload, - scheduleTextTrackReload, - scheduleVideoTrackReload, -} from '../../behaviors/schedule-track-reload'; import { type FailoverMonitorConfig, setupFailoverMonitor } from '../../behaviors/setup-failover-monitor'; import { syncPreload } from '../../behaviors/sync-preload'; import { switchAudioTrack, switchTextTrack, switchVideoTrack } from '../../behaviors/track-switching'; @@ -110,15 +115,6 @@ export interface SimpleHlsEngineState { failedCdns?: string[]; currentTime?: number; loadActivated?: boolean; - /** - * Per-type live-reload triggers. Owned by `scheduleTrackReload` (composed - * only by live engines), which bumps them on a target-duration cadence; - * `resolveTrack` watches them to re-fetch the media playlist. Inert for VoD - * (no scheduler → never bumped → loader resolves once). - */ - videoReloadEpoch?: number; - audioReloadEpoch?: number; - textReloadEpoch?: number; } /** @@ -194,6 +190,15 @@ export interface SimpleHlsEngineConfig extends ShareSignalsConfig; /** * Manifest parser handed to `resolvePresentation`. Defaults to the HLS * multivariant-playlist parser; supply your own for alternate format @@ -314,6 +319,11 @@ export function createSimpleHlsEngine( const finalConfig = { ...config, canPlayTrack: config.canPlayTrack ?? canPlayTrack, + // The resolve* loaders' RecurringRunner re-runs on this `reschedule`: the pure + // target-duration cadence, start-anchored + made awaitable by `delayedReschedule`. + // Inert for VoD (the cadence returns null once a playlist is complete), so it + // composes always. + reschedule: config.reschedule ?? delayedReschedule(mediaPlaylistReloadDelay), resolveTextTrackSegment: config.resolveTextTrackSegment ?? resolveVttSegment, resolveDuration: config.resolveDuration ?? getResolvedSelectedTrackDuration, parsePresentation: config.parsePresentation ?? parseMultivariantPlaylist, @@ -352,18 +362,13 @@ export function createSimpleHlsEngine( // Resolve selected tracks (fetch media playlists). Composed before the // switch* slot owners; selection is reactive, so a resolve* re-fires once // its switch* sets the id (same convergence for all three types). Also the - // live loader: re-fetches when its reload epoch advances (below). + // live loader: its `RecurringRunner` re-resolves on the playlist's + // target-duration cadence (`reschedule` in config) until `#EXT-X-ENDLIST`. + // Inert for VoD (a complete playlist stops the policy after one resolve). resolveVideoTrack, resolveAudioTrack, resolveTextTrack, - // Live refetch policy: bump each type's reload epoch on a target-duration - // cadence until `#EXT-X-ENDLIST`. Inert for VoD (a complete playlist never - // reloads), so these compose unconditionally. - scheduleVideoTrackReload, - scheduleAudioTrackReload, - scheduleTextTrackReload, - // Re-base selected live tracks' timelines to the estimated stream origin // (segment.startTime ≈ native PTS). No-op for VoD (no PDT / shift 0). anchorLiveTracks, diff --git a/packages/spf/src/playback/engines/live-hls/engine.ts b/packages/spf/src/playback/engines/live-hls/engine.ts index 74e82f69..5d1af96f 100644 --- a/packages/spf/src/playback/engines/live-hls/engine.ts +++ b/packages/spf/src/playback/engines/live-hls/engine.ts @@ -13,8 +13,10 @@ */ import { type Composition, createComposition } from '../../../core/composition/create-composition'; import { makeShareSignals } from '../../../core/composition/share-signals'; +import { delayedReschedule } from '../../../core/tasks/delayed-reschedule'; import { canPlayTrack } from '../../../media/dom/capabilities'; import { parseMultivariantPlaylist } from '../../../media/hls/parse-multivariant'; +import { mediaPlaylistReloadDelay } from '../../../media/hls/reload-policy'; import { anchorLiveTracks } from '../../behaviors/anchor-live-tracks'; import { calculatePresentationDuration } from '../../behaviors/calculate-presentation-duration'; import { deriveCdnPriority } from '../../behaviors/derive-cdn-priority'; @@ -28,7 +30,6 @@ import { trackLoadTriggers } from '../../behaviors/dom/track-load-triggers'; import { updateMediaSourceDuration } from '../../behaviors/dom/update-mediasource-duration'; import { resolvePresentation } from '../../behaviors/resolve-presentation'; import { resolveAudioTrack, resolveVideoTrack } from '../../behaviors/resolve-track'; -import { scheduleAudioTrackReload, scheduleVideoTrackReload } from '../../behaviors/schedule-track-reload'; import { setupFailoverMonitor } from '../../behaviors/setup-failover-monitor'; import { syncPreload } from '../../behaviors/sync-preload'; import { switchAudioTrack, switchVideoTrack } from '../../behaviors/track-switching'; @@ -74,6 +75,9 @@ export function createLiveHlsEngine( // Infinity to `mediaSource.duration` per the MSE spec. resolveDuration: config.resolveDuration ?? (() => Number.POSITIVE_INFINITY), startSequence: config.startSequence ?? 0, + // Reload the selected playlists via the loaders' RecurringRunner — the + // target-duration cadence, start-anchored + made awaitable by `delayedReschedule`. + reschedule: config.reschedule ?? delayedReschedule(mediaPlaylistReloadDelay), }; return createComposition( @@ -85,15 +89,11 @@ export function createLiveHlsEngine( deriveCdnPriority, setupFailoverMonitor, - // Loader (category [1]): resolves the selected track and re-fetches it - // whenever the scheduler bumps its reload epoch, carrying the timeline - // forward. + // Loader (category [1]): resolves the selected track and, via its + // RecurringRunner + `reschedule`, re-fetches it on a target-duration + // cadence until #EXT-X-ENDLIST, carrying the timeline forward. resolveVideoTrack, resolveAudioTrack, - // Scheduler (category [3]): bumps the per-type reload epoch on a - // target-duration cadence until #EXT-X-ENDLIST. - scheduleVideoTrackReload, - scheduleAudioTrackReload, // Anchor selected tracks' timelines to the estimated stream origin so // segment.startTime ≈ native PTS (what the loader matches currentTime diff --git a/packages/spf/src/playback/engines/live-playlist-spike/engine.ts b/packages/spf/src/playback/engines/live-playlist-spike/engine.ts index b7315218..f5ff6376 100644 --- a/packages/spf/src/playback/engines/live-playlist-spike/engine.ts +++ b/packages/spf/src/playback/engines/live-playlist-spike/engine.ts @@ -12,6 +12,7 @@ * Drive it via `onSignalsReady`: set `presentation = { url }`. Selection and * reloading then run on their own. */ + import { type Composition, type ContextSignals, @@ -19,18 +20,18 @@ import { type StateSignals, } from '../../../core/composition/create-composition'; import { makeShareSignals, type ShareSignalsConfig } from '../../../core/composition/share-signals'; +import { delayedReschedule } from '../../../core/tasks/delayed-reschedule'; import { parseMultivariantPlaylist } from '../../../media/hls/parse-multivariant'; +import { mediaPlaylistReloadDelay } from '../../../media/hls/reload-policy'; import { pickHighestResolutionVideoTrack, type TrackPicker } from '../../../media/primitives/select-tracks'; import type { MaybeResolvedPresentation } from '../../../media/types'; import { type ParsePresentation, resolvePresentation } from '../../behaviors/resolve-presentation'; import { resolveVideoTrack } from '../../behaviors/resolve-track'; -import { scheduleVideoTrackReload } from '../../behaviors/schedule-track-reload'; import { type SelectVideoTrackConfig, selectVideoTrack } from '../../behaviors/select-tracks'; export interface LivePlaylistSpikeState { presentation?: MaybeResolvedPresentation; selectedVideoTrackId?: string; - videoReloadEpoch?: number; preload?: 'auto' | 'metadata' | 'none'; loadActivated?: boolean; } @@ -58,14 +59,15 @@ export function createLivePlaylistSpikeEngine( ...config, picker: config.picker ?? pickHighestResolutionVideoTrack, parsePresentation: config.parsePresentation ?? parseMultivariantPlaylist, + // Reload the resolved video playlist (the spike's whole point) via the + // loader's RecurringRunner — target-duration cadence, start-anchored + made + // awaitable by `delayedReschedule`. + reschedule: delayedReschedule(mediaPlaylistReloadDelay), }; - return createComposition( - [resolvePresentation, selectVideoTrack, resolveVideoTrack, scheduleVideoTrackReload, shareSignals], - { - config: finalConfig, - // Spike skips the preload gate — resolve as soon as a url is set. - initialState: { loadActivated: true }, - } - ); + return createComposition([resolvePresentation, selectVideoTrack, resolveVideoTrack, shareSignals], { + config: finalConfig, + // Spike skips the preload gate — resolve as soon as a url is set. + initialState: { loadActivated: true }, + }); } diff --git a/packages/utils/src/time/index.ts b/packages/utils/src/time/index.ts index 16c5b2b5..200f480e 100644 --- a/packages/utils/src/time/index.ts +++ b/packages/utils/src/time/index.ts @@ -1 +1,2 @@ export * from './format'; +export * from './sleep'; diff --git a/packages/utils/src/time/sleep.ts b/packages/utils/src/time/sleep.ts new file mode 100644 index 00000000..2205eb7b --- /dev/null +++ b/packages/utils/src/time/sleep.ts @@ -0,0 +1,25 @@ +/** + * Resolve after `ms` milliseconds. Pass a `signal` to make it cancellable: the + * timer is cleared and the promise rejects with the signal's reason as soon as + * the signal aborts (including if it's already aborted). + */ +export function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(signal.reason); + return; + } + + const onAbort = () => { + clearTimeout(timer); + reject(signal?.reason); + }; + + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} diff --git a/packages/utils/src/time/tests/sleep.test.ts b/packages/utils/src/time/tests/sleep.test.ts new file mode 100644 index 00000000..26019c69 --- /dev/null +++ b/packages/utils/src/time/tests/sleep.test.ts @@ -0,0 +1,42 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { sleep } from '../sleep'; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('sleep', () => { + it('resolves after the given delay', async () => { + vi.useFakeTimers(); + const resolved = vi.fn(); + const promise = sleep(100).then(resolved); + + await vi.advanceTimersByTimeAsync(99); + expect(resolved).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + await promise; + expect(resolved).toHaveBeenCalledTimes(1); + }); + + it('rejects with the signal reason and clears the timer when aborted mid-wait', async () => { + vi.useFakeTimers(); + const controller = new AbortController(); + const reason = new DOMException('Aborted', 'AbortError'); + const promise = sleep(100, controller.signal); + + controller.abort(reason); + await expect(promise).rejects.toBe(reason); + + // Timer was cleared — advancing past the delay does nothing further. + await vi.advanceTimersByTimeAsync(200); + }); + + it('rejects immediately when the signal is already aborted', async () => { + const controller = new AbortController(); + const reason = new DOMException('Aborted', 'AbortError'); + controller.abort(reason); + + await expect(sleep(100, controller.signal)).rejects.toBe(reason); + }); +});