feat: add live SABR media support

.github/workflows/release.yml

@@ -26,7 +26,7 @@ permissions:
 jobs:
   publish-npm:
     if: github.event_name == 'push' || inputs.registry == 'npm' || inputs.registry == 'both'
-    runs-on: [self-hosted, Linux, X64, arko, typetype]
+    runs-on: ubuntu-latest
     timeout-minutes: 15
     environment: npm
     permissions:

jsr.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://jsr.io/schema/config-file.v1.json",
   "name": "@typetype/mse",
-  "version": "0.1.31",
+  "version": "0.1.35",
   "exports": "./src/index.ts",
   "publish": {
     "include": ["LICENSE", "README.md", "src/**/*.ts"]

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@typetype/mse",
-  "version": "0.1.31",
+  "version": "0.1.35",
   "description": "MSE playback engine for TypeType",
   "license": "MIT",
   "type": "module",

src/buffer-policy.ts

@@ -10,11 +10,12 @@ export type BufferPolicy = {
 };
 
 export function resolveBufferPolicy(config: TypeTypeMseConfig): BufferPolicy {
+  const live = config.isLive === true;
   return {
-    bufferGoalMs: positive(config.bufferGoalMs, 30_000),
+    bufferGoalMs: positive(config.bufferGoalMs, live ? 8_000 : 30_000),
     backBufferMs: positive(config.backBufferMs, 30_000),
-    pollIntervalMs: positive(config.pollIntervalMs, 500),
-    manifestRefreshMs: positive(config.manifestRefreshMs, 8_000),
+    pollIntervalMs: positive(config.pollIntervalMs, live ? 250 : 500),
+    manifestRefreshMs: positive(config.manifestRefreshMs, live ? 1_000 : 8_000),
     manifestPollLimit: integer(config.manifestPollLimit, 60),
     segmentPollLimit: integer(config.segmentPollLimit, 60),
   };

src/index.ts

@@ -1,5 +1,10 @@
 /** Browser Media Source Extensions engine for TypeType SABR playback. */
-export type { ManifestSegment, ManifestTrack, PlaybackManifest } from "./manifest";
+export type {
+  LivePlaybackWindow,
+  ManifestSegment,
+  ManifestTrack,
+  PlaybackManifest,
+} from "./manifest";
 export type {
   PlaybackBufferedRange,
   PlaybackWindow,

src/live-edge-follower.ts

@@ -0,0 +1,60 @@
+import type { LivePlaybackWindow } from "./manifest";
+
+type CatchUpContext = {
+  positionMs: number;
+  live: LivePlaybackWindow | null | undefined;
+  paused: boolean;
+  busy: boolean;
+  nowMs: number;
+};
+
+const MAX_TARGET_DRIFT_MS = 20_000;
+const REJOIN_TOLERANCE_MS = 5_000;
+const CATCH_UP_COOLDOWN_MS = 15_000;
+
+export class LiveEdgeFollower {
+  private initialized = false;
+  private following = false;
+  private nextCatchUpAtMs = 0;
+
+  constructor(private readonly enabled: boolean) {}
+
+  initialize(positionMs: number, live: LivePlaybackWindow | null | undefined): void {
+    if (this.initialized || !this.enabled || live?.active !== true) return;
+    this.initialized = true;
+    this.following = live.atLiveEdge || this.isNearTarget(positionMs, live);
+  }
+
+  observeUserSeek(positionMs: number, live: LivePlaybackWindow | null | undefined): void {
+    this.nextCatchUpAtMs = 0;
+    this.following = this.enabled && live?.active === true && this.isNearTarget(positionMs, live);
+  }
+
+  nextTarget(context: CatchUpContext): number | null {
+    if (
+      !this.following ||
+      context.live?.active !== true ||
+      context.paused ||
+      context.busy ||
+      context.nowMs < this.nextCatchUpAtMs
+    ) {
+      return null;
+    }
+    const targetMs = liveTargetMs(context.live);
+    if (targetMs - context.positionMs <= MAX_TARGET_DRIFT_MS) return null;
+    this.nextCatchUpAtMs = context.nowMs + CATCH_UP_COOLDOWN_MS;
+    return targetMs;
+  }
+
+  get isFollowing(): boolean {
+    return this.following;
+  }
+
+  private isNearTarget(positionMs: number, live: LivePlaybackWindow): boolean {
+    return positionMs >= liveTargetMs(live) - REJOIN_TOLERANCE_MS;
+  }
+}
+
+function liveTargetMs(live: LivePlaybackWindow): number {
+  return Math.max(live.seekableStartMs, live.seekableEndMs - live.targetLatencyMs);
+}

src/manifest.ts

@@ -15,10 +15,24 @@ export type ManifestTrack = {
   segments: ManifestSegment[];
 };
 
+/** Dynamic timing information for an active live stream or a completed live DVR. */
+export type LivePlaybackWindow = {
+  active: boolean;
+  postLiveDvr: boolean;
+  headSequence: number;
+  headTimeMs: number;
+  seekableStartMs: number;
+  seekableEndMs: number;
+  atLiveEdge: boolean;
+  targetLatencyMs: number;
+};
+
 /** Browser-ready tracks for the current playback window. */
 export type PlaybackManifest = {
   durationMs: number;
   endOfStream: boolean;
+  startTimeMs?: number;
+  live?: LivePlaybackWindow | null;
   audio: ManifestTrack;
   video: ManifestTrack | null;
 };

src/media-source-controller.ts

@@ -81,11 +81,11 @@ export class MediaSourceController {
     manifest: PlaybackManifest,
   ): Promise<void> {
     await Promise.all([this.audioQueue?.reset(), this.videoQueue?.reset()]);
-    mediaSource.duration = manifest.durationMs > 0 ? manifest.durationMs / 1000 : Number.NaN;
+    this.applyTiming(mediaSource, manifest);
   }
 
   private createSourceBuffers(mediaSource: MediaSource, manifest: PlaybackManifest): void {
-    mediaSource.duration = manifest.durationMs > 0 ? manifest.durationMs / 1000 : Number.NaN;
+    this.applyTiming(mediaSource, manifest);
     this.audioQueue = new AppendQueue(mediaSource.addSourceBuffer(manifest.audio.mime));
     this.videoQueue = manifest.video
       ? new AppendQueue(mediaSource.addSourceBuffer(manifest.video.mime))
@@ -94,6 +94,12 @@ export class MediaSourceController {
     this.videoMime = manifest.video?.mime ?? null;
   }
 
+  updateTiming(manifest: PlaybackManifest): void {
+    const mediaSource = this.mediaSource;
+    if (mediaSource?.readyState !== "open") return;
+    this.applyTiming(mediaSource, manifest);
+  }
+
   append(kind: TrackKind, data: ArrayBuffer): Promise<void> {
     const queue = kind === "audio" ? this.audioQueue : this.videoQueue;
     if (!queue) return Promise.reject(new Error(`${kind} SourceBuffer is not ready`));
@@ -173,4 +179,19 @@ export class MediaSourceController {
     this.audioMime = null;
     this.videoMime = null;
   }
+
+  private applyTiming(mediaSource: MediaSource, manifest: PlaybackManifest): void {
+    const live = manifest.live;
+    if (live?.active) {
+      mediaSource.duration = Number.POSITIVE_INFINITY;
+      if (typeof mediaSource.setLiveSeekableRange === "function") {
+        mediaSource.setLiveSeekableRange(live.seekableStartMs / 1000, live.seekableEndMs / 1000);
+      }
+      return;
+    }
+    if (typeof mediaSource.clearLiveSeekableRange === "function") {
+      mediaSource.clearLiveSeekableRange();
+    }
+    mediaSource.duration = manifest.durationMs > 0 ? manifest.durationMs / 1000 : Number.NaN;
+  }
 }

src/playback-client.ts

@@ -1,7 +1,9 @@
 import type { HttpClient } from "./http-client";
+import type { LivePlaybackWindow } from "./manifest";
 import {
   type PlaybackWindow,
   type PlaybackWindowRequest,
+  parseLivePlaybackWindow,
   parsePlaybackWindow,
 } from "./playback-window";
 
@@ -11,6 +13,8 @@ export type PlaybackResponse = {
   generation: number | null;
   ready: boolean;
   retryAfterMs: number | null;
+  startTimeMs?: number | null;
+  live?: LivePlaybackWindow | null;
 };
 
 export type CreatePlaybackRequest = {
@@ -20,6 +24,7 @@ export type CreatePlaybackRequest = {
   audioTrackId: string | null;
   startTimeMs: number;
   audioOnly: boolean;
+  isLive?: boolean;
 };
 
 export type SeekPlaybackOptions = {
@@ -54,6 +59,8 @@ function parsePlaybackResponse(value: unknown): PlaybackResponse {
     generation: numberField(value, "generation"),
     ready: field(value, "ready") === true,
     retryAfterMs: numberField(value, "retryAfterMs"),
+    startTimeMs: numberField(value, "startTimeMs"),
+    live: parseLivePlaybackWindow(field(value, "live")),
   };
 }
 
@@ -68,6 +75,7 @@ export class PlaybackClient {
     });
     if (request.audioTrackId) params.set("audioTrackId", request.audioTrackId);
     if (request.audioOnly) params.set("audioOnly", "true");
+    if (request.isLive) params.set("isLive", "true");
     const videoId = encodeURIComponent(request.videoId);
     const init = signal ? { method: "POST", signal } : { method: "POST" };
     const response = await this.http.json(`/sabr/playback/${videoId}?${params}`, init);

src/playback-loop.ts

@@ -8,7 +8,7 @@ import type { LoadedSession } from "./session-loader";
 import { refreshPlaybackWindow } from "./session-loader";
 
 type PlaybackLoopArgs = {
-  video: { currentTime: number };
+  video: { currentTime: number; paused: boolean; readyState: number };
   playback: Pick<PlaybackClient, "position" | "prefetch" | "segments">;
   media: Pick<MediaSourceController, "bufferedRanges" | "endOfStream" | "trim">;
   scheduler: Pick<SegmentScheduler, "fill">;
@@ -93,9 +93,26 @@ export class PlaybackLoop {
   ): Promise<void> {
     const currentMs = currentTimeMs(this.args.video);
     const bufferGoalMs = this.args.policy.bufferGoalMs;
-    const goalMs = currentMs + bufferGoalMs;
+    const goalMs = currentMs + bufferGoalMs + this.liveStallRecoveryMs(bufferGoalMs);
     await this.args.scheduler.fill(session.manifest, currentMs, goalMs, signal);
     this.ensureCurrent(revision, signal);
+    if (
+      !session.manifest.endOfStream &&
+      this.args.bufferedEndMs() < currentMs + refreshThresholdMs(bufferGoalMs)
+    ) {
+      this.requestManifestRefresh(revision);
+      const refresh = this.refreshTask;
+      if (refresh) {
+        try {
+          await refresh;
+        } catch {
+          return;
+        }
+        this.ensureCurrent(revision, signal);
+        await this.args.scheduler.fill(session.manifest, currentMs, goalMs, signal);
+        this.ensureCurrent(revision, signal);
+      }
+    }
     await this.args.media.trim(currentMs, this.args.policy.backBufferMs);
     this.ensureCurrent(revision, signal);
     const bufferedEndMs = this.args.bufferedEndMs();
@@ -108,9 +125,6 @@ export class PlaybackLoop {
       this.stop();
       return;
     }
-    if (bufferedEndMs < currentMs + refreshThresholdMs(bufferGoalMs)) {
-      this.requestManifestRefresh(revision);
-    }
   }
 
   private async refreshManifest(session: LoadedSession, signal: AbortSignal): Promise<void> {
@@ -146,11 +160,25 @@ export class PlaybackLoop {
     if (revision !== this.revision) return;
     const currentMs = currentTimeMs(this.args.video);
     const thresholdMs = refreshThresholdMs(this.args.policy.bufferGoalMs);
-    if (this.args.bufferedEndMs() < currentMs + thresholdMs) {
+    const session = this.args.session();
+    const waitingForLiveData = this.waitingForLiveData(session);
+    if (waitingForLiveData || this.args.bufferedEndMs() < currentMs + thresholdMs) {
       this.requestManifestRefresh(revision);
     }
   }
 
+  private liveStallRecoveryMs(bufferGoalMs: number): number {
+    return this.waitingForLiveData(this.args.session()) ? refreshThresholdMs(bufferGoalMs) : 0;
+  }
+
+  private waitingForLiveData(session: LoadedSession | null): boolean {
+    return (
+      session?.manifest.live?.active === true &&
+      !this.args.video.paused &&
+      this.args.video.readyState < HAVE_FUTURE_DATA
+    );
+  }
+
   private failureContext(): PlaybackLoopFailureContext {
     return {
       sessionId: this.args.session()?.response.sessionId ?? null,
@@ -174,3 +202,5 @@ export class PlaybackLoop {
 function refreshThresholdMs(bufferGoalMs: number): number {
   return Math.min(bufferGoalMs, Math.max(5_000, Math.round((bufferGoalMs * 2) / 3)));
 }
+
+const HAVE_FUTURE_DATA = 3;

src/playback-window.ts

@@ -1,4 +1,9 @@
-import type { ManifestSegment, ManifestTrack, PlaybackManifest } from "./manifest";
+import type {
+  LivePlaybackWindow,
+  ManifestSegment,
+  ManifestTrack,
+  PlaybackManifest,
+} from "./manifest";
 import type { TrackKind } from "./types";
 
 /** Buffered media interval reported to the backend for a specific format. */
@@ -39,6 +44,8 @@ export type PlaybackWindow = {
   status: string | null;
   blockedBy: string | null;
   bufferedEdgeMs: number | null;
+  startTimeMs?: number | null;
+  live?: LivePlaybackWindow | null;
   manifest: PlaybackManifest | null;
 };
 
@@ -65,6 +72,34 @@ function booleanField(value: object, key: string): boolean {
   return field(value, key) === true;
 }
 
+export function parseLivePlaybackWindow(value: unknown): LivePlaybackWindow | null {
+  if (!value || typeof value !== "object") return null;
+  const headSequence = numberField(value, "headSequence");
+  const headTimeMs = numberField(value, "headTimeMs");
+  const seekableStartMs = numberField(value, "seekableStartMs");
+  const seekableEndMs = numberField(value, "seekableEndMs");
+  const targetLatencyMs = numberField(value, "targetLatencyMs");
+  if (
+    headSequence === null ||
+    headTimeMs === null ||
+    seekableStartMs === null ||
+    seekableEndMs === null ||
+    targetLatencyMs === null
+  ) {
+    return null;
+  }
+  return {
+    active: booleanField(value, "active"),
+    postLiveDvr: booleanField(value, "postLiveDvr"),
+    headSequence,
+    headTimeMs,
+    seekableStartMs,
+    seekableEndMs,
+    atLiveEdge: booleanField(value, "atLiveEdge"),
+    targetLatencyMs,
+  };
+}
+
 function arrayField(value: object, key: string): unknown[] {
   const result = field(value, key);
   return Array.isArray(result) ? result : [];
@@ -109,13 +144,17 @@ function parseTrack(kind: TrackKind, value: object, baseUrl: string): ManifestTr
 
 function parseManifest(value: object, baseUrl: string): PlaybackManifest | null {
   const durationMs = numberField(value, "durationMs") ?? 0;
+  const startTimeMs = numberField(value, "startTimeMs") ?? 0;
   const endOfStream = booleanField(value, "endOfStream");
+  const live = parseLivePlaybackWindow(field(value, "live"));
   const audioValue = objectField(value, "audio");
   const videoValue = objectField(value, "video");
   if (!audioValue) return null;
   const audio = parseTrack("audio", audioValue, baseUrl);
   const video = videoValue ? parseTrack("video", videoValue, baseUrl) : null;
-  return audio && (!videoValue || video) ? { durationMs, endOfStream, audio, video } : null;
+  return audio && (!videoValue || video)
+    ? { durationMs, endOfStream, startTimeMs, live, audio, video }
+    : null;
 }
 
 export function parsePlaybackWindow(value: unknown, baseUrl: string): PlaybackWindow {
@@ -134,6 +173,8 @@ export function parsePlaybackWindow(value: unknown, baseUrl: string): PlaybackWi
     status: stringField(value, "status"),
     blockedBy: stringField(value, "blockedBy"),
     bufferedEdgeMs: numberField(value, "bufferedEdgeMs"),
+    startTimeMs: numberField(value, "startTimeMs"),
+    live: parseLivePlaybackWindow(field(value, "live")),
     manifest: parseManifest(manifestValue, baseUrl),
   };
 }