diff --git a/js/hang/src/util/opus.ts b/js/hang/src/util/opus.ts index 28ec3aba4d..a6ba348aa1 100644 --- a/js/hang/src/util/opus.ts +++ b/js/hang/src/util/opus.ts @@ -6,6 +6,9 @@ // 44100 is never a valid decoder config. Chrome hides this by ignoring the configured rate; Safari // trusts it and fails every decode. +/** The decoder pre-roll in milliseconds a joining Opus viewer discards, per RFC 7845 section 4.6. */ +export const WARMUP = 80; + /** Full-band Opus: its highest rate, and the one to use when the source rate is unknown. */ export const DEFAULT_SAMPLE_RATE = 48_000; diff --git a/js/justfile b/js/justfile index ce574a707f..4fad3263e3 100644 --- a/js/justfile +++ b/js/justfile @@ -101,3 +101,8 @@ arrivals-bench: clock-bench: bun install --frozen-lockfile bun clock/bench/timers.ts + +# Sweep decoder count against PCM callback size during audio warmup. +audio-warmup-bench: + bun install --frozen-lockfile + bun watch/bench/warmup.ts diff --git a/js/publish/src/audio/encoder.test.ts b/js/publish/src/audio/encoder.test.ts index 4adaaa70a9..a4efb84b9a 100644 --- a/js/publish/src/audio/encoder.test.ts +++ b/js/publish/src/audio/encoder.test.ts @@ -25,6 +25,7 @@ describe("resolve", () => { test("defaults Opus to 20ms", () => { const resolved = resolve(captured, "opus"); expect(resolved.frameDuration).toBe(Time.Micro(20_000)); + expect(Number(resolved.catalog.warmup)).toBe(80); expect(resolved.catalog.jitter).toBeUndefined(); }); @@ -54,6 +55,7 @@ describe("resolve", () => { test("leaves AAC without a frame duration", () => { const resolved = resolve(captured, "aac"); expect(resolved.frameDuration).toBeUndefined(); + expect(resolved.catalog.warmup).toBeUndefined(); expect(resolved.catalog.jitter).toBeUndefined(); }); }); diff --git a/js/publish/src/audio/encoder.ts b/js/publish/src/audio/encoder.ts index 7ee7fc56af..bc3804c289 100644 --- a/js/publish/src/audio/encoder.ts +++ b/js/publish/src/audio/encoder.ts @@ -813,6 +813,7 @@ export function resolve(captured: Format, selected: Codec): Resolved { return { catalog: { codec: "opus", + warmup: Catalog.u53(Util.Opus.WARMUP), sampleRate, numberOfChannels, bitrate: Catalog.u53(codec.bitrate ?? captured.channelCount * OPUS_BITRATE_PER_CHANNEL), diff --git a/js/watch/bench/warmup.ts b/js/watch/bench/warmup.ts new file mode 100644 index 0000000000..43a8c56718 --- /dev/null +++ b/js/watch/bench/warmup.ts @@ -0,0 +1,37 @@ +import { Time } from "@moq/net"; +import { Terminal } from "../src/audio/terminal"; + +// Decoder count × callback size checks that trimming costs a span calculation, +// independent of the PCM payload size. Periodic gaps exercise the priming path. +const CALLBACKS = 4096; +console.log("decoders,frames,warmup_us,ns_per_callback"); +let retained = 0; +for (const count of [1, 16, 64]) { + for (const frames of [120, 960, 2880]) { + for (const warmup of [0, 80_000]) { + const timings = []; + for (let repeat = 0; repeat < 3; repeat++) { + const decoders = Array.from({ length: count }, () => { + const decoder = new Terminal(); + decoder.clear(0, Time.Micro(warmup)); + return decoder; + }); + const start = performance.now(); + for (let callback = 0; callback < CALLBACKS; callback++) { + const timestamp = Time.Micro((callback * frames * 1_000_000) / 48_000); + const read = { kind: "frame" as const, group: callback, frame: { timestamp } }; + const sample = { timestamp, sampleRate: 48_000, numberOfFrames: frames }; + for (const decoder of decoders) { + if (callback % 64 === 0) decoder.update({ kind: "gap" }); + decoder.update(read); + retained += decoder.span(sample).frames; + } + } + timings.push(((performance.now() - start) * 1e6) / (CALLBACKS * count)); + } + timings.sort((a, b) => a - b); + console.log(`${count},${frames},${warmup},${timings[1]?.toFixed(1)}`); + } + } +} +if (retained === 0) throw new Error("warmup discarded the whole stream"); diff --git a/js/watch/src/audio/config.ts b/js/watch/src/audio/config.ts index 7b53512b69..abaad956a3 100644 --- a/js/watch/src/audio/config.ts +++ b/js/watch/src/audio/config.ts @@ -13,7 +13,7 @@ const MP3_MPEG1_MIN_SAMPLE_RATE = 32000; /** The catalog fields that determine the demuxer and WebCodecs decoder instance. */ export type DecoderConfig = Pick< Catalog.AudioConfig, - "codec" | "container" | "description" | "sampleRate" | "numberOfChannels" + "codec" | "container" | "description" | "sampleRate" | "numberOfChannels" | "warmup" >; /** The routing and decoder state whose changes require a replacement subscription. */ @@ -31,6 +31,7 @@ export function decoderConfig(config: Catalog.AudioConfig): DecoderConfig { description: config.description, sampleRate: config.sampleRate, numberOfChannels: config.numberOfChannels, + warmup: config.warmup, }; } diff --git a/js/watch/src/audio/decoder.test.ts b/js/watch/src/audio/decoder.test.ts index c4b42779e7..2cc1260b00 100644 --- a/js/watch/src/audio/decoder.test.ts +++ b/js/watch/src/audio/decoder.test.ts @@ -439,13 +439,12 @@ describe("Decoder across a rendition's absence", () => { const before = playback.inserted().length; const timestamps = await playback.play(6); - // One graph throughout, and the return decodes into the same ring from its first frame past - // the legacy warm-up. + // One graph throughout; a rendition without warmup keeps its entire return. expect(contexts).toHaveLength(1); expect(context.calls.close).toBe(0); expect(playback.subscriptions()).toBe(2); expect(playback.rings()).toEqual([ring]); - expect(playback.inserted().slice(before)).toEqual(timestamps.slice(3)); + expect(playback.inserted().slice(before)).toEqual(timestamps); } finally { playback.close(); } @@ -563,8 +562,8 @@ describe("Decoder across a rendition's absence", () => { load(); await microtasks(); - // Everything past the legacy warm-up reaches the ring, none of it dropped for want of one. - expect(playback.inserted()).toEqual(timestamps.slice(3)); + // Every frame reaches the ring once the worklet loads. + expect(playback.inserted()).toEqual(timestamps); } finally { playback.close(); } diff --git a/js/watch/src/audio/decoder.ts b/js/watch/src/audio/decoder.ts index fee44b1e31..e69a2eb376 100644 --- a/js/watch/src/audio/decoder.ts +++ b/js/watch/src/audio/decoder.ts @@ -26,10 +26,7 @@ import { AUTO_MAX_DELAY, ringSamples, target } from "./latency"; // A blob: URL, or a hosted file when assets() is set; see vite-plugin-worklet. import RenderWorklet from "./render-worklet.ts?worklet"; import type { Source } from "./source"; -import { type DecodedSpan, Terminal } from "./terminal"; -import { Warmup } from "./warmup"; - -const LEGACY_WARMUP_CALLBACKS = 3; +import { Terminal } from "./terminal"; export type DecoderInput = { // Whether to download the audio track. Defaults to true. @@ -128,6 +125,7 @@ export class Decoder { // The subscription's max delay: the shared budget, raised to the estimator's ceiling in "auto". #subscribeMaxDelay = new Signal(Time.Milli.zero); + #maxDelay = new Signal(Time.Milli.zero); // The codec's frame duration: the catalog constant, refined by each frame's own duration. #frame = new Signal(undefined); @@ -189,7 +187,9 @@ export class Decoder { #runSubscribeMaxDelay(effect: Effect): void { const maxDelay = effect.get(this.sync.out.maxDelay); const auto = effect.get(this.sync.in.delay) === "auto"; - this.#subscribeMaxDelay.set(auto ? Time.Milli.max(maxDelay, AUTO_MAX_DELAY) : maxDelay); + const warmup = Time.Milli(effect.get(this.source.out.config)?.warmup ?? 0); + this.#maxDelay.set(Time.Milli.add(maxDelay, warmup)); + this.#subscribeMaxDelay.set(Time.Milli.add(auto ? Time.Milli.max(maxDelay, AUTO_MAX_DELAY) : maxDelay, warmup)); } #runShape(effect: Effect): void { @@ -412,14 +412,14 @@ export class Decoder { ): void { const preSkip = config.codec === "opus" && config.description ? Util.Opus.preSkip(Util.Hex.toBytes(config.description)) : 0; - this.#terminal.clear(preSkip); + this.#terminal.clear(preSkip, Time.Micro.fromMilli(Time.Milli(config.warmup ?? 0))); const format = config.container.kind === "loc" ? new Container.Loc.Format("audio") : new Container.Legacy.Format(config); // Create consumer with slightly less latency than the render worklet to avoid underflowing. // TODO include JITTER_UNDERHEAD const consumer = new Container.Consumer(sub, { format, - maxDelay: this.sync.out.maxDelay, + maxDelay: this.#maxDelay, }); effect.cleanup(() => consumer.close()); @@ -439,18 +439,8 @@ export class Decoder { const loaded = await Util.Libav.polyfill(); if (!loaded) return; // cancelled - const warmup = new Warmup(LEGACY_WARMUP_CALLBACKS); - const decoder = new AudioDecoder({ - output: (data) => { - const decoded = this.#terminal.span(data); - if (warmup.drop()) { - // Drop initial callbacks to prime the decoder. - data.close(); - return; - } - this.#emit(data, decoded); - }, + output: (data) => this.#emit(data), error: (error) => failure.set(error), }); effect.cleanup(() => { @@ -529,7 +519,7 @@ export class Decoder { const init = Container.Cmaf.decodeInitSegment(initSegment); const opusDescription = config.description ? Util.Hex.toBytes(config.description) : init.description; const preSkip = config.codec === "opus" && opusDescription ? Util.Opus.preSkip(opusDescription) : 0; - this.#terminal.clear(preSkip); + this.#terminal.clear(preSkip, Time.Micro.fromMilli(Time.Milli(config.warmup ?? 0))); // Opus in CMAF uses raw packets (not OGG-wrapped), so description must be omitted. // The dOps box from the init segment is not a valid OGG Identification Header. const description = @@ -541,7 +531,7 @@ export class Decoder { const consumer = new Container.Consumer(sub, { format: new Container.Cmaf.Format(init), - maxDelay: this.sync.out.maxDelay, + maxDelay: this.#maxDelay, }); effect.cleanup(() => consumer.close()); @@ -634,8 +624,8 @@ export class Decoder { } } - #emit(sample: AudioData, decoded: DecodedSpan = this.#terminal.span(sample)) { - const { timestamp, frameOffset, frames } = decoded; + #emit(sample: AudioData) { + const { timestamp, frameOffset, frames } = this.#terminal.span(sample); const timestampMilli = Time.Milli.fromMicro(timestamp); if (frames === 0) { sample.close(); diff --git a/js/watch/src/audio/terminal.test.ts b/js/watch/src/audio/terminal.test.ts index f766ef247e..d626a9b4d4 100644 --- a/js/watch/src/audio/terminal.test.ts +++ b/js/watch/src/audio/terminal.test.ts @@ -10,6 +10,27 @@ const frame = (group: number, timestamp: number) => ({ const end = (group: number, timestamp: number) => ({ kind: "end" as const, group, end: timestamp as Time.Micro }); const gap = { kind: "gap" as const }; +test("warmup trims by timestamp across callback sizes, not ordinary groups or declared returns", () => { + const terminal = new Terminal(); + terminal.clear(0, 80_000 as Time.Micro); + terminal.update(frame(10, 1_000_000)); + expect(terminal.span({ timestamp: 1_000_000, sampleRate: 48_000, numberOfFrames: 2_880 }).frames).toBe(0); + terminal.update(frame(11, 1_060_000)); + expect(terminal.span({ timestamp: 1_060_000, sampleRate: 48_000, numberOfFrames: 2_880 })).toEqual({ + timestamp: 1_080_000 as Time.Micro, + frameOffset: 960, + frames: 1_920, + }); + terminal.update(frame(12, 1_120_000)); + expect(terminal.span({ timestamp: 1_120_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(960); + terminal.update(end(13, 1_140_000)); + terminal.update(frame(14, 2_000_000)); + expect(terminal.span({ timestamp: 2_000_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(960); + terminal.update(gap); + terminal.update(frame(16, 3_000_000)); + expect(terminal.span({ timestamp: 3_000_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(0); +}); + test("Terminal trims a partial packet to the source endpoint", () => { const terminal = new Terminal(); terminal.update(end(0, 17_916)); diff --git a/js/watch/src/audio/terminal.ts b/js/watch/src/audio/terminal.ts index 59e501ae17..2b2964bfb3 100644 --- a/js/watch/src/audio/terminal.ts +++ b/js/watch/src/audio/terminal.ts @@ -32,24 +32,33 @@ export class Terminal { #epoch?: Time.Micro; #preSkip = 0; #preSkipRemaining?: number; + #warmup = 0 as Time.Micro; + #prime = true; + #before?: Time.Micro; /** The exclusive source endpoint most recently received. */ get end(): Time.Micro | undefined { return this.#end; } - /** Reset state for a new container consumer with Opus pre-skip in 48 kHz frames. */ - clear(preSkip = 0): void { + /** Reset with Opus pre-skip in 48 kHz frames and warmup in microseconds. */ + clear(preSkip = 0, warmup = 0 as Time.Micro): void { this.#started = false; this.#end = undefined; this.#endGroup = undefined; this.#preSkip = preSkip; + this.#warmup = warmup; + this.#prime = true; + this.#before = undefined; this.#resetEpoch(); } /** Apply one ordered consumer read and report whether a codec epoch ended before it. */ update(read: Update): boolean { - if (read.kind === "gap") return this.#restart(); + if (read.kind === "gap") { + this.#prime = true; + return this.#restart(); + } if (read.kind === "end") { this.#end = read.end; this.#endGroup = read.group; @@ -60,6 +69,10 @@ export class Terminal { const reset = this.#end !== undefined && read.group !== this.#endGroup && this.#restart(); this.#started = true; this.#epoch ??= read.frame.timestamp; + if (this.#prime) { + this.#before = (read.frame.timestamp + this.#warmup) as Time.Micro; + this.#prime = false; + } return reset; } @@ -69,6 +82,7 @@ export class Terminal { this.#started = false; this.#end = undefined; this.#endGroup = undefined; + this.#before = undefined; this.#resetEpoch(); return started; } @@ -80,12 +94,18 @@ export class Terminal { const delayFrames = Math.floor((this.#preSkip * sample.sampleRate) / 48_000); const remaining = this.#preSkipRemaining ?? delayFrames; - const frameOffset = Math.min(remaining, sample.numberOfFrames); + let frameOffset = Math.min(remaining, sample.numberOfFrames); this.#preSkipRemaining = remaining - frameOffset; const delay = Math.round((delayFrames * 1_000_000) / sample.sampleRate); - const timestamp = Math.max(epoch, sample.timestamp - delay) as Time.Micro; + let timestamp = Math.max(epoch, sample.timestamp - delay) as Time.Micro; let frames = sample.numberOfFrames - frameOffset; + if (this.#before !== undefined && timestamp < this.#before) { + const trim = Math.min(frames, Math.ceil(((this.#before - timestamp) * sample.sampleRate) / 1_000_000)); + frameOffset += trim; + frames -= trim; + timestamp = (timestamp + Math.round((trim * 1_000_000) / sample.sampleRate)) as Time.Micro; + } if (this.#end !== undefined) { if (this.#end <= timestamp) { frames = 0; diff --git a/js/watch/src/audio/warmup.test.ts b/js/watch/src/audio/warmup.test.ts deleted file mode 100644 index acf5bc559e..0000000000 --- a/js/watch/src/audio/warmup.test.ts +++ /dev/null @@ -1,14 +0,0 @@ -import { expect, test } from "bun:test"; -import { Warmup } from "./warmup"; - -test("decoder warm-up drops only the configured initial callbacks", () => { - const warmup = new Warmup(3); - - expect([warmup.drop(), warmup.drop(), warmup.drop(), warmup.drop(), warmup.drop()]).toEqual([ - true, - true, - true, - false, - false, - ]); -}); diff --git a/js/watch/src/audio/warmup.ts b/js/watch/src/audio/warmup.ts deleted file mode 100644 index a7d7485073..0000000000 --- a/js/watch/src/audio/warmup.ts +++ /dev/null @@ -1,16 +0,0 @@ -/** Tracks decoder callbacks discarded during initial playback setup. */ -export class Warmup { - #remaining: number; - - /** Create a warm-up window containing `callbacks` decoder callbacks. */ - constructor(callbacks: number) { - this.#remaining = callbacks; - } - - /** Consume one callback and report whether it belongs to the warm-up window. */ - drop(): boolean { - if (this.#remaining === 0) return false; - this.#remaining--; - return true; - } -} diff --git a/js/watch/src/msf.test.ts b/js/watch/src/msf.test.ts index 1a7d876a36..94b3e925b0 100644 --- a/js/watch/src/msf.test.ts +++ b/js/watch/src/msf.test.ts @@ -30,6 +30,26 @@ test("copies delay onto the hang rendition", () => { expect(toHang(catalog).audio?.renditions.audio?.delay).toBe(u53(80)); }); +test.each(["loc", "legacy"])("derives 80 ms Opus decoder warmup from %s MSF audio", (packaging) => { + for (const samplerate of [8000, 48000]) { + const catalog: Msf.Catalog = { + tracks: [{ name: "audio", packaging, role: "audio", codec: "opus", samplerate, delay: 3, jitter: 4 }], + }; + const audio = toHang(catalog).audio?.renditions.audio; + expect(audio?.warmup).toBe(u53(80)); + expect(audio?.sampleRate).toBe(u53(samplerate)); + expect(audio?.delay).toBe(u53(3)); + expect(audio?.jitter).toBe(u53(4)); + } +}); + +test.each(["mp4a.40.2", "flac", "future-audio"])("keeps MSF decoder warmup absent for %s", (codec) => { + const audio = toHang({ tracks: [{ name: "audio", packaging: "loc", role: "audio", codec }] }).audio?.renditions + .audio; + expect(audio?.codec).toBe(codec); + expect(audio?.warmup).toBeUndefined(); +}); + test("rounds a fractional MSF delay up, and drops zero", () => { const catalog: Msf.Catalog = { tracks: [ diff --git a/js/watch/src/msf.ts b/js/watch/src/msf.ts index 80080af6a6..cd5e233611 100644 --- a/js/watch/src/msf.ts +++ b/js/watch/src/msf.ts @@ -1,5 +1,6 @@ import type * as Catalog from "@moq/hang/catalog"; import { u53 } from "@moq/hang/catalog"; +import * as Util from "@moq/hang/util"; import type * as Msf from "@moq/msf"; import { Path } from "@moq/net"; import { base64ToBytes } from "./base64"; @@ -92,6 +93,8 @@ function toAudioConfig(track: Msf.Track): Catalog.AudioConfig | undefined { sampleRate: u53(track.samplerate ?? DEFAULT_SAMPLE_RATE), numberOfChannels: u53(channels), bitrate: track.bitrate != null ? u53(track.bitrate) : undefined, + // MSF cannot advertise decoder warmup, so derive Opus's seek pre-roll. + warmup: track.codec === "opus" ? u53(Util.Opus.WARMUP) : undefined, jitter: track.jitter != null ? u53(track.jitter) : undefined, delay: track.delay != null ? delayMillis(track.delay) : undefined, }; diff --git a/quest/m1/README.md b/quest/m1/README.md index 9a70d02d35..3b10a98df6 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -146,7 +146,6 @@ above them. - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the supported T-STD buffer models, starting with a fixed `--delay` - [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - a viewer joining at a recovery point drops the leading pictures it cannot decode; continuous viewers keep them -- [Audio warmup](/quest/m1/audio-warmup.md) - a viewer joining an Opus rendition mid-stream never hears the unconverged first 80 ms - [CLI stdin shutdown on release](/quest/m1/cli-stdin-release.md) - backport only the runtime-teardown fix so a stopped CLI exits while its stdin writer stays open - [Draft 14-16 updates](/quest/m1/ietf-legacy-updates.md) - Rust and JS apply and answer moq-transport 14-16 request updates without ending or leaking the request - [Tail arrivals](/quest/m1/tail-arrivals.md) - a track judges its pending tail from recorded arrivals, not a cache scan, and errors a reader whose tail ends short diff --git a/quest/m1/audio-warmup.md b/quest/m1/audio-warmup.md deleted file mode 100644 index c68e2f3397..0000000000 --- a/quest/m1/audio-warmup.md +++ /dev/null @@ -1,38 +0,0 @@ -# [S] Audio warmup: Opus converges before a joined viewer hears it - -## Goal - -A viewer joining an Opus rendition mid-stream, or skipping within it, never -hears the decoder's first unconverged output. Publishers set the rendition's -`warmup` to the Opus pre-roll (80 ms, RFC 7845 section 4.6), import sets it -for Opus tracks, and both audio consumers join that much earlier and discard -decoded samples stamped before the join group's start plus `warmup`. AAC-LC -frames decode independently and set nothing; HE-AAC is out of scope. - -## Plan - -- `rs/moq-audio/src/encode/producer.rs` and `rs/moq-mux/src/codec/opus` - publish `warmup` for Opus; `js/publish` does the same for its Opus track. -- `rs/moq-audio/src/decode` already trims Opus `pre_skip` at stream start; - the warmup trim is the same mechanism keyed on the first frame read and the - container read's gap item (`container::Read::Gap` in Rust, `{ kind: "gap" }` - in JS; a join yields no gap), never on a - declared end, which would cut 80 ms at every pause, and the - subscription's maximum age grows by `warmup` as the video consumer quest - does. `js/watch` audio mirrors it. -- The trim replaces `LEGACY_WARMUP_CALLBACKS` in `js/watch/src/audio/decoder.ts`, - which drops the first three decoded frames of every legacy or LOC - subscription, CMAF excepted. That is the start loss left on a rendition's - return after an absence: 60 ms of 20 ms Opus. -- Name the 80 ms pre-roll once in Rust and once in JS. Every site that - advertises or derives it uses that constant: the encoder catalog, the - shared Opus conversion, and the MSF consumer. Add a test that the audio - catalog `Producer` publishes exactly `Encoder::catalog()`, because the two - have drifted before. -- Tests in both languages: a mid-stream join discards exactly the warmup span - and a continuous listener loses nothing. - -## Related - -- [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - trims video at tune-in from the same signal -- [Intra-refresh GOPs](/quest/m2/intra-refresh/README.md) - the video side of the same field diff --git a/quest/m1/open-gop-leading-pictures.md b/quest/m1/open-gop-leading-pictures.md index cdaf674ec1..8b5b347e8b 100644 --- a/quest/m1/open-gop-leading-pictures.md +++ b/quest/m1/open-gop-leading-pictures.md @@ -79,4 +79,3 @@ so the first frame read is its own trigger. ## Related - [Consumer warmup](/quest/m2/intra-refresh/consumer-warmup.md) - the `recovery_frame_cnt > 0` case this rule does not cover -- [Audio warmup](/quest/m1/audio-warmup.md) - keys its Opus pre-roll trim on the same signal diff --git a/quest/m2/intra-refresh/README.md b/quest/m2/intra-refresh/README.md index 540d1128d4..8e36de2316 100644 --- a/quest/m2/intra-refresh/README.md +++ b/quest/m2/intra-refresh/README.md @@ -59,5 +59,4 @@ make these presentation guarantees true even when sweep lengths change. ## Related -- [Audio warmup](/quest/m1/audio-warmup.md) - Opus convergence after a mid-stream join uses the same `warmup` field - [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - frames stamped before the group's keyframe are the other tune-in trim diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index da73a8d355..d2b30e0eb1 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -19,7 +19,7 @@ pub enum Start { /// Decode every cached group. #[default] Oldest, - /// Start at the newest cached group. + /// Start at the newest cached group, including earlier groups for decoder warmup. Latest, } @@ -78,6 +78,12 @@ pub struct Consumer { remix: Option, options: Options, max_delay: std::time::Duration, + /// Prime after a join or loss, never after a declared endpoint. + warmup: Duration, + warmup_pending: bool, + warmup_until: Option, + /// Initial subscription reaches only the warmup window before restoring the skip budget. + joining_latest: bool, resolved_sample_rate: u32, resolved_layout: Layout, /// Where the next packet's timestamp should land: the last packet's timestamp @@ -145,29 +151,26 @@ impl Consumer { }; let name = name.into(); + let warmup = catalog.warmup.unwrap_or_default(); + let joining_latest = options.start == Start::Latest && !warmup.is_zero(); + let budget = if joining_latest { + warmup + } else { + options.max_delay.saturating_add(warmup) + }; let track = broadcast.track(hang::catalog::Reference::track(catalog, &name))?; let mut subscriber = track .subscribe( moq_net::track::Subscription::default() .with_priority(hang::catalog::PRIORITY.audio) - .with_max_delay(options.max_delay), + .with_max_delay(budget), ) .await?; - // A decoder often opens on a track that is already cached: a replacement - // decoder subscribes while its predecessor still holds groups, and a - // rendition switched away from and back to stays warm on the origin for - // `TRACK_IDLE_LINGER` (cached groups, not an upstream subscription). A - // caller that asked for `Start::Latest` wants none of that backlog, - // because a cursor starting at sequence zero replays every cached group - // at decode speed before reaching live media, which on a thirty-second - // retention is half a minute of sound raced through. - // - // This moves the local read cursor and deliberately not - // `Subscription::group_start`. That field is a request to the publisher, - // aggregated across every live subscriber, so naming a stale cached - // sequence there asks the publisher to rewind the track for everyone - // reading it. What a player wants is to skip what it already has. + // Latest skips cached history unless decoder warmup needs it. Move the + // local cursor, not Subscription::group_start: the latter is aggregated + // across subscribers and can ask the publisher to rewind for everyone. if options.start == Start::Latest + && warmup.is_zero() && let Some(live_edge) = track.latest() { subscriber.set_groups(live_edge..); @@ -198,6 +201,10 @@ impl Consumer { remix, options, max_delay, + warmup, + warmup_pending: !warmup.is_zero(), + warmup_until: None, + joining_latest, resolved_sample_rate: sample_rate, resolved_layout: layout, next_start: None, @@ -224,7 +231,7 @@ impl Consumer { &self.options } - /// The effective delay budget after clamping to the publisher's retention window. + /// The effective skip budget after clamping to retention, excluding decoder warmup. pub fn max_delay(&self) -> std::time::Duration { self.max_delay } @@ -280,6 +287,7 @@ impl Consumer { } Some(Read::Gap { .. }) => { self.restart()?; + self.warmup_pending = !self.warmup.is_zero(); continue; } Some(Read::End { group, end }) => { @@ -296,6 +304,11 @@ impl Consumer { // decoder is about to refuse, so it is observed before anything can fail. let arrival = jitter::millis(now().duration_since(self.origin)); let timestamp = mux_frame.timestamp; + if self.joining_latest { + self.track + .set_max_delay(self.options.max_delay.saturating_add(self.warmup)); + self.joining_latest = false; + } let media = timestamp.value() as f64 * 1000.0 / timestamp.scale().as_u64() as f64; self.target.observe(arrival, media); @@ -319,6 +332,12 @@ impl Consumer { let epoch = *self.epoch.get_or_insert(mux_frame.timestamp); let delay = self.decoder.delay_remaining(); let decoded = self.decoder.decode(&mux_frame.payload)?; + // Anchored only once a packet decodes: a refused one primed nothing. + if self.warmup_pending { + let warmup = moq_net::Timestamp::try_from(self.warmup)?.convert(timestamp.scale())?; + self.warmup_until = Some(timestamp.checked_add(warmup)?); + self.warmup_pending = false; + } // Codec delay trimmed off the front is media this packet covered even // though no samples came out, so it still moves the packet after it along. let trimmed = delay - self.decoder.delay_remaining(); @@ -407,8 +426,9 @@ impl Consumer { }; // Queued rather than returned, so a tail drained at a gap earlier in this // same iteration still comes out first. The next turn of the loop pops it. - let frame = self.frame(pcm, timestamp, activity)?; - self.ready.push_back(frame); + if let Some(frame) = self.frame(pcm, timestamp, activity)? { + self.ready.push_back(frame); + } } } @@ -429,6 +449,7 @@ impl Consumer { self.terminal_start = None; self.epoch = None; self.delay_trimmed = 0; + self.warmup_until = None; self.decoder.reapply_delay(); Ok(()) } @@ -451,6 +472,8 @@ impl Consumer { self.trailing = Activity::Active; self.epoch = None; self.delay_trimmed = 0; + self.warmup_pending = !self.warmup.is_zero(); + self.warmup_until = None; Ok(frame) } @@ -498,7 +521,7 @@ impl Consumer { let timestamp = rewind(held, skipped, self.resolved_sample_rate)?; let activity = self.activity_at(timestamp); - Ok(Some(self.frame(pcm, timestamp, activity)?)) + self.frame(pcm, timestamp, activity) } /// The codec activity covering `timestamp`, dropping the spans it has passed. @@ -512,7 +535,27 @@ impl Consumer { } /// Remix and pack decoded PCM into an output frame. - fn frame(&self, pcm: Vec, timestamp: moq_net::Timestamp, activity: Activity) -> Result { + fn frame( + &self, + mut pcm: Vec, + mut timestamp: moq_net::Timestamp, + activity: Activity, + ) -> Result, Error> { + // Warmup must prime the resampler too; trim only what reaches presentation. + if let Some(until) = self.warmup_until + && timestamp < until + { + let channels = self.decoder.layout().channels() as usize; + let frames = pcm.len() / channels; + let duration = Duration::from(until.checked_sub(timestamp)?); + let trim = (duration.as_nanos() * u128::from(self.resolved_sample_rate)).div_ceil(1_000_000_000); + let trim = usize::try_from(trim).unwrap_or(usize::MAX).min(frames); + pcm.drain(..trim * channels); + timestamp = advance(timestamp, trim, self.resolved_sample_rate)?; + } + if pcm.is_empty() { + return Ok(None); + } let pcm = match &self.remix { Some(remix) => remix.process(&pcm), None => pcm, @@ -523,11 +566,11 @@ impl Consumer { .output .format .from_interleaved_f32(&pcm, self.resolved_layout.channels())?; - Ok(Frame { + Ok(Some(Frame { timestamp, data: Bytes::from(bytes), activity, - }) + })) } } @@ -611,6 +654,174 @@ mod tests { use crate::encode::{Encoder, Input, Options as EncodeOptions, Producer, Settings}; use crate::{Format, Layout}; + async fn warmup_output(start: Start, declared: bool) -> (Vec, usize) { + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let mut catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, 48_000, 1); + catalog.warmup = Some(Duration::from_millis(80)); + let mut producer = moq_mux::container::Producer::new( + track, + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio), + ); + for epoch in 0..2 { + if epoch != 0 && declared { + producer.discontinuity().unwrap(); + } + for packet in 0..6 { + producer + .write(moq_mux::container::Frame { + timestamp: Timestamp::from_micros(epoch * 1_000_000 + packet * 20_000).unwrap(), + duration: None, + payload: vec![0; 960 * size_of::()].into(), + keyframe: true, + }) + .unwrap(); + producer.cut(None).unwrap(); + } + } + producer.finish().unwrap(); + let mut consumer = Consumer::new( + &broadcast.consume(), + &catalog, + "audio", + Options { + start, + max_delay: Duration::from_secs(10), + ..Options::new() + }, + ) + .await + .unwrap(); + let mut stamps = Vec::new(); + let mut frames = 0; + while let Some(frame) = consumer.read().await.unwrap() { + stamps.push(frame.timestamp.as_micros()); + frames += frame.data.len() / size_of::(); + } + (stamps, frames) + } + + #[tokio::test(start_paused = true)] + async fn warmup_trims_the_join_without_trimming_groups_or_declared_returns() { + let (stamps, frames) = warmup_output(Start::Oldest, true).await; + assert_eq!( + stamps, + [ + 80_000, 100_000, 1_000_000, 1_020_000, 1_040_000, 1_060_000, 1_080_000, 1_100_000 + ] + ); + assert_eq!(frames, 8 * 960); + } + + #[tokio::test(start_paused = true)] + async fn warmup_restarts_after_an_undeclared_gap() { + let (stamps, frames) = warmup_output(Start::Oldest, false).await; + assert_eq!(stamps, [80_000, 100_000, 1_080_000, 1_100_000]); + assert_eq!(frames, 4 * 960); + } + + #[tokio::test(start_paused = true)] + async fn warmup_latest_joins_early_without_replaying_the_skip_budget() { + let (stamps, frames) = warmup_output(Start::Latest, true).await; + assert_eq!(stamps, [1_100_000]); + assert_eq!(frames, 960); + } + + #[tokio::test(start_paused = true)] + async fn warmup_primes_the_resampler_before_presenting_samples() { + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let mut catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, 48_000, 1); + catalog.warmup = Some(Duration::from_millis(80)); + let mut producer = moq_mux::container::Producer::new( + track, + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio), + ); + producer + .write(moq_mux::container::Frame { + timestamp: Timestamp::from_micros(1_000_000).unwrap(), + duration: None, + payload: [0, 0, 0x80, 0x3e].repeat(7680).into(), + keyframe: true, + }) + .unwrap(); + producer.finish().unwrap(); + let mut consumer = Consumer::new( + &broadcast.consume(), + &catalog, + "audio", + Options { + max_delay: Duration::from_secs(1), + output: Output { + sample_rate: Some(44_100), + ..Output::default() + }, + ..Options::new() + }, + ) + .await + .unwrap(); + let frame = consumer.read().await.unwrap().unwrap(); + assert!(frame.timestamp.as_micros() >= 1_080_000); + let first = f32::from_le_bytes(frame.data[..4].try_into().unwrap()); + assert!( + (first - 0.25).abs() < 0.0001, + "the resampler must be converged too: {first}" + ); + } + + /// A refused packet primed nothing, so the window starts at the first one that decodes. + #[tokio::test(start_paused = true)] + async fn warmup_starts_at_the_first_decoded_packet() { + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let mut catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, 48_000, 1); + catalog.warmup = Some(Duration::from_millis(80)); + let mut producer = moq_mux::container::Producer::new( + track, + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio), + ); + for packet in 0..7 { + let payload = if packet == 0 { + vec![0; 3] + } else { + vec![0; 960 * size_of::()] + }; + producer + .write(moq_mux::container::Frame { + timestamp: Timestamp::from_micros(packet * 20_000).unwrap(), + duration: None, + payload: payload.into(), + keyframe: true, + }) + .unwrap(); + } + producer.finish().unwrap(); + let mut consumer = Consumer::new( + &broadcast.consume(), + &catalog, + "audio", + Options { + max_delay: Duration::from_secs(10), + ..Options::new() + }, + ) + .await + .unwrap(); + assert!(consumer.read().await.is_err()); + let mut stamps = Vec::new(); + while let Some(frame) = consumer.read().await.unwrap() { + stamps.push(frame.timestamp.as_micros()); + } + assert_eq!(stamps, [100_000, 120_000]); + } + /// What the resampler holds at a marker belongs before it: draining it after the next /// packet would splice it onto the resumed audio and stamp that late. #[tokio::test(start_paused = true)] @@ -682,9 +893,11 @@ mod tests { ..EncodeOptions::default() }; let mut producer = Producer::new(&mut broadcast, catalog, input.clone(), &options).unwrap(); - let catalog = Encoder::new(&Settings::new(input.sample_rate, input.layout)) + let mut catalog = Encoder::new(&Settings::new(input.sample_rate, input.layout)) .unwrap() .catalog(); + // Every packet counts here; the warmup tests cover trimming. + catalog.warmup = None; let mut consumer = Consumer::new( &subscriber, &catalog, @@ -1116,7 +1329,9 @@ mod tests { ..Settings::new(48_000, Layout::Mono) }) .unwrap(); - let catalog = encoder.catalog(); + let mut catalog = encoder.catalog(); + // Every packet counts here; the warmup tests cover trimming. + catalog.warmup = None; // One coded packet, then a withheld one to follow it. Taken from the same // encoder rather than published in between, so nothing fills the resampler's @@ -1230,7 +1445,9 @@ mod tests { layout: Layout::Mono, }; let mut encoder = Encoder::new(&Settings::new(input.sample_rate, input.layout)).unwrap(); - let catalog = encoder.catalog(); + let mut catalog = encoder.catalog(); + // Every packet counts here; the warmup tests cover trimming. + catalog.warmup = None; let broadcast = moq_net::broadcast::Info::new().produce(); let track = broadcast @@ -1396,7 +1613,9 @@ mod tests { #[tokio::test] async fn opus_states_its_own_frame_duration() { let mut encoder = Encoder::new(&Settings::new(48_000, Layout::Mono)).unwrap(); - let catalog = encoder.catalog(); + let mut catalog = encoder.catalog(); + // Every packet counts here; the warmup tests cover trimming. + catalog.warmup = None; let broadcast = moq_net::broadcast::Info::new().produce(); let track = broadcast @@ -1490,7 +1709,9 @@ mod tests { layout: Layout::Mono, }; let mut encoder = Encoder::new(&Settings::new(input.sample_rate, input.layout)).unwrap(); - let catalog = encoder.catalog(); + let mut catalog = encoder.catalog(); + // Every packet counts here; the warmup tests cover trimming. + catalog.warmup = None; let broadcast = moq_net::broadcast::Info::new().produce(); let track = broadcast @@ -1541,7 +1762,9 @@ mod tests { layout: Layout::Mono, }; let mut encoder = Encoder::new(&Settings::new(input.sample_rate, input.layout)).unwrap(); - let catalog = encoder.catalog(); + let mut catalog = encoder.catalog(); + // Every packet counts here; the warmup tests cover trimming. + catalog.warmup = None; let frame_size = encoder.frame_size(); let broadcast = moq_net::broadcast::Info::new().produce(); @@ -1658,6 +1881,8 @@ mod tests { // One real Opus packet, so a mis-framed read can't accidentally decode. let mut encoder = Encoder::new(&Settings::new(input.sample_rate, input.layout)).unwrap(); let mut catalog = encoder.catalog(); + // Every packet counts here; the warmup tests cover trimming. + catalog.warmup = None; let pcm = vec![0.0f32; encoder.frame_size() * encoder.codec_channels() as usize]; let packet = encoder.encode(&pcm).unwrap(); diff --git a/rs/moq-audio/src/encode/encoder.rs b/rs/moq-audio/src/encode/encoder.rs index fff254ab36..4788c1eba1 100644 --- a/rs/moq-audio/src/encode/encoder.rs +++ b/rs/moq-audio/src/encode/encoder.rs @@ -655,6 +655,9 @@ impl Encoder { config.bitrate = bitrate; config.description = self.description.clone(); config.container = hang::catalog::Container::Legacy; + if matches!(self.settings.codec, Codec::Opus) { + config.warmup = Some(moq_mux::codec::opus::WARMUP); + } config } } @@ -806,6 +809,7 @@ mod tests { assert_eq!(desc.len(), 19); let head = moq_mux::codec::opus::Config::parse(&mut desc.as_ref()).unwrap(); assert_eq!(head.pre_skip, 312); + assert_eq!(cfg.warmup, Some(Duration::from_millis(80))); } #[test] diff --git a/rs/moq-audio/src/encode/producer.rs b/rs/moq-audio/src/encode/producer.rs index bc68c6a703..85465c01af 100644 --- a/rs/moq-audio/src/encode/producer.rs +++ b/rs/moq-audio/src/encode/producer.rs @@ -633,7 +633,9 @@ mod tests { }, ..Options::default() }; - let decoder_config = Encoder::new(&options.settings).unwrap().catalog(); + let mut decoder_config = Encoder::new(&options.settings).unwrap().catalog(); + // Every packet counts here; the consumer's warmup tests cover trimming. + decoder_config.warmup = None; let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); @@ -811,7 +813,9 @@ mod tests { settings: Settings::new(48_000, Layout::Mono), ..Options::default() }; - let decoder_config = Encoder::new(&options.settings).unwrap().catalog(); + let mut decoder_config = Encoder::new(&options.settings).unwrap().catalog(); + // Every packet counts here; the consumer's warmup tests cover trimming. + decoder_config.warmup = None; let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); @@ -863,7 +867,9 @@ mod tests { }, ..Options::default() }; - let decoder_config = Encoder::new(&options.settings).unwrap().catalog(); + let mut decoder_config = Encoder::new(&options.settings).unwrap().catalog(); + // Every packet counts here; the consumer's warmup tests cover trimming. + decoder_config.warmup = None; let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); @@ -1276,4 +1282,28 @@ mod tests { let second = Producer::new(&mut broadcast, catalog, Input::default(), &Options::default()).unwrap(); assert_eq!(second.track_name(), "1.opus"); } + + /// The rendition a producer publishes is the encoder's own catalog, so neither can drift. + #[tokio::test] + async fn publishes_the_encoder_catalog() { + for codec in [crate::encode::Codec::Opus, crate::encode::Codec::Pcm] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); + let options = Options { + track: Some("audio".into()), + settings: Settings { + codec, + ..Settings::default() + }, + ..Options::default() + }; + let _producer = Producer::new(&mut broadcast, catalog.clone(), Input::default(), &options).unwrap(); + let published = catalog.snapshot().audio.renditions.remove("audio").unwrap(); + assert_eq!( + published, + Encoder::new(&options.settings).unwrap().catalog(), + "{codec:?}" + ); + } + } } diff --git a/rs/moq-mux/src/catalog/msf/consumer.rs b/rs/moq-mux/src/catalog/msf/consumer.rs index 2d85ac2a7b..cc79e9bb79 100644 --- a/rs/moq-mux/src/catalog/msf/consumer.rs +++ b/rs/moq-mux/src/catalog/msf/consumer.rs @@ -333,6 +333,10 @@ fn audio_config_from_msf(track: &moq_msf::Track) -> Result> }; let mut config = AudioConfig::new(codec, sample_rate, channel_count); + // MSF cannot advertise decoder warmup, so derive Opus's seek pre-roll. + if matches!(config.codec, AudioCodec::Opus) { + config.warmup = Some(crate::codec::opus::WARMUP); + } config.bitrate = track.bitrate; config.description = legacy_description(track)?; config.container = container; @@ -667,6 +671,22 @@ mod test { ); } + #[test] + fn opus_msf_warmup_is_derived_without_changing_other_codecs() { + for packaging in [moq_msf::Packaging::Legacy, moq_msf::Packaging::Loc] { + for (codec, warmup) in [("opus", Some(80)), ("mp4a.40.2", None)] { + let mut track = audio_track("audio0", packaging.clone()); + track.codec = Some(codec.into()); + let catalog = from_msf::<()>(&moq_msf::Catalog::new(vec![track]), &moq_net::Path::empty()).unwrap(); + assert_eq!( + catalog.audio.renditions["audio0"].warmup, + warmup.map(std::time::Duration::from_millis), + "{codec} {packaging:?}" + ); + } + } + } + #[test] fn loc_audio_yields_loc_container() { let msf = moq_msf::Catalog::new(vec![audio_track("audio0", moq_msf::Packaging::Loc)]); diff --git a/rs/moq-mux/src/codec/opus/import.rs b/rs/moq-mux/src/codec/opus/import.rs index 1abbbed143..1dbd98f120 100644 --- a/rs/moq-mux/src/codec/opus/import.rs +++ b/rs/moq-mux/src/codec/opus/import.rs @@ -124,6 +124,7 @@ impl From for hang::catalog::AudioConfig { /// informational input rate in [`Config::sample_rate`] claims. fn from(config: Config) -> Self { let mut audio = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, config.channel_count); + audio.warmup = Some(super::WARMUP); audio.description = config.encode().ok(); audio.container = hang::catalog::Container::Legacy; audio @@ -143,6 +144,7 @@ mod tests { let config = super::config(&head).unwrap(); assert_eq!(config.sample_rate, 48_000); assert_eq!(config.description.as_deref(), Some(head.as_ref())); + assert_eq!(config.warmup, Some(Duration::from_millis(80))); let config: hang::catalog::AudioConfig = super::Config::new(44_100, 2).into(); assert_eq!(config.sample_rate, 48_000); diff --git a/rs/moq-mux/src/codec/opus/mod.rs b/rs/moq-mux/src/codec/opus/mod.rs index 7d37a8095a..679dcbf68b 100644 --- a/rs/moq-mux/src/codec/opus/mod.rs +++ b/rs/moq-mux/src/codec/opus/mod.rs @@ -13,6 +13,9 @@ use bytes::{Buf, Bytes}; const OPUS_HEAD: u64 = u64::from_be_bytes(*b"OpusHead"); +/// The decoder pre-roll a joining Opus viewer discards, per RFC 7845 section 4.6. +pub const WARMUP: std::time::Duration = std::time::Duration::from_millis(80); + /// Opus parsing errors. #[derive(Debug, Clone, thiserror::Error)] #[non_exhaustive] diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index a0f07f49e6..6f63d9383e 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -598,10 +598,11 @@ impl Import { .with_pre_skip(dops.pre_skip); head.output_gain = dops.output_gain; - // The catalog describes the decoder's output: Opus always decodes at 48 kHz, whatever - // the sample entry or the informational input rate claims. - let mut config = AudioConfig::new(AudioCodec::Opus, 48_000, head.channel_count); - config.description = Some(head.encode()?); + // The shared conversion sets the decoder's 48 kHz output rate and Opus warmup, + // whatever the sample entry or the informational input rate claims. + let description = head.encode()?; + let mut config: AudioConfig = head.into(); + config.description = Some(description); config.container = container; config } diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 871eb35b32..9e80e98ba6 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -737,6 +737,7 @@ fn opus_catalog_uses_the_decode_rate() { let catalog = run_fmp4(&data); let a = catalog.audio.renditions.values().next().expect("opus rendition"); assert_eq!(a.sample_rate, 48_000); + assert_eq!(a.warmup, Some(std::time::Duration::from_millis(80))); } /// A dOps with a channel mapping table is refused rather than imported without it. diff --git a/rs/moq-mux/src/container/mkv/import.rs b/rs/moq-mux/src/container/mkv/import.rs index bdf1094660..2f1faa5d6c 100644 --- a/rs/moq-mux/src/container/mkv/import.rs +++ b/rs/moq-mux/src/container/mkv/import.rs @@ -742,7 +742,7 @@ fn build_audio_config( } config } - None => AudioConfig::new(AudioCodec::Opus, 48_000, channels), + None => crate::codec::opus::Config::new(48_000, channels).into(), }; config.description = codec_private.cloned(); Ok(config) diff --git a/rs/moq-mux/src/container/mkv/import_test.rs b/rs/moq-mux/src/container/mkv/import_test.rs index cf0cef4204..a5d692df88 100644 --- a/rs/moq-mux/src/container/mkv/import_test.rs +++ b/rs/moq-mux/src/container/mkv/import_test.rs @@ -308,19 +308,29 @@ fn test_vp9_opus_catalog() { #[test] fn test_opus_catalog_rate_is_decoder_rate() { // A 44.1 kHz input rate is informational; the catalog still reports the decoder's 48 kHz. - let data = MkvBuilder::new() - .header("webm") - .segment_start() - .info(1_000_000) - .tracks(vec![track_entry_audio_opus(1, 44100.0, 2)]) - .cluster(0, || vec![simple_block(1, 0, true, b"opus")]) - .segment_end() - .build(); + for header in [true, false] { + let MatroskaSpec::TrackEntry(Master::Full(mut fields)) = track_entry_audio_opus(1, 44100.0, 2) else { + unreachable!() + }; + if !header { + fields.retain(|field| !matches!(field, MatroskaSpec::CodecPrivate(_))); + } + let data = MkvBuilder::new() + .header("webm") + .segment_start() + .info(1_000_000) + .tracks(vec![MatroskaSpec::TrackEntry(Master::Full(fields))]) + .cluster(0, || vec![simple_block(1, 0, true, b"opus")]) + .segment_end() + .build(); - let catalog = run(&data); - let a = catalog.audio.renditions.values().next().unwrap(); - assert_eq!(a.sample_rate, 48000); - assert_eq!(a.channel_count, 2); + let catalog = run(&data); + let a = catalog.audio.renditions.values().next().unwrap(); + assert_eq!(a.sample_rate, 48000); + assert_eq!(a.channel_count, 2); + assert_eq!(a.description.is_some(), header); + assert_eq!(a.warmup, Some(std::time::Duration::from_millis(80))); + } } #[test]