diff --git a/doc/concept/hang.md b/doc/concept/hang.md index a35b1e5a834..82e6b754342 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -139,6 +139,13 @@ The `container.kind` on each rendition says how frames are framed: A consumer skips renditions with a kind it doesn't recognize and carries them through when republishing the catalog. +A CMAF frame's timestamp is its fragment's earliest presentation time, and the +broadcast timeline: `tfdt` only places the samples relative to each other. A +publisher can move a passthrough track to another timeline by changing the frame +timestamp alone, without rewriting the fragment. An untimed track's frames carry +no timestamp, so their samples present at `tfdt`. A timed CMAF track declares the +init segment's `mdhd` timescale, so the frame timestamp lands on a media tick. + A Legacy or LOC video publisher can close the last frame's duration with an empty codec payload whose timestamp is that frame's exclusive end. Consumers treat it as metadata and never pass it to a decoder. This lets a group close immediately diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 038e3b204f1..6c3c91247a4 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -230,6 +230,17 @@ These land with the next breaking release, not the 2026-09-23 train. On a moq-transport track with `TIMESCALE`, an object without a Timestamp is malformed. A new subscriber with no start on an untimed track starts at the latest group. +- **CMAF decodes at the frame timestamp.** A fragment's earliest sample + presents at its moq-net frame timestamp; `tfdt` only orders the samples. In + TypeScript, `Container.Format.decode` and `Cmaf.decodeDataSegment` take that + timestamp: pass `frame.timestamp` alongside `frame.payload`, or `undefined` + for an untimed frame, whose samples present at `tfdt`. + `Cmaf.decodeTimestamp` is gone; the frame timestamp is the fragment's time. + moq-mux's CMAF `Wire::write` refuses a track whose timescale isn't the init's + `mdhd` timescale with `fmp4::Error::TimescaleMismatch`. `import::Track` and + `TrackStream` accept a CMAF rendition's track at that timescale; a track you + create yourself declares it with `track::Info::with_timescale`. Both decoders refuse a `trun` whose + `data_offset` doesn't start at the next sample in the `mdat`. ## Wire diff --git a/drafts/draft-lcurley-moq-hang.md b/drafts/draft-lcurley-moq-hang.md index b3666b5520b..732165b13fd 100644 --- a/drafts/draft-lcurley-moq-hang.md +++ b/drafts/draft-lcurley-moq-hang.md @@ -609,7 +609,12 @@ For example, h.264 with no `description` field would be annex.b encoded, while h For a text track, the remainder is the cue in the track's declared `format` (for example a `WEBVTT` segment). ## cmaf -Each frame is a complete fragmented MP4 fragment (`moof`+`mdat`), carrying its own timestamps. +Each frame is a complete fragmented MP4 fragment (`moof`+`mdat`). +The frame timestamp is the presentation time of the fragment's earliest sample; the fragment's own timestamps only place its samples relative to that sample. +A consumer MUST present each sample at the frame timestamp plus its offset from the earliest sample, even when `tfdt` says otherwise. +A frame on an untimed track has no timestamp, so its samples present at the fragment's own timestamps. +A timed track's timescale MUST equal the init segment's `mdhd` timescale, so the frame timestamp carries the earliest sample's time exactly. +A fragment's runs MUST lie back to back in the `mdat`, in `trun` order, with each `data_offset` counting from the first byte of the `moof`. Audio samples MUST be marked as sync samples, including samples inside a group. A sync sample does not declare an audio group boundary; the publisher chooses those boundaries. @@ -1082,6 +1087,7 @@ This document has no IANA actions. - Defined encoder `jitter` as flush lateness above the rendition's own recent minimum, replacing fixed frame-duration hints; container batches retain media-span estimates. - Clarified that CMAF audio samples are sync samples independently of publisher group boundaries. +- Timed CMAF samples from the frame timestamp, which is the fragment's earliest presentation time; `tfdt` is only relative within the fragment, and places the samples of an untimed frame. A timed CMAF track counts in its `mdhd` ticks. - Clarified that container importers can estimate jitter from batch media spans without measuring input wait time. - Specified the `jitter` field's computation: the publisher's own structure rather than the network, rounded up to whole milliseconds, never `0` (a consumer treats `0` as absent), and never lowered once advertised. - For video, an empty codec payload is the exclusive end of the frame before it. A video publisher SHOULD end each group with one when the exclusive end is known. Audio retains its terminal-trimming marker before codec drain packets. diff --git a/js/hang/src/container/cmaf/decode.ts b/js/hang/src/container/cmaf/decode.ts index 24d7df3eb2a..841169e5e7e 100644 --- a/js/hang/src/container/cmaf/decode.ts +++ b/js/hang/src/container/cmaf/decode.ts @@ -298,25 +298,6 @@ function extractAudioSpecificConfig(esds: Uint8Array): Uint8Array | undefined { return undefined; } -/** - * Extract just the base media decode time from a data segment (moof + mdat). - * This is a lighter-weight function when you only need the timestamp. - * - * @param segment - The moof + mdat data - * @param init - Parsed init segment (provides timescale) - * @returns The base media decode time in microseconds - */ -export function decodeTimestamp(segment: Uint8Array, init: InitSegment): Time.Micro { - const boxes = readIsoBoxes(toArrayBuffer(segment), { readers: DATA_READERS }) as ParsedIsoBox[]; - - // Find moof > traf > tfdt for base media decode time - const tfdt = findBox(boxes, isBoxType("tfdt")); - const baseDecodeTime = tfdt?.baseMediaDecodeTime ?? 0; - - // Convert to microseconds - return ((baseDecodeTime * 1_000_000) / init.timescale) as Time.Micro; -} - /** * Parse a data segment (moof + mdat) to extract raw samples. * @@ -324,11 +305,22 @@ export function decodeTimestamp(segment: Uint8Array, init: InitSegment): Time.Mi * per ISO/IEC 14496-12 §8.8.7. The init segment's trex defaults are required for * fragments where the encoder only set them once in moov (e.g. gstreamer passthrough). * + * The moq-net `timestamp` is the broadcast timeline: the fragment's earliest sample presents at + * it. `tfdt` and the composition offsets only place the samples relative to each other, since a + * publisher may move a passthrough track to another timeline without rewriting the payload. + * An untimed frame (`undefined`) has no broadcast time, so its samples present at the time + * `tfdt` gives them. + * * @param segment - The moof + mdat data * @param init - Parsed init segment (provides timescale and trex defaults) + * @param timestamp - The moq-net frame timestamp carrying this segment, if its track is timed * @returns Array of decoded samples */ -export function decodeDataSegment(segment: Uint8Array, init: InitSegment): Sample[] { +export function decodeDataSegment( + segment: Uint8Array, + init: InitSegment, + timestamp: Time.Timestamp | undefined, +): Sample[] { // Cast to ParsedIsoBox[] since the library's return type changes with readers const boxes = readIsoBoxes(toArrayBuffer(segment), { readers: DATA_READERS }) as ParsedIsoBox[]; @@ -342,9 +334,12 @@ export function decodeDataSegment(segment: Uint8Array, init: InitSegment): Sampl const defaultSize = tfhd?.defaultSampleSize ?? init.defaultSampleSize; const defaultFlags = tfhd?.defaultSampleFlags ?? init.defaultSampleFlags; - // Find moof > traf > trun for sample info - const trun = findBox(boxes, isBoxType("trun")); - if (!trun) { + // Find moof > traf > trun for sample info. A traf may split its samples across several runs, + // which continue one decode timeline and one mdat. + const traf = findBox(boxes, isBoxType("traf")); + // biome-ignore lint/suspicious/noExplicitAny: ISO box structure varies + const truns = ((traf as any)?.boxes ?? []).filter(isBoxType("trun")); + if (truns.length === 0) { throw new Error("No trun box found in data segment"); } @@ -361,70 +356,96 @@ export function decodeDataSegment(segment: Uint8Array, init: InitSegment): Sampl throw new Error("No data in mdat box"); } + // Samples are read from the mdat front to back, so each run must start where the previous one + // ended. A run's dataOffset counts from the moof's first byte (CMAF's default-base-is-moof). + if (tfhd?.baseDataOffset !== undefined) { + throw new Error("tfhd base_data_offset is unsupported: CMAF data offsets count from the moof"); + } + let position = 0; + let moofStart: number | undefined; + let mdatDataStart: number | undefined; + for (const box of boxes) { + const size = box.largesize ?? box.size; + if (box.type === "moof") moofStart ??= position; + if (box === mdat) mdatDataStart = position + size - mdatData.byteLength; + position += size; + } + if (moofStart === undefined || mdatDataStart === undefined || mdatDataStart < moofStart) { + throw new Error("mdat must follow the moof"); + } + const dataStart = mdatDataStart - moofStart; + const samples: Sample[] = []; + const ptss: number[] = []; - // trun.dataOffset is an offset from the base data offset (typically moof start) to the first sample. - // For simple CMAF segments where moof is followed immediately by mdat, this equals moof.size + 8. - // Since mdat.data is the mdat payload (excluding the 8-byte header), we need to compute the - // offset within mdatData. For now, we assume samples start at the beginning of mdat.data - // when dataOffset is not specified or when it points to the start of mdat payload. - // TODO: For complex cases with base_data_offset in tfhd, this needs additional handling. let dataOffset = 0; let decodeTime = baseDecodeTime; - for (let i = 0; i < trun.sampleCount; i++) { - const sample: TrackRunSample = trun.samples[i] ?? {}; + for (const trun of truns) { + if (trun.dataOffset !== undefined && trun.dataOffset !== dataStart + dataOffset) { + throw new Error( + `trun data_offset ${trun.dataOffset} doesn't start at the next sample (${dataStart + dataOffset})`, + ); + } - const sampleSize = sample.sampleSize ?? defaultSize; - const sampleDuration = sample.sampleDuration ?? defaultDuration; + for (let i = 0; i < trun.sampleCount; i++) { + const sample: TrackRunSample = trun.samples[i] ?? {}; - // Validate sample size - must be positive to produce valid data - if (sampleSize <= 0) { - throw new Error(`Invalid sample size ${sampleSize} for sample ${i} in trun`); - } + const sampleSize = sample.sampleSize ?? defaultSize; + const sampleDuration = sample.sampleDuration ?? defaultDuration; - // Duration 0 is valid for single-sample CMAF fragments where duration - // is implicit. Negative duration would indicate corrupt data. - if (sampleDuration < 0) { - throw new Error(`Invalid sample duration ${sampleDuration} for sample ${i} in trun`); - } + // Validate sample size - must be positive to produce valid data + if (sampleSize <= 0) { + throw new Error(`Invalid sample size ${sampleSize} for sample ${i} in trun`); + } - // Bounds check before slicing to prevent reading past mdat data - if (dataOffset + sampleSize > mdatData.length) { - throw new Error( - `Sample ${i} would overflow mdat: offset=${dataOffset}, size=${sampleSize}, mdatLength=${mdatData.length}`, - ); + // Duration 0 is valid for single-sample CMAF fragments where duration + // is implicit. Negative duration would indicate corrupt data. + if (sampleDuration < 0) { + throw new Error(`Invalid sample duration ${sampleDuration} for sample ${i} in trun`); + } + + // Bounds check before slicing to prevent reading past mdat data + if (dataOffset + sampleSize > mdatData.length) { + throw new Error( + `Sample ${i} would overflow mdat: offset=${dataOffset}, size=${sampleSize}, mdatLength=${mdatData.length}`, + ); + } + + const sampleFlags = + i === 0 && trun.firstSampleFlags !== undefined + ? trun.firstSampleFlags + : (sample.sampleFlags ?? defaultFlags); + const compositionOffset = sample.sampleCompositionTimeOffset ?? 0; + + // Extract sample data + const data = new Uint8Array(mdatData.slice(dataOffset, dataOffset + sampleSize)); + dataOffset += sampleSize; + + ptss.push(decodeTime + compositionOffset); + const duration = Math.round((sampleDuration * 1_000_000) / init.timescale); + + // Check if keyframe (sample_is_non_sync_sample flag is bit 16) + // If flag is 0, treat as keyframe for safety. Audio never reports one: every + // audio sample is a sync sample, and the group start is the consumer's to mark. + const keyframe = init.kind === "video" && (sampleFlags === 0 || (sampleFlags & 0x00010000) === 0); + + // Set below, once the earliest presentation time is known. + samples.push({ data, timestamp: 0, keyframe, duration }); + + decodeTime += sampleDuration; } + } - const sampleFlags = - i === 0 && trun.firstSampleFlags !== undefined - ? trun.firstSampleFlags - : (sample.sampleFlags ?? defaultFlags); - const compositionOffset = sample.sampleCompositionTimeOffset ?? 0; - - // Extract sample data - const data = new Uint8Array(mdatData.slice(dataOffset, dataOffset + sampleSize)); - dataOffset += sampleSize; - - // Calculate presentation timestamp in microseconds - // PTS = (decode_time + composition_offset) * 1_000_000 / timescale - const pts = decodeTime + compositionOffset; - const timestamp = Math.round((pts * 1_000_000) / init.timescale); - const duration = Math.round((sampleDuration * 1_000_000) / init.timescale); - - // Check if keyframe (sample_is_non_sync_sample flag is bit 16) - // If flag is 0, treat as keyframe for safety. Audio never reports one: every - // audio sample is a sync sample, and the group start is the consumer's to mark. - const keyframe = init.kind === "video" && (sampleFlags === 0 || (sampleFlags & 0x00010000) === 0); - - samples.push({ - data, - timestamp, - keyframe, - duration, - }); - - decodeTime += sampleDuration; + // A loop, not `Math.min(...ptss)`: a long fragment's sample count can exceed the argument limit. + let earliest = 0; + if (timestamp !== undefined) { + earliest = Number.POSITIVE_INFINITY; + for (const pts of ptss) earliest = Math.min(earliest, pts); + } + const anchor = timestamp?.asMicros() ?? 0; + for (const [i, sample] of samples.entries()) { + sample.timestamp = Math.round(anchor + ((ptss[i] - earliest) * 1_000_000) / init.timescale); } return samples; diff --git a/js/hang/src/container/cmaf/format.test.ts b/js/hang/src/container/cmaf/format.test.ts new file mode 100644 index 00000000000..ecf42602699 --- /dev/null +++ b/js/hang/src/container/cmaf/format.test.ts @@ -0,0 +1,125 @@ +import { expect, test } from "bun:test"; +import { Time } from "@moq/net"; +import type { InitSegment } from "./decode.ts"; +import { Format } from "./format.ts"; + +const TIMESCALE = 90_000; +const INIT: InitSegment = { + timescale: TIMESCALE, + trackId: 1, + kind: "video", + defaultSampleDuration: 0, + defaultSampleSize: 0, + defaultSampleFlags: 0, +}; + +function box(type: string, ...parts: Uint8Array[]): Uint8Array { + const size = 8 + parts.reduce((sum, part) => sum + part.byteLength, 0); + const out = new Uint8Array(size); + const view = new DataView(out.buffer); + view.setUint32(0, size); + out.set(new TextEncoder().encode(type), 4); + let offset = 8; + for (const part of parts) { + out.set(part, offset); + offset += part.byteLength; + } + return out; +} + +function u32(...values: number[]): Uint8Array { + const out = new Uint8Array(values.length * 4); + const view = new DataView(out.buffer); + for (const [i, value] of values.entries()) view.setInt32(i * 4, value); + return out; +} + +type Run = [duration: number, cts: number][]; + +/** A moof+mdat at `tfdt` ticks whose runs list one-byte samples in decode order. */ +function fragment(tfdt: number, ...runs: Run[]): Uint8Array { + const count = runs.reduce((sum, run) => sum + run.length, 0); + const moof = (dataOffset: number) => { + let offset = dataOffset; + const truns = runs.map((run) => { + // Version 1 (signed CTS): data-offset, sample-duration, sample-size, and CTS present. + const trun = box("trun", u32(0x01000b01, run.length, offset), ...run.map(([d, cts]) => u32(d, 1, cts))); + offset += run.length; + return trun; + }); + return box( + "moof", + box("mfhd", u32(0, 0)), + box( + "traf", + box("tfhd", u32(0x020000, 1)), + box("tfdt", u32(0x01000000, Math.floor(tfdt / 2 ** 32), tfdt % 2 ** 32)), + ...truns, + ), + ); + }; + const header = moof(0); + return new Uint8Array([...moof(header.byteLength + 8), ...box("mdat", new Uint8Array(count))]); +} + +// The frame timestamp is the broadcast timeline: a passthrough fragment whose `tfdt` still +// carries its source PTS decodes at the frame timestamp, keeping its B-frame order. +test("CmafFormat times samples from the frame timestamp", () => { + const source = 3_600 * TIMESCALE; + // I, P, B in decode order, presenting at +0, +2, and +1 frames. + const segment = fragment(source, [ + [3000, 3000], + [3000, 6000], + [3000, 0], + ]); + + const frames = new Format(INIT).decode(segment, Time.Timestamp.fromMillis(10_000)); + expect(frames.map((f) => f.timestamp)).toEqual([10_000_000, 10_066_667, 10_033_333] as Time.Micro[]); +}); + +// A fragment can open on a sample that presents after a later one (open-GOP leading pictures, +// or a cut mid-GOP). The earliest presentation time is the anchor, not the first sample. +test("CmafFormat anchors the earliest presentation time", () => { + const segment = fragment(0, [ + [3000, 6000], + [3000, 0], + ]); + + const frames = new Format(INIT).decode(segment, Time.Timestamp.fromMillis(5_000)); + expect(frames.map((f) => f.timestamp)).toEqual([5_033_333, 5_000_000] as Time.Micro[]); +}); + +// Runs continue one decode timeline, so the anchor is the earliest sample across all of them. +test("CmafFormat anchors the earliest sample across runs", () => { + const segment = fragment(0, [[3000, 6000]], [[3000, 0]]); + + const frames = new Format(INIT).decode(segment, at(3000)); + expect(frames.map((f) => f.timestamp)).toEqual([66_667, 33_333] as Time.Micro[]); +}); + +// An untimed track's frames carry no broadcast time, so their samples present at `tfdt`. +test("CmafFormat times an untimed fragment from its tfdt", () => { + const segment = fragment(TIMESCALE, [ + [3000, 3000], + [3000, 6000], + [3000, 0], + ]); + + const frames = new Format(INIT).decode(segment, undefined); + expect(frames.map((f) => f.timestamp)).toEqual([1_033_333, 1_100_000, 1_066_667] as Time.Micro[]); +}); + +// Samples are read front to back, so a run that skips bytes is refused rather than sliced wrong. +test("CmafFormat refuses a run that doesn't start at the next sample", () => { + const segment = fragment(0, [[3000, 0]], [[3000, 0]]); + const view = new DataView(segment.buffer); + // The second trun's data_offset follows its type, version/flags, and sample count. + const trun = segment.findLastIndex((_, i) => new TextDecoder().decode(segment.subarray(i, i + 4)) === "trun"); + view.setInt32(trun + 12, view.getInt32(trun + 12) + 1); + + expect(() => new Format(INIT).decode(segment, undefined)).toThrow(/data_offset/); +}); + +function at(ticks: number): Time.Timestamp { + return new Time.Timestamp(ticks, Time.Timescale(TIMESCALE)); +} diff --git a/js/hang/src/container/cmaf/format.ts b/js/hang/src/container/cmaf/format.ts index 7cfc4726485..3ecc4572b3c 100644 --- a/js/hang/src/container/cmaf/format.ts +++ b/js/hang/src/container/cmaf/format.ts @@ -12,9 +12,9 @@ export class Format implements ContainerFormat { this.#init = init; } - /** Decode one CMAF fragment into its media frames. */ - decode(frame: Uint8Array): Frame[] { - return decodeDataSegment(frame, this.#init).map((s) => ({ + /** Decode one CMAF fragment, presenting its earliest sample at the moq-net `timestamp`, or at `tfdt` if untimed. */ + decode(payload: Uint8Array, timestamp: Time.Timestamp | undefined): Frame[] { + return decodeDataSegment(payload, this.#init, timestamp).map((s) => ({ payload: s.data, timestamp: s.timestamp as Time.Micro, keyframe: s.keyframe, diff --git a/js/hang/src/container/consumer.test.ts b/js/hang/src/container/consumer.test.ts index a072ea869b4..0443a095d38 100644 --- a/js/hang/src/container/consumer.test.ts +++ b/js/hang/src/container/consumer.test.ts @@ -28,9 +28,14 @@ function encodeLegacyFrame(timestamp: Time.Micro, payload: Uint8Array): Uint8Arr return data; } +/** A moq-net timestamp of `ticks` at the test track's timescale. */ +function at(ticks: number): Time.Timestamp { + return new Time.Timestamp(ticks, Time.Timescale(TIMESCALE)); +} + /** A one-byte CMAF sample at `timestamp` ticks, lasting one 3000-tick (33_333µs) frame. */ -function encodeCmafFrame(data: number, timestamp: number, sequence: number): Uint8Array { - return encodeDataSegment({ +function cmafFrame(data: number, timestamp: number, sequence: number): Group.Frame { + const payload = encodeDataSegment({ kind: "video", data: new Uint8Array([data]), timestamp, @@ -38,6 +43,7 @@ function encodeCmafFrame(data: number, timestamp: number, sequence: number): Uin keyframe: true, sequence, }); + return { payload, timestamp: at(timestamp) }; } /** Yield long enough for the consumer's spawned group readers to drain what's been written. */ @@ -388,7 +394,7 @@ test("CmafFormat decodes a valid keyframe segment", () => { sequence: 0, }); - const result = format.decode(segment); + const result = format.decode(segment, at(0)); expect(result).toHaveLength(1); expect(result[0].payload).toEqual(new Uint8Array([0xca, 0xfe])); @@ -418,7 +424,7 @@ test("CmafFormat never reports an audio keyframe", () => { sequence: 0, }); - const result = format.decode(segment); + const result = format.decode(segment, at(0)); expect(result).toHaveLength(1); expect(result[0].keyframe).toBe(false); }); @@ -448,7 +454,7 @@ test("CmafFormat decodes a delta frame segment", () => { sequence: 1, }); - const result = format.decode(segment); + const result = format.decode(segment, at(3000)); expect(result).toHaveLength(1); expect(result[0].keyframe).toBe(false); @@ -466,13 +472,13 @@ test("CmafFormat converts timescale units to microseconds", () => { sequence: 0, }); - const result = format.decode(segment); + const result = format.decode(segment, at(TIMESCALE)); expect(result[0].timestamp).toBe(1_000_000 as Time.Micro); }); test("CmafFormat throws on corrupt segment", () => { const format = new CmafFormat(TEST_INIT); - expect(() => format.decode(new Uint8Array([0x00, 0x01, 0x02]))).toThrow(); + expect(() => format.decode(new Uint8Array([0x00, 0x01, 0x02]), at(0))).toThrow(); }); // --- Consumer --- @@ -1079,7 +1085,7 @@ test("Consumer with CmafFormat delivers correct timestamps", async () => { keyframe: true, sequence: 0, }), - timestamp: Time.Timestamp.now(), + timestamp: at(0), }); group.writeFrame({ payload: encodeDataSegment({ @@ -1090,7 +1096,7 @@ test("Consumer with CmafFormat delivers correct timestamps", async () => { keyframe: false, sequence: 0, }), - timestamp: Time.Timestamp.now(), + timestamp: at(3000), }); group.close(); track.writeGroup(group); @@ -1105,6 +1111,25 @@ test("Consumer with CmafFormat delivers correct timestamps", async () => { consumer.close(); }); +test("Consumer with CmafFormat times an untimed track from tfdt", async () => { + const track = new Track.Producer("test"); + const consumer = new Consumer(replay(track), { + format: new CmafFormat(TEST_INIT), + maxDelay: 500 as Time.Milli, + }); + + const group = new Group.Producer(0); + group.writeFrame({ payload: cmafFrame(0xca, TIMESCALE, 0).payload }); + group.writeFrame({ payload: cmafFrame(0xbe, TIMESCALE + 3000, 1).payload }); + group.close(); + track.writeGroup(group); + track.close(); + + const frames = await drainFrames(consumer, 200); + expect(frames.map((f) => f.timestamp)).toEqual([1_000_000, 1_033_333] as Time.Micro[]); + consumer.close(); +}); + test("CmafFormat decodes the per-sample duration", () => { const format = new CmafFormat(TEST_INIT); const segment = encodeDataSegment({ @@ -1116,7 +1141,7 @@ test("CmafFormat decodes the per-sample duration", () => { sequence: 0, }); - const [frame] = format.decode(segment); + const [frame] = format.decode(segment, at(0)); // 3000 ticks / 90000 timescale * 1_000_000 = 33333µs expect(frame.duration).toBe(33_333 as Time.Micro); }); @@ -1226,7 +1251,7 @@ test("Consumer delivers a PTS-contiguous next group whose sequence jumped (CMAF) keyframe: true, sequence: 0, }), - timestamp: Time.Timestamp.now(), + timestamp: at(0), }); a.close(); track.writeGroup(a); @@ -1252,7 +1277,7 @@ test("Consumer delivers a PTS-contiguous next group whose sequence jumped (CMAF) keyframe: true, sequence: 1, }), - timestamp: Time.Timestamp.now(), + timestamp: at(3045), }); const result = await Promise.race([ @@ -1287,7 +1312,7 @@ test("Consumer waits on a PTS gap instead of skipping to a later buffered group keyframe: true, sequence: 0, }), - timestamp: Time.Timestamp.now(), + timestamp: at(0), }); a.close(); track.writeGroup(a); @@ -1311,7 +1336,7 @@ test("Consumer waits on a PTS gap instead of skipping to a later buffered group keyframe: true, sequence: 1, }), - timestamp: Time.Timestamp.now(), + timestamp: at(90_000), }); const gap = await Promise.race([ @@ -1333,7 +1358,7 @@ test("Consumer waits on a PTS gap instead of skipping to a later buffered group keyframe: true, sequence: 2, }), - timestamp: Time.Timestamp.now(), + timestamp: at(3000), }); const result = await pending; @@ -1356,19 +1381,19 @@ test("Consumer delivers a contiguous group after one that completed out of order // A (seq 1_000_000) stays open so it remains the active group. const a = new Group.Producer(1_000_000); track.writeGroup(a); - a.writeFrame({ payload: encodeCmafFrame(0x01, 0, 0), timestamp: Time.Timestamp.now() }); + a.writeFrame(cmafFrame(0x01, 0, 0)); expect((await consumer.next())?.frame?.payload).toEqual(new Uint8Array([0x01])); // A's second frame stays buffered so A can't be duration-skipped. A ends at 6000 ticks (66_666µs). - a.writeFrame({ payload: encodeCmafFrame(0x02, 3000, 1), timestamp: Time.Timestamp.now() }); + a.writeFrame(cmafFrame(0x02, 3000, 1)); await settle(); // B (seq 1_045_000) starts at A's end and is CLOSED while A is still active, so B's own finally // block sees sequence !== #active and never records anything. B ends at 9000 ticks (100_000µs). const b = new Group.Producer(1_045_000); track.writeGroup(b); - b.writeFrame({ payload: encodeCmafFrame(0x03, 6000, 2), timestamp: Time.Timestamp.now() }); + b.writeFrame(cmafFrame(0x03, 6000, 2)); b.close(); await settle(); @@ -1382,7 +1407,7 @@ test("Consumer delivers a contiguous group after one that completed out of order const pending = consumer.next(); const c = new Group.Producer(1_090_000); track.writeGroup(c); - c.writeFrame({ payload: encodeCmafFrame(0x04, 9000, 3), timestamp: Time.Timestamp.now() }); + c.writeFrame(cmafFrame(0x04, 9000, 3)); const result = await Promise.race([pending, settle(300).then(() => "timeout" as const)]); expect(result).not.toBe("timeout"); @@ -1404,7 +1429,7 @@ test("Consumer plays the head once a waited-out gap exceeds the budget (CMAF)", // A (seq 1000): one frame, ends at 3000 ticks (33_333µs). const a = new Group.Producer(1000); track.writeGroup(a); - a.writeFrame({ payload: encodeCmafFrame(0x01, 0, 0), timestamp: Time.Timestamp.now() }); + a.writeFrame(cmafFrame(0x01, 0, 0)); expect((await consumer.next())?.frame?.payload).toEqual(new Uint8Array([0x01])); a.close(); expect((await consumer.next())?.frame).toBeUndefined(); // #active falls back to the 1001 phantom @@ -1416,7 +1441,7 @@ test("Consumer plays the head once a waited-out gap exceeds the budget (CMAF)", // could only ever arrive far beyond the 100ms budget, so B plays from its first frame. const b = new Group.Producer(2000); track.writeGroup(b); - b.writeFrame({ payload: encodeCmafFrame(0x02, 90_000, 1), timestamp: Time.Timestamp.now() }); + b.writeFrame(cmafFrame(0x02, 90_000, 1)); const result = await Promise.race([pending, settle(300).then(() => "timeout" as const)]); expect(result).not.toBe("timeout"); @@ -1439,7 +1464,7 @@ test("Consumer waits on a gap within the budget, then plays the head once it exc // A (seq 1000): one frame, ends at 3000 ticks (33_333µs). const a = new Group.Producer(1000); track.writeGroup(a); - a.writeFrame({ payload: encodeCmafFrame(0x01, 0, 0), timestamp: Time.Timestamp.now() }); + a.writeFrame(cmafFrame(0x01, 0, 0)); expect((await consumer.next())?.frame?.payload).toEqual(new Uint8Array([0x01])); a.close(); expect((await consumer.next())?.frame).toBeUndefined(); @@ -1447,7 +1472,7 @@ test("Consumer waits on a gap within the budget, then plays the head once it exc // B (seq 2000) starts at 6000 ticks (66_667µs): a gap, but only 33ms past A's end. const b = new Group.Producer(2000); track.writeGroup(b); - b.writeFrame({ payload: encodeCmafFrame(0x02, 6000, 1), timestamp: Time.Timestamp.now() }); + b.writeFrame(cmafFrame(0x02, 6000, 1)); const pending = consumer.next(); const early = await Promise.race([pending, settle(50).then(() => "waiting" as const)]); @@ -1455,7 +1480,7 @@ test("Consumer waits on a gap within the budget, then plays the head once it exc // B alone grows past the budget (up to 12_000 ticks, 100ms past A's end and beyond). for (let i = 1; i <= 3; i++) { - b.writeFrame({ payload: encodeCmafFrame(0x02, 6000 + i * 3000, 1 + i), timestamp: Time.Timestamp.now() }); + b.writeFrame(cmafFrame(0x02, 6000 + i * 3000, 1 + i)); await settle(10); } @@ -1600,7 +1625,7 @@ test("Consumer reports continuity across a PTS-contiguous group id jump (CMAF)", const a = new Group.Producer(1_000_000); track.writeGroup(a); - a.writeFrame({ payload: encodeCmafFrame(0x01, 0, 0), timestamp: Time.Timestamp.now() }); + a.writeFrame(cmafFrame(0x01, 0, 0)); a.close(); expect((await consumer.next())?.frame?.payload).toEqual(new Uint8Array([0x01])); @@ -1609,7 +1634,7 @@ test("Consumer reports continuity across a PTS-contiguous group id jump (CMAF)", // Sequence jumps +90_000 but the first PTS meets A's end, so nothing is missing. const b = new Group.Producer(1_090_000); track.writeGroup(b); - b.writeFrame({ payload: encodeCmafFrame(0x02, 3000, 1), timestamp: Time.Timestamp.now() }); + b.writeFrame(cmafFrame(0x02, 3000, 1)); const result = await consumer.next(); expect(result?.frame?.payload).toEqual(new Uint8Array([0x02])); diff --git a/js/hang/src/container/consumer.ts b/js/hang/src/container/consumer.ts index d15f94aaa6d..a7e563d6fd2 100644 --- a/js/hang/src/container/consumer.ts +++ b/js/hang/src/container/consumer.ts @@ -188,7 +188,7 @@ export class Consumer { group.empty = false; const arrival = Moq.Time.Milli.now(); - const decoded = this.#format.decode(next.payload); + const decoded = this.#format.decode(next.payload, next.timestamp); for (const sample of decoded) { const marker = this.#format.end?.(sample) !== undefined; diff --git a/js/hang/src/container/format.ts b/js/hang/src/container/format.ts index 9a77ccb2a24..e8dbdbf2e4d 100644 --- a/js/hang/src/container/format.ts +++ b/js/hang/src/container/format.ts @@ -3,8 +3,8 @@ import type { Frame } from "./types"; /** A container format that decodes raw MoQ frames into media frames. */ export interface Format { - /** Parse one MoQ frame (raw bytes) into decoded media frames. */ - decode(frame: Uint8Array): Frame[]; + /** Parse one MoQ frame's payload, given its moq-net timestamp (absent on an untimed track), into decoded media frames. */ + decode(payload: Uint8Array, timestamp: Time.Timestamp | undefined): Frame[]; /** Return the endpoint timestamp carried by empty-payload metadata. */ end?(frame: Frame): Time.Micro | undefined; } diff --git a/js/watch/src/text/renderer.ts b/js/watch/src/text/renderer.ts index 5e0d8405ee7..0f48076a695 100644 --- a/js/watch/src/text/renderer.ts +++ b/js/watch/src/text/renderer.ts @@ -293,7 +293,7 @@ export class Renderer { for (;;) { const frame = await group.readFrame(); if (!frame) break; - for (const sample of format.decode(frame.payload)) { + for (const sample of format.decode(frame.payload, frame.timestamp)) { await this.#ingest(config.format, sample, store); } } diff --git a/quest/m1/README.md b/quest/m1/README.md index d39e15b6875..3ace48e2444 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -37,7 +37,6 @@ were done or not worth their cost. - [Upstream position regression](/quest/m1/largest-regression.md) - a relay copy that sees upstream's largest group go backwards on moq-transport or epochless lite-07 fails loud instead of serving the old instance's cache - [Untimed decisions](/quest/m1/untimed-decisions.md) - the maintainer decides whether moq-archive keeps refusing untimed tracks, and whether a malformed FETCH object ends its track - [Untimed by default in Rust](/quest/m1/rust-untimed-default.md) - an undeclared Rust timescale means untimed, and shared-clock publishers declare milliseconds -- [CMAF frame timestamp](/quest/m1/cmaf-frame-timestamp.md) - CMAF decoders time samples from the moq-lite frame timestamp, using `tfdt` only within the fragment - [CMAF sample defaults](/quest/m1/cmaf-sample-defaults.md) - one trun, tfhd, trex resolver and one keyframe rule in the importer, the Rust decoder, and JS - [Shared import clock](/quest/m1/shared-clock.md) - an importer joining a clock already in use offsets its PTS instead of moving it, so captures and imports share one timeline - [FFI shape](/quest/m1/ffi-shape/README.md) - the bindings mirror Rust's layers: net at the root, then media, json, flate, audio, and video namespaces built from the handle below diff --git a/quest/m1/cmaf-frame-timestamp.md b/quest/m1/cmaf-frame-timestamp.md deleted file mode 100644 index dc8ff7d2137..00000000000 --- a/quest/m1/cmaf-frame-timestamp.md +++ /dev/null @@ -1,56 +0,0 @@ -# [S] CMAF decoders time samples from the frame timestamp - -## Goal - -A CMAF fragment decodes at its moq-lite frame timestamp, even when its `tfdt` -says otherwise. Today both decoders take sample times from `tfdt` and ignore -the frame timestamp: `fmp4::decode` in `rs/moq-mux/src/container/fmp4/mod.rs` -and `Format.decode` in `js/hang/src/container/cmaf/format.ts`. A publisher -that moves a passthrough track onto another timeline (an importer joining a -shared clock, ad insertion splicing unrelated PTS bases) would play at the -source's PTS. - -## Plan - -Decided (2026-10-01): the frame timestamp is the broadcast timeline. A -payload's own timestamps are only relative within the frame. Rewriting `tfdt` -on the publisher was rejected: it can force a v0 to v1 box upgrade, which -shifts `trun` data offsets. - -- A sample's time is the frame timestamp plus its PTS offset from the - fragment's earliest sample, not its first: the passthrough importer already - stamps the earliest PTS, and the two differ on open-GOP leading pictures or - a fragment cut mid-GOP. `fmp4::encode` stamps the first sample today, so it - moves to the earliest too. -- Rust's decode gets the frame timestamp from the caller. The fMP4 exporter - and moq-hls re-fragment decoded frames, so they follow without changes; - confirm with a test. -- The TS exporter needs no change: verbatim PES tracks carry only the PES - payload, and export writes a fresh PES header from the frame timestamp. -- An untimed CMAF frame decodes at its `tfdt` (decided 2026-10-05, replacing - the 2026-10-02 decision to refuse it). Rejected: refusing it, and parking - the quest. -- JS `Format.decode` takes the frame timestamp, a breaking `@moq/hang` - change; moq-mux's decode is crate-private. - -Open PR [#4826](https://github.com/moq-dev/moq/pull/4826) implements this -(held since 2026-10-06, branch still under `m2`). Its blocker, -[#4822](https://github.com/moq-dev/moq/pull/4822) (the untimed model), has -merged. Timedness is per track there, so the decoder can check the track -rather than each frame. Decided 2026-10-08: no forced order among the CMAF -decode changes. This, #5015 (catalog init), and -[CMAF sample defaults](/quest/m1/cmaf-sample-defaults.md) touch the same -module; whichever lands second rebases. - -Test: in Rust and JS, a fragment whose `tfdt` disagrees with its frame -timestamp decodes at the frame timestamp, with B-frame offsets preserved, and -an untimed fragment decodes at its `tfdt`. - -Public API: breaking in `@moq/hang` (`Format.decode` and -`decodeDataSegment` take the frame timestamp). Wire: none; this states what -the frame timestamp already means. - -## Related - -- [Shared clock](/quest/m1/shared-clock.md) - the first publisher to rely on it -- [CMAF sample defaults](/quest/m1/cmaf-sample-defaults.md) - the same `decode` functions; whichever lands second rebases diff --git a/quest/m1/cmaf-sample-defaults.md b/quest/m1/cmaf-sample-defaults.md index 76aa11b47a2..bde566e63ad 100644 --- a/quest/m1/cmaf-sample-defaults.md +++ b/quest/m1/cmaf-sample-defaults.md @@ -37,5 +37,4 @@ Public API: none. Wire: none. ## Related -- [CMAF frame timestamp](/quest/m1/cmaf-frame-timestamp.md) - #4826, the same `decode` functions; no forced order (decided 2026-10-08), whichever lands second rebases - [Export sync flags](/quest/m2/intra-refresh/export-sync-flags.md) - the export side of the same flags diff --git a/quest/m1/lite-untimed.md b/quest/m1/lite-untimed.md index b5e288c68a5..e9a021ece14 100644 --- a/quest/m1/lite-untimed.md +++ b/quest/m1/lite-untimed.md @@ -17,6 +17,13 @@ Timescale in TRACK_INFO and no Timestamp fields on an untimed track. The absent timestamp inside a timed track, which no longer exists (settled 2026-10-06: a mismatched frame is refused). +Decided (2026-10-08, with [#4826](https://github.com/moq-dev/moq/pull/4826)): +CMAF decoders anchor a fragment to its frame timestamp and fall back to `tfdt` +only on an untimed frame. Until this lands, an untimed CMAF track (a +draft-14 publisher such as MOQtail) relayed to a lite-05+ subscriber anchors +to the relay's send times, and lite-05/06 peers keep doing so. Shipped anyway: +the invented timestamp is this quest's to remove. + Decided (2026-10-02): one PR for both languages. Both model quests have landed (#4822). Shipping one language first would break Rust-JS interop on lite-07-wip in between. diff --git a/quest/m1/shared-clock.md b/quest/m1/shared-clock.md index 2606b209a2e..04e09b06758 100644 --- a/quest/m1/shared-clock.md +++ b/quest/m1/shared-clock.md @@ -137,8 +137,6 @@ Guidance: the clock in place of the data track, importing an fMP4 at PTS 0: both tracks advance from the capture's timeline with no rewind. Also cover the reverse order: an importer first on a default clock keeps its PTS verbatim. - A passthrough fragment whose `tfdt` disagrees with its frame timestamp - decodes at the frame timestamp, in Rust and JS. Public API: `catalog::Input`, `catalog::Producer::input`, and `Input::reserve` are new, and `Input` accepts an input-derived anchor in @@ -147,10 +145,6 @@ now fix the mapping, and importers no longer publish verbatim PTS when the clock was already taken or set with `Config::with_clock`. Wire: none. -## Required - -- [CMAF frame timestamp](/quest/m1/cmaf-frame-timestamp.md) - decoders honour an offset frame timestamp on passthrough tracks - ## Related - [Same-epoch importers](/quest/m1/hop-aligned-import.md) - supplies the input-derived anchor through this quest's API diff --git a/quest/m2/fmp4-edit-lists.md b/quest/m2/fmp4-edit-lists.md index 464f93d84c3..931ae0e605e 100644 --- a/quest/m2/fmp4-edit-lists.md +++ b/quest/m2/fmp4-edit-lists.md @@ -25,10 +25,6 @@ Public API: none. Wire: none, but a behavior change: frame timestamps now include the edit and the init loses `edts`, so a consumer that offsets AAC priming on its own must stop, or it shifts twice. Note it in the changelog. -## Required - -- [CMAF frame timestamp](/quest/m1/cmaf-frame-timestamp.md) - decoders time samples from the frame timestamp, which is where the edit lands - ## Related - [MP4 export](/quest/m2/mp4-export.md) - its open "edit lists" item is the output side diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index d594dafe0d1..8875b579724 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -1596,9 +1596,9 @@ mod tests { let broadcast = moq_net::broadcast::Info::new().produce(); let subscriber = broadcast.consume(); - let track = broadcast - .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) - .unwrap(); + // A CMAF track counts in its init's ticks. + let info = hang::container::track_info(hang::catalog::PRIORITY.audio).with_timescale(muxer.timescale()); + let track = broadcast.create_track("audio", info).unwrap(); let container = moq_mux::catalog::hang::Container::try_from(&catalog).unwrap(); let mut producer = moq_mux::container::Producer::new(track, container); diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 03fffe10f2c..2cc1d089497 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1147,7 +1147,9 @@ async fn fetch_media_group_decodes_multiple_cmaf_samples() { moq_mux::catalog::hang::Container::new(&catalog_container, moq_mux::container::Kind::Video).unwrap(); let broadcast = moq_net::broadcast::Info::new().produce(); - let track = broadcast.create_track("video", None).unwrap(); + // A CMAF track counts in its init's ticks. + let info = moq_net::track::Info::default().with_timescale(muxer.timescale()); + let track = broadcast.create_track("video", info).unwrap(); let consumer = MoqBroadcastConsumer::new(broadcast.consume()); // Buffer both samples into one moof+mdat, which is what this decodes. let mut media = moq_mux::container::Producer::new(track, container).with_buffer(Duration::from_secs(1)); diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index f61177cc525..387f5be7699 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -688,7 +688,7 @@ async fn unusable_framerate_uses_the_standard_fallback_rate() { let fragment = chunk_now(&mut exporter).await.fragment().expect("a media fragment"); assert_eq!(fragment.duration, std::time::Duration::from_secs_f64(1.0 / 30.0)); let timescale = moq_net::Timescale::new(90_000).unwrap(); - let decoded = super::decode(fragment.data, timescale, crate::container::fmp4::Kind::Video).unwrap(); + let decoded = super::decode(fragment.data, None, timescale, crate::container::fmp4::Kind::Video).unwrap(); assert_eq!(decoded[0].duration.unwrap().as_scale(timescale), 3_000); } @@ -1297,6 +1297,7 @@ fn annexb_frame(timestamp_us: u64, nals: &[&[u8]]) -> crate::container::Frame { fn video_payloads(fragment: &crate::container::fmp4::Fragment) -> Vec { super::decode( fragment.data.clone(), + None, moq_net::Timescale::new(30_000).unwrap(), super::Kind::Video, ) @@ -1853,6 +1854,7 @@ async fn split_parameter_sets_case( .flat_map(|fragment| { super::decode( fragment.data.clone(), + None, moq_net::Timescale::new(30_000).unwrap(), super::Kind::Video, ) @@ -2123,6 +2125,96 @@ async fn hevc_main10_is_described_from_its_sps() { assert_eq!(hev1.hvcc.bit_depth_chroma_minus8, 2, "10-bit chroma"); } +/// The exporter re-fragments decoded samples, so a passthrough CMAF fragment whose `tfdt` +/// still carries its source PTS exports at its moq-lite frame timestamp, B-frame order kept. +#[tokio::test(start_paused = true)] +async fn cmaf_source_exports_at_the_frame_timestamp() { + use hang::catalog::{Container, VideoCodec, VideoConfig}; + + let mut config = VideoConfig::new(VideoCodec::VP8); + config.coded_width = Some(320); + config.coded_height = Some(240); + let trak = super::synthesize_video_trak(1, 1_000_000, &config, None).unwrap(); + let trex = mp4_atom::Trex { + track_id: 1, + default_sample_description_index: 1, + ..Default::default() + }; + config.container = Container::Cmaf { + init: super::encode_init(None, vec![trak], vec![trex]).unwrap(), + }; + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + catalog + .modify() + .unwrap() + .video + .renditions + .insert("video".into(), config); + + let source = 3_600_000_000; + let sample = |micros: u64, keyframe: bool| crate::container::Frame { + timestamp: moq_net::Timestamp::from_micros(micros).unwrap(), + payload: Bytes::from_static(&[0x10, 0x00, 0x00, 0x9d, 0x01, 0x2a, 0x40, 0x01, 0xf0, 0x00]), + keyframe, + duration: Some(moq_net::Timestamp::from_micros(33_000).unwrap()), + }; + let samples = [ + sample(source, true), + sample(source + 66_000, false), + sample(source + 33_000, false), + ]; + let info = super::FragmentInfo { + timescale: moq_net::Timescale::MICRO, + track_id: 1, + sequence_number: 0, + kind: super::Kind::Video, + }; + let fragment = super::encode_fragment(info, &samples).unwrap(); + + let mut group = track.append_group().unwrap(); + let mut frame = group + .create_frame(moq_net::frame::Info { + size: fragment.len() as u64, + timestamp: Some(moq_net::Timestamp::from_secs(10).unwrap()), + }) + .unwrap(); + frame.write(fragment).unwrap(); + frame.finish().unwrap(); + group.finish().unwrap(); + track.finish().unwrap(); + + let catalog_stream = crate::catalog::Consumer::<()>::new(&consumer, crate::catalog::CatalogFormat::Hang) + .await + .expect("catalog consumer"); + let mut exporter = crate::container::fmp4::Export::new(crate::source::announced(&consumer), catalog_stream) + .with_max_delay(RECORDING_MAX_DELAY); + let init = chunk_now(&mut exporter).await.init().expect("init"); + let fragments = drain_now(&mut exporter).await; + + let scale = moq_net::Timescale::new(timescale(&init).into()).unwrap(); + let presented: Vec = fragments + .into_iter() + .flat_map(|fragment| super::decode(fragment.data, None, scale, super::Kind::Video).unwrap()) + .map(|sample| sample.timestamp.as_micros()) + .collect(); + assert_eq!(presented, [10_000_000, 10_066_000, 10_033_000]); +} + +/// The single track's media timescale, from an init segment. +fn timescale(init: &Bytes) -> u32 { + let mut cursor = Cursor::new(init.as_ref()); + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).expect("decode init") { + if let mp4_atom::Any::Moov(moov) = atom { + return moov.trak[0].mdia.mdhd.timescale; + } + } + panic!("no moov"); +} + /// The catalog codec string and dimensions are enough, so the init is written /// before any media frame arrives. #[tokio::test(start_paused = true)] diff --git a/rs/moq-mux/src/container/fmp4/fragmenter.rs b/rs/moq-mux/src/container/fmp4/fragmenter.rs index cfc4c67a7e5..5a87c3a949d 100644 --- a/rs/moq-mux/src/container/fmp4/fragmenter.rs +++ b/rs/moq-mux/src/container/fmp4/fragmenter.rs @@ -302,8 +302,13 @@ mod tests { ); for (fragment, expected) in fragments.iter().zip(&input) { - let decoded = - super::super::decode(fragment.data.clone(), timescale, crate::container::fmp4::Kind::Video).unwrap(); + let decoded = super::super::decode( + fragment.data.clone(), + None, + timescale, + crate::container::fmp4::Kind::Video, + ) + .unwrap(); assert_eq!(decoded.len(), 1); assert_eq!(decoded[0].timestamp, expected.timestamp, "pts survives the reorder"); } diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 144d9eeb717..49daa4034b3 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -274,10 +274,9 @@ impl Import { continue; } - // Declare the track at the fMP4's native timescale. Frame timestamps are - // emitted at this same scale (see below), so they satisfy the track's - // timescale invariant and ride the wire for the relay, redundant with the - // timing already inside each CMAF fragment. + // Declare the track at the fMP4's native timescale, which the moq-hang draft requires + // of a CMAF track: frame timestamps are emitted at this same scale (see below), so + // each one carries its fragment's earliest presentation time exactly. let timescale = moq_net::Timescale::new(trak.mdia.mdhd.timescale as u64)?; let track = self.broadcast.create_track( self.broadcast.unique_name(suffix), @@ -887,8 +886,8 @@ impl Import { let fragment_bytes = Bytes::from(moof_buf); // Carry the fragment's earliest presentation time as the frame timestamp, - // in the track's native timescale. The relay reads it off the wire; the - // consumer still drives playback from the fragment's internal timing. + // in the track's native timescale. Consumers present the fragment at it; + // `tfdt` only places the samples relative to each other. let timestamp = min_timestamp.ok_or(Error::MissingTrun)?; // The first fragment of the import is live on arrival. Anchor before releasing the // reservation, so the first snapshot carries the final clock; the moov declared every diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 69a386de1e1..45f377805d9 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -564,7 +564,7 @@ fn sample(timestamp_us: u64, keyframe: bool, duration_us: Option) -> crate: fn decode_rejects_durationless_multisample() { let frames = vec![sample(0, true, None), sample(33_000, false, None)]; let frag = super::encode_fragment(info(1, scale(), 0), &frames).unwrap(); - let err = super::decode(frag, scale(), crate::container::fmp4::Kind::Video).unwrap_err(); + let err = super::decode(frag, None, scale(), crate::container::fmp4::Kind::Video).unwrap_err(); assert!(matches!(err, super::Error::MissingSampleDuration), "got {err:?}"); } @@ -572,7 +572,7 @@ fn decode_rejects_durationless_multisample() { #[test] fn decode_single_sample_no_duration_ok() { let frag = super::encode_fragment(info(1, scale(), 0), &[sample(0, true, None)]).unwrap(); - let out = super::decode(frag, scale(), crate::container::fmp4::Kind::Video).unwrap(); + let out = super::decode(frag, None, scale(), crate::container::fmp4::Kind::Video).unwrap(); assert_eq!(out.len(), 1); assert_eq!(out[0].timestamp.as_micros(), 0); } @@ -583,7 +583,7 @@ fn decode_single_sample_no_duration_ok() { fn decode_multisample_with_durations_roundtrips() { let frames = vec![sample(0, true, Some(33_000)), sample(33_000, false, Some(33_000))]; let frag = super::encode_fragment(info(1, scale(), 0), &frames).unwrap(); - let out = super::decode(frag, scale(), crate::container::fmp4::Kind::Video).unwrap(); + let out = super::decode(frag, None, scale(), crate::container::fmp4::Kind::Video).unwrap(); assert_eq!(out.len(), 2); assert_eq!(out[0].timestamp.as_micros(), 0); assert_eq!(out[1].timestamp.as_micros(), 33_000); diff --git a/rs/moq-mux/src/container/fmp4/mod.rs b/rs/moq-mux/src/container/fmp4/mod.rs index a8e5b2f4dcf..48843bceaff 100644 --- a/rs/moq-mux/src/container/fmp4/mod.rs +++ b/rs/moq-mux/src/container/fmp4/mod.rs @@ -144,6 +144,14 @@ pub enum Error { #[error("invalid data offset")] InvalidDataOffset, + /// The frame timestamp carries a fragment's earliest presentation time exactly only when the + /// track counts in the same ticks as the media. + #[error("track timescale {track:?} doesn't match the CMAF mdhd timescale {mdhd:?}")] + TimescaleMismatch { + track: Option, + mdhd: moq_net::Timescale, + }, + #[error("unknown track {0}")] UnknownTrack(u32), @@ -308,6 +316,11 @@ impl Wire { &self.trak } + /// The media timescale, which a CMAF track must count in. + pub(crate) fn timescale(&self) -> Result { + Ok(moq_net::Timescale::new(self.trak.mdia.mdhd.timescale as u64)?) + } + /// The media kind declared by the track handler. fn kind(&self) -> Result { Kind::from_handler(self.trak.mdia.hdlr.handler) @@ -326,7 +339,13 @@ impl Container for Wire { type Error = Error; fn write(&self, group: &mut moq_net::group::Producer, frames: &[Frame]) -> std::result::Result<(), Self::Error> { - let timescale = moq_net::Timescale::new(self.trak.mdia.mdhd.timescale as u64)?; + let timescale = self.timescale()?; + if group.timescale() != Some(timescale) { + return Err(Error::TimescaleMismatch { + track: group.timescale(), + mdhd: timescale, + }); + } let track_id = self.trak.tkhd.track_id; encode( group, @@ -351,8 +370,13 @@ impl Container for Wire { return Poll::Ready(Ok(None)); }; - let timescale = moq_net::Timescale::new(self.trak.mdia.mdhd.timescale as u64)?; - Poll::Ready(Ok(Some(decode(frame.payload, timescale, self.kind()?)?))) + let timescale = self.timescale()?; + Poll::Ready(Ok(Some(decode( + frame.payload, + frame.timestamp, + timescale, + self.kind()?, + )?))) } } @@ -361,24 +385,46 @@ impl Container for Wire { /// `Kind::Audio` says the track's samples are all independently decodable. Packagers flag every /// audio sample a sync sample, which as a [`Frame::keyframe`] would open a group per sample, /// so an audio sample decodes with the bit clear and the consumer marks the group start. -pub(crate) fn decode(data: Bytes, timescale: moq_net::Timescale, kind: Kind) -> Result> { +/// +/// `timestamp` is the moq-lite frame timestamp, which is the broadcast timeline: the fragment's +/// earliest sample presents at it. `tfdt` and the `trun` offsets only place the samples relative +/// to each other, since a publisher may move a passthrough track to another timeline without +/// rewriting the payload. An untimed frame (`None`) has no broadcast time, so its samples +/// present at the time `tfdt` gives them. +pub(crate) fn decode( + data: Bytes, + timestamp: Option, + timescale: moq_net::Timescale, + kind: Kind, +) -> Result> { use mp4_atom::DecodeMaybe; let mut cursor = std::io::Cursor::new(&data); let mut moof = None; - let mut mdat_data = None; + let mut mdat = None; - while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor)? { + loop { + let start = cursor.position() as usize; + let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor)? else { + break; + }; match atom { - mp4_atom::Any::Moof(m) => moof = Some(m), - mp4_atom::Any::Mdat(m) => mdat_data = Some(m.data), + mp4_atom::Any::Moof(m) => moof = Some((start, m)), + mp4_atom::Any::Mdat(m) => mdat = Some((cursor.position() as usize - m.data.len(), m.data)), _ => {} } } - let moof = moof.ok_or(Error::NoMoof)?; - let mdat_data = mdat_data.ok_or(Error::NoMdat)?; + let (moof_start, moof) = moof.ok_or(Error::NoMoof)?; + let (mdat_start, mdat_data) = mdat.ok_or(Error::NoMdat)?; let traf = moof.traf.first().ok_or(Error::NoTraf)?; + + // Samples are read from the mdat front to back, so each run must start where the previous + // one ended. A run's data_offset counts from the moof's first byte (CMAF's default-base-is-moof). + if traf.tfhd.base_data_offset.is_some() { + return Err(Error::InvalidDataOffset); + } + let data_start = mdat_start.checked_sub(moof_start).ok_or(Error::InvalidDataOffset)?; let tfdt = traf.tfdt.as_ref().ok_or(Error::NoTfdt)?; let base_dts = tfdt.base_media_decode_time; @@ -390,12 +436,23 @@ pub(crate) fn decode(data: Bytes, timescale: moq_net::Timescale, kind: Kind) -> // DTS, silently collapsing their timestamps, so reject that fragment instead. let total_samples: usize = traf.trun.iter().map(|t| t.entries.len()).sum(); + // Rounded the way `encode` stamps it, so a timestamp the track carried at another scale + // lands back on the tick it was written at. + let anchor = timestamp.map(|t| timestamp_ticks(t, timescale)).transpose()?; + let mut frames = Vec::new(); + let mut ptss = Vec::with_capacity(total_samples); let mut offset = 0usize; - let mut dts = base_dts; + let mut dts = i128::from(base_dts); let mut sample_index = 0usize; for trun in &traf.trun { + if let Some(data_offset) = trun.data_offset + && usize::try_from(data_offset).ok() != Some(data_start + offset) + { + return Err(Error::InvalidDataOffset); + } + for entry in &trun.entries { let size = entry.size.or(default_size).unwrap_or(0) as usize; let end = offset + size; @@ -408,10 +465,7 @@ pub(crate) fn decode(data: Bytes, timescale: moq_net::Timescale, kind: Kind) -> }); } - let cts = entry.cts.unwrap_or_default() as i64; - let pts = dts.checked_add_signed(cts).ok_or(Error::PtsOverflow)?; - // Preserve the fmp4 track's native scale through the pipeline. - let timestamp = Timestamp::new(pts, timescale)?; + ptss.push(dts + i128::from(entry.cts.unwrap_or_default())); let payload = Bytes::copy_from_slice(&mdat_data[offset..end]); let flags = entry.flags.unwrap_or(0); // depends_on_no_other (bits 24-25 == 0x2) means keyframe @@ -433,32 +487,48 @@ pub(crate) fn decode(data: Bytes, timescale: moq_net::Timescale, kind: Kind) -> .transpose()?; frames.push(Frame { - timestamp, + // Placeholder until the earliest PTS is known. + timestamp: Timestamp::ZERO, payload, keyframe, duration, }); offset = end; - dts += sample_duration.unwrap_or(0) as u64; + dts += i128::from(sample_duration.unwrap_or(0)); sample_index += 1; } } + let (anchor, earliest) = match anchor { + Some(anchor) => (anchor, ptss.iter().copied().min().unwrap_or_default()), + None => (0, 0), + }; + for (frame, pts) in frames.iter_mut().zip(ptss) { + let ticks = u64::try_from(pts - earliest) + .ok() + .and_then(|offset| anchor.checked_add(offset)) + .ok_or(Error::PtsOverflow)?; + // Preserve the fmp4 track's native scale through the pipeline. + frame.timestamp = Timestamp::new(ticks, timescale)?; + } + Ok(frames) } fn encode(group: &mut moq_net::group::Producer, frames: &[Frame], info: FragmentInfo) -> Result<()> { - if frames.is_empty() { + // The fragment may carry several samples; the net frame's timestamp is the + // fragment's earliest presentation time, which `decode` anchors its samples to. + // It is rounded to the track's ticks the same way the samples are, so the two agree. + let Some(earliest) = frames.iter().map(|f| f.timestamp).min() else { return Ok(()); - } + }; + let timestamp = Timestamp::new(timestamp_ticks(earliest, info.timescale)?, info.timescale)?; let bytes = encode_fragment(info, frames)?; - // The fragment may carry several samples; the net frame's timestamp is the - // fragment's earliest presentation time so a relay can order it. let mut writer = group.create_frame(moq_net::frame::Info { size: bytes.len() as u64, - timestamp: Some(frames[0].timestamp), + timestamp: Some(timestamp), })?; writer.write(bytes)?; writer.finish()?; @@ -1291,7 +1361,7 @@ mod tests { use mp4_atom::Encode; // Microsecond timescale so each tick maps 1:1 to the Timestamp's µs. - // decode() walks the mdat by sample size and ignores data_offset, so a + // Without a data_offset, decode() reads the run from the start of the mdat, so a // hand-built moof+mdat with explicit per-sample durations is enough. let timescale = moq_net::Timescale::MICRO; let moof = mp4_atom::Moof { @@ -1305,7 +1375,7 @@ mod tests { base_media_decode_time: 0, }), trun: vec![mp4_atom::Trun { - data_offset: Some(0), + data_offset: None, entries: vec![ mp4_atom::TrunEntry { size: Some(2), @@ -1331,7 +1401,7 @@ mod tests { .encode(&mut buf) .unwrap(); - let frames = decode(Bytes::from(buf), timescale, Kind::Video).unwrap(); + let frames = decode(Bytes::from(buf), None, timescale, Kind::Video).unwrap(); assert_eq!(frames.len(), 2); assert_eq!(frames[0].timestamp, ts(0)); assert_eq!(frames[0].duration, Some(ts(33_333))); @@ -1351,7 +1421,7 @@ mod tests { }]; let fragment = encode_fragment(info(1, timescale, 0), &input).unwrap(); - let frames = decode(fragment, timescale, Kind::Video).unwrap(); + let frames = decode(fragment, None, timescale, Kind::Video).unwrap(); assert_eq!(frames.len(), 1); assert_eq!(frames[0].duration, Some(ts(33_333))); @@ -1466,7 +1536,7 @@ mod tests { ]; let fragment = encode_fragment(info(1, timescale, 0), &input).unwrap(); - let frames = decode(fragment, timescale, Kind::Video).unwrap(); + let frames = decode(fragment, None, timescale, Kind::Video).unwrap(); assert_eq!(frames.len(), input.len()); for (actual, expected) in frames.iter().zip(&input) { @@ -1507,12 +1577,12 @@ mod tests { .collect(); assert_eq!(flags, vec![0x0200_0000; 3]); - let audio = decode(fragment.clone(), timescale, Kind::Audio).unwrap(); + let audio = decode(fragment.clone(), None, timescale, Kind::Audio).unwrap(); assert!( audio.iter().all(|frame| !frame.keyframe), "audio never decodes a keyframe" ); - let video = decode(fragment, timescale, Kind::Video).unwrap(); + let video = decode(fragment, None, timescale, Kind::Video).unwrap(); assert!( video.iter().all(|frame| frame.keyframe), "the sync flag is a video keyframe" @@ -1540,6 +1610,139 @@ mod tests { assert_eq!(video.kind().unwrap(), Kind::Video); } + fn sample(micros: u64) -> Frame { + Frame { + timestamp: ts(micros), + payload: Bytes::from_static(&[0xAB]), + keyframe: false, + duration: Some(ts(33_000)), + } + } + + fn micros(frames: &[Frame]) -> Vec { + frames.iter().map(|f| f.timestamp.as_micros()).collect() + } + + /// The frame timestamp is the broadcast timeline: a passthrough fragment whose `tfdt` still + /// carries its source PTS decodes at the frame timestamp, keeping its B-frame order. + #[test] + fn decode_times_samples_from_the_frame_timestamp() { + let timescale = moq_net::Timescale::MICRO; + let source = 3_600_000_000; + let input = [sample(source), sample(source + 66_000), sample(source + 33_000)]; + let fragment = encode_fragment(info(1, timescale, 0), &input).unwrap(); + + let frames = decode(fragment, Some(ts(10_000_000)), timescale, Kind::Video).unwrap(); + assert_eq!(micros(&frames), [10_000_000, 10_066_000, 10_033_000]); + } + + /// A fragment can open on a sample that presents after a later one (open-GOP leading + /// pictures, or a cut mid-GOP). The earliest presentation time is the anchor, on both sides. + #[tokio::test] + async fn the_frame_timestamp_is_the_earliest_presentation_time() { + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.coded_width = Some(320); + config.coded_height = Some(240); + let timescale = moq_net::Timescale::MICRO; + let wire = Wire::new(synthesize_video_trak(1, 1_000_000, &config, None).unwrap()); + + let broadcast = moq_net::broadcast::Info::new().produce(); + let info = moq_net::track::Info::default().with_timescale(timescale); + let track = broadcast.create_track("video", info).unwrap(); + let mut group = track.append_group().unwrap(); + let input = [sample(100_000), sample(33_000), sample(66_000)]; + wire.write(&mut group, &input).unwrap(); + + let frame = group.consume().read_frame().await.unwrap().unwrap(); + assert_eq!(frame.timestamp.unwrap().as_micros(), 33_000); + + let frames = decode(frame.payload.clone(), frame.timestamp, timescale, Kind::Video).unwrap(); + assert_eq!(micros(&frames), [100_000, 33_000, 66_000]); + + let shifted = decode(frame.payload, Some(ts(5_000_000)), timescale, Kind::Video).unwrap(); + assert_eq!(micros(&shifted), [5_067_000, 5_000_000, 5_033_000]); + } + + /// An untimed track's frames carry no broadcast time, so their samples present at `tfdt`. + #[test] + fn an_untimed_fragment_decodes_at_its_tfdt() { + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.coded_width = Some(320); + config.coded_height = Some(240); + let timescale = moq_net::Timescale::MICRO; + let wire = Wire::new(synthesize_video_trak(1, 1_000_000, &config, None).unwrap()); + + let broadcast = moq_net::broadcast::Info::new().produce(); + let untimed = moq_net::track::Info::default().with_timescale(None); + let track = broadcast.create_track("video", untimed).unwrap(); + let mut group = track.append_group().unwrap(); + let source = 3_600_000_000; + let input = [sample(source), sample(source + 66_000), sample(source + 33_000)]; + let fragment = encode_fragment(info(1, timescale, 0), &input).unwrap(); + group.write_frame(None, fragment).unwrap(); + + let Poll::Ready(Ok(Some(frames))) = wire.poll_read(&mut group.consume(), &kio::Waiter::noop()) else { + panic!("the fragment is ready"); + }; + let source = u128::from(source); + assert_eq!(micros(&frames), [source, source + 66_000, source + 33_000]); + } + + /// A track at another scale than the media couldn't carry the earliest presentation time + /// exactly, so writing a fragment to it fails. + #[test] + fn write_refuses_a_track_at_another_timescale() { + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.coded_width = Some(320); + config.coded_height = Some(240); + let wire = Wire::new(synthesize_video_trak(1, 90_000, &config, None).unwrap()); + + let broadcast = moq_net::broadcast::Info::new().produce(); + let info = moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI); + let track = broadcast.create_track("video", info).unwrap(); + let mut group = track.append_group().unwrap(); + + let err = wire.write(&mut group, &[sample(0)]).unwrap_err(); + assert!(matches!(err, Error::TimescaleMismatch { .. }), "got {err:?}"); + } + + /// Samples are read front to back, so a run whose data_offset points anywhere but the next + /// sample is refused rather than sliced from the wrong bytes. + #[test] + fn decode_refuses_a_run_that_skips_bytes() { + use mp4_atom::{DecodeMaybe, Encode}; + + let timescale = moq_net::Timescale::MICRO; + let fragment = encode_fragment(info(1, timescale, 0), &[sample(0), sample(33_000)]).unwrap(); + + let mut cursor = std::io::Cursor::new(fragment.as_ref()); + let Some(mp4_atom::Any::Moof(mut moof)) = mp4_atom::Any::decode_maybe(&mut cursor).unwrap() else { + panic!("a moof first"); + }; + let mdat = &fragment[cursor.position() as usize..]; + let offset = moof.traf[0].trun[0].data_offset.as_mut().unwrap(); + *offset += 1; + + let mut skewed = Vec::new(); + moof.encode(&mut skewed).unwrap(); + skewed.extend_from_slice(mdat); + + let err = decode(Bytes::from(skewed), None, timescale, Kind::Video).unwrap_err(); + assert!(matches!(err, Error::InvalidDataOffset), "got {err:?}"); + } + + /// A frame timestamp the track carries at another scale anchors to the nearest tick, the way + /// `encode` stamps it. + #[test] + fn decode_rounds_the_frame_timestamp_to_the_nearest_tick() { + let timescale = moq_net::Timescale::new(90_000).unwrap(); + let fragment = encode_fragment(info(1, timescale, 0), &[sample(0)]).unwrap(); + + // Tick 100 is 1111.1 µs, which a microsecond track truncates to 1111. + let frames = decode(fragment, Some(ts(1_111)), timescale, Kind::Video).unwrap(); + assert_eq!(frames[0].timestamp, Timestamp::new(100, timescale).unwrap()); + } + #[test] fn decode_without_duration_reports_none() { // encode_fragment writes no sample-duration for a duration-less frame, @@ -1553,7 +1756,7 @@ mod tests { }]; let fragment = encode_fragment(info(1, timescale, 0), &frames).unwrap(); - let frames = decode(fragment, timescale, Kind::Video).unwrap(); + let frames = decode(fragment, None, timescale, Kind::Video).unwrap(); assert_eq!(frames.len(), 1); assert_eq!(frames[0].duration, None); @@ -1577,7 +1780,7 @@ mod tests { base_media_decode_time: 2_000, }), trun: vec![mp4_atom::Trun { - data_offset: Some(0), + data_offset: None, entries: vec![mp4_atom::TrunEntry { size: None, duration: None, @@ -1592,7 +1795,7 @@ mod tests { moof.encode(&mut buf).unwrap(); mp4_atom::Mdat { data: vec![0xDE, 0xAD] }.encode(&mut buf).unwrap(); - let frames = decode(Bytes::from(buf), timescale, Kind::Video).unwrap(); + let frames = decode(Bytes::from(buf), None, timescale, Kind::Video).unwrap(); assert_eq!(frames.len(), 1); assert_eq!(frames[0].timestamp.as_micros(), 83_333); assert_eq!(frames[0].duration, None); diff --git a/rs/moq-mux/src/container/fmp4/muxer.rs b/rs/moq-mux/src/container/fmp4/muxer.rs index d9089d90b47..8a67a5edee9 100644 --- a/rs/moq-mux/src/container/fmp4/muxer.rs +++ b/rs/moq-mux/src/container/fmp4/muxer.rs @@ -385,7 +385,7 @@ mod tests { // Decode it back: timestamps survive at the muxer's timescale (framerate * 1000). let timescale = moq_net::Timescale::new(30_000).unwrap(); - let decoded = super::super::decode(fragment, timescale, crate::container::fmp4::Kind::Video).unwrap(); + let decoded = super::super::decode(fragment, None, timescale, crate::container::fmp4::Kind::Video).unwrap(); assert_eq!(decoded.len(), 2); assert_eq!(decoded[0].timestamp.as_micros(), 10_000_000); assert!(decoded[0].keyframe); @@ -575,6 +575,7 @@ mod tests { }; let decoded = super::super::decode( muxer.fragment(0, &[frame]).unwrap(), + None, timescale, crate::container::fmp4::Kind::Video, ) @@ -605,6 +606,7 @@ mod tests { }; let decoded = super::super::decode( muxer.fragment(0, &[frame]).unwrap(), + None, timescale, crate::container::fmp4::Kind::Video, ) @@ -680,6 +682,7 @@ mod tests { let decoded = super::super::decode( muxer.fragment(0, &[frame]).unwrap(), + None, timescale, crate::container::fmp4::Kind::Video, ) @@ -766,7 +769,7 @@ mod tests { let fragment = muxer.fragment(0, &frames).unwrap(); let timescale = moq_net::Timescale::new(48_000).unwrap(); - let decoded = super::super::decode(fragment, timescale, crate::container::fmp4::Kind::Video).unwrap(); + let decoded = super::super::decode(fragment, None, timescale, crate::container::fmp4::Kind::Video).unwrap(); assert_eq!(decoded.len(), 4); for f in &decoded { assert_eq!( @@ -800,7 +803,7 @@ mod tests { let fragment = muxer.fragment(0, &frames).unwrap(); let timescale = moq_net::Timescale::new(48_000).unwrap(); - let decoded = super::super::decode(fragment, timescale, crate::container::fmp4::Kind::Video).unwrap(); + let decoded = super::super::decode(fragment, None, timescale, crate::container::fmp4::Kind::Video).unwrap(); let first = decoded[0].duration.unwrap().as_micros(); assert_eq!(first, 20_000, "the pause is a discontinuity, not a 2405 second sample"); } diff --git a/rs/moq-mux/src/container/group.rs b/rs/moq-mux/src/container/group.rs index c7888e09c63..92c756af347 100644 --- a/rs/moq-mux/src/container/group.rs +++ b/rs/moq-mux/src/container/group.rs @@ -194,7 +194,9 @@ mod tests { let format = Hang::new(&cmaf, crate::container::Kind::Video).unwrap(); let broadcast = moq_net::broadcast::Info::new().produce(); - let track = broadcast.create_track("video", None).unwrap(); + // A CMAF track counts in its init's ticks. + let info = moq_net::track::Info::default().with_timescale(muxer.timescale()); + let track = broadcast.create_track("video", info).unwrap(); let consumer = broadcast.consume(); // Buffer both samples into one moof+mdat, which is what this decodes. diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index 7a440c93507..9aca70feae6 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -23,6 +23,23 @@ fn video_hint(init: &VideoInit, default_codec: Option hint } +/// The track info for a rendition in `container`. A CMAF track counts in its init's `mdhd` ticks, +/// as the moq-hang draft requires, so each frame timestamp lands on a media tick. +fn track_info( + reserved: &crate::catalog::Reserved, + priority: u8, + container: &hang::catalog::Container, +) -> Result { + let info = reserved.track_info(priority); + match container { + hang::catalog::Container::Cmaf { init, .. } => { + let wire = crate::container::fmp4::Wire::from_init(init)?; + Ok(info.with_timescale(wire.timescale()?)) + } + _ => Ok(info), + } +} + /// The codec parser fills everything from the init bytes, so the label and container are all the /// caller adds. fn with_init(init: &AudioInit, mut config: hang::catalog::AudioConfig) -> hang::catalog::AudioConfig { @@ -190,16 +207,14 @@ impl Track { /// /// The caller reserves the track (by name) with /// [`BroadcastProducer::reserve_track`](moq_net::broadcast::Producer::reserve_track); the - /// importer accepts it here, which is where the track's timescale is set. The rendition is + /// importer accepts it here, at the microsecond timescale or a CMAF init's own. The rendition is /// published from the init bytes, since audio has no in-band config to wait for. pub fn audio( request: moq_net::track::Request, reserved: crate::catalog::Reserved, init: AudioInit, ) -> Result { - // Accept at the legacy microsecond timescale, matching the frame timestamps the container - // stamps. A codec-specific timescale (e.g. the opus sample rate) would be chosen here. - let track = request.accept(reserved.track_info(hang::catalog::PRIORITY.audio)); + let track = request.accept(track_info(&reserved, hang::catalog::PRIORITY.audio, &init.container)?); let data = init.data.as_ref(); let kind = match init.format { AudioFormat::Aac => { @@ -236,7 +251,11 @@ impl Track { ) -> Result { use hang::catalog::VideoCodec; - let track = request.accept(reserved.track_info(hang::catalog::PRIORITY.video)); + let track = request.accept(track_info( + &reserved, + hang::catalog::PRIORITY.video, + &init.hint.container, + )?); let data = init.data.as_ref(); let kind = match init.format { VideoFormat::Avc1 => { @@ -611,15 +630,18 @@ impl TrackStream { /// /// The caller reserves the track with /// [`BroadcastProducer::reserve_track`](moq_net::broadcast::Producer::reserve_track); - /// the importer accepts it here at the legacy microsecond timescale (where a codec-specific - /// timescale would be chosen). A [`VideoHint`] carrying a codec publishes the catalog before the + /// the importer accepts it here, at the microsecond timescale or a CMAF init's own. A [`VideoHint`] carrying a codec publishes the catalog before the /// first frame; any [`VideoInit::data`] seeds the stream (as a call to [`initialize`](Self::initialize)). pub fn video( request: moq_net::track::Request, reserved: crate::catalog::Reserved, init: VideoInit, ) -> Result { - let track = request.accept(reserved.track_info(hang::catalog::PRIORITY.video)); + let track = request.accept(track_info( + &reserved, + hang::catalog::PRIORITY.video, + &init.hint.container, + )?); let hint = video_hint(&init, None); // Only the self-delimiting codecs can be recovered from a raw byte stream. let kind = match init.format { @@ -983,6 +1005,29 @@ mod tests { assert!(!catalog.snapshot().audio.renditions.contains_key("0.opus")); } + /// A CMAF rendition counts in its init's ticks, so its frames can carry the earliest + /// presentation time exactly. + #[tokio::test(start_paused = true)] + async fn cmaf_audio_counts_in_the_init_timescale() { + let (broadcast, catalog) = new_broadcast(); + let config = crate::codec::opus::config(&opus_head()).unwrap(); + let cmaf = crate::container::fmp4::Muxer::audio(&config) + .unwrap() + .init() + .unwrap() + .unwrap(); + let mut init = AudioInit::new(AudioFormat::Opus, opus_head()); + init.container = hang::catalog::Container::Cmaf { init: cmaf }; + + // The CMAF writer refuses a track at any other scale. + let request = broadcast.reserve_track("audio").unwrap(); + let mut import = Track::audio(request, catalog.reserve(), init).unwrap(); + import + .decode(b"opus payload", Some(Timestamp::from_micros(20_000).unwrap())) + .unwrap(); + import.finish().unwrap(); + } + #[tokio::test(start_paused = true)] async fn opus_import_delivers_frames() { let (broadcast, catalog) = new_broadcast(); diff --git a/test/audio-quality/clients/js/src/capture.ts b/test/audio-quality/clients/js/src/capture.ts index a68c5e16f84..9c48c7df79c 100644 --- a/test/audio-quality/clients/js/src/capture.ts +++ b/test/audio-quality/clients/js/src/capture.ts @@ -117,7 +117,7 @@ export async function* arrivals(group: Group.Consumer, decoder: Container.Format const next = await group.readFrame(); const at = performance.now(); if (!next) return; - for (const frame of decoder.decode(next.payload)) { + for (const frame of decoder.decode(next.payload, next.timestamp)) { if (decoder.end?.(frame) !== undefined) continue; yield [at, Time.Milli.fromMicro(frame.timestamp), group.sequence]; }