diff --git a/js/hang/api/container.api.md b/js/hang/api/container.api.md index 0db0d680e8d..f3f01b44471 100644 --- a/js/hang/api/container.api.md +++ b/js/hang/api/container.api.md @@ -6,6 +6,7 @@ import { Format as Format_1 } from "@moq/hang/container"; import { Micro } from "@moq/clock"; import { Milli } from "@moq/clock"; import { Net } from "@moq/hang"; +import { Recorder } from "@moq/hang/hls"; import { Signals } from "@moq/hang"; import { Video } from "@moq/hang/catalog"; @@ -64,7 +65,7 @@ export namespace Legacy { } export { Frame }; export class Producer { - constructor(track: Net.Track.Producer, format: Legacy.Format); + constructor(track: Net.Track.Producer, format: Legacy.Format, recorder?: Recorder); close(err?: Error): void; discontinuity(end?: Micro): void; encode(data: Uint8Array | Legacy.Source, timestamp: Micro, keyframe: boolean): void; diff --git a/js/hang/api/hls.api.md b/js/hang/api/hls.api.md new file mode 100644 index 00000000000..a65b60469dd --- /dev/null +++ b/js/hang/api/hls.api.md @@ -0,0 +1,72 @@ +# `@moq/hang/hls` + +```ts +import { Hls } from "@moq/hang/catalog"; +import { Micro } from "@moq/clock"; +import { Net } from "@moq/hang"; +import { Signals } from "@moq/hang"; +import { Window } from "@moq/json"; + +export interface Config { + duration?: Micro; + durationHint?: Micro; +} + +export class Consumer implements AsyncIterable { + [Symbol.asyncIterator](): AsyncIterator; + close(): void; + next(): Promise; + static subscribe(broadcast: Net.Broadcast.Consumer, hls: Hls): Consumer; +} + +export const DEFAULT_DURATION: Micro; + +export type Entry = Omit & { + duration?: Micro; + pts: Micro; +}; + +export type Event = { + push: { + entry: Entry; + index: number; + }; +} | { + pop: Window.Span; +} | { + skip: Window.Span; +}; + +export interface Record { + duration?: number; + group: number; + pts: number; + sequence: number; +} + +export class Recorder { + constructor(track: Net.Track.Producer, config?: Config); + static readonly INFO: Partial; + abort(err: Error): void; + end(pts: Micro): void; + finish(): void; + frame(group: number, pts: Micro): void; + readonly hls: Signals.Getter; + stop(group: number): void; + readonly track: Net.Track.Producer; +} + +export const SUFFIX: ".hls"; + +export class Segmenter { + constructor(config?: Config); + close(): void; + get declared(): Micro | undefined; + end(pts: Micro): void; + frame(group: number, pts: Micro): void; + stop(group: number): void; + take(): Record[]; +} + +export function defaultName(track: string): string; +``` diff --git a/js/hang/api/index.api.md b/js/hang/api/index.api.md index 4b26a93bb4b..5c748b48db5 100644 --- a/js/hang/api/index.api.md +++ b/js/hang/api/index.api.md @@ -5,8 +5,6 @@ import { $ZodBranded } from "zod/v4/core"; import { $brand } from "@zod/mini"; import { $loose } from "zod/v4/core"; import { $strip } from "zod/v4/core"; -import { Micro } from "@moq/clock"; -import { Window } from "@moq/json"; import { ZodMiniNullable } from "@zod/mini"; import { ZodMiniNumber } from "@zod/mini"; import { ZodMiniObject } from "@zod/mini"; @@ -158,35 +156,7 @@ export namespace Health { export function viewer(sample: Health.Sample, thresholds?: Health.Thresholds): Health.Health; } -export namespace Hls { - export class Consumer implements AsyncIterable { - [Symbol.asyncIterator](): AsyncIterator; - close(): void; - next(): Promise; - static subscribe(broadcast: Net.Broadcast.Consumer, hls: import("@moq/hang/catalog").Hls): Hls.Consumer; - } - export type Entry = Omit & { - duration?: Micro; - pts: Micro; - }; - export type Event = { - push: { - entry: Hls.Entry; - index: number; - }; - } | { - pop: Window.Span; - } | { - skip: Window.Span; - }; - export interface Record { - duration?: number; - group: number; - pts: number; - sequence: number; - } - export const SUFFIX: ".hls"; -} +export * as Hls from "@moq/hang/hls"; export * as Net from "@moq/net"; diff --git a/js/hang/package.json b/js/hang/package.json index 1a0f714d2c8..c1001c7733d 100644 --- a/js/hang/package.json +++ b/js/hang/package.json @@ -15,6 +15,7 @@ "./cmaf": "./src/cmaf/index.ts", "./catalog": "./src/catalog/index.ts", "./container": "./src/container/index.ts", + "./hls": "./src/hls.ts", "./util": "./src/util/index.ts" }, "scripts": { diff --git a/js/hang/src/container/legacy.ts b/js/hang/src/container/legacy.ts index dd2bc91c7cf..717259a7866 100644 --- a/js/hang/src/container/legacy.ts +++ b/js/hang/src/container/legacy.ts @@ -5,6 +5,7 @@ export type { BufferedRange, BufferedRanges, Frame } from "./types"; import type { Micro } from "@moq/clock"; import type { AudioConfig, VideoConfig } from "../catalog"; +import type { Recorder } from "../hls"; import type { Format as ContainerFormat } from "./format"; import type { Frame } from "./types"; @@ -78,11 +79,18 @@ export class Producer { #interval?: Micro; // A discontinuity's marker is the newest group, so another one would say nothing new. #marked = false; + #recorder?: Recorder; - /** Wrap a track to publish legacy-container frames into it. */ - constructor(track: Moq.Track.Producer, format: Format) { + /** + * Wrap a track to publish legacy-container frames into it. + * + * With a `recorder`, each written group is reported into the track's HLS records. The recorder + * outlives this producer, so another producer taking over the track continues its records. + */ + constructor(track: Moq.Track.Producer, format: Format, recorder?: Recorder) { this.#format = format; this.#track = track; + this.#recorder = recorder; } /** Encode and append a frame; a keyframe starts a new group. Throws if the first frame is not a keyframe, a group start goes backwards, or a frame is below the previous group start. */ @@ -107,6 +115,7 @@ export class Producer { payload: encodeFrame(data, timestamp), timestamp: Time.Timestamp.fromMicros(timestamp), }); + if (this.#group) this.#recorder?.frame(this.#group.sequence, timestamp); this.#reordered ||= this.#previous !== undefined && timestamp < this.#previous; if (this.#previous !== undefined && timestamp > this.#previous) { @@ -147,6 +156,8 @@ export class Producer { timestamp: Time.Timestamp.fromMicros(timestamp), }); group.finish(); + // The marker is contiguous, so without this the open HLS segment would absorb the pause. + this.#recorder?.stop(group.sequence); this.#liveEdge = this.#liveEdge === undefined ? timestamp : (Math.max(this.#liveEdge, timestamp) as Micro); this.#marked = true; } @@ -167,6 +178,11 @@ export class Producer { this.#end !== undefined && this.#interval !== undefined ? ((this.#end + this.#interval) as Micro) : undefined; + // Where this group's content stops: the duration marker when there is one, else the furthest + // point written. + const reached = + end === undefined ? this.#end : this.#end === undefined ? end : (Math.max(end, this.#end) as Micro); + if (reached !== undefined) this.#recorder?.end(reached); // A presentation endpoint cannot bound the decode-order tail after reordering. this.#format.finishGroup(this.#group, this.#reordered ? undefined : end); this.#group.finish(); @@ -179,9 +195,11 @@ export class Producer { this.#reordered = false; } - /** Close the track and current group, optionally with an error. */ + /** Close the track and current group, optionally with an error. Finishes or aborts the HLS records with it. */ close(err?: Error) { if (!err) this.#close(); + if (err) this.#recorder?.abort(err); + else this.#recorder?.finish(); if (err) this.#group?.abort(err); else this.#group?.finish(); if (err) this.#track.abort(err); diff --git a/js/hang/src/hls.test.ts b/js/hang/src/hls.test.ts index 62c31cf74a2..497ee5be9a5 100644 --- a/js/hang/src/hls.test.ts +++ b/js/hang/src/hls.test.ts @@ -3,7 +3,8 @@ import type { Micro } from "@moq/clock"; import * as Json from "@moq/json"; import { Broadcast, Time } from "@moq/net"; import { u53 } from "./catalog"; -import { Consumer, type Record } from "./hls.ts"; +import * as Legacy from "./container/legacy.ts"; +import { Consumer, DEFAULT_DURATION, defaultName, type Record, Recorder, Segmenter } from "./hls.ts"; const us = (ms: number): Micro => (ms * 1000) as Micro; @@ -49,3 +50,183 @@ test("a record without its group is refused", async () => { await expect(consumer.next()).rejects.toThrow(); close(); }); + +// The segmenter cases mirror `rs/moq-mux/src/hls.rs`, so both cut the same records. + +const hinted = { durationHint: us(4000) }; + +// Each record as `[group, pts, duration]` in wire units, checking the numbering. +function drain(segmenter: Segmenter, first = 0): [number, number, number | undefined][] { + return segmenter.take().map((record, index) => { + expect(record.sequence).toBe(first + index); + return [record.group, record.pts, record.duration]; + }); +} + +// Groups `start..end`, one frame each, `step` ms apart. +function groups(segmenter: Segmenter, start: number, end: number, step: number) { + for (let group = start; group < end; group++) segmenter.frame(group, us(group * step)); +} + +test("long groups get a segment each", () => { + const segmenter = new Segmenter(); + groups(segmenter, 0, 3, 5000); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [1, 5000, undefined], + [2, 10000, undefined], + ]); + expect(segmenter.declared).toBe(us(5000)); +}); + +test("short groups pack into a grid cell", () => { + const segmenter = new Segmenter(); + groups(segmenter, 0, 10, 1000); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [4, 4000, undefined], + [8, 8000, undefined], + ]); +}); + +test("a hint declares its smallest multiple at or past the duration", () => { + expect(new Segmenter({ durationHint: us(1500) }).declared).toBe(us(4500)); + expect(new Segmenter(hinted).declared).toBe(us(4000)); +}); + +test("a stopped first segment declares the grid", () => { + const segmenter = new Segmenter(); + groups(segmenter, 0, 2, 1000); + segmenter.close(); + expect(segmenter.declared).toBe(DEFAULT_DURATION); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [2, 1000, 0], + ]); +}); + +test("the end extends the stop", () => { + const segmenter = new Segmenter(hinted); + groups(segmenter, 0, 2, 1000); + segmenter.end(us(2000)); + segmenter.close(); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [2, 2000, 0], + ]); + // Nothing reports after the close. + segmenter.frame(5, us(9000)); + expect(drain(segmenter, 2)).toEqual([]); +}); + +test("a skipped sequence stops the segment", () => { + const segmenter = new Segmenter(hinted); + groups(segmenter, 0, 2, 1000); + segmenter.end(us(1500)); + segmenter.frame(5, us(60000)); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [2, 1500, 0], + [5, 60000, undefined], + ]); +}); + +test("a stop ends the segment and the next group starts one", () => { + const segmenter = new Segmenter(hinted); + groups(segmenter, 0, 2, 1000); + segmenter.end(us(2000)); + segmenter.stop(2); + segmenter.frame(3, us(10000)); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [2, 2000, 0], + [3, 10000, undefined], + ]); +}); + +test("a report that goes backwards is refused", () => { + const segmenter = new Segmenter(hinted); + segmenter.frame(3, us(0)); + segmenter.frame(3, us(10)); + expect(() => segmenter.frame(2, us(20))).toThrow("group 2 after group 3"); +}); + +test("an uneven GOP cuts on the grid", () => { + const segmenter = new Segmenter(); + // 3s GOPs on a 4s grid: a group starts a segment only in a later cell than the one before. + groups(segmenter, 0, 9, 3000); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [2, 6000, undefined], + [3, 9000, undefined], + [4, 12000, undefined], + [6, 18000, undefined], + [7, 21000, undefined], + [8, 24000, undefined], + ]); + expect(segmenter.declared).toBe(us(6000)); +}); + +test("a sub-millisecond pts truncates on the wire", () => { + // 120 frames of 30000/1001 fps video: a 4004.004 ms grid. + const segmenter = new Segmenter({ durationHint: 33366.7 as Micro }); + expect(segmenter.declared).toBeCloseTo(4_004_004); + groups(segmenter, 0, 1, 0); + segmenter.frame(1, 4_004_999 as Micro); + expect(drain(segmenter)).toEqual([ + [0, 0, undefined], + [1, 4004, undefined], + ]); +}); + +test("a recorder advertises its declaration rounded up to whole milliseconds", () => { + const broadcast = new Broadcast.Producer(); + const recorder = new Recorder(broadcast.createTrack(defaultName("video"), Recorder.INFO), { + durationHint: 33366.7 as Micro, + }); + expect(recorder.hls.peek()).toEqual({ track: "video.hls", duration: u53(4005) }); + recorder.finish(); + broadcast.close(); +}); + +// A legacy producer reports each group it writes, and a recorder publishes and advertises them. +test("a recorded legacy track publishes its records and declares its duration", async () => { + const broadcast = new Broadcast.Producer(); + const media = broadcast.createTrack("video", { timescale: Time.Timescale.MICRO }); + const recorder = new Recorder(broadcast.createTrack(defaultName("video"), Recorder.INFO)); + expect(recorder.hls.peek()).toBeUndefined(); + + const producer = new Legacy.Producer(media, new Legacy.Format("video"), recorder); + const payload = new Uint8Array([1]); + // 2s GOPs of 30 fps-ish frames, then a pause and two more GOPs. + for (let ms = 0; ms < 6000; ms += 100) producer.encode(payload, us(ms), ms % 2000 === 0); + producer.discontinuity(us(6000)); + for (let ms = 20000; ms < 24000; ms += 100) producer.encode(payload, us(ms), ms % 2000 === 0); + producer.close(); + expect(recorder.hls.peek()).toEqual({ track: "video.hls", duration: u53(4000) }); + + const consumer = Consumer.subscribe(broadcast.consume(), { track: "video.hls", duration: u53(4000) }); + const entries = []; + for await (const event of consumer) if ("push" in event) entries.push(event.push.entry); + expect(entries).toEqual([ + { sequence: 0, group: 0, pts: us(0) }, + { sequence: 1, group: 2, pts: us(4000) }, + { sequence: 2, group: 3, pts: us(6000), duration: us(0) }, + { sequence: 3, group: 4, pts: us(20000) }, + { sequence: 4, group: 6, pts: us(24000), duration: us(0) }, + ]); + broadcast.close(); +}); + +test("a report out of order aborts the records but not the media", async () => { + const broadcast = new Broadcast.Producer(); + const records = broadcast.createTrack(defaultName("video"), Recorder.INFO); + const recorder = new Recorder(records); + recorder.frame(3, us(0)); + recorder.frame(2, us(10)); + expect(records.closed.peek()).toBeInstanceOf(Error); + // Later reports are ignored. + recorder.frame(4, us(20)); + recorder.finish(); + broadcast.close(); +}); diff --git a/js/hang/src/hls.ts b/js/hang/src/hls.ts index 4a3cc106d47..840362102fc 100644 --- a/js/hang/src/hls.ts +++ b/js/hang/src/hls.ts @@ -8,21 +8,41 @@ * zero `duration` marks where content stopped: that segment holds no media, and what follows it * is a discontinuity. * - * Every `pts` and `duration` is in milliseconds. The track is an uncompressed `@moq/json` window. + * Every `pts` and `duration` is in milliseconds. The track is a compressed `@moq/json` window + * whose groups restate the recent records. + * + * A {@link Recorder} writes one rendition's records from reports of the frames its media track + * publishes, as `rs/moq-mux`'s does. Segments are cut on a grid of {@link Config.duration} (4s by + * default) in pts: a group starts a segment when its pts falls in a later multiple of the duration + * than the previous group's. Short groups (audio) pack into one segment per grid cell, and long ones + * (video GOPs) get one each. A hint replaces the duration with its smallest multiple at or beyond + * it, which is also the nominal duration declared; without one, the first segment declares it. + * Each rendition cuts on its own grid and numbers its own records, sharing no state with the + * others, so renditions whose keyframes share timestamps cut the same boundaries. + * + * Content stopping (a skipped group sequence, a discontinuity marker, or the end of the track) + * publishes a zero-duration record where it stopped, so the segment before it ends there rather + * than absorbing the pause. * * @module */ -import type { Micro } from "@moq/clock"; +import { Micro } from "@moq/clock"; import * as Json from "@moq/json"; -import type * as Moq from "@moq/net"; +import * as Moq from "@moq/net"; +import { type Getter, Signal } from "@moq/signals"; import * as z from "@zod/mini"; import type * as Catalog from "./catalog"; -import { u53Schema } from "./catalog"; +import { PRIORITY, u53, u53Schema } from "./catalog"; /** The conventional suffix appended to a media track's name to name its HLS record track. */ export const SUFFIX = ".hls"; +/** The conventional HLS record track name for `track`: the name plus {@link SUFFIX}. */ +export function defaultName(track: string): string { + return `${track}${SUFFIX}`; +} + /** One HLS segment: its first group and pts, in milliseconds. */ export interface Record { /** The segment's number, consecutive within the track and equal to the record's window index. */ @@ -99,3 +119,275 @@ function micros(millis: number): Micro { if (!Number.isSafeInteger(value)) throw new Error("HLS record timestamp overflow"); return value as Micro; } + +/** The default {@link Config.duration} (4 seconds). */ +export const DEFAULT_DURATION = Micro(4_000_000); + +// Recent records repeated when the window track rolls to a new group: one more than the 256 +// segments an HLS edge lists, since a segment is listed only once the record after it arrives. +const CHECKPOINT_RECORDS = 257; + +/** How a track's HLS segments are cut. */ +export interface Config { + /** + * The spacing of the pts grid segments are cut on. Defaults to {@link DEFAULT_DURATION}. + * + * A grid rather than a running length on purpose: renditions whose keyframes share timestamps + * cut the same boundaries without sharing any state. + */ + duration?: Micro; + + /** The application's nominal group duration, when known. */ + durationHint?: Micro; +} + +// A pts in wire milliseconds, truncated as a timescale conversion does. +function wire(pts: Micro): number { + return Math.floor(pts / 1000); +} + +/** + * Builds one track's records from reports of the frames it published. + * + * Reports arrive in group order: {@link frame} for each frame (or at least the first of each group), + * {@link end} wherever the content is known to extend, and {@link stop} where it stops. Records + * queue until {@link take} drains them. + */ +export class Segmenter { + readonly #duration: Micro; + // The grid's spacing: the hint's multiple, else the configured duration. + readonly #grid: Micro; + // Fixed nominal duration; absent until an unhinted track's first segment closes. Kept unrounded; + // only the wire declaration rounds. + #declared?: Micro; + // The number the next record gets. + #sequence = 0; + // The open segment's start: absent before the first segment and after a stop. + #open?: Micro; + // The grid cell of the open segment's newest group. + #cell = 0; + // The newest reported group. + #last?: number; + // The newest reported content time in the open segment. + #frontier?: Micro; + #ready: Record[] = []; + #closed = false; + + /** A segmenter numbering its records from zero. */ + constructor(config: Config = {}) { + this.#duration = config.duration ?? DEFAULT_DURATION; + const hint = config.durationHint; + if (hint !== undefined && hint > 0) { + this.#declared = Micro(hint * Math.max(1, Math.ceil(this.#duration / hint))); + } + this.#grid = this.#declared ?? this.#duration; + } + + /** The fixed nominal segment duration, once known. */ + get declared(): Micro | undefined { + return this.#declared; + } + + // The grid cell `pts` falls in. A zero grid gives every distinct pts its own. + #cellOf(pts: Micro): number { + return this.#grid > 0 ? Math.floor(pts / this.#grid) : pts; + } + + /** + * Report a frame of group `group`, presented at `pts`. + * + * Throws when `group` precedes the previous report's: segments hold whole groups in order, so that + * frame could never be listed. + */ + frame(group: number, pts: Micro): void { + if (this.#closed) return; + if (this.#last !== undefined) { + if (group < this.#last) throw new Error(`HLS report for group ${group} after group ${this.#last}`); + if (group !== this.#last) this.#group(group, this.#last, pts); + } else { + this.#begin(group, pts); + } + this.#last = group; + this.#advance(pts); + } + + // Group `group` starts at `pts`, after group `last`. + #group(group: number, last: number, pts: Micro): void { + const start = this.#open; + if (start === undefined) { + this.#begin(group, pts); + return; + } + if (group !== last + 1) { + // A skipped sequence stops the segment where its content reached, so the skipped time is + // not absorbed into it and the reader never has to guess what lies between. + this.#stopAt(last + 1); + this.#begin(group, pts); + return; + } + // Rust's segmenter also cuts on source-declared boundaries, held to a ceiling; nothing here + // declares them, so the grid alone decides. + const cell = this.#cellOf(pts); + if (cell > this.#cell) { + // A grid's first segment starts wherever the content does, so it may be short. + this.#declared ??= Micro(Math.max(pts - start, this.#grid)); + this.#begin(group, pts); + } + this.#cell = cell; + } + + // Start a segment with group `group` at `pts`. + #begin(group: number, pts: Micro): void { + this.#open = pts; + this.#frontier = pts; + this.#cell = this.#cellOf(pts); + this.#ready.push({ sequence: this.#sequence++, group, pts: wire(pts) }); + } + + /** + * Report that the content extends to `pts`, without a frame starting there. + * + * A frame report says where content starts; the stop after the last segment would otherwise land + * at its last frame's start. + */ + end(pts: Micro): void { + if (!this.#closed) this.#advance(pts); + } + + // Content in the open segment reached `pts`. + #advance(pts: Micro): void { + if (this.#open !== undefined && (this.#frontier === undefined || pts > this.#frontier)) this.#frontier = pts; + } + + /** + * Content stopped before group `group`: queue a zero-duration record where it reached. + * + * The next group starts a new segment. Nothing happens while no segment is open. + */ + stop(group: number): void { + if (!this.#closed) this.#stopAt(group); + } + + #stopAt(group: number): void { + const start = this.#open; + if (start === undefined) return; + this.#open = undefined; + // One stopped early says nothing about the cadence, so the target falls back to the grid. + this.#declared ??= this.#grid; + const end = this.#frontier ?? start; + this.#frontier = undefined; + this.#ready.push({ sequence: this.#sequence++, group, pts: wire(end), duration: 0 }); + } + + /** Stop after the newest group, then ignore every later report. */ + close(): void { + if (this.#closed) return; + if (this.#last !== undefined) this.#stopAt(this.#last + 1); + this.#closed = true; + } + + /** Take the records queued since the last call. */ + take(): Record[] { + const ready = this.#ready; + this.#ready = []; + return ready; + } +} + +/** + * Cuts one media track's segments and publishes their records onto its HLS record track. + * + * The records are an optional sidecar, so nothing here throws at the media: a report it cannot place + * or a failed publish aborts the record track instead, and the media carries on. + */ +export class Recorder { + /** The properties a record track is created with. */ + static readonly INFO: Partial = { + timescale: Moq.Time.Timescale.MILLI, + priority: PRIORITY.catalog, + // A joiner asks for a fresh group once the ops outweigh eight headers. + refresh: 8, + // Records are small, repetitive JSON. + compress: true, + }; + + /** The record track. */ + readonly track: Moq.Track.Producer; + + readonly #segmenter: Segmenter; + readonly #sink: Json.Window.Producer; + readonly #declared = new Signal(undefined); + #done = false; + + /** + * The `hls` field to advertise for the media track, once its nominal duration is known: the + * record track and that duration in whole milliseconds, rounded up. + */ + readonly hls: Getter = this.#declared; + + /** Publish onto `track`, created with {@link INFO}, starting at record zero. */ + constructor(track: Moq.Track.Producer, config?: Config) { + this.track = track; + this.#segmenter = new Segmenter(config); + this.#sink = new Json.Window.Producer({ track, checkpointRecords: CHECKPOINT_RECORDS }); + this.#publish(); + } + + /** Report a frame of group `group`, presented at `pts`; see {@link Segmenter.frame}. */ + frame(group: number, pts: Micro): void { + if (this.#done) return; + try { + this.#segmenter.frame(group, pts); + } catch (err) { + // Records that silently omit frames would misdirect the playlist. + this.abort(err instanceof Error ? err : new Error(String(err))); + return; + } + this.#publish(); + } + + /** Report that the content extends to `pts`; see {@link Segmenter.end}. */ + end(pts: Micro): void { + this.#segmenter.end(pts); + } + + /** Report that content stopped before group `group`; see {@link Segmenter.stop}. */ + stop(group: number): void { + this.#segmenter.stop(group); + this.#publish(); + } + + /** Stop after the newest group and finish the record track. Idempotent. */ + finish(): void { + if (this.#done) return; + this.#segmenter.close(); + this.#publish(); + this.#done = true; + this.#sink.finish(); + } + + /** Abort the record track with `err`, so readers see the cause rather than a complete record. Idempotent. */ + abort(err: Error): void { + if (this.#done) return; + this.#done = true; + this.#segmenter.close(); + this.#sink.abort(err); + } + + #publish(): void { + if (this.#done) return; + for (const value of this.#segmenter.take()) { + try { + this.#sink.push({ value, at: Moq.Time.Timestamp.now() }); + } catch (err) { + console.warn("HLS record publish failed; aborting the record track", err); + this.abort(err instanceof Error ? err : new Error(String(err))); + return; + } + } + const declared = this.#segmenter.declared; + if (declared !== undefined && this.#declared.peek() === undefined) { + this.#declared.set({ track: this.track.name, duration: u53(Math.ceil(declared / 1000)) }); + } + } +} diff --git a/js/json/api/index.api.md b/js/json/api/index.api.md index 6933f7664f9..d952e2bdcea 100644 --- a/js/json/api/index.api.md +++ b/js/json/api/index.api.md @@ -166,6 +166,7 @@ export namespace Window { } export class Producer { constructor(config: Window.ProducerConfig); + abort(err: Error): void; cut(): void; finish(): void; get offset(): number; diff --git a/js/json/src/window/producer.ts b/js/json/src/window/producer.ts index 2cd04e8fb07..698f5f86c75 100644 --- a/js/json/src/window/producer.ts +++ b/js/json/src/window/producer.ts @@ -161,6 +161,11 @@ export class Producer { this.#close(); } + /** Abort the track with `err`, so readers see the cause rather than a complete window. Idempotent. */ + abort(err: Error): void { + this.#close(err); + } + #close(abort?: Error): void { if (this.#finished) return; this.#finished = true; diff --git a/js/publish/api/index.api.md b/js/publish/api/index.api.md index 7f85518b9c5..496011d832f 100644 --- a/js/publish/api/index.api.md +++ b/js/publish/api/index.api.md @@ -134,6 +134,7 @@ export type BroadcastInput = { } | undefined>; enabled: Signals.Getter; flip: Signals.Getter; + hls: Signals.Getter; maxAge: Signals.Getter; name: Signals.Getter; onDemand: Signals.Getter; @@ -213,6 +214,8 @@ export namespace Preview { export class Rendition { close(): void; readonly config: Signals.Signal; + hls(track: Net.Track.Producer): Hang.Hls.Recorder | undefined; + readonly hlsHint: Signals.Signal; readonly kind: Kind; readonly name: string; readonly track: Signals.Getter; diff --git a/js/publish/src/audio/capture.test.ts b/js/publish/src/audio/capture.test.ts index b5a3cba0a2f..330f295fc0d 100644 --- a/js/publish/src/audio/capture.test.ts +++ b/js/publish/src/audio/capture.test.ts @@ -562,6 +562,8 @@ test("a failed encoder stays failed until the capture source changes", async () slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => {}, }; const encoder = new Encoder("audio", { diff --git a/js/publish/src/audio/encoder.test.ts b/js/publish/src/audio/encoder.test.ts index 70db56f10c7..07099de5e0c 100644 --- a/js/publish/src/audio/encoder.test.ts +++ b/js/publish/src/audio/encoder.test.ts @@ -294,6 +294,8 @@ async function setup(order = new Order(), codec?: Codec, groupDuration?: Milli) slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -1025,6 +1027,37 @@ test.each([undefined, false, true])( }, ); +// An HLS export subscribes only to the records the catalog advertises, so their reader must start the +// on-demand audio encode just as a media subscriber does. +test("a record subscriber starts on-demand audio", async () => { + using _webcodecs = installFakeWebCodecs(); + const broadcast = new Broadcast({ origin: new Moq.Origin.Producer(), name: Moq.Path.from("records.hang") }); + const feed = new Feed(); + const configuring = configured(); + const encoder = new Encoder("audio", { broadcast, capture: fakeCapture(feed) as never }); + let records: Moq.Track.Subscriber | undefined; + try { + await configuring; + await settle(); + const net = broadcast.net.peek(); + if (!net) throw new Error("expected network broadcast"); + expect(encoder.out.active.peek()).toBe(false); + + records = net.track("audio.hls").subscribe(); + await settle(); + expect(encoder.out.active.peek()).toBe(true); + + records.close(); + records = undefined; + await settle(); + expect(encoder.out.active.peek()).toBe(false); + } finally { + records?.close(); + encoder.close(); + broadcast.close(); + } +}); + test("a backlog encode refusal aborts its rendition when demand arrives", async () => { using _webcodecs = installFakeWebCodecs(); const broadcast = new Broadcast({ origin: new Moq.Origin.Producer(), name: Moq.Path.from("refused.hang") }); diff --git a/js/publish/src/audio/encoder.ts b/js/publish/src/audio/encoder.ts index ce6a1e25b95..c824a0b0f12 100644 --- a/js/publish/src/audio/encoder.ts +++ b/js/publish/src/audio/encoder.ts @@ -13,7 +13,7 @@ import { Signal, } from "@moq/signals"; import type { Broadcast } from "../broadcast"; -import { watched } from "../internal"; +import { hlsHint, wanted, watched } from "../internal"; import { Estimator, type Order } from "../order"; import { Slot } from "../publication"; import type { AudioFrame, Capture, Format } from "./capture"; @@ -128,7 +128,7 @@ export type EncoderOutput = { // PCM rather than in the graph, so this is pre-gain. Undefined for a source that isn't a track. root: Signal; // True while encoding: enabled with a captured format and a track, which also takes a - // subscriber unless the broadcast's `onDemand` is false. + // subscriber to it or its HLS records unless the broadcast's `onDemand` is false. active: Signal; // Cumulative output totals (frames, bytes) measured while serving. stats: Signal; @@ -146,7 +146,8 @@ type Pipeline = { * A single audio rendition encoder. * * Registers itself on the {@link Broadcast} under {@link name} (via `broadcast.audio(name)`), pumps PCM - * off the source via a {@link Capture}, and retains raw PCM while enabled. A subscriber triggers encoding of that history followed by + * off the source via a {@link Capture}, and retains raw PCM while enabled. A subscriber, to the track or its HLS records, + * triggers encoding of that history followed by * live audio; leaving resumes buffering without ending the epoch. The broadcast's `onDemand: false` * keeps encoding continuously instead. Rename by constructing a new encoder; the name is not a signal. */ @@ -403,6 +404,9 @@ export class Encoder { const rendition = broadcast.audio(this.name); effect.cleanup(() => rendition.close()); + // Audio groups are far shorter than a segment, so any hint declares the grid's own duration up front. + rendition.hlsHint.set(hlsHint(this.groupDuration.peek())); + effect.subscribe(this.groupDuration, (duration) => rendition.hlsHint.set(hlsHint(duration))); effect.proxy(this.#slot, rendition.slot); effect.cleanup(() => this.#slot.set(undefined)); @@ -433,7 +437,11 @@ export class Encoder { const live = { track, - producer: new Container.Legacy.Producer(track, new Container.Legacy.Format("audio")), + producer: new Container.Legacy.Producer( + track, + new Container.Legacy.Format("audio"), + rendition.hls(track), + ), }; this.#live = live; effect.cleanup(() => { @@ -452,8 +460,8 @@ export class Encoder { const fatal = effect.get(this.#out.error); const onDemand = effect.get(broadcast.in.onDemand) ?? true; - const demanded = track && effect.get(track.demand().wanted) !== undefined; - this.#encoding = enabled && !!format && !!track && !fatal && (!onDemand || !!demanded); + const demanded = !!track && wanted(effect, track, rendition.hls(track)); + this.#encoding = enabled && !!format && !!track && !fatal && (!onDemand || demanded); effect.set(this.#out.active, this.#encoding, false); try { if (this.#encoding) this.#pipeline?.drain(); diff --git a/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts index 307ac81a94a..0fc7b910269 100644 --- a/js/publish/src/broadcast.test.ts +++ b/js/publish/src/broadcast.test.ts @@ -1,6 +1,7 @@ -import { expect, test } from "bun:test"; -import { Manual, Milli, Nano } from "@moq/clock"; +import { expect, spyOn, test } from "bun:test"; +import { Manual, Micro, Milli, Nano } from "@moq/clock"; import * as Catalog from "@moq/hang/catalog"; +import * as Container from "@moq/hang/container"; import * as Json from "@moq/json"; import { Origin, Path, Track } from "@moq/net"; import { Effect, Signal } from "@moq/signals"; @@ -762,6 +763,167 @@ test("identity changes before the first announce take a fresh track name", async broadcast.close(); }); +test("each video and audio track publishes HLS records and advertises them once declared", async () => { + const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name: Path.from("hls.hang") }); + const video = broadcast.video("video/hd"); + video.config.set(videoConfig); + const audio = broadcast.audio("audio/data"); + audio.config.set(audioConfig); + const text = broadcast.text("captions/en"); + await settle(); + + const track = video.track.peek(); + if (!track) throw new Error("expected the video track"); + const recorder = video.hls(track); + expect(recorder?.track.name).toBe("video/hd.hls"); + const audioTrack = audio.track.peek(); + expect(audioTrack && audio.hls(audioTrack)?.track.name).toBe("audio/data.hls"); + const textTrack = text.track.peek(); + expect(textTrack && text.hls(textTrack)).toBeUndefined(); + + // Nothing declared a duration yet, so the catalog doesn't advertise one. + expect((await readCatalog(broadcast))?.video?.renditions["video/hd"]?.hls).toBeUndefined(); + + const producer = new Container.Legacy.Producer(track, new Container.Legacy.Format("video"), recorder); + for (let ms = 0; ms <= 4000; ms += 2000) producer.encode(new Uint8Array([1]), Micro(ms * 1000), true); + await settle(); + const catalog = await readCatalog(broadcast); + expect(catalog?.video?.renditions["video/hd"]?.hls).toEqual({ track: "video/hd.hls", duration: Catalog.u53(4000) }); + expect(catalog?.audio?.renditions["audio/data"]?.hls).toBeUndefined(); + + broadcast.close(); + // Closing the broadcast finishes the records with their media. + expect(recorder?.track.closed.peek()).toBeNull(); +}); + +test("records end with their media however it closes", async () => { + const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name: Path.from("ended.hang") }); + const video = broadcast.video("video"); + video.config.set(videoConfig); + await settle(); + + const track = video.track.peek(); + if (!track) throw new Error("expected the video track"); + const recorder = video.hls(track); + if (!recorder) throw new Error("expected HLS records"); + + // An application aborting the media directly, not through an encoder, still ends the records. + const cause = new Error("media failed"); + track.abort(cause); + await settle(); + expect(recorder.track.closed.peek()).toBe(cause); + + broadcast.close(); +}); + +test("hls: false publishes no records", async () => { + const broadcast = new Broadcast({ + enabled: true, + origin: new Origin.Producer(), + name: Path.from("nohls.hang"), + hls: false, + }); + const video = broadcast.video("video"); + await settle(); + const track = video.track.peek(); + if (!track) throw new Error("expected the video track"); + expect(video.hls(track)).toBeUndefined(); + broadcast.close(); +}); + +test("on demand, a record subscriber wants the media too", async () => { + const broadcast = new Broadcast({ + enabled: true, + origin: new Origin.Producer(), + name: Path.from("lend.hang"), + onDemand: true, + }); + await settle(); + const net = broadcast.net.peek(); + if (!net) throw new Error("expected a network producer"); + + const rendition = broadcast.video("video"); + await settle(); + expect(rendition.track.peek()).toBeUndefined(); + + // An HLS export reads only the records, yet the media they describe has to be produced. + const records = net.track("video.hls").subscribe(); + await settle(); + expect(rendition.track.peek()).toBeDefined(); + + records.close(); + await settle(); + expect(rendition.track.peek()).toBeUndefined(); + broadcast.close(); +}); + +test("a hinted rendition advertises its records before encoding, so an on-demand export finds them", async () => { + const broadcast = new Broadcast({ + enabled: true, + origin: new Origin.Producer(), + name: Path.from("hinted.hang"), + onDemand: true, + }); + const video = broadcast.video("video"); + video.hlsHint.set(Micro(2_000_000)); + video.config.set(videoConfig); + await settle(); + + expect(video.track.peek()).toBeUndefined(); + const catalog = await readCatalog(broadcast); + expect(catalog?.video?.renditions.video?.hls).toEqual({ track: "video.hls", duration: Catalog.u53(4000) }); + broadcast.close(); +}); + +test("registering a name another rendition's HLS records hold throws", async () => { + const broadcast = new Broadcast({ enabled: true, origin: new Origin.Producer(), name: Path.from("taken.hang") }); + broadcast.video("video"); + expect(() => broadcast.audio("video.hls")).toThrow("HLS record track"); + await settle(); + expect(() => broadcast.audio("video.hls")).toThrow("HLS record track"); + // A text rendition publishes no records, so its sidecar name stays free. + broadcast.text("captions"); + expect(() => broadcast.audio("captions.hls")).not.toThrow(); + broadcast.close(); +}); + +test("with hls off, a name ending in the record suffix is free", async () => { + const broadcast = new Broadcast({ + enabled: true, + origin: new Origin.Producer(), + name: Path.from("free.hang"), + hls: false, + }); + broadcast.video("video"); + expect(() => broadcast.audio("video.hls")).not.toThrow(); + broadcast.close(); +}); + +test("turning hls on leaves a rendition holding a record name its media", async () => { + const hls = new Signal(false); + const broadcast = new Broadcast({ + enabled: true, + origin: new Origin.Producer(), + name: Path.from("flip.hang"), + hls, + }); + const video = broadcast.video("video"); + const audio = broadcast.audio("video.hls"); + const later = broadcast.audio("audio"); + await settle(); + const warn = spyOn(console, "warn").mockImplementation(() => {}); + hls.set(true); + await settle(); + warn.mockRestore(); + + const videoTrack = video.track.peek(); + expect(videoTrack && video.hls(videoTrack)).toBeUndefined(); + expect(audio.track.peek()?.name).toBe("video.hls"); + const laterTrack = later.track.peek(); + expect(laterTrack && later.hls(laterTrack)?.track.name).toBe("audio.hls"); + broadcast.close(); +}); + // Identity follows hang's `same_identity`: interlacing reinitializes a video decoder. test("an interlacing change takes a fresh track name", async () => { const broadcast = new Broadcast({ origin: new Origin.Producer(), name: Path.from("interlaced.hang") }); diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 3057e931f64..c084e45b8ea 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -1,10 +1,12 @@ import { Milli } from "@moq/clock"; import * as Catalog from "@moq/hang/catalog"; import * as Container from "@moq/hang/container"; +import * as Hls from "@moq/hang/hls"; import * as Moq from "@moq/net"; import { Effect, type Getter, getter, type Inputs, type Readonlys, Signal } from "@moq/signals"; import { rotateVideoDimensions } from "@moq/video"; import { CatalogProducer } from "./catalog"; +import { wanted } from "./internal"; import { Order } from "./order"; import { type Identity, identity, type Listing, Publication, type Slot } from "./publication"; import type { Kind, Rendition } from "./rendition"; @@ -65,6 +67,14 @@ export type BroadcastInput = { * its past. Unless false, audio keeps raw history up to `maxAge`. Capture stays open. */ onDemand: Getter; + + /** + * Whether each video and audio track publishes HLS segment records, advertised in its catalog + * entry's `hls` field, so an edge can export the broadcast as HLS. Defaults to true. A record + * track's subscribers count as demand for its media under {@link onDemand}, since an export reads + * the records rather than the media. Changing this publishes a new network broadcast. + */ + hls: Getter; }; /** @@ -123,6 +133,10 @@ export class Broadcast { // The renditions of the current network broadcast, undefined while there is none. readonly #publication = new Signal(undefined); + // Each track name's HLS recorder in the current network broadcast, whose `hls` field the catalog + // advertises once its duration is known. + readonly #records = new Signal>({}); + // Each catalog ID in the latest catalog fold, which seeds a new network broadcast's publication. #listed = new Map(); @@ -139,6 +153,7 @@ export class Broadcast { rotation: getter(props?.rotation ?? 0), maxAge: getter(props?.maxAge), onDemand: getter(props?.onDemand), + hls: getter(props?.hls ?? true), }; this.#signals.run(this.#runCatalog.bind(this)); @@ -178,6 +193,14 @@ export class Broadcast { throw new Error(`rendition already registered: ${name}`); } this.#publication.peek()?.check(name, kind); + // A video or audio rendition's HLS records hold or will hold the name. + if (this.in.hls.peek() && name.endsWith(Hls.SUFFIX)) { + const media = name.slice(0, -Hls.SUFFIX.length); + const owner = this.#renditions.peek()[media]; + if (this.#records.peek()[media] || (owner && owner.kind !== "text")) { + throw new Error(`rendition ${name} names another rendition's HLS record track`); + } + } const handle = { track: new Signal(undefined), @@ -187,7 +210,14 @@ export class Broadcast { // The slot is the one its producer fills, so it holds this rendition's config. const slot = handle.slot as Getter | undefined>; - const rendition = makeRendition(name, kind, handle.track, slot, () => this.#unregister(name)); + const rendition = makeRendition( + name, + kind, + handle.track, + slot, + (served) => this.#records.peek()[served.name], + () => this.#unregister(name), + ); this.#renditions.update((renditions) => ({ ...renditions, [name]: rendition as Rendition })); return rendition; @@ -217,6 +247,7 @@ export class Broadcast { const renditions = effect.get(this.#renditions); const publication = effect.get(this.#publication); if (publication) effect.get(publication.renames); + const records = effect.get(this.#records); const video: Record = {}; const audio: Record = {}; @@ -236,7 +267,11 @@ export class Broadcast { listed.set(rendition.name, { kind: rendition.kind, identity: media }); const track = publication?.name(rendition, media) ?? rendition.name; served.push([rendition, track, media]); - const config = track !== rendition.name ? { ...(source as object), track } : source; + let config = track !== rendition.name ? { ...(source as object), track } : source; + // Filled in here rather than by the producer, so one rewriting its config can't drop it. + const recorder = records[track]; + const hls = recorder && effect.get(recorder.hls); + if (hls) config = { ...(config as object), hls }; if (rendition.kind === "video") { video[rendition.name] = config as Catalog.VideoConfig; @@ -313,6 +348,7 @@ export class Broadcast { const name = effect.get(this.in.name); const maxAge = effect.get(this.in.maxAge); + const hls = effect.get(this.in.hls); if (Catalog.detectFormat(name) === undefined) { console.warn( `You should append .hang to broadcast name ${JSON.stringify(name)} to make the catalog format explicit.`, @@ -339,6 +375,7 @@ export class Broadcast { // group sequence under the same epoch. The publication keeps rendition tracks by catalog ID, // so registering an ID again finds its track; the catalog's stay out of reach of a colliding // rendition. + this.#records.set({}); const publication = new Publication(this.#listed, () => this.#reserved()); this.#publication.set(publication); effect.cleanup(() => { @@ -395,12 +432,16 @@ export class Broadcast { if (entry.owner && entry.owner !== rendition) entry.slot.error.set(undefined); const cause = inner.get(entry.slot.error); // Including an idle track, which no producer holds to abort. - if (entry.track && cause) entry.track.abort(cause); + if (entry.track && cause) { + entry.track.abort(cause); + entry.recorder?.abort(cause); + } if (entry.track && entry.track.name !== entry.name) { // The catalog moved this rendition to a new name for new media. End the old track // cleanly on its last identity, so its subscribers follow the catalog. entry.track.finish(); entry.track.close(); + entry.recorder?.finish(); entry.track = undefined; } else if (entry.track && entry.track.closed.peek() !== undefined) { // A failure leaves this rendition down until it clears. It then recovers under a fresh @@ -414,9 +455,20 @@ export class Broadcast { entry.name, rendition.kind === "video" ? { ...info, refresh: VIDEO_REFRESH } : info, ); + entry.recorder = hls ? this.#record(broadcast, rendition, entry.name) : undefined; } entry.owner = rendition; const served = entry.track; + const recorder = entry.recorder; + if (recorder) { + // The records end with their media, however it closes, so a reader of only the + // records sees the same finish or failure. + inner.run((ended) => { + const closed = ended.get(served.closed); + if (closed instanceof Error) recorder.abort(closed); + else if (closed === null) recorder.finish(); + }); + } inner.run((gate) => { const closed = gate.get(served.closed); if (closed !== undefined) { @@ -424,11 +476,7 @@ export class Broadcast { if (closed && !entry.slot.error.peek()) entry.slot.error.set(closed); return; } - if ( - rendition.kind !== "audio" && - gate.get(this.in.onDemand) && - gate.get(served.demand().wanted) === undefined - ) + if (rendition.kind !== "audio" && gate.get(this.in.onDemand) && !wanted(gate, served, recorder)) return; gate.set(handle.track, served); }); @@ -436,6 +484,28 @@ export class Broadcast { }); } + // Start the HLS records of `rendition`'s video or audio track `name`. A record track name already + // taken, by a rendition registered while hls was off or an application's own track, leaves the + // rendition without records rather than down. + #record(broadcast: Moq.Broadcast.Producer, rendition: Rendition, name: string): Hls.Recorder | undefined { + if (rendition.kind === "text") return undefined; + const records = Hls.defaultName(name); + if (records in this.#renditions.peek() || this.#publication.peek()?.has(records)) { + console.warn(`rendition ${name} publishes no HLS records: ${records} is a rendition`); + return undefined; + } + let track: Moq.Track.Producer; + try { + track = broadcast.createTrack(records, Hls.Recorder.INFO); + } catch (err) { + console.warn(`rendition ${name} publishes no HLS records`, err); + return undefined; + } + const recorder = new Hls.Recorder(track, { durationHint: rendition.hlsHint.peek() }); + this.#records.update((records) => ({ ...records, [name]: recorder })); + return recorder; + } + /** Stop and release everything this holds. */ close() { this.#signals.close(); diff --git a/js/publish/src/internal.ts b/js/publish/src/internal.ts index a3222a88221..89918de4818 100644 --- a/js/publish/src/internal.ts +++ b/js/publish/src/internal.ts @@ -1,3 +1,5 @@ +import { Micro, Milli } from "@moq/clock"; +import type * as Hls from "@moq/hang/hls"; import type * as Moq from "@moq/net"; import type { Effect, Getter } from "@moq/signals"; @@ -8,3 +10,21 @@ import type { Effect, Getter } from "@moq/signals"; export function watched(track: Moq.Track.Producer, effect: Effect): Getter { return effect.computed((demand) => demand.get(track.demand().wanted) !== undefined); } + +/** + * @internal Whether anyone wants `track`'s media: a subscriber to it, or to its HLS records, since an + * export reads only the records. + */ +export function wanted(effect: Effect, track: Moq.Track.Producer, recorder: Hls.Recorder | undefined): boolean { + if (effect.get(track.demand().wanted) !== undefined) return true; + return recorder !== undefined && effect.get(recorder.track.demand().wanted) !== undefined; +} + +/** + * @internal A rendition's HLS duration hint for groups of `duration`. A zero or unbounded group still + * hints a millisecond, which declares the grid's own duration up front: without a hint, an on-demand + * rendition never advertises the records whose subscription would start it encoding. + */ +export function hlsHint(duration: Milli): Micro { + return Micro.fromMilli(Number.isFinite(duration) && duration > 0 ? duration : Milli(1)); +} diff --git a/js/publish/src/publication.ts b/js/publish/src/publication.ts index 0cb40493265..27afb6a5d17 100644 --- a/js/publish/src/publication.ts +++ b/js/publish/src/publication.ts @@ -1,3 +1,4 @@ +import type * as Hls from "@moq/hang/hls"; import * as Json from "@moq/json"; import type * as Moq from "@moq/net"; import { type Getter, Signal } from "@moq/signals"; @@ -35,6 +36,8 @@ export type Entry = { // The network track serving it. Kept after its rendition leaves, so registering the ID again // continues it. track?: Moq.Track.Producer; + // The HLS records accompanying this track. + recorder?: Hls.Recorder; // The registration that last served `track`. owner?: Rendition; // What the registration's producer keeps for this ID. @@ -78,6 +81,11 @@ export class Publication { } } + /** Whether a catalog ID holds or held a rendition here. */ + has(id: string): boolean { + return this.#entries.has(id); + } + /** Throw unless `id` may register as `kind`. */ check(id: string, kind: Kind): void { // An ID keeps its track, so it keeps its kind: different media takes a different ID. @@ -150,7 +158,10 @@ export class Publication { /** End every track. */ close(): void { - for (const { track } of this.#entries.values()) track?.finish(); + for (const { track, recorder } of this.#entries.values()) { + track?.finish(); + recorder?.finish(); + } for (const { track } of this.#entries.values()) track?.close(); } } diff --git a/js/publish/src/rendition-internal.ts b/js/publish/src/rendition-internal.ts index 9ad9efaf973..fad0053b62c 100644 --- a/js/publish/src/rendition-internal.ts +++ b/js/publish/src/rendition-internal.ts @@ -1,3 +1,5 @@ +import type { Micro } from "@moq/clock"; +import type * as Hls from "@moq/hang/hls"; import type * as Moq from "@moq/net"; import { type Getter, Signal } from "@moq/signals"; import type { Slot } from "./publication"; @@ -11,6 +13,7 @@ export let makeRendition: ( kind: Kind, track: Getter, slot: Getter | undefined>, + hls: (track: Moq.Track.Producer) => Hls.Recorder | undefined, close: () => void, ) => Rendition; @@ -37,9 +40,18 @@ export class Rendition { */ readonly config = new Signal(undefined); + /** + * The rendition's longest group duration, when known (a video encoder's keyframe interval). Read + * when each of its tracks is created, it fixes that track's HLS segment duration up front, so the + * catalog advertises the records before any media is encoded, as an on-demand broadcast needs. + * Without one, the first segment declares the duration. + */ + readonly hlsHint = new Signal(undefined); + /** * The open track producer, or `undefined` while the network broadcast is down, after the track - * closes, and, for non-audio tracks with the broadcast's `onDemand`, while no subscriber is attached. + * closes, and, for non-audio tracks with the broadcast's `onDemand`, while no subscriber is attached + * to it or its HLS records. * Producers should * encode only while this is set, and only media matching the current {@link config}: a new * identity swaps in a new track. @@ -52,6 +64,7 @@ export class Rendition { */ readonly slot: Getter | undefined>; + readonly #hls: (track: Moq.Track.Producer) => Hls.Recorder | undefined; readonly #close: () => void; #closed = false; @@ -60,17 +73,28 @@ export class Rendition { kind: Kind, track: Getter, slot: Getter | undefined>, + hls: (track: Moq.Track.Producer) => Hls.Recorder | undefined, close: () => void, ) { this.name = name; this.kind = kind; this.track = track; this.slot = slot; + this.#hls = hls; this.#close = close; } static { - makeRendition = (name, kind, track, slot, close) => new Rendition(name, kind, track, slot, close); + makeRendition = (name, kind, track, slot, hls, close) => new Rendition(name, kind, track, slot, hls, close); + } + + /** + * The HLS recorder of `track`, one of this rendition's {@link Rendition.track | tracks}, or `undefined` when it + * publishes no records. Pass it to the container producer writing the track, so each group it + * writes is reported into the records the catalog advertises in the rendition's `hls` field. + */ + hls(track: Moq.Track.Producer): Hls.Recorder | undefined { + return this.#hls(track); } /** diff --git a/js/publish/src/video/capture.test.ts b/js/publish/src/video/capture.test.ts index e1f64567de9..cf8bdb8e347 100644 --- a/js/publish/src/video/capture.test.ts +++ b/js/publish/src/video/capture.test.ts @@ -457,7 +457,14 @@ test("an encoder resuming on a still source encodes a keyframe of the current pi const track = new Moq.Track.Producer("video").accept({ timescale: Moq.Time.Timescale.MILLI }); // No subscriber yet, so the encoder idles while the source sends its only frame. const live = new Signal(undefined); - const rendition = { config: new Signal(undefined), track: live, slot: new Signal(undefined), close: () => {} }; + const rendition = { + slot: new Signal(undefined), + config: new Signal(undefined), + track: live, + hls: () => undefined, + hlsHint: new Signal(undefined), + close: () => {}, + }; const encoder = new Encoder("video", { enabled: true, broadcast: { video: () => rendition, order: new Order() } as never, @@ -564,7 +571,14 @@ test("an encoder torn down while a frame is in flight closes that frame", async const capture = new Capture({ source: input.source }); const track = new Moq.Track.Producer("video").accept({ timescale: Moq.Time.Timescale.MILLI }); const live = new Signal(undefined); - const rendition = { config: new Signal(undefined), track: live, slot: new Signal(undefined), close: () => {} }; + const rendition = { + slot: new Signal(undefined), + config: new Signal(undefined), + track: live, + hls: () => undefined, + hlsHint: new Signal(undefined), + close: () => {}, + }; const encoder = new Encoder("video", { enabled: true, broadcast: { video: () => rendition, order: new Order() } as never, diff --git a/js/publish/src/video/encoder-stall.test.ts b/js/publish/src/video/encoder-stall.test.ts index f03a2f95540..454a5593c47 100644 --- a/js/publish/src/video/encoder-stall.test.ts +++ b/js/publish/src/video/encoder-stall.test.ts @@ -113,6 +113,8 @@ test("a slower-than-realtime encoder drops input instead of ratcheting catalog l slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); diff --git a/js/publish/src/video/encoder.test.ts b/js/publish/src/video/encoder.test.ts index a69d7f1b2ff..8f5e6165e55 100644 --- a/js/publish/src/video/encoder.test.ts +++ b/js/publish/src/video/encoder.test.ts @@ -127,6 +127,8 @@ test("encoding tracks encoder config in its child effect", async () => { slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -169,6 +171,8 @@ test("an on-demand gap marks a discontinuity and leaves the broadcast-owned trac slot: new Signal(undefined), config: new Signal(undefined), track: live, + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -224,6 +228,8 @@ test("a bandwidth estimate updates the bitrate without blanking the config or re slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -310,6 +316,8 @@ test("every published config fits a probe for its own codec", async () => { slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -663,6 +671,8 @@ test("disabling encodes one black keyframe and keeps the rendition with enabled: slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -831,6 +841,8 @@ test("the probe encodes with the frame rate and bitrate ceiling", async () => { slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -1039,6 +1051,8 @@ async function recordKeyframes( slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -1196,6 +1210,46 @@ test.each([false, true])( }, ); +// An HLS export subscribes only to the records the catalog advertises, so an on-demand rendition has +// to advertise them before anything is encoded, even with every frame a keyframe. +test("an on-demand all-keyframe rendition advertises its records, whose subscriber starts it", async () => { + using recording = await installRecording(); + const broadcast = new Broadcast({ + origin: new Moq.Origin.Producer(), + name: Moq.Path.from("keyframes.hang"), + onDemand: true, + }); + const encoder = new Encoder("video", { + enabled: true, + broadcast, + capture: recording.capture as never, + config: { keyframeInterval: Milli(0) }, + }); + const subscribers: Moq.Track.Subscriber[] = []; + try { + await settle(); + const net = broadcast.net.peek(); + if (!net) throw new Error("expected network broadcast"); + expect(encoder.out.active.peek()).toBe(false); + + const catalogSubscriber = net.track(Broadcast.CATALOG_TRACK).subscribe(); + subscribers.push(catalogSubscriber); + const catalog = await new Json.Snapshot.Consumer({ track: catalogSubscriber }).latest(); + expect(catalog?.value.video?.renditions.video?.hls).toEqual({ + track: "video.hls", + duration: Catalog.u53(4000), + }); + + subscribers.push(net.track("video.hls").subscribe()); + await settle(); + expect(encoder.out.active.peek()).toBe(true); + } finally { + for (const subscriber of subscribers) subscriber.close(); + encoder.close(); + broadcast.close(); + } +}); + test("an unwatched rendition reserves nothing and encodes at its configured bitrate", async () => { using _videoEncoder = installFakeVideoEncoder(); const broadcast = new Broadcast({ origin: new Moq.Origin.Producer(), name: Moq.Path.from("unwatched.hang") }); @@ -1459,6 +1513,8 @@ test("a shrinking capture resizes in band below the advertised ceiling", async ( slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -1532,6 +1588,8 @@ test("the catalog bitrate and frame rate only rise", async () => { slot: new Signal(undefined), config: new Signal(undefined), track: new Signal(track), + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => { track.finish(); track.close(); @@ -1592,6 +1650,8 @@ test("each slot keeps its own ceiling and maxima", async () => { config: new Signal(undefined), track: new Signal(undefined), slot, + hls: () => undefined, + hlsHint: new Signal(undefined), close: () => {}, }); return { video: rendition, order: new Order() }; diff --git a/js/publish/src/video/encoder.ts b/js/publish/src/video/encoder.ts index 186d7395690..5eee7e40d44 100644 --- a/js/publish/src/video/encoder.ts +++ b/js/publish/src/video/encoder.ts @@ -1,6 +1,7 @@ import { Micro, Milli, type Second } from "@moq/clock"; import * as Catalog from "@moq/hang/catalog"; import * as Container from "@moq/hang/container"; +import type * as Hls from "@moq/hang/hls"; import type * as Moq from "@moq/net"; import { type Computed, @@ -13,13 +14,16 @@ import { Signal, } from "@moq/signals"; import type { Broadcast } from "../broadcast"; -import { watched } from "../internal"; +import { hlsHint, watched } from "../internal"; import { Estimator, type Order } from "../order"; import { Slot } from "../publication"; import { hardwareReliable } from "../support/video"; import type { Capture } from "./capture"; import { normalizeSource, type Source } from "./types"; +// The keyframe interval when the config sets none. +const DEFAULT_KEYFRAME_INTERVAL = Milli.fromSecond(2 as Second); + /** Cumulative encoder output totals, measured from the chunks the encoder produces. */ export interface Stats { /** Total frames encoded while serving. Monotonic; diff over an interval for a frame rate. */ @@ -336,6 +340,11 @@ export class Encoder { effect.proxy(this.#slot, rendition.slot); effect.cleanup(() => this.#slot.set(undefined)); + // The longest GOP fixes each new track's HLS segment duration, as moq-video's encoders hint theirs. + const hint = (config: Config | undefined) => hlsHint(config?.keyframeInterval ?? DEFAULT_KEYFRAME_INTERVAL); + rendition.hlsHint.set(hint(this.config.peek())); + effect.subscribe(this.config, (config) => rendition.hlsHint.set(hint(config))); + // Publish the resolved catalog config; undefined (until one resolves) leaves it out of the catalog. effect.proxy(rendition.config, this.out.catalog); @@ -347,7 +356,7 @@ export class Encoder { effect.set(this.#out.active, enabled && !!track && !error, false); if (!enabled || !track || error) return; - this.#encode(track, broadcast.order, effect); + this.#encode(track, rendition.hls(track), broadcast.order, effect); }); // Reserve against the connection while a subscriber is attached. Wait for a ceiling so we @@ -383,11 +392,11 @@ export class Encoder { } // Encode captured frames into the track producer, reconfiguring when the resolved config changes. - #encode(track: Moq.Track.Producer, order: Order, effect: Effect): void { + #encode(track: Moq.Track.Producer, hls: Hls.Recorder | undefined, order: Order, effect: Effect): void { const capture = effect.get(this.in.capture); if (!capture) return; - const producer = new Container.Legacy.Producer(track, new Container.Legacy.Format("video")); + const producer = new Container.Legacy.Producer(track, new Container.Legacy.Format("video"), hls); const demand = track.demand(); let lastKeyframe: Micro | undefined; @@ -505,7 +514,7 @@ export class Encoder { } lastEncoded = frame.timestamp as Micro; - const interval = config?.keyframeInterval ?? Milli.fromSecond(2 as Second); + const interval = config?.keyframeInterval ?? DEFAULT_KEYFRAME_INTERVAL; // Force a keyframe if this is the first frame (no group yet), the GOP elapsed, the // caller asked for one, or a join wants a group past the largest. No spacing: the diff --git a/quest/m2/README.md b/quest/m2/README.md index 468b7da5df4..59b33afd3d2 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -187,8 +187,6 @@ lite-07; bitrate enforcement is not part of this release. - [DVR rewind](/quest/m2/archive-dvr.md) - follow-on capability after the release contract is stable - [Archive lookups cancel when abandoned](/quest/m2/archive-lookup-withdrawal.md) - a public withdrawal signal on `Fill` stops a time lookup's store read once its readers leave -- [JS HLS field](/quest/m2/js-hls.md) - follow-on capability after the release contract is stable - - [Demo dashboard shows viewer lag](/quest/m2/qos-lag-dashboard.md) - follow-on capability after the release contract is stable - [Publisher timeliness at relay ingest](/quest/m2/qos-publisher-timeliness.md) - follow-on capability after the release contract is stable diff --git a/quest/m2/js-hls.md b/quest/m2/js-hls.md deleted file mode 100644 index b9d7b25365b..00000000000 --- a/quest/m2/js-hls.md +++ /dev/null @@ -1,23 +0,0 @@ -# [M] JS HLS field - -## Goal - -`@moq/publish` publishes the per-rendition `hls` field and record tracks by -default, with the same records and declared duration Rust does, so a -browser-published broadcast exports as HLS like a native one. - -## Plan - -Decided 2026-10-27 (maintainer): records are on by default in Rust -(`moq_mux::catalog::Config` publishes them unless `with_hls(None)`), and JS -matches, until [time addressing](/quest/m1/time-addressing/consumers.md) lets -the export list segments from group starts and deletes the records. The -archive does not use the field, so the old idle `flush()` and catalog and data -track timelines stay out of this port. - -Each rendition cuts its own segments on a pts grid and numbers its own -records, sharing no state with the others, as `rs/moq-mux/src/hls.rs` does. - -The read side follows Catalog HLS field. Port the Rust producer -(`rs/moq-mux`) to `js/hang`, publish it from `@moq/publish` by default, and -check its records against Rust output in the interop suite. diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 9adf4962d33..daa31ee58fc 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -203,6 +203,10 @@ path = "../../test/interop/clients/rust/goaway-client.rs" name = "interop-finish" path = "../../test/interop/clients/rust/finish.rs" +[[example]] +name = "interop-hls" +path = "../../test/interop/clients/rust/hls.rs" + [[example]] name = "interop-audio" path = "../../test/interop/clients/rust/audio.rs" diff --git a/rs/moq-relay/src/auth/caps.rs b/rs/moq-relay/src/auth/caps.rs index 0c3fe335f65..0c8d11329e8 100644 --- a/rs/moq-relay/src/auth/caps.rs +++ b/rs/moq-relay/src/auth/caps.rs @@ -88,6 +88,7 @@ pub struct Pacer(Arc); impl Pacer { /// Pace a live connection by calling `apply` with each new set of caps. + #[cfg(any(test, feature = "quic", all(target_os = "linux", feature = "_uring")))] pub(crate) fn new(apply: impl Fn(Caps) + Send + Sync + 'static) -> Self { Self(Arc::new(apply)) } diff --git a/test/interop/README.md b/test/interop/README.md index 95e9144fc32..767e037b8df 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -545,6 +545,7 @@ clients/ rust/audio.rs the Rust subscriber's audio decode via rs/moq-audio (`interop-audio`) rust/tail.rs the finite raw-track client (`interop-tail`) rust/catalog.rs the compressed-catalog publisher (`interop-catalog`) + rust/hls.rs the HLS-records publisher and reader (`interop-hls`) python/interop.py publish/subscribe via py/moq-rs (import moq) go/main.go publish/subscribe via go/wrapper (import moq-go/moq) cpp/main.cpp publish/subscribe via cpp/moq (find_package(moq-cpp)) @@ -562,6 +563,7 @@ clients/ src/close.ts the refused session, read off `WebTransport.closed` js-native/subscribe.ts subscribe via @moq/net + @moq/hang + the WebTransport polyfill js-native/catalog.ts the compressed-catalog reader, pinned to one version + js-native/hls.ts the HLS-records publisher via @moq/hang c/subscribe.c subscribe via rs/moq-c ``` @@ -646,3 +648,9 @@ readers at once, each pinned to lite-07, lite-06, or moq-transport-22 and readin `@moq/watch` does. Only the lite-07 reader sees the flag; the other two share the relay's inflated copy, so `moq_relay_inflated_bytes_total` must grow by exactly the catalog's inflated size. The harness enables the relay's `[stats]` and `[internal]` listener for it. + +The HLS-records lane has the Rust publisher (moq-mux) and each native JS runtime +(`@moq/hang`'s recorder) publish one fixed frame schedule, with GOPs on and off the 4 s +grid, a discontinuity, and a finish, and the Rust reader prints each broadcast's +declared `hls` durations and every record. The two readouts must be identical, so a +browser-published broadcast exports as HLS like a native one. diff --git a/test/interop/clients/js-native/hls.ts b/test/interop/clients/js-native/hls.ts new file mode 100644 index 00000000000..a71613872b5 --- /dev/null +++ b/test/interop/clients/js-native/hls.ts @@ -0,0 +1,110 @@ +/** + * HLS record interop publisher: publish the schedule `interop-hls publish` does through + * `@moq/hang`'s recorder, finish every track, and stay up until the harness acknowledges. The + * harness reads both publishers' records with `interop-hls read` and requires them identical. + * + * node --import tsx hls.ts URL BROADCAST + * + * Keep the schedule in step with `test/interop/clients/rust/hls.rs`. + * + * @module + */ +import { strict as assert } from "node:assert"; +import { createInterface } from "node:readline"; +import { Micro } from "@moq/clock"; +import * as Catalog from "@moq/hang/catalog"; +import * as Container from "@moq/hang/container"; +import * as Hls from "@moq/hang/hls"; +import * as Json from "@moq/json"; +import * as Moq from "@moq/net"; +import { install } from "@moq/web-transport"; + +install(); +console.debug = console.error; +console.info = console.error; +const [url, path] = process.argv.slice(2); +if (!url || !path) throw new Error("usage: hls.ts URL BROADCAST"); + +/** One frame: its pts in milliseconds and whether it opens a group. */ +type Frame = { ms: number; key: boolean }; + +/** Frames every `step` ms over `[start, end)`, opening a group every `gop` ms. */ +function frames(start: number, end: number, step: number, gop: number): Frame[] { + const out: Frame[] = []; + for (let ms = start; ms < end; ms += step) out.push({ ms, key: (ms - start) % gop === 0 }); + return out; +} + +// Video: 2s GOPs, a pause, then 3s GOPs off the grid. Audio: 100ms groups of 20ms frames, a +// pause, then more. Each part ends in a discontinuity but the last, which the finish ends. +const VIDEO = [frames(0, 6000, 100, 2000), frames(20000, 30000, 100, 3000)]; +const AUDIO = [frames(0, 9000, 20, 100), frames(15000, 17000, 20, 100)]; + +const origin = new Moq.Origin.Producer(); +const connection = await Moq.Connection.connect({ + url: new URL(url), + websocket: { enabled: false }, + publish: origin.consume(), +}); +try { + const publication = origin.createBroadcast(Moq.Path.from(path)); + const broadcast = publication.broadcast; + const renditions: { video: Record; audio: Record } = { + video: {}, + audio: {}, + }; + const publish = (kind: "video" | "audio", parts: Frame[][]) => { + const media = broadcast.createTrack(kind, Container.trackInfo({ priority: Catalog.PRIORITY[kind] })); + const recorder = new Hls.Recorder(broadcast.createTrack(Hls.defaultName(kind), Hls.Recorder.INFO)); + const producer = new Container.Legacy.Producer(media, new Container.Legacy.Format(kind), recorder); + for (const [index, part] of parts.entries()) { + if (index > 0) producer.discontinuity(); + for (const { ms, key } of part) producer.encode(new Uint8Array([1]), Micro(ms * 1000), key); + } + producer.close(); + const hls = recorder.hls.peek(); + assert(hls, `${kind} declared no HLS duration`); + if (kind === "video") renditions.video[kind] = { codec: "avc1.64001f", container: { kind: "legacy" }, hls }; + else + renditions.audio[kind] = { + codec: "opus", + sampleRate: Catalog.u53(48000), + numberOfChannels: Catalog.u53(2), + container: { kind: "legacy" }, + hls, + }; + }; + publish("video", VIDEO); + publish("audio", AUDIO); + + const track = broadcast.createTrack(Catalog.TRACK, { + timescale: Moq.Time.Timescale.MILLI, + priority: Catalog.PRIORITY.catalog, + compress: true, + }); + const catalog = new Json.Snapshot.Producer({ track, deltas: false }); + catalog.update({ + value: Catalog.encode({ + video: { renditions: renditions.video }, + audio: { renditions: renditions.audio }, + }), + at: Moq.Time.Timestamp.now(), + }); + catalog.finish(); + publication.announce(); + console.log("hls ready"); + + const lines = createInterface({ input: process.stdin }); + try { + const ack = await lines[Symbol.asyncIterator]().next(); + assert.equal(ack.value, "clean end", "reader did not acknowledge"); + } finally { + lines.close(); + } + broadcast.close(); + console.log("hls acknowledged"); +} finally { + connection.close(); + origin.close(); +} +process.exit(0); diff --git a/test/interop/clients/rust/hls.rs b/test/interop/clients/rust/hls.rs new file mode 100644 index 00000000000..a8908bf3be4 --- /dev/null +++ b/test/interop/clients/rust/hls.rs @@ -0,0 +1,170 @@ +//! HLS record interop: `publish` writes a fixed frame schedule through moq-mux, whose catalog +//! publishes HLS records by default, and `read` prints a broadcast's declared durations and +//! records. The harness reads this publisher and the native JS one (`js-native/hls.ts`), which +//! publishes the same schedule through `@moq/hang`, and requires the two readouts identical. +//! +//! interop-hls publish URL BROADCAST +//! interop-hls read URL BROADCAST +//! +//! Keep the schedule in step with `test/interop/clients/js-native/hls.ts`. +use std::time::Duration; + +use anyhow::{Context, ensure}; +use moq_mux::catalog::Stream; +use moq_mux::catalog::hang::Container; +use moq_mux::container::{Frame, Kind}; +use moq_tokio::moq_net; +use tokio::io::AsyncBufReadExt; + +/// One frame: its pts in milliseconds and whether it opens a group. +type Schedule = Vec>; + +/// Frames every `step` ms over `start..end`, opening a group every `gop` ms. +fn frames(start: u64, end: u64, step: u64, gop: u64) -> Vec<(u64, bool)> { + (start..end) + .step_by(step as usize) + .map(|ms| (ms, (ms - start).is_multiple_of(gop))) + .collect() +} + +/// Video: 2s GOPs, a pause, then 3s GOPs off the grid. Audio: 100ms groups of 20ms frames, a +/// pause, then more. Each part ends in a discontinuity but the last, which the finish ends. +fn video() -> Schedule { + vec![frames(0, 6000, 100, 2000), frames(20000, 30000, 100, 3000)] +} + +fn audio() -> Schedule { + vec![frames(0, 9000, 20, 100), frames(15000, 17000, 20, 100)] +} + +fn write( + producer: &mut moq_mux::container::Producer, + schedule: Schedule, +) -> anyhow::Result<()> { + for (index, part) in schedule.into_iter().enumerate() { + if index > 0 { + producer.discontinuity()?; + } + for (ms, keyframe) in part { + producer.write(Frame { + timestamp: moq_net::Timestamp::from_millis(ms)?, + payload: bytes::Bytes::from_static(&[1]), + keyframe, + duration: None, + })?; + } + } + producer.finish()?; + Ok(()) +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> anyhow::Result<()> { + let args: Vec<_> = std::env::args().collect(); + ensure!( + args.len() == 4 && ["publish", "read"].contains(&args[1].as_str()), + "usage: interop-hls publish|read URL BROADCAST" + ); + let url: url::Url = args[2].parse()?; + let path = &args[3]; + let mut config = moq_tokio::connect::Config::default(); + config.websocket.enabled = Some(false); + let client = config.init(Default::default())?.with_reconnect(false); + let origin = moq_tokio::origin::spawn(); + + if args[1] == "publish" { + let publication = origin.create_broadcast(path)?; + let mut broadcast = publication.broadcast().clone(); + let mut catalog = moq_mux::catalog::Producer::<()>::new(&mut broadcast, Default::default())?; + let track = broadcast.create_track("video", catalog.track_info(hang::catalog::PRIORITY.video))?; + let config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8); + // Each producer owns its catalog entry, so both stay alive until the reader is done. + let mut video_track = catalog.video(track, Container::Legacy(Kind::Video), config)?; + write(&mut video_track, video())?; + let track = broadcast.create_track("audio", catalog.track_info(hang::catalog::PRIORITY.audio))?; + let config = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2); + let mut audio_track = catalog.audio(track, Container::Legacy(Kind::Audio), config)?; + write(&mut audio_track, audio())?; + catalog.finish()?; + + let session = client + .clone() + .with_publisher(&origin) + .connect(url) + .established() + .await?; + publication.announce(Default::default())?; + println!("hls ready"); + let mut commands = tokio::io::BufReader::new(tokio::io::stdin()).lines(); + ensure!( + commands.next_line().await?.as_deref() == Some("clean end"), + "reader did not acknowledge" + ); + println!("hls acknowledged"); + drop((video_track, audio_track, catalog)); + session.close().await?; + client.close().await; + return Ok(()); + } + + let session = client + .clone() + .with_subscriber(origin.clone()) + .connect(url) + .established() + .await?; + kio::clock::timeout(Duration::from_secs(30), async { + let broadcast = origin.consume().follow(path, None)?.available().await?; + let mut catalogs = + moq_mux::catalog::Consumer::<()>::new(&broadcast, moq_mux::catalog::CatalogFormat::Hang).await?; + // Both renditions, each advertising its records. + let renditions = loop { + let catalog = catalogs + .next() + .await? + .context("catalog ended before advertising records")?; + let video = catalog + .video + .renditions + .iter() + .map(|(name, config)| ("video", name, &config.hls)); + let audio = catalog + .audio + .renditions + .iter() + .map(|(name, config)| ("audio", name, &config.hls)); + let renditions: Vec<_> = video + .chain(audio) + .filter_map(|(kind, name, hls)| Some((kind, name.clone(), hls.clone()?))) + .collect(); + if renditions.len() == 2 { + break renditions; + } + }; + for (kind, name, hls) in renditions { + let mut records = moq_mux::hls::Consumer::subscribe(&broadcast, &hls).await?; + let mut out = Vec::new(); + while let Some(event) = records.next().await? { + if let moq_mux::hls::Event::Push { entry, .. } = event { + let mut record = format!("{}:{}:{}", entry.sequence, entry.group, entry.pts.as_millis()); + if let Some(duration) = entry.duration { + record += &format!(":{}", duration.as_millis()); + } + out.push(record); + } + } + ensure!(!out.is_empty(), "{kind} {name} published no records"); + println!( + "hls {kind} {name} track={} duration={} records={}", + hls.track, + hls.duration.as_millis(), + out.join(",") + ); + } + Ok::<_, anyhow::Error>(()) + }) + .await??; + session.close().await?; + client.close().await; + Ok(()) +} diff --git a/test/interop/interop.sh b/test/interop/interop.sh index f113695ef0f..42fbdf34eaa 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -525,6 +525,7 @@ plan_pass() { record "not run" case=compress publisher="$runtime" subscriber=rust version="$version" done record "not run" case="compressed catalog" publisher=rust subscriber="$runtime" + record "not run" case="hls records" publisher="$runtime" subscriber=rust done needs js-native-node || return 0 for version in "${DROP_VERSIONS[@]}"; do @@ -602,7 +603,7 @@ build_relay_cli() { exit 1 } if [[ "$MEDIA" -eq 0 ]]; then - (cd "$WORKSPACE" && cargo build --locked ${flag[@]+"${flag[@]}"} -p moq-cli --example interop-tail --example interop-catalog --example interop-audio --example interop-control --example interop-goaway --example interop-goaway-client --example interop-finish) + (cd "$WORKSPACE" && cargo build --locked ${flag[@]+"${flag[@]}"} -p moq-cli --example interop-tail --example interop-catalog --example interop-audio --example interop-hls --example interop-control --example interop-goaway --example interop-goaway-client --example interop-finish) fi [[ -n "$RELAY" ]] || RELAY="$TARGET_BASE/$PROFILE/moq-relay" # The `moq-cli` crate ships its binary as `moq` (a `[[bin]]` override). @@ -2353,6 +2354,62 @@ run_catalog_fanout() { rm "$fifo" } +# shellcheck disable=SC2329 # invoked indirectly via harness_spawn +run_hls_publisher() { + local fifo="$1" runtime="$2" + shift 2 + case "$runtime" in + rust) timeout -k 3 $((TIMEOUT * 2)) "$TARGET_BASE/$PROFILE/examples/interop-hls" publish "$@" <"$fifo" 9>&- ;; + js-native-node) (cd "$CLIENTS/js-native" && timeout -k 3 $((TIMEOUT * 2)) node --import tsx hls.ts "$@" <"$fifo" 9>&-) ;; + js-native-bun) (cd "$CLIENTS/js-native" && timeout -k 3 $((TIMEOUT * 2)) bun hls.ts "$@" <"$fifo" 9>&-) ;; + esac +} + +# A native JS publisher's HLS records and declared durations match Rust's for the same frames: +# moq-mux and @moq/hang each publish one schedule through the relay, and the Rust reader's +# readout of the two broadcasts must be identical. +run_hls_records() { + local runtime="$1" started=$SECONDS publisher pid why="" want got + local name="hls-$1" fifo="$LOGS/hls-ack" broadcast="hls-$1-$$-$RANDOM" + local cell=(case="hls records" publisher="$runtime" subscriber=rust log="$LOGS/$name-$runtime-read.log") + if is_broken "$runtime"; then + echo " FAIL hls records $runtime -> rust (client unavailable)" + record fail "${cell[@]}" detail="client unavailable" + overall=1 + return + fi + mkfifo "$fifo" + exec 9<>"$fifo" + for publisher in rust "$runtime"; do + harness_spawn "$name-$publisher-pub" "$LOGS/$name-$publisher-pub.log" run_hls_publisher "$fifo" "$publisher" "$(token_url --publish "$broadcast-$publisher/**")" "$broadcast-$publisher" + pid="$HARNESS_PID" + harness_spawn "$name-$publisher-read" "$LOGS/$name-$publisher-read.log" timeout -k 3 "$TIMEOUT" "$TARGET_BASE/$PROFILE/examples/interop-hls" read "$(token_url --subscribe "$broadcast-$publisher/**")" "$broadcast-$publisher" + harness_wait "$HARNESS_PID" || why="${why:-$publisher records unreadable}" + printf 'clean end\n' >&9 + harness_wait "$pid" || true + grep -qx 'hls acknowledged' "$LOGS/$name-$publisher-pub.log" || why="${why:-$publisher publisher did not acknowledge}" + done + want=$(grep '^hls ' "$LOGS/$name-rust-read.log" || true) + got=$(grep '^hls ' "$LOGS/$name-$runtime-read.log" || true) + if [[ -z "$why" && -z "$want" ]]; then + why="rust published no records" + elif [[ -z "$why" && "$want" != "$got" ]]; then + why="records differ from rust" + fi + if [[ -z "$why" ]]; then + echo " PASS hls records $runtime -> rust (match moq-mux)" + record pass "${cell[@]}" seconds=$((SECONDS - started)) + else + echo " FAIL hls records $runtime -> rust ($why)" + record fail "${cell[@]}" seconds=$((SECONDS - started)) detail="$why" + overall=1 + diff <(echo "$want") <(echo "$got") || true + for publisher in rust "$runtime"; do cat "$LOGS/$name-$publisher-pub.log" "$LOGS/$name-$publisher-read.log"; done + fi + exec 9>&- + rm "$fifo" +} + # This checkout's CLI FETCH and finite track tails in this pass, then the lite drop lanes. run_tails() { local runtime version @@ -2394,6 +2451,7 @@ run_tails() { run_tail_pair "$runtime" rust compress "$version" done run_catalog_fanout "$runtime" + run_hls_records "$runtime" done if needs js-native-node; then for version in "${DROP_VERSIONS[@]}"; do