v10/packages/spf/src/playback/actors/dom/segment-loader.ts

import { createMachineActor, type HandlerContext, type MessageActor } from '../../../core/actors/create-machine-actor';
import type { AnySlotMap } from '../../../core/composition/create-composition';
import { effect } from '../../../core/signals/effect';
import { peek } from '../../../core/signals/primitives';
import { SerialRunner, Task } from '../../../core/tasks/task';
import {
  type BackBufferConfig,
  calculateBackBufferFlushPoint,
  DEFAULT_BACK_BUFFER_CONFIG,
} from '../../../media/buffer/back-buffer';
import {
  calculateForwardFlushPoint,
  DEFAULT_FORWARD_BUFFER_CONFIG,
  type ForwardBufferConfig,
  getSegmentsToLoad,
} from '../../../media/buffer/forward-buffer';
import {
  type AddressableObject,
  type AudioTrack,
  SEGMENT_TIME_EPSILON,
  type Segment,
  type VideoTrack,
} from '../../../media/types';
import type {
  AppendInitMessage,
  AppendSegmentMessage,
  IndividualSourceBufferMessage,
  RemoveMessage,
  SourceBufferActor,
} from './source-buffer';

// ============================================================================
// BUFFER STATE TYPES
// ============================================================================

/**
 * Buffer state for a single SourceBuffer.
 * Tracks which init segment and media segments are loaded.
 */
export interface SourceBufferState {
  /** Track ID of the loaded init segment */
  initTrackId?: string;
  /** Loaded media segments (unordered - selectors derive ordering) */
  segments: Array<{ id: string; trackId: string }>;
}

/**
 * Buffer state for all SourceBuffers.
 */
export interface BufferState {
  video?: SourceBufferState;
  audio?: SourceBufferState;
}

// ============================================================================
// MESSAGE PROTOCOL
// ============================================================================

/** Track types that have SourceBuffers (video and audio only). */
export type SegmentLoaderTrack = VideoTrack | AudioTrack;

/**
 * Message sent to a SegmentLoaderActor.
 *
 * `range` is optional to distinguish loading modes:
 * - No range: load init segment only (metadata preload mode)
 * - With range: load init + all segments overlapping [start, end]
 *
 * `start` and `end` are raw time values — no segment snapping.
 * The actor maps them onto segment boundaries internally.
 */
export type SegmentLoaderMessage = {
  type: 'load';
  track: SegmentLoaderTrack;
  range?: { start: number; end: number };
};

// ============================================================================
// LOAD TASK
// ============================================================================

/**
 * A LoadTask is the intent to perform one unit of SegmentLoader work.
 * Unlike SourceBufferMessage, fetch-based tasks carry a URL rather than
 * pre-fetched data — the runner fetches and appends them in sequence.
 *
 * Derived from SourceBufferMessage types by removing `data` and adding
 * a fetch URL via AddressableObject.
 *
 * @todo Rename — "LoadTask" risks confusion with the `Task` class used for
 * SourceBufferActor scheduling. These are closer to operation descriptors or
 * messages than tasks in that sense.
 */
export type LoadTask =
  | (Omit<AppendInitMessage, 'data'> & AddressableObject)
  | (Omit<AppendSegmentMessage, 'data'> & AddressableObject)
  | RemoveMessage;

// ============================================================================
// ACTOR INTERFACE
// ============================================================================

/** Finite states of the actor. */
export type SegmentLoaderActorState = 'idle' | 'loading' | 'destroyed';

/** Non-finite (extended) data managed by the actor. */
export interface SegmentLoaderActorContext {
  /** Track ID of the init segment currently being fetched/appended, or null. */
  inFlightInitTrackId: string | null;
  /** Segment ID currently being fetched/appended, or null. */
  inFlightSegmentId: string | 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, and a step that needs composition signals
 * (relocation's discover/stamp) reads them from `deps` at call time.
 */
export type LoadStep = (frame: Frame, signal: AbortSignal, deps: StepDeps) => void | Promise<void>;

/**
 * The uniform passthrough handed to every {@link LoadStep} on every call — the
 * composition triple, nothing more. `state`/`context` are the full composition signal
 * maps; `config` is the threaded config with the loader's own wiring folded in (see
 * {@link stepWiring} + `createSegmentLoaderActor`). All three are typed loose: the
 * loader is a conduit and never reads them; a step asserts the slots it knows are
 * present (composition steps read `state`; base steps read the folded wiring off
 * `config`).
 */
export interface StepDeps {
  state: AnySlotMap;
  context: AnySlotMap;
  config: object;
}

/**
 * Base-step view of the loader's own wiring. `createSegmentLoaderActor` folds its
 * `sourceBufferActor` + `fetch` into the threaded `config` so base steps read them
 * from the uniform passthrough — present whether the loader runs inside a composition
 * or standalone. `config` is loose (`object`), so assert the shape here (one cast, like
 * relocation's `containerSlot`).
 */
function stepWiring(deps: StepDeps): { sourceBufferActor: SourceBufferActor; fetch: FetchBytes } {
  return deps.config as { sourceBufferActor: SourceBufferActor; fetch: 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 `establishStartMediaTime`), 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
 * override individual fields.
 */
export interface SegmentLoaderActorConfig {
  forwardBuffer?: Partial<ForwardBufferConfig>;
  backBuffer?: Partial<BackBufferConfig>;
  /** Per-message-type step pipelines. Defaults to {@link DEFAULT_MESSAGE_PIPELINES} (`fetch → dispatch`). */
  messagePipelines?: MessagePipelines;
}

// ============================================================================
// HELPERS
// ============================================================================

type FetchBytes = (
  addressable: AddressableObject,
  options?: RequestInit & { minChunkSize?: number }
) => Promise<AsyncIterable<Uint8Array>>;

/**
 * Resolves when the SourceBufferActor snapshot reaches 'idle'.
 * Rejects if the signal is aborted or the actor is destroyed.
 *
 * Used to sequence SourceBufferActor operations without awaiting send()
 * directly — send() is fire-and-forget; callers observe completion via
 * state transition.
 */
function waitForIdle(snapshot: SourceBufferActor['snapshot'], signal: AbortSignal): Promise<void> {
  return new Promise((resolve, reject) => {
    if (snapshot.get().value === 'idle') {
      resolve();
      return;
    }
    if (snapshot.get().value === 'destroyed') {
      reject(new DOMException('Aborted', 'AbortError'));
      return;
    }
    if (signal.aborted) {
      reject(signal.reason);
      return;
    }

    let stop: (() => void) | undefined;

    const cleanup = (fn: () => void) => {
      stop?.();
      signal.removeEventListener('abort', onAbort);
      fn();
    };

    const onAbort = () => cleanup(() => reject(signal.reason));

    stop = effect(() => {
      const value = snapshot.get().value;
      if (value === 'idle') cleanup(resolve);
      else if (value === 'destroyed') cleanup(() => reject(new DOMException('Aborted', 'AbortError')));
    });

    signal.addEventListener('abort', onAbort, { once: true });
  });
}

// ============================================================================
// 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 };
  }
}

/**
 * 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
  const { fetch } = stepWiring(deps);
  frame.data = await fetch(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) => {
  const { sourceBufferActor } = stepWiring(deps);
  sourceBufferActor.send(toMessage(frame));
  await waitForIdle(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
// ============================================================================

interface LoadTaskOptions {
  getContext: () => SegmentLoaderActorContext;
  setContext: (ctx: SegmentLoaderActorContext) => void;
  pipelines: Record<LoadTask['type'], LoadStep[]>;
  deps: StepDeps;
}

/**
 * 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, pipelines, deps }: LoadTaskOptions): Task<void> {
  return new Task(async (taskSignal) => {
    if (taskSignal.aborted) return;

    const frame: Frame = op.type === 'append-segment' ? { op, meta: op.meta } : { op };

    // 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 {
      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 {
      if (op.type === 'append-init') setContext({ ...getContext(), inFlightInitTrackId: null });
      else if (op.type === 'append-segment') setContext({ ...getContext(), inFlightSegmentId: null });
    }
  });
}

// ============================================================================
// IMPLEMENTATION
// ============================================================================

/**
 * Creates a SegmentLoaderActor for one track type (video or audio).
 *
 * Receives load assignments via `send()` and owns all execution: planning,
 * removes, fetches, and appends. Coordinates with the SourceBufferActor for
 * all physical SourceBuffer operations.
 *
 * Planning (Cases 1–3) happens in the `load` handler on every incoming
 * message, producing an ordered LoadTask list. The runner drains that list
 * sequentially via SerialRunner. When a new message arrives mid-run, the
 * handler replans and either continues the in-flight operation (abortPending
 * + schedule new remainder) or preempts it (abortAll + cancel SourceBuffer
 * if needed + schedule new plan).
 *
 * @param sourceBufferActor - Shared SourceBufferActor reference (not owned)
 * @param fetchBytes - Tracked fetch closure (owns throughput sampling for segments).
 *   Accepts an optional `minChunkSize` in options; init segments pass `Infinity`
 *   so the entire body accumulates as one chunk before appending.
 * @param compositionDeps - The composition's `state`/`context`/`config`, threaded
 *   opaquely into each step's {@link StepDeps} (the loader never reads them). Lets
 *   injected steps (relocation) read composition signals at call time. Defaults to
 *   empty for standalone / base-pipeline use.
 */
export function createSegmentLoaderActor(
  sourceBufferActor: SourceBufferActor,
  fetchBytes: FetchBytes,
  config: SegmentLoaderActorConfig = {},
  compositionDeps: StepDeps = { state: {}, context: {}, config: {} }
): SegmentLoaderActor {
  type UserState = Exclude<SegmentLoaderActorState, 'destroyed'>;
  type Ctx = HandlerContext<UserState, SegmentLoaderActorContext, () => SerialRunner>;

  const forwardBufferConfig: ForwardBufferConfig = { ...DEFAULT_FORWARD_BUFFER_CONFIG, ...config.forwardBuffer };
  const backBufferConfig: BackBufferConfig = { ...DEFAULT_BACK_BUFFER_CONFIG, ...config.backBuffer };
  // Fold the loader's own wiring into the passthrough `config` (see `stepWiring`) so
  // base steps read it from the uniform `{state,context,config}` — present in both
  // composition and standalone use.
  const deps: StepDeps = {
    state: compositionDeps.state,
    context: compositionDeps.context,
    config: { ...compositionDeps.config, sourceBufferActor, fetch: 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
    // treated as fully buffered for load planning or buffer window calculations.
    //
    // `peek` defensively: `load` handlers run synchronously inside `send()`,
    // which is called from inside the dispatcher reactor's `effects:` body.
    // A tracked `.snapshot.get()` here would leak the source-buffer-actor's
    // snapshot into the dispatcher's dep set, causing the dispatcher to re-
    // fire on every SourceBufferActor state change. Mirrors the fix applied
    // to the text-track loader in `b3f44efe`.
    const bufferedIds = new Set(
      peek(sourceBufferActor.snapshot)
        .context.segments.filter((s) => !s.partial)
        .map((s) => s.id)
    );
    return allSegments.filter((s) => bufferedIds.has(s.id));
  };

  /**
   * Translate a load message into an ordered LoadTask list based on committed
   * actor state. In-flight awareness is handled separately in the load handler.
   *
   * @todo Rename alongside LoadTask (e.g. planOps).
   *
   * Case 1 — Removes: forward and back buffer flush points, segment-aligned.
   *   ABR-style track switches (same content, different bitrate) do not flush:
   *   appending new content overwrites existing buffer ranges, and the actor's
   *   time-aligned deduplication keeps the segment model accurate as new
   *   segments arrive.
   *
   *   Cross-rendition track switches (audio language change, text language
   *   change) do flush: the buffered content is semantically incompatible with
   *   the newly-selected track, so overwrite-on-append would leave stale
   *   content playing until each replacement segment lands. Today's predicate:
   *   `actorCtx.initTrackLanguage !== track.language` — fires for language
   *   changes, no-ops for video / same-language audio bitrate switches.
   *   Future stage: pluggable predicate / strategy at actor construction time
   *   for codec-change (5.1 surround) and other cross-rendition shapes.
   *
   * Case 2 — Init: schedule if not yet committed for this track.
   *
   * Case 3 — Segments: all segments in the load window not yet committed.
   */
  const planTasks = (message: SegmentLoaderMessage): LoadTask[] => {
    const { track, range } = message;
    // `peek` for the same reason as `getBufferedSegments` above — avoid
    // leaking the SourceBufferActor snapshot into the calling dispatcher's
    // tracking scope.
    const actorCtx = peek(sourceBufferActor.snapshot).context;
    const bufferedSegments = getBufferedSegments(track.segments);
    const currentTime = range?.start ?? 0;
    const tasks: LoadTask[] = [];

    // Cross-rendition switch check (mid-stream language change). Fires when
    // (a) an init segment has already been committed for some track,
    // (b) the newly-selected track is a different track, and
    // (c) the languages differ. The buffered range from the current segment
    // boundary forward is treated as stale (new track's same-timestamp
    // segments will overwrite it via MSE append-at-same-timestamp); no
    // explicit `remove` task is emitted for the cross-rendition range.
    // Computed against the currently-buffered segments (stable reference;
    // new track's playlist may not be resolved yet at first load).
    const isCrossRenditionSwitch =
      actorCtx.initTrackId !== undefined &&
      actorCtx.initTrackId !== track.id &&
      actorCtx.initTrackLanguage !== track.language;

    // Two categories of "buffered content that should not gate planning":
    //
    // - `removes` — content that needs an explicit `remove` task (out-of-window
    //   forward content, back-buffer content beyond the keep window). Emitted
    //   as `{ type: 'remove' }` tasks.
    // - `staleRanges` — content that the new appends will overwrite at the
    //   same timestamps (cross-rendition switch). No explicit `remove` —
    //   MSE's overwrite-on-append handles it, which gives a much smaller
    //   perceived audio gap than `remove`-then-fetch-then-append.
    //
    // Both categories affect `effectiveBuffered` so `getSegmentsToLoad`
    // re-plans new-track segments inside them.
    const removes: Array<{ start: number; end: number }> = [];
    const staleRanges: Array<{ start: number; end: number }> = [];
    if (range) {
      if (isCrossRenditionSwitch) {
        // Mark current-segment-start onward as stale. Falls back to the
        // first buffered segment after currentTime when the playhead sits
        // in a buffer gap.
        const currentSeg = actorCtx.segments.find(
          (s) => s.startTime <= currentTime && s.startTime + s.duration > currentTime
        );
        const staleStart = currentSeg?.startTime ?? actorCtx.segments.find((s) => s.startTime > currentTime)?.startTime;
        if (staleStart !== undefined) {
          staleRanges.push({ start: staleStart, end: Infinity });
        }
      }
      const forwardFlushStart = calculateForwardFlushPoint(bufferedSegments, currentTime, forwardBufferConfig);
      if (forwardFlushStart < Infinity) {
        removes.push({ start: forwardFlushStart, end: Infinity });
      }
      const backFlushEnd = calculateBackBufferFlushPoint(bufferedSegments, currentTime, backBufferConfig);
      if (backFlushEnd > 0) {
        removes.push({ start: 0, end: backFlushEnd });
      }
      for (const r of removes) tasks.push({ type: 'remove', start: r.start, end: r.end });
    }

    // Treat any segment overlapping a planned remove OR a stale range as
    // not-buffered. Without this, sibling renditions that share segment IDs
    // and startTimes (audio language variants) would see the new track's
    // same-startTime segments marked "buffered" by the pre-flush snapshot,
    // skip them in `getSegmentsToLoad`, and leave a permanent gap from
    // `currentSegmentStart` to the end of the old buffer window — stalling
    // playback.
    const overlapsStale = (seg: { startTime: number; duration: number }): boolean => {
      const segEnd = seg.startTime + seg.duration;
      return (
        removes.some((r) => seg.startTime < r.end && segEnd > r.start) ||
        staleRanges.some((r) => seg.startTime < r.end && segEnd > r.start)
      );
    };
    const effectiveBuffered =
      removes.length + staleRanges.length > 0 ? bufferedSegments.filter((s) => !overlapsStale(s)) : bufferedSegments;

    // Case 2: Init
    if (actorCtx.initTrackId !== track.id) {
      tasks.push({
        type: 'append-init',
        meta: { trackId: track.id, language: track.language },
        url: track.initialization.url,
        ...(track.initialization.byteRange !== undefined && { byteRange: track.initialization.byteRange }),
      });
    }

    // Case 3: Segments
    if (range) {
      const segmentsToLoad = getSegmentsToLoad(
        track.segments,
        effectiveBuffered,
        currentTime,
        forwardBufferConfig
      ).filter((seg) => {
        // Quality-aware filter: skip segments already covered by equal-or-higher-quality
        // content in the actor context. Preserves buffered high-quality content during
        // ABR downgrades; loads during upgrades and for uncovered positions.
        // Actor entries that overlap a planned remove OR a stale range are treated
        // as nonexistent here — they're about to be flushed or overwritten, so the
        // new-track segment must load regardless of the existing entry's bandwidth.
        const existing = actorCtx.segments.find(
          (s) => !overlapsStale(s) && Math.abs(s.startTime - seg.startTime) < SEGMENT_TIME_EPSILON
        );
        // Partial segments are still streaming — treat as not buffered so they
        // are always re-planned (avoids relying on incomplete data).
        if (existing?.partial) return true;
        if (!existing?.trackBandwidth || !track.bandwidth) return true;
        return track.bandwidth > existing.trackBandwidth;
      });
      for (const segment of segmentsToLoad) {
        tasks.push({
          type: 'append-segment',
          meta: {
            id: segment.id,
            startTime: segment.startTime,
            duration: segment.duration,
            trackId: track.id,
            trackBandwidth: track.bandwidth,
          },
          url: segment.url,
          ...(segment.byteRange !== undefined && { byteRange: segment.byteRange }),
        });
      }
    }

    return tasks;
  };

  const scheduleAll = (tasks: LoadTask[], { getContext, setContext, runner }: Ctx): void => {
    tasks.forEach((op) => {
      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 },
    states: {
      idle: {
        on: {
          load: (msg, ctx) => {
            const allTasks = planTasks(msg);
            if (allTasks.length === 0) return;
            ctx.transition('loading');
            scheduleAll(allTasks, ctx);
          },
        },
      },
      loading: {
        onSettled: 'idle',
        on: {
          load: (msg, ctx) => {
            const { context, runner } = ctx;
            const allTasks = planTasks(msg);

            // Determine whether the in-flight operation is still needed.
            const inFlightStillNeeded =
              (context.inFlightSegmentId !== null &&
                allTasks.some((t) => t.type === 'append-segment' && t.meta.id === context.inFlightSegmentId)) ||
              (context.inFlightInitTrackId !== null &&
                allTasks.some((t) => t.type === 'append-init' && t.meta.trackId === context.inFlightInitTrackId));

            if (inFlightStillNeeded) {
              // Continue: abort only the pending queue, let the in-flight task finish.
              // Schedule everything except the in-flight item — it covers that slot.
              runner.abortPending();
              scheduleAll(
                allTasks.filter(
                  (t) =>
                    !(t.type === 'append-segment' && t.meta.id === context.inFlightSegmentId) &&
                    !(t.type === 'append-init' && t.meta.trackId === context.inFlightInitTrackId)
                ),
                ctx
              );
            } else {
              // Preempt: abort everything and replan.
              runner.abortAll();
              // Cancel SourceBufferActor tasks when a segment is in-flight (always
              // discard) or when a track switch is happening (new track's init follows).
              // For a same-track seek with an in-flight init, skip cancel — the task's
              // signal is not aborted (abortAll was called on the runner, but the init
              // task already completed or will complete via its own signal path).
              const cancelSourceBuffer =
                context.inFlightSegmentId !== null ||
                (context.inFlightInitTrackId !== null &&
                  allTasks.some((t) => t.type === 'append-init' && t.meta.trackId !== context.inFlightInitTrackId));
              if (cancelSourceBuffer) {
                sourceBufferActor.send({ type: 'cancel' });
              }
              scheduleAll(allTasks, ctx);
            }
          },
        },
      },
    },
  });
}