From 2520373f2f04625bb19ce889bf9393ea8a9f673e Mon Sep 17 00:00:00 2001 From: Christian Pillsbury Date: Thu, 18 Jun 2026 16:09:34 -0700 Subject: [PATCH] refactor(spf): make RecurringRunner a self-recursive schedule(), require reschedule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the imperative `#loop()` with a single chained promise: each cycle's `.then` returns the next cycle's `schedule(clone())`, so the recurrence is the method calling itself. The slot is released just before re-scheduling the same-id clone so the call advances rather than dedup-returning; ownership is tracked by `#active === task`. Error handling moves downstream — the runner no longer invents a retry policy: - A genuine run/reschedule failure rejects schedule()'s promise (propagates to the caller); no swallowing. - The runner's own cancellation (abort/supersede/destroy) is not a failure, so an aborted recurrence settles quietly — callers don't `.catch` routine teardown. Consequences: - `reschedule` is now required; `runOnce` expresses run-exactly-once explicitly (a missing reschedule is a bug, not a silent run-once). - Reschedule-driven retry-on-transient-error is dropped (a rejected run is terminal). The retry logic in delayedReschedule / mediaPlaylistReloadDelay is now vestigial — to be cleaned up or relocated to the fetch layer next. - `resolve-track` catches schedule()'s promise (abort settles quietly; genuine resolve failures end the recurrence, TODO surface to state). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/spf/src/core/tasks/task.ts | 109 ++++++++++-------- .../spf/src/core/tasks/tests/task.test.ts | 22 ++-- .../src/playback/behaviors/resolve-track.ts | 17 ++- .../behaviors/tests/resolve-track.test.ts | 19 ++- 4 files changed, 94 insertions(+), 73 deletions(-) diff --git a/packages/spf/src/core/tasks/task.ts b/packages/spf/src/core/tasks/task.ts index e1c1c85a..d259626c 100644 --- a/packages/spf/src/core/tasks/task.ts +++ b/packages/spf/src/core/tasks/task.ts @@ -366,6 +366,13 @@ export class SerialRunner { */ export type Reschedule = (task: TaskLike) => PromiseLike; +/** + * A {@link Reschedule} that never recurs — the task runs exactly once. Pass it to + * a {@link RecurringRunner} for non-recurring, run-once work (e.g. resolving a + * complete VoD playlist that can never go stale). + */ +export const runOnce: Reschedule = () => Promise.resolve(false); + /** * 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 @@ -373,8 +380,8 @@ export type Reschedule = (task: TaskLike) => PromiseLike = (task: TaskLike) => PromiseLike { - readonly #reschedule: Reschedule | undefined; + readonly #reschedule: Reschedule; // The identified active task — the one being (re)run. Held across cycles // (including the inter-cycle wait); null once the recurrence stops/aborts. Also - // serves as the loop's ownership token: a loop frees the slot only while + // serves as the ownership token: a cycle advances or frees the slot only while // `#active` still points at its own task (a supersede/abortAll swaps it). #active: TaskLike | null = null; #destroyed = false; - constructor(reschedule?: Reschedule) { + constructor(reschedule: Reschedule) { this.#reschedule = reschedule; } - schedule(task: TaskLike): void { - if (this.#destroyed) return; + /** + * Run `task` and recur per the `reschedule` verdict, as a single promise. + * Resolves with the *final* cycle's value when the recurrence stops; **rejects** + * if a run (or reschedule) genuinely fails — the rejection propagates to the + * caller, who owns error handling; the runner only frees its slot (no + * swallowing). The runner's *own* cancellation (abort/supersede/destroy) is not + * a failure, so an aborted recurrence settles quietly rather than rejecting — + * callers don't have to `.catch` routine teardown. + * + * Each cycle runs the task and consults `reschedule` concurrently (so the delay + * can be measured from the run's start); when both settle and this cycle still + * owns the slot, a `true` verdict re-schedules a `clone()` whose promise is + * *returned* — so the recurrence is the method calling itself, threaded into one + * promise, no separate loop. The clone shares the id, so the slot's identity is + * stable across cycles; it's released just before the re-schedule so the call + * advances rather than dedup-returning. + * + * Note: because each cycle's promise adopts the next, the chain retains every + * prior cycle for the life of the recurrence — bounded for finite recurrences, + * an unbounded (small per-cycle) cost for a long-lived one (e.g. live reload). + */ + schedule(task: TaskLike): Promise { + if (this.#destroyed) return Promise.resolve() as Promise; // 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. + // recurrence continues uninterrupted (don't restart it) — hand back its + // in-flight run. + if (this.#active?.id === task.id) return this.#active.run(); + // Different id supersedes: abort the prior recurrence, then take over. this.#cancel(); this.#active = task; - void this.#loop(task); - } - async #loop(task: TaskLike): Promise { - let current = task; - while (true) { - let again = false; - try { - // Start the run and the reschedule together: reschedule observes the run - // (via the memoized `current.run()`) and owns the inter-run delay, so the - // interval can be measured from the run's *start*. `Promise.all` waits - // for both — the run (so the clone can carry its value forward as - // `previous`) and the verdict + delay. The run's own result is swallowed - // here; reschedule reads it (and decides retry-vs-stop on error). - [, again] = await Promise.all([ - current.run().then( - () => {}, - () => {} - ), - this.#reschedule ? this.#reschedule(current) : Promise.resolve(false), - ]); - } catch { - return; // reschedule rejected (aborted during its delay) — stop without freeing + // Drive the run (in case `reschedule` doesn't observe it — e.g. `runOnce`) + // and the verdict together; a rejected run rejects the whole cycle. + return Promise.all([task.run(), this.#reschedule(task)]).then( + ([value, again]) => { + // Only act while we still own the slot — a supersede/abortAll swapped + // `#active`, in which case this stale cycle does nothing. + if (this.#active === task && again && !task.signal.aborted) { + // Release first so the same-id clone advances (isn't dedup-returned), + // then chain the next cycle into this promise. + this.#active = null; + return this.schedule(task.clone()); + } + if (this.#active === task) this.#active = null; // natural stop / superseded + return value; + }, + (error) => { + // The recurrence ended on a rejection; free the slot if we still own it. + if (this.#active === task) this.#active = null; + // The runner's own cancellation isn't a failure — settle quietly so + // routine teardown (abort/supersede/destroy) needs no caller `.catch`. + // A genuine run/reschedule failure propagates to the caller. + if (task.signal.aborted) return undefined as TValue; + throw error; } - // The task is the cancellation channel; an aborted task means we were - // superseded/aborted even if reschedule still resolved true. - if (current.signal.aborted || !again) break; - // 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) and carries `previous`; 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 `#active`). - if (this.#active === current) this.#active = null; + ); } #cancel(): void { diff --git a/packages/spf/src/core/tasks/tests/task.test.ts b/packages/spf/src/core/tasks/tests/task.test.ts index b1f4329f..18144fb2 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, RecurringRunner, type Reschedule, SerialRunner, Task } from '../task'; +import { ConcurrentRunner, RecurringRunner, type Reschedule, runOnce, SerialRunner, Task } from '../task'; // ============================================================================= // Task @@ -670,10 +670,10 @@ describe('RecurringRunner', () => { /** 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 () => { + it('runs the task exactly once with the runOnce reschedule', async () => { let runs = 0; const task = new Task(async () => ++runs, { id: 'x' }); - const runner = new RecurringRunner(); + const runner = new RecurringRunner(runOnce); runner.schedule(task); await vi.waitFor(() => expect(runs).toBe(1)); @@ -714,21 +714,21 @@ describe('RecurringRunner', () => { ]); }); - it('keeps the loop alive across a transient error (observed value is undefined)', async () => { + it('rejects and stops the recurrence when a run errors (no swallowing/retry)', 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; - }); + // A reschedule that would otherwise keep going — but a run error ends it. + const runner = new RecurringRunner(async () => true); - runner.schedule(task); - await vi.waitFor(() => expect(n).toBe(3)); + // The first cycle's error propagates out of schedule(); the recurrence stops. + await expect(runner.schedule(task)).rejects.toThrow('boom'); + + await flush(); + expect(n).toBe(1); // no retry — the errored run was terminal runner.destroy(); }); diff --git a/packages/spf/src/playback/behaviors/resolve-track.ts b/packages/spf/src/playback/behaviors/resolve-track.ts index 068aa020..eb91b213 100644 --- a/packages/spf/src/playback/behaviors/resolve-track.ts +++ b/packages/spf/src/playback/behaviors/resolve-track.ts @@ -1,7 +1,7 @@ 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 { RecurringRunner, type Reschedule, Task } from '../../core/tasks/task'; +import { RecurringRunner, type Reschedule, runOnce, 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'; @@ -107,10 +107,11 @@ function setupTrackResolution({ config: TrackResolutionConfig; }) { // 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); + // task whenever reschedule resolves true, until it returns false; with none + // configured (VoD) `runOnce` makes it resolve exactly 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 ?? runOnce); // 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 — @@ -205,7 +206,11 @@ function setupTrackResolution({ // is the runner's job, driven by `reschedule`. if (!track || !shouldLoadTrack(track)) return; - runner.schedule(createResolveTask(trackId)); + // Abort (selection/source change, via `abortAll`) settles quietly; + // `schedule` rejects only on a genuine resolve failure. We don't + // surface those to state yet, so end quietly. + // TODO: surface unrecoverable resolve errors. + runner.schedule(createResolveTask(trackId)).catch(() => {}); }, ], }, 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 d23ea34c..465891fa 100644 --- a/packages/spf/src/playback/behaviors/tests/resolve-track.test.ts +++ b/packages/spf/src/playback/behaviors/tests/resolve-track.test.ts @@ -498,24 +498,23 @@ http://example.com/seg0.m4s`; reactor.destroy(); }); - it('retries an unresolved track after a transient failure (errored run → reschedule retry)', async () => { + it('stops resolving on a fetch failure (an errored run is terminal — no retry)', async () => { const state = makeState({ presentation: liveVideoPresentation(), selectedVideoTrackId: 'track-1' }); let calls = 0; vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { calls += 1; - if (calls === 1) throw new TypeError('Failed to fetch'); - return new Response(LIVE_PLAYLIST); + throw new TypeError('Failed to fetch'); }); - // 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; - }; + // Even a reschedule that would keep going can't revive an errored run — the + // rejected run ends the recurrence (retry, if wanted, belongs in the fetch layer). + const reschedule = async () => true; const reactor = resolveVideoTrack.setup({ state, config: { reschedule } }); - await vi.waitFor(() => expect(isResolvedTrack(findTrackById(state.presentation.get()!, 'track-1')!)).toBe(true)); - expect(calls).toBe(2); + await vi.waitFor(() => expect(calls).toBe(1)); + await flush(); + expect(calls).toBe(1); // no retry + expect(isResolvedTrack(findTrackById(state.presentation.get()!, 'track-1')!)).toBe(false); reactor.destroy(); });