mirror of
https://github.com/zoriya/v10.git
synced 2026-08-15 10:23:32 +00:00
wip(spf): relocation via messagePipelines step model
Rework the segment loader's per-message execution into composable step pipelines (fetch → dispatch by default); relocation injects discover/stamp steps between them, so loader + SourceBuffer actor stay relocation-oblivious and Tier 0 imports no relocation code. - segment-loader: Frame/LoadStep/StepDeps/MessagePipelines, fetchStep/ dispatchStep (deps as 3rd arg), step-runner makeLoadTask + inFlight wrapper. - source-buffer: drop CreateAppendMeta; keep idempotent timestampOffset guard. - relocation.ts / origin-discoverer.ts: TRANSITIONAL — slated for rebuild as the establishStartMediaTime reactor. WIP checkpoint; outstanding work tracked in .claude/plans/spf-non-zero-pts-timestamp-offset-spike.md. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Christian Pillsbury
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
6a20bade9b
commit
041cecca6a
@@ -13,7 +13,6 @@ import {
|
||||
type ForwardBufferConfig,
|
||||
getSegmentsToLoad,
|
||||
} from '../../../media/buffer/forward-buffer';
|
||||
import { readFirstBaseMediaDecodeTime, readFirstMediaTimescale } from '../../../media/mp4/timestamp-origin';
|
||||
import {
|
||||
type AddressableObject,
|
||||
type AudioTrack,
|
||||
@@ -22,9 +21,9 @@ import {
|
||||
type VideoTrack,
|
||||
} from '../../../media/types';
|
||||
import type {
|
||||
AppendData,
|
||||
AppendInitMessage,
|
||||
AppendSegmentMessage,
|
||||
IndividualSourceBufferMessage,
|
||||
RemoveMessage,
|
||||
SourceBufferActor,
|
||||
} from './source-buffer';
|
||||
@@ -109,18 +108,51 @@ export interface SegmentLoaderActorContext {
|
||||
inFlightInitTrackId: string | null;
|
||||
/** Segment ID currently being fetched/appended, or null. */
|
||||
inFlightSegmentId: string | null;
|
||||
/**
|
||||
* Relocation offset in seconds (`−baseMediaDecodeTime / timescale`), or null
|
||||
* until established / when relocation is off (spike). Exposed on the snapshot
|
||||
* so the text path can consume the same offset to rebase cues — Mux cues carry
|
||||
* no `X-TIMESTAMP-MAP`, so text can't self-derive the origin and reads the
|
||||
* A/V-established one instead.
|
||||
*/
|
||||
relocationOffset: number | null;
|
||||
}
|
||||
|
||||
export type SegmentLoaderActor = MessageActor<SegmentLoaderActorState, SegmentLoaderActorContext, SegmentLoaderMessage>;
|
||||
|
||||
/**
|
||||
* A {@link LoadTask} mid-reassembly into its append message. `LoadTask` is a
|
||||
* message with `data` omitted and a URL added; the pipeline reverses that —
|
||||
* `fetchStep` produces `data`, `dispatchStep` reassembles the message. `data` is
|
||||
* the omitted payload (kept as the fetched stream — the loader never produces the
|
||||
* `ArrayBuffer` arm of `AppendData` — widened back at dispatch). `meta` overrides
|
||||
* `op.meta` for `append-segment` (e.g. a stamped `timestampOffset`); an
|
||||
* `append-init` dispatches `op.meta` directly.
|
||||
*/
|
||||
export interface Frame {
|
||||
readonly op: LoadTask;
|
||||
data?: AsyncIterable<Uint8Array>;
|
||||
meta?: AppendSegmentMessage['meta'];
|
||||
}
|
||||
|
||||
/**
|
||||
* One stage of a message pipeline. Mutates the {@link Frame} in place and may be
|
||||
* async; the runner checks `signal.aborted` before each step and passes the
|
||||
* actor's {@link StepDeps} on every call, so a stateless step (e.g.
|
||||
* {@link fetchStep}) is a plain value — only a *parameterized or stateful* step
|
||||
* (relocation's `tapOrigin`) needs to be a factory.
|
||||
*/
|
||||
export type LoadStep = (frame: Frame, signal: AbortSignal, deps: StepDeps) => void | Promise<void>;
|
||||
|
||||
/** Per-actor runtime dependencies, passed to each {@link LoadStep} on every call. */
|
||||
export interface StepDeps {
|
||||
sourceBufferActor: SourceBufferActor;
|
||||
fetchBytes: FetchBytes;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds the ordered step list for each message type. Called **once per actor**
|
||||
* — its role is per-actor instantiation, so stateful steps (relocation's origin
|
||||
* discoverer) get fresh state per source reset; deps arrive at step-call time,
|
||||
* not here. The default ({@link DEFAULT_MESSAGE_PIPELINES}) is `fetch → dispatch`;
|
||||
* a non-zero-PTS composition returns a map that inserts its own discover/stamp
|
||||
* steps between them (see `createRelocation`), so the loader stays oblivious to
|
||||
* relocation and the Tier 0 pipeline carries no relocation vocabulary at all.
|
||||
*/
|
||||
export type MessagePipelines = () => Record<LoadTask['type'], LoadStep[]>;
|
||||
|
||||
/**
|
||||
* Configuration for `createSegmentLoaderActor`. Each sub-config is
|
||||
* spread over the corresponding `DEFAULT_*_CONFIG` so callers can
|
||||
@@ -129,23 +161,8 @@ export type SegmentLoaderActor = MessageActor<SegmentLoaderActorState, SegmentLo
|
||||
export interface SegmentLoaderActorConfig {
|
||||
forwardBuffer?: Partial<ForwardBufferConfig>;
|
||||
backBuffer?: Partial<BackBufferConfig>;
|
||||
/**
|
||||
* Non-zero-PTS relocation (spike). When true, this loader reads its track's
|
||||
* decode-time origin (`tfdt`/`mdhd`) from the init + first media segment and
|
||||
* relocates the buffer to 0-based via `timestampOffset`. Per-track single-origin
|
||||
* (Tier-1); off by default.
|
||||
*/
|
||||
relocateTimestampOrigin?: boolean;
|
||||
}
|
||||
|
||||
/** Per-track relocation working state (spike). Established once, on first media segment. */
|
||||
interface RelocationState {
|
||||
readonly enabled: boolean;
|
||||
/** `mdhd` timescale from the init segment. */
|
||||
timescale?: number;
|
||||
/** `−(baseMediaDecodeTime / timescale)` — the SourceBuffer timestampOffset. */
|
||||
offset?: number;
|
||||
established: boolean;
|
||||
/** Per-message-type step pipelines. Defaults to {@link DEFAULT_MESSAGE_PIPELINES} (`fetch → dispatch`). */
|
||||
messagePipelines?: MessagePipelines;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
@@ -200,27 +217,46 @@ function waitForIdle(snapshot: SourceBufferActor['snapshot'], signal: AbortSigna
|
||||
});
|
||||
}
|
||||
|
||||
/** Drain an async byte-iterable into one contiguous buffer (relocation spike). */
|
||||
async function collect(iterable: AsyncIterable<Uint8Array>): Promise<Uint8Array> {
|
||||
const chunks: Uint8Array[] = [];
|
||||
let total = 0;
|
||||
for await (const chunk of iterable) {
|
||||
chunks.push(chunk);
|
||||
total += chunk.length;
|
||||
// ============================================================================
|
||||
// STEPS
|
||||
// ============================================================================
|
||||
|
||||
/** Build the SourceBuffer message a completed frame dispatches. `fetchStep` always precedes `dispatchStep` in append pipelines, so `data` is set by now. */
|
||||
function toMessage({ op, data, meta }: Frame): IndividualSourceBufferMessage {
|
||||
switch (op.type) {
|
||||
case 'remove':
|
||||
return op;
|
||||
case 'append-init':
|
||||
return { type: 'append-init', data: data!, meta: op.meta };
|
||||
case 'append-segment':
|
||||
return { type: 'append-segment', data: data!, meta: meta ?? op.meta };
|
||||
}
|
||||
const out = new Uint8Array(total);
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
out.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** A collected buffer spans its backing ArrayBuffer exactly — hand it to MSE as one append. */
|
||||
function toArrayBuffer(bytes: Uint8Array): ArrayBuffer {
|
||||
return bytes.buffer as ArrayBuffer;
|
||||
}
|
||||
/**
|
||||
* Fetch this op's bytes into the frame. Init segments need the full body
|
||||
* (`minChunkSize: Infinity`) before appending; media segments stream so chunks
|
||||
* append as they arrive. Awaiting headers eagerly also starts the HTTP
|
||||
* connection (and records the fetch in observers like tests).
|
||||
*/
|
||||
export const fetchStep: LoadStep = async (frame, signal, deps) => {
|
||||
const { op } = frame;
|
||||
if (op.type === 'remove') return; // fetchStep only appears in append pipelines
|
||||
frame.data = await deps.fetchBytes(op, op.type === 'append-init' ? { signal, minChunkSize: Infinity } : { signal });
|
||||
};
|
||||
|
||||
/** Dispatch the frame's message to the SourceBufferActor and await its return to idle. */
|
||||
export const dispatchStep: LoadStep = async (frame, signal, deps) => {
|
||||
deps.sourceBufferActor.send(toMessage(frame));
|
||||
await waitForIdle(deps.sourceBufferActor.snapshot, signal);
|
||||
};
|
||||
|
||||
/** Tier 0 default: fetch (for ops that carry bytes) then dispatch. No relocation vocabulary. */
|
||||
const DEFAULT_MESSAGE_PIPELINES: MessagePipelines = () => ({
|
||||
remove: [dispatchStep],
|
||||
'append-init': [fetchStep, dispatchStep],
|
||||
'append-segment': [fetchStep, dispatchStep],
|
||||
});
|
||||
|
||||
// ============================================================================
|
||||
// LOAD TASK FACTORY
|
||||
@@ -229,88 +265,36 @@ function toArrayBuffer(bytes: Uint8Array): ArrayBuffer {
|
||||
interface LoadTaskOptions {
|
||||
getContext: () => SegmentLoaderActorContext;
|
||||
setContext: (ctx: SegmentLoaderActorContext) => void;
|
||||
fetchBytes: FetchBytes;
|
||||
sourceBufferActor: SourceBufferActor;
|
||||
relocation: RelocationState;
|
||||
pipelines: Record<LoadTask['type'], LoadStep[]>;
|
||||
deps: StepDeps;
|
||||
}
|
||||
|
||||
/**
|
||||
* Wraps a LoadTask descriptor into a Task that fetches (if needed) and
|
||||
* forwards to SourceBufferActor. Updates in-flight context around async
|
||||
* operations so the loading handler can make accurate continue/preempt
|
||||
* decisions at any point.
|
||||
* Wraps a LoadTask descriptor into a Task that runs the op's message pipeline
|
||||
* (fetch/discover/stamp/dispatch, per the composition's `messagePipelines`).
|
||||
* Updates in-flight context around the async region so the loading handler can
|
||||
* make accurate continue/preempt decisions at any point, and checks the abort
|
||||
* signal before each step.
|
||||
*/
|
||||
function makeLoadTask(
|
||||
op: LoadTask,
|
||||
{ getContext, setContext, fetchBytes, sourceBufferActor, relocation }: LoadTaskOptions
|
||||
): Task<void> {
|
||||
function makeLoadTask(op: LoadTask, { getContext, setContext, pipelines, deps }: LoadTaskOptions): Task<void> {
|
||||
return new Task(async (taskSignal) => {
|
||||
if (taskSignal.aborted) return;
|
||||
|
||||
if (op.type === 'remove') {
|
||||
sourceBufferActor.send(op);
|
||||
await waitForIdle(sourceBufferActor.snapshot, taskSignal);
|
||||
return;
|
||||
}
|
||||
const frame: Frame = op.type === 'append-segment' ? { op, meta: op.meta } : { op };
|
||||
|
||||
if (op.type === 'append-init') {
|
||||
setContext({ ...getContext(), inFlightInitTrackId: op.meta.trackId });
|
||||
try {
|
||||
// Init segments are small and need the full body before appending.
|
||||
// minChunkSize: Infinity accumulates all chunks into one before yielding.
|
||||
const body = await fetchBytes(op, { signal: taskSignal, minChunkSize: Infinity });
|
||||
// Relocation (spike): read this track's mdhd timescale from the init —
|
||||
// half of the decode-time origin (paired with the first segment's tfdt).
|
||||
let data: AppendData = body;
|
||||
if (relocation.enabled) {
|
||||
const bytes = await collect(body);
|
||||
relocation.timescale = readFirstMediaTimescale(bytes);
|
||||
data = toArrayBuffer(bytes);
|
||||
}
|
||||
if (!taskSignal.aborted) {
|
||||
sourceBufferActor.send({ type: 'append-init', data, meta: op.meta });
|
||||
await waitForIdle(sourceBufferActor.snapshot, taskSignal);
|
||||
}
|
||||
} finally {
|
||||
setContext({ ...getContext(), inFlightInitTrackId: null });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// append-segment: await headers eagerly (starts the HTTP connection and
|
||||
// records the fetch in observers like tests), then pass the body stream
|
||||
// directly to the actor so chunks are appended as they arrive.
|
||||
setContext({ ...getContext(), inFlightSegmentId: op.meta.id });
|
||||
// In-flight bookkeeping brackets the async region; the `finally` resets it
|
||||
// even if a step aborts or throws mid-pipeline. Only append ops track it.
|
||||
try {
|
||||
// Relocation (spike): on the FIRST media segment, fetch the whole body,
|
||||
// read its tfdt baseMediaDecodeTime, pair with the init timescale to get
|
||||
// this track's decode-time origin, and carry `timestampOffset = −origin`
|
||||
// in the meta so the actor sets it before appending. Established once;
|
||||
// subsequent segments stream as usual (the offset persists on the buffer).
|
||||
if (relocation.enabled && !relocation.established) {
|
||||
const bytes = await collect(await fetchBytes(op, { signal: taskSignal, minChunkSize: Infinity }));
|
||||
const baseMediaDecodeTime = readFirstBaseMediaDecodeTime(bytes);
|
||||
if (baseMediaDecodeTime !== undefined && relocation.timescale) {
|
||||
relocation.offset = -(baseMediaDecodeTime / relocation.timescale);
|
||||
// Publish on the actor snapshot so the text path can read the same
|
||||
// offset for cue rebasing (see SegmentLoaderActorContext.relocationOffset).
|
||||
setContext({ ...getContext(), relocationOffset: relocation.offset });
|
||||
}
|
||||
relocation.established = true;
|
||||
if (!taskSignal.aborted) {
|
||||
const meta = relocation.offset !== undefined ? { ...op.meta, timestampOffset: relocation.offset } : op.meta;
|
||||
sourceBufferActor.send({ type: 'append-segment', data: toArrayBuffer(bytes), meta });
|
||||
await waitForIdle(sourceBufferActor.snapshot, taskSignal);
|
||||
}
|
||||
} else {
|
||||
const stream = await fetchBytes(op, { signal: taskSignal });
|
||||
if (!taskSignal.aborted) {
|
||||
sourceBufferActor.send({ type: 'append-segment', data: stream, meta: op.meta });
|
||||
await waitForIdle(sourceBufferActor.snapshot, taskSignal);
|
||||
}
|
||||
if (op.type === 'append-init') setContext({ ...getContext(), inFlightInitTrackId: op.meta.trackId });
|
||||
else if (op.type === 'append-segment') setContext({ ...getContext(), inFlightSegmentId: op.meta.id });
|
||||
|
||||
for (const step of pipelines[op.type]) {
|
||||
if (taskSignal.aborted) return;
|
||||
await step(frame, taskSignal, deps);
|
||||
}
|
||||
} finally {
|
||||
setContext({ ...getContext(), inFlightSegmentId: null });
|
||||
if (op.type === 'append-init') setContext({ ...getContext(), inFlightInitTrackId: null });
|
||||
else if (op.type === 'append-segment') setContext({ ...getContext(), inFlightSegmentId: null });
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -348,8 +332,9 @@ export function createSegmentLoaderActor(
|
||||
|
||||
const forwardBufferConfig: ForwardBufferConfig = { ...DEFAULT_FORWARD_BUFFER_CONFIG, ...config.forwardBuffer };
|
||||
const backBufferConfig: BackBufferConfig = { ...DEFAULT_BACK_BUFFER_CONFIG, ...config.backBuffer };
|
||||
// Per-track relocation state (spike), established once on the first media segment.
|
||||
const relocation: RelocationState = { enabled: config.relocateTimestampOrigin ?? false, established: false };
|
||||
const deps: StepDeps = { sourceBufferActor, fetchBytes };
|
||||
// Built once per actor (fresh stateful steps per source); default is `fetch → dispatch`.
|
||||
const pipelines = (config.messagePipelines ?? DEFAULT_MESSAGE_PIPELINES)();
|
||||
|
||||
const getBufferedSegments = (allSegments: readonly Segment[]): Segment[] => {
|
||||
// Exclude partial segments — they are still being streamed and must not be
|
||||
@@ -527,22 +512,20 @@ export function createSegmentLoaderActor(
|
||||
|
||||
const scheduleAll = (tasks: LoadTask[], { getContext, setContext, runner }: Ctx): void => {
|
||||
tasks.forEach((op) => {
|
||||
runner
|
||||
.schedule(makeLoadTask(op, { getContext, setContext, fetchBytes, sourceBufferActor, relocation }))
|
||||
.then(undefined, (e: unknown) => {
|
||||
if (e instanceof Error && e.name === 'AbortError') return;
|
||||
// On unexpected fetch/append errors, abort remaining tasks so a failed
|
||||
// init doesn't cause segment fetches to proceed with no init segment.
|
||||
console.error('Unexpected error in segment loader:', e);
|
||||
runner.abortPending();
|
||||
});
|
||||
runner.schedule(makeLoadTask(op, { getContext, setContext, pipelines, deps })).then(undefined, (e: unknown) => {
|
||||
if (e instanceof Error && e.name === 'AbortError') return;
|
||||
// On unexpected fetch/append errors, abort remaining tasks so a failed
|
||||
// init doesn't cause segment fetches to proceed with no init segment.
|
||||
console.error('Unexpected error in segment loader:', e);
|
||||
runner.abortPending();
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
return createMachineActor<UserState, SegmentLoaderActorContext, SegmentLoaderMessage, () => SerialRunner>({
|
||||
runner: () => new SerialRunner(),
|
||||
initial: 'idle',
|
||||
context: { inFlightInitTrackId: null, inFlightSegmentId: null, relocationOffset: null },
|
||||
context: { inFlightInitTrackId: null, inFlightSegmentId: null },
|
||||
states: {
|
||||
idle: {
|
||||
on: {
|
||||
|
||||
@@ -18,10 +18,10 @@ export type AppendSegmentMeta = Pick<Segment, 'id' | 'startTime' | 'duration'> &
|
||||
/** Declared track bandwidth in bps (from playlist BANDWIDTH attribute). */
|
||||
trackBandwidth?: number;
|
||||
/**
|
||||
* Non-zero-PTS relocation (spike): when present, set as
|
||||
* `SourceBuffer.timestampOffset` before this append so native PTS is relocated
|
||||
* onto a 0-based presentation timeline. Constant per source; the loader
|
||||
* includes it once (on the first media segment). Absent = no relocation.
|
||||
* Non-zero-PTS relocation: when present, applied as `SourceBuffer.timestampOffset`
|
||||
* before this append so native PTS is relocated onto a 0-based presentation
|
||||
* timeline. A relocating composition stamps it (constant per source) onto each
|
||||
* media segment's meta; the apply is idempotent-guarded. Absent = no relocation.
|
||||
*/
|
||||
timestampOffset?: number;
|
||||
};
|
||||
@@ -158,10 +158,10 @@ function appendSegmentTask(
|
||||
});
|
||||
}
|
||||
|
||||
// Relocation (non-zero-PTS spike): set the offset before the coded frames
|
||||
// are appended. The SerialRunner guarantees the buffer is idle here, so the
|
||||
// assignment is safe; it's constant per source (loader sends it once).
|
||||
if (meta.timestampOffset != null) {
|
||||
// Relocation: set the offset before the coded frames are appended. The
|
||||
// SerialRunner guarantees the buffer is idle here, so the assignment is safe.
|
||||
// Guarded so re-stamping the (constant) offset on later appends is a no-op.
|
||||
if (meta.timestampOffset != null && sourceBuffer.timestampOffset !== meta.timestampOffset) {
|
||||
sourceBuffer.timestampOffset = meta.timestampOffset;
|
||||
}
|
||||
await appendSegment(sourceBuffer, message.data, taskSignal);
|
||||
|
||||
@@ -72,6 +72,7 @@ import type { BandwidthState } from '../../../network/bandwidth-estimator';
|
||||
import { createTrackedFetch, type FetchBytes, fetchStream } from '../../../network/fetch';
|
||||
import {
|
||||
createSegmentLoaderActor,
|
||||
type MessagePipelines,
|
||||
type SegmentLoaderActor,
|
||||
type SegmentLoaderActorConfig,
|
||||
} from '../../actors/dom/segment-loader';
|
||||
@@ -146,7 +147,7 @@ function setupBufferActors<K extends SelectedTrackKey, A extends BufferActorKey,
|
||||
fetch: FetchBytes;
|
||||
} & SegmentLoaderActorConfig;
|
||||
}): Reactor<BufferActorsFsmState | 'destroying' | 'destroyed'> {
|
||||
const { type, selectedKey, actorKey, loaderKey, fetch, forwardBuffer, backBuffer, relocateTimestampOrigin } = config;
|
||||
const { type, selectedKey, actorKey, loaderKey, fetch, forwardBuffer, backBuffer, messagePipelines } = config;
|
||||
const derivedStateSignal = computed<BufferActorsFsmState>(() => {
|
||||
if (!context.mediaSource.get()) return 'preconditions-unmet';
|
||||
const selection: TrackSelectionState = {
|
||||
@@ -175,7 +176,7 @@ function setupBufferActors<K extends SelectedTrackKey, A extends BufferActorKey,
|
||||
const segmentLoader = createSegmentLoaderActor(bufferActor, fetch, {
|
||||
forwardBuffer,
|
||||
backBuffer,
|
||||
relocateTimestampOrigin,
|
||||
messagePipelines,
|
||||
});
|
||||
|
||||
// Synchronous slot writes — load-bearing for the Firefox
|
||||
@@ -227,7 +228,11 @@ export const setupVideoBufferActors = defineBehavior({
|
||||
bandwidthState: Signal<BufferActorsState['bandwidthState']>;
|
||||
};
|
||||
context: BufferActorsContextMap<'videoBufferActor', 'videoSegmentLoaderActor'>;
|
||||
config?: SegmentLoaderActorConfig & { getCdnId?: GetCdnId };
|
||||
config?: SegmentLoaderActorConfig & {
|
||||
getCdnId?: GetCdnId;
|
||||
/** Optional non-zero-PTS relocation pipelines (Tier-1); inert when absent. */
|
||||
videoMessagePipelines?: MessagePipelines;
|
||||
};
|
||||
}) => {
|
||||
// Bandwidth-sampling fetch. The factory accumulates EWMA state
|
||||
// internally; the callback bridges samples to engine state for ABR.
|
||||
@@ -250,7 +255,11 @@ export const setupVideoBufferActors = defineBehavior({
|
||||
return setupBufferActors({
|
||||
state,
|
||||
context,
|
||||
config: { ...typeConfig, fetch: failoverFetch(trackedFetch, state, typeConfig) },
|
||||
config: {
|
||||
...typeConfig,
|
||||
fetch: failoverFetch(trackedFetch, state, typeConfig),
|
||||
messagePipelines: config.videoMessagePipelines,
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
@@ -283,14 +292,22 @@ export const setupAudioBufferActors = defineBehavior({
|
||||
}: {
|
||||
state: BufferActorsStateMap<'selectedAudioTrackId'>;
|
||||
context: BufferActorsContextMap<'audioBufferActor', 'audioSegmentLoaderActor'>;
|
||||
config?: SegmentLoaderActorConfig & { getCdnId?: GetCdnId };
|
||||
config?: SegmentLoaderActorConfig & {
|
||||
getCdnId?: GetCdnId;
|
||||
/** Optional non-zero-PTS relocation pipelines (Tier-1); inert when absent. */
|
||||
audioMessagePipelines?: MessagePipelines;
|
||||
};
|
||||
}) => {
|
||||
// Key order mirrors setupVideoBufferActors.
|
||||
const typeConfig = { ...AUDIO_TYPE_CONFIG, ...config };
|
||||
return setupBufferActors({
|
||||
state,
|
||||
context,
|
||||
config: { ...typeConfig, fetch: failoverFetch(fetchStream, state, typeConfig) },
|
||||
config: {
|
||||
...typeConfig,
|
||||
fetch: failoverFetch(fetchStream, state, typeConfig),
|
||||
messagePipelines: config.audioMessagePipelines,
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
@@ -20,9 +20,7 @@
|
||||
*/
|
||||
import { defineBehavior } from '../../../core/composition/create-composition';
|
||||
import { effect } from '../../../core/signals/effect';
|
||||
import { peek, type ReadonlySignal, type Signal } from '../../../core/signals/primitives';
|
||||
import { resolveVttSegmentWithMetadata } from '../../../media/dom/text/resolve-vtt-segment';
|
||||
import type { SegmentLoaderActor } from '../../actors/dom/segment-loader';
|
||||
import type { ReadonlySignal, Signal } from '../../../core/signals/primitives';
|
||||
import { createTextTracksActor } from '../../actors/dom/text-tracks';
|
||||
import {
|
||||
createTextTrackSegmentLoaderActor,
|
||||
@@ -36,75 +34,10 @@ export interface TextTrackActorsContext {
|
||||
mediaElement?: HTMLMediaElement | undefined;
|
||||
textTracksActor?: TextTracksActor<VTTCue> | undefined;
|
||||
textTrackSegmentLoaderActor?: TextTrackSegmentLoaderActor | undefined;
|
||||
/**
|
||||
* Read-only, for cue rebasing under non-zero-PTS relocation (spike). The video
|
||||
* loader establishes the shared decode-time offset and publishes it on its
|
||||
* snapshot; the relocating resolver below reads it so text cues land on the
|
||||
* same 0-based timeline as the relocated A/V.
|
||||
*/
|
||||
videoSegmentLoaderActor?: SegmentLoaderActor | undefined;
|
||||
}
|
||||
|
||||
export interface TextTrackActorsConfig extends TextTrackSegmentLoaderActorConfig {
|
||||
resolveTextTrackSegment: TextTrackSegmentResolver<VTTCue>;
|
||||
/**
|
||||
* Non-zero-PTS relocation (spike). When on, cues are rebased onto the relocated
|
||||
* 0-based presentation timeline: `cueFinal = cueNative + timestampOffset`, where
|
||||
* `cueNative = LOCAL + MPEGTS/90000` (`X-TIMESTAMP-MAP`) or the absolute cue time
|
||||
* (no map), and `timestampOffset` is the video loader's established relocation
|
||||
* offset. Off → the injected resolver is used unchanged (Tier 0).
|
||||
*/
|
||||
relocateTimestampOrigin?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve when the video loader has published its relocation offset (spike). Text
|
||||
* cues can't self-derive the origin for map-less sources (Mux), so the first text
|
||||
* segment must wait for the A/V ground truth rather than land ~60s off.
|
||||
*/
|
||||
function awaitRelocationOffset(videoLoader: ReadonlySignal<SegmentLoaderActor | undefined>): Promise<number> {
|
||||
const read = (): number | null => {
|
||||
const actor = peek(videoLoader);
|
||||
return actor ? peek(actor.snapshot).context.relocationOffset : null;
|
||||
};
|
||||
const current = read();
|
||||
if (current !== null) return Promise.resolve(current);
|
||||
return new Promise((resolve) => {
|
||||
let stop: (() => void) | undefined;
|
||||
stop = effect(() => {
|
||||
const actor = videoLoader.get();
|
||||
const offset = actor ? actor.snapshot.get().context.relocationOffset : null;
|
||||
if (offset !== null) {
|
||||
stop?.();
|
||||
resolve(offset);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap a cue resolver so cues are rebased onto the relocated 0-based timeline
|
||||
* (spike). `X-TIMESTAMP-MAP` (Apple) puts cues at LOCAL time and the map's
|
||||
* `mpegts/90000 − local` corrects them to the media timeline; absolute cues (Mux)
|
||||
* need no correction. Both then shift by the video loader's relocation offset.
|
||||
*/
|
||||
function makeRelocatingResolver(
|
||||
videoLoader: ReadonlySignal<SegmentLoaderActor | undefined>
|
||||
): TextTrackSegmentResolver<VTTCue> {
|
||||
return async (url) => {
|
||||
const { cues, metadata } = await resolveVttSegmentWithMetadata(url);
|
||||
const offset = await awaitRelocationOffset(videoLoader);
|
||||
const map = metadata.timestampMap;
|
||||
const mapCorrection = map?.local !== undefined ? map.mpegts / 90000 - map.local : 0;
|
||||
const delta = mapCorrection + offset;
|
||||
if (delta !== 0) {
|
||||
for (const cue of cues) {
|
||||
cue.startTime += delta;
|
||||
cue.endTime += delta;
|
||||
}
|
||||
}
|
||||
return cues;
|
||||
};
|
||||
}
|
||||
|
||||
function setupTextTrackActorsSetup({
|
||||
@@ -115,7 +48,6 @@ function setupTextTrackActorsSetup({
|
||||
mediaElement: ReadonlySignal<TextTrackActorsContext['mediaElement']>;
|
||||
textTracksActor: Signal<TextTrackActorsContext['textTracksActor']>;
|
||||
textTrackSegmentLoaderActor: Signal<TextTrackActorsContext['textTrackSegmentLoaderActor']>;
|
||||
videoSegmentLoaderActor: ReadonlySignal<TextTrackActorsContext['videoSegmentLoaderActor']>;
|
||||
};
|
||||
config: TextTrackActorsConfig;
|
||||
}): () => void {
|
||||
@@ -123,14 +55,12 @@ function setupTextTrackActorsSetup({
|
||||
const mediaElement = context.mediaElement.get();
|
||||
if (!mediaElement) return;
|
||||
|
||||
const resolveTextTrackSegment = config.relocateTimestampOrigin
|
||||
? makeRelocatingResolver(context.videoSegmentLoaderActor)
|
||||
: config.resolveTextTrackSegment;
|
||||
|
||||
const textTracksActor = createTextTracksActor(mediaElement);
|
||||
const textTrackSegmentLoaderActor = createTextTrackSegmentLoaderActor(textTracksActor, resolveTextTrackSegment, {
|
||||
forwardBuffer: config.forwardBuffer,
|
||||
});
|
||||
const textTrackSegmentLoaderActor = createTextTrackSegmentLoaderActor(
|
||||
textTracksActor,
|
||||
config.resolveTextTrackSegment,
|
||||
{ forwardBuffer: config.forwardBuffer }
|
||||
);
|
||||
context.textTracksActor.set(textTracksActor);
|
||||
context.textTrackSegmentLoaderActor.set(textTrackSegmentLoaderActor);
|
||||
|
||||
@@ -145,6 +75,6 @@ function setupTextTrackActorsSetup({
|
||||
|
||||
export const setupTextTrackActors = defineBehavior({
|
||||
stateKeys: [],
|
||||
contextKeys: ['mediaElement', 'textTracksActor', 'textTrackSegmentLoaderActor', 'videoSegmentLoaderActor'],
|
||||
contextKeys: ['mediaElement', 'textTracksActor', 'textTrackSegmentLoaderActor'],
|
||||
setup: setupTextTrackActorsSetup,
|
||||
});
|
||||
|
||||
@@ -10,7 +10,6 @@ import type {
|
||||
Segment,
|
||||
TextTrack,
|
||||
} from '../../../../media/types';
|
||||
import type { SegmentLoaderActor } from '../../../actors/dom/segment-loader';
|
||||
import type { TextTrackSegmentLoaderActor } from '../../../actors/text-track-segment-loader';
|
||||
import type { TextTracksActor } from '../../../actors/text-tracks';
|
||||
import { loadTextTrackSegments } from '../load-segments';
|
||||
@@ -45,9 +44,6 @@ vi.mock('../../../../media/dom/text/resolve-vtt-segment', () => ({
|
||||
}
|
||||
return Promise.resolve([new VTTCue(0, 5, `Subtitle from ${url}`)]);
|
||||
}),
|
||||
resolveVttSegmentWithMetadata: vi.fn((url: string) =>
|
||||
Promise.resolve({ cues: [new VTTCue(0, 5, `Subtitle from ${url}`)], metadata: {} })
|
||||
),
|
||||
destroyVttResolver: vi.fn(),
|
||||
}));
|
||||
|
||||
@@ -70,7 +66,6 @@ function makeContext(initial: ComposedContext = {}): ContextSignals<ComposedCont
|
||||
initial.textTracksActor as TextTracksActor<VTTCue & Cue> | undefined
|
||||
) as ContextSignals<ComposedContext>['textTracksActor'],
|
||||
textTrackSegmentLoaderActor: signal<TextTrackSegmentLoaderActor | undefined>(initial.textTrackSegmentLoaderActor),
|
||||
videoSegmentLoaderActor: signal<SegmentLoaderActor | undefined>(initial.videoSegmentLoaderActor),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ import type {
|
||||
import type { GetCdnId } from '../../../media/utils/cdn';
|
||||
import { getResolvedSelectedTrackDuration } from '../../../media/utils/track-selection';
|
||||
import type { BandwidthConfig, BandwidthState } from '../../../network/bandwidth-estimator';
|
||||
import type { SegmentLoaderActor } from '../../actors/dom/segment-loader';
|
||||
import type { MessagePipelines, SegmentLoaderActor } from '../../actors/dom/segment-loader';
|
||||
import type { SourceBufferActor } from '../../actors/dom/source-buffer';
|
||||
import type { TextTracksActor } from '../../actors/dom/text-tracks';
|
||||
import type { TextTrackSegmentLoaderActor, TextTrackSegmentResolver } from '../../actors/text-track-segment-loader';
|
||||
@@ -275,15 +275,15 @@ export interface SimpleHlsEngineConfig extends ShareSignalsConfig<SimpleHlsEngin
|
||||
*/
|
||||
getCdnId?: GetCdnId;
|
||||
/**
|
||||
* Non-zero-PTS relocation (spike). When on, the engine reads each source's
|
||||
* decode-time origin (`tfdt` baseMediaDecodeTime ÷ `mdhd` timescale, via
|
||||
* `media/mp4`) and relocates the buffer onto a 0-based presentation timeline
|
||||
* with `SourceBuffer.timestampOffset = −sharedOrigin`, so `currentTime` /
|
||||
* `seekable` / `duration` stay 0-based with no adapter translation. **Off by
|
||||
* default** — zero-PTS VOD needs none of this and pays nothing (Tier 0). See
|
||||
* `internal/design/spf/presentation-timeline-model.md`.
|
||||
* Non-zero-PTS relocation seams (Tier-1). Per-type segment-loader pipelines
|
||||
* that weave discover + stamp steps between fetch and dispatch; generic and
|
||||
* inert when absent (Tier 0 pays nothing and imports no relocation code).
|
||||
* Rather than set these by hand, spread `createRelocation()` (`./relocation`)
|
||||
* into config — it builds both pipelines and the cue resolver against shared
|
||||
* offset signals. See `internal/design/spf/presentation-timeline-model.md`.
|
||||
*/
|
||||
relocateTimestampOrigin?: boolean;
|
||||
videoMessagePipelines?: MessagePipelines;
|
||||
audioMessagePipelines?: MessagePipelines;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
@@ -352,7 +352,6 @@ export function createSimpleHlsEngine(
|
||||
addSubtitlesTracksToMedia: config.addSubtitlesTracksToMedia ?? addSubtitlesTracksToMedia,
|
||||
getShowingSubtitlesTrackFromMedia: config.getShowingSubtitlesTrackFromMedia ?? getShowingSubtitlesTrackFromMedia,
|
||||
removeAllSubtitlesTracksFromMedia: config.removeAllSubtitlesTracksFromMedia ?? removeAllSubtitlesTracksFromMedia,
|
||||
relocateTimestampOrigin: config.relocateTimestampOrigin ?? false,
|
||||
};
|
||||
|
||||
const composition = createComposition(
|
||||
|
||||
@@ -25,3 +25,5 @@ export type {
|
||||
SimpleHlsAudioOnlyEngineState,
|
||||
} from './engine-audio-only';
|
||||
export { createHlsAudioOnlyEngine } from './engine-audio-only';
|
||||
export type { Relocation } from './relocation';
|
||||
export { createRelocation } from './relocation';
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
/**
|
||||
* Non-zero-PTS relocation assembly (Tier-1). Bundles the seam values a
|
||||
* composition injects to relocate a non-zero-PTS source onto a 0-based
|
||||
* presentation timeline — kept in one module so a Tier-0 composition that never
|
||||
* calls {@link createRelocation} tree-shakes away the mp4 parser, the VTT
|
||||
* metadata scraper, and the origin discovery.
|
||||
*
|
||||
* Relocation is expressed as extra **steps** woven into the segment loader's
|
||||
* per-message pipelines (see `MessagePipelines`), not as bespoke loader config.
|
||||
* The loader ships a Tier-0 `fetch → dispatch` pipeline and stays oblivious to
|
||||
* relocation; this module returns pipelines that insert two steps between them:
|
||||
* - **tapOrigin** — a head-peek on the fetched byte stream that reads the
|
||||
* decode-time origin (`tfdt` baseMediaDecodeTime ÷ `mdhd` timescale) and
|
||||
* publishes the offset to a per-track signal (kept out of `state.presentation`
|
||||
* per the lost-update hazard in the presentation-timeline model). One shared
|
||||
* discoverer spans a track's init (timescale) and first media segment (tfdt).
|
||||
* - **stampOffset** — stamps `timestampOffset` onto the append meta, so the
|
||||
* SourceBufferActor relocates the buffer via `SourceBuffer.timestampOffset`.
|
||||
* Synchronous: `tapOrigin` runs earlier in the same segment's pipeline, so the
|
||||
* offset is already published by the time it reads.
|
||||
*
|
||||
* Text has no SourceBuffer, so its offset is applied as cue arithmetic in the
|
||||
* resolver: `cueFinal = cueNative + offset`, `cueNative = LOCAL + MPEGTS/90000`
|
||||
* (`X-TIMESTAMP-MAP`) or the absolute cue time (no map). Text reads the *video*
|
||||
* offset because map-less sources (Mux) can't self-derive the origin, and awaits
|
||||
* it (unlike stampOffset) because a text segment may resolve before video
|
||||
* establishes.
|
||||
*
|
||||
* Per-track offsets (not one shared min): relocating every track by its own
|
||||
* origin keeps each earliest DTS ≥ 0 (a video-primary offset would push Mux's
|
||||
* slightly-earlier audio negative). This is the single-origin coordination axis;
|
||||
* a `min`-reduce across A/V is a later capability.
|
||||
*/
|
||||
import { effect } from '../../../core/signals/effect';
|
||||
import { peek, type ReadonlySignal, type Signal, signal } from '../../../core/signals/primitives';
|
||||
import { resolveVttSegmentWithMetadata } from '../../../media/dom/text/resolve-vtt-segment';
|
||||
import { dispatchStep, fetchStep, type LoadStep, type MessagePipelines } from '../../actors/dom/segment-loader';
|
||||
import type { TextTrackSegmentResolver } from '../../actors/text-track-segment-loader';
|
||||
import { createOriginDiscoverer } from '../../primitives/origin-discoverer';
|
||||
|
||||
/** Head-peek the fetched stream, reading the decode-time origin and publishing it (see {@link createOriginDiscoverer}). */
|
||||
function tapOrigin(publish: (offsetSeconds: number) => void): LoadStep {
|
||||
const discover = createOriginDiscoverer(publish);
|
||||
return async (frame) => {
|
||||
if (frame.data) frame.data = await discover(frame.data);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Stamp the established offset onto the append meta (no-op until it resolves).
|
||||
* Synchronous: `tapOrigin` runs earlier in the same segment's pipeline, so the
|
||||
* offset is published by now. A Tier-2 shared-`min` variant would await the
|
||||
* cross-track reduce here instead.
|
||||
*/
|
||||
function stampOffset(offset: ReadonlySignal<number | undefined>): LoadStep {
|
||||
return (frame) => {
|
||||
const timestampOffset = peek(offset);
|
||||
if (timestampOffset != null && frame.meta) frame.meta = { ...frame.meta, timestampOffset };
|
||||
};
|
||||
}
|
||||
|
||||
/** Resolve once `source` holds a number — so the first text cue waits for the A/V ground truth. */
|
||||
function awaitDefined(source: ReadonlySignal<number | undefined>): Promise<number> {
|
||||
const current = peek(source);
|
||||
if (current !== undefined) return Promise.resolve(current);
|
||||
return new Promise((resolve) => {
|
||||
let stop: (() => void) | undefined;
|
||||
stop = effect(() => {
|
||||
const value = source.get();
|
||||
if (value !== undefined) {
|
||||
stop?.();
|
||||
resolve(value);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function createRelocatingTextResolver(offset: ReadonlySignal<number | undefined>): TextTrackSegmentResolver<VTTCue> {
|
||||
return async (url) => {
|
||||
const { cues, metadata } = await resolveVttSegmentWithMetadata(url);
|
||||
const relocationOffset = await awaitDefined(offset);
|
||||
const map = metadata.timestampMap;
|
||||
// The LOCAL→native correction is `mpegts/90000 − local`; absent map → 0.
|
||||
const mapCorrection = map?.local !== undefined ? map.mpegts / 90000 - map.local : 0;
|
||||
const delta = mapCorrection + relocationOffset;
|
||||
if (delta !== 0) {
|
||||
for (const cue of cues) {
|
||||
cue.startTime += delta;
|
||||
cue.endTime += delta;
|
||||
}
|
||||
}
|
||||
return cues;
|
||||
};
|
||||
}
|
||||
|
||||
/** Build a track's pipelines around its own offset signal. `tapOrigin`'s discoverer is created once per actor (shared across init + segments). */
|
||||
function relocatingPipelines(offset: Signal<number | undefined>): MessagePipelines {
|
||||
const publish = (offsetSeconds: number) => offset.set(offsetSeconds);
|
||||
return () => {
|
||||
// One discoverer per actor: the init establishes the timescale, the first
|
||||
// media segment establishes `tfdt` + publishes — so the same `tapOrigin`
|
||||
// step must be shared across both pipelines.
|
||||
const tap = tapOrigin(publish);
|
||||
return {
|
||||
remove: [dispatchStep],
|
||||
'append-init': [fetchStep, tap, dispatchStep],
|
||||
'append-segment': [fetchStep, tap, stampOffset(offset), dispatchStep],
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
/** Seam values a composition injects to enable non-zero-PTS relocation. */
|
||||
export interface Relocation {
|
||||
videoMessagePipelines: MessagePipelines;
|
||||
audioMessagePipelines: MessagePipelines;
|
||||
resolveTextTrackSegment: TextTrackSegmentResolver<VTTCue>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the relocation seam bundle. Spread into a `SimpleHlsEngineConfig` to
|
||||
* compose a non-zero-PTS engine:
|
||||
*
|
||||
* ```ts
|
||||
* createSimpleHlsEngine({ ...config, ...createRelocation() });
|
||||
* ```
|
||||
*/
|
||||
export function createRelocation(): Relocation {
|
||||
const videoOffset = signal<number | undefined>(undefined);
|
||||
const audioOffset = signal<number | undefined>(undefined);
|
||||
return {
|
||||
videoMessagePipelines: relocatingPipelines(videoOffset),
|
||||
audioMessagePipelines: relocatingPipelines(audioOffset),
|
||||
resolveTextTrackSegment: createRelocatingTextResolver(videoOffset),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
/**
|
||||
* Non-zero-PTS **discover**: a slim, eager head-peek decoration on a segment's
|
||||
* byte stream that reads the decode-time origin and publishes the relocation
|
||||
* offset — the content-inspection half of relocation, kept out of the fetch
|
||||
* (transport) abstraction.
|
||||
*
|
||||
* Tier-1-only — the always-present segment loader never imports this; a
|
||||
* relocating composition injects it as the loader's `discover` seam, so the mp4
|
||||
* parser tree-shakes out of a Tier-0 build.
|
||||
*
|
||||
* Unlike a throughput/failover fetch tap (which observes chunks *post-hoc*, as
|
||||
* the appender pulls them), discovery feeds the *same* segment's append —
|
||||
* `SourceBuffer.timestampOffset` must be set before the frames append — so it
|
||||
* reads the head **eagerly**: pull chunks only until the boxes parse (usually
|
||||
* one), publish, then re-emit the pulled head followed by the untouched tail, so
|
||||
* the append still **streams**. The init segment carries `mdhd` timescale, a
|
||||
* media segment carries `tfdt` baseMediaDecodeTime (both readers return
|
||||
* `undefined` when their box is absent, so one discoverer handles both and
|
||||
* self-discriminates). Once established it's a pure pass-through.
|
||||
*/
|
||||
import { readFirstBaseMediaDecodeTime, readFirstMediaTimescale } from '../../media/mp4/timestamp-origin';
|
||||
|
||||
function concat(chunks: Uint8Array[]): Uint8Array {
|
||||
if (chunks.length === 1) return chunks[0]!;
|
||||
const total = chunks.reduce((n, c) => n + c.length, 0);
|
||||
const out = new Uint8Array(total);
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
out.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Re-emit the eagerly-pulled head chunks, then stream the untouched tail. */
|
||||
async function* reassemble(head: Uint8Array[], tail: AsyncIterator<Uint8Array>): AsyncIterable<Uint8Array> {
|
||||
yield* head;
|
||||
for (let next = await tail.next(); !next.done; next = await tail.next()) yield next.value;
|
||||
}
|
||||
|
||||
export function createOriginDiscoverer(
|
||||
publish: (offsetSeconds: number) => void
|
||||
): (data: AsyncIterable<Uint8Array>) => Promise<AsyncIterable<Uint8Array>> {
|
||||
let timescale: number | undefined;
|
||||
let established = false;
|
||||
return async (data) => {
|
||||
if (established) return data;
|
||||
const iterator = data[Symbol.asyncIterator]();
|
||||
const head: Uint8Array[] = [];
|
||||
for (let next = await iterator.next(); !next.done; next = await iterator.next()) {
|
||||
head.push(next.value);
|
||||
const bytes = concat(head);
|
||||
timescale ??= readFirstMediaTimescale(bytes);
|
||||
const baseMediaDecodeTime = readFirstBaseMediaDecodeTime(bytes);
|
||||
if (baseMediaDecodeTime !== undefined && timescale !== undefined) {
|
||||
publish(-(baseMediaDecodeTime / timescale));
|
||||
established = true;
|
||||
break;
|
||||
}
|
||||
// Init data (timescale but no `moof`) — nothing more to peek here.
|
||||
if (timescale !== undefined) break;
|
||||
}
|
||||
return reassemble(head, iterator);
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user