feat(audio): pack several frames per group with a minimum group duration

.github/workflows/nightly.yml

@@ -83,6 +83,10 @@ jobs:
         env:
           NEXTEST_PROFILE: ci
 
+      - name: Audio grouping relay benchmark
+        if: ${{ !cancelled() }}
+        run: nix develop --command just bench-audio
+
       - name: Audio mixer benchmark
         if: ${{ !cancelled() }}
         run: nix develop --command just rs bench-playback

bench/README.md

@@ -15,6 +15,20 @@ Run every Criterion target plus the local relay workloads:
 nix develop --command just bench
 ```
 
+Measure audio grouping at 0, 100 and 200 ms with 50 fps, 200-byte frames:
+
+```bash
+nix develop --command just bench-audio
+```
+
+This sweeps room connections (16, 32) and subscriptions per connection (2, 8)
+independently, plus one publisher serving 64 and 200 subscribers. It uses the
+normal relay sampler and runs nightly. Each shape reports delivered frames,
+p99 latency, relay CPU and RSS; the group-size suffix is the number of frames
+after the keyframe (0, 4, 9), so each group carries 1, 5 or 10 frames. These
+synthetic payloads measure group overhead; they do not measure codec loss
+concealment.
+
 Compare the current tree with another revision:
 
 ```bash
@@ -43,6 +57,8 @@ interval, and output paths so every runtime receives the same load.
 - `fanout`: light one-to-many traffic.
 - `video-heavy`: multicore many-to-many video traffic.
 - `fanout-heavy`: multicore one-to-many traffic near saturation.
+- `audio` and `audio-fanout`: small, 50 fps audio-shaped frames; `bench-audio`
+  sweeps grouping and load shape through command-line overrides.
 
 The runtime matrix rotates execution order between rounds, then reports the
 median throughput, loss, latency, CPU split, context switches, RSS, and thread

bench/relay.sh

@@ -217,6 +217,7 @@ run_workload() {
     local workload=$3
     local runtime=${4:-}
     local workers=${5:-1}
+    local -a overrides=("${@:6}")
     local directory=$RUN/relay/$label/$workload
     local stats=$directory/load.jsonl
     local host=$directory/host.jsonl
@@ -245,7 +246,7 @@ run_workload() {
         --startup 2s \
         --duration 10s \
         --report 500ms \
-        --output "$stats" >"$directory/load.log" 2>&1; then
+        --output "$stats" ${overrides[@]+"${overrides[@]}"} >"$directory/load.log" 2>&1; then
         printf 'load benchmark failed: %s/%s\n' "$label" "$workload" >&2
         cat "$directory/load.log" >&2
         return 1

bench/run.sh

@@ -6,10 +6,13 @@ set -euo pipefail
 # base/current delta. `--runtime` instead compares one multi-threaded Tokio
 # runtime with the same number of independent Tokio/epoll and io_uring workers
 # on the current tree. The current load generator drives every relay so the
-# workloads stay identical.
+# workloads stay identical. `--audio` isolates the audio grouping/load-shape matrix.
 
 MODE=${1:-}
 case $MODE in
+    --audio)
+        BASE=
+        ;;
     --runtime)
         BASE=
         RUNTIME_ROUNDS=${MOQ_BENCH_RUNTIME_ROUNDS:-3}
@@ -373,6 +376,42 @@ print_runtime_comparison() {
     done
 }
 
+run_audio_comparison() {
+    printf 'Building relay and audio load generator...\n'
+    CARGO_TARGET_DIR=$CURRENT_TARGET cargo build --locked --release -p moq-relay -p moq-bench
+    LOAD_BIN=$CURRENT_TARGET/release/moq-bench
+    HOST_BIN=$CURRENT_TARGET/release/moq-bench-host
+
+    printf '\nAudio grouping: 50 fps, 200-byte frames, 0 / 100 / 200 ms groups\n'
+    printf 'shape\trecv-fps\tp99-ms\tCPU\tRSS-MiB\n'
+    local connections subscribers group_size label
+    for group_size in 0 4 9; do
+        for connections in 16 32; do
+            for subscribers in 2 8; do
+                label=room-$connections-$subscribers-$group_size
+                run_workload "$label" "$CURRENT_TARGET/release/moq-relay" audio '' 1 \
+                    --connections "$connections" --subscribe "$subscribers" --group-size "$group_size"
+                print_audio_sample "$label" audio
+            done
+        done
+        for connections in 65 201; do
+            label=fanout-$connections-$group_size
+            run_workload "$label" "$CURRENT_TARGET/release/moq-relay" audio-fanout '' 1 \
+                --connections "$connections" --group-size "$group_size"
+            print_audio_sample "$label" audio-fanout
+        done
+    done
+}
+
+print_audio_sample() {
+    local label=$1
+    local workload=$2
+    jq -r --arg shape "$label" '[
+        $shape, .recv_fps, .latency_p99_ms, (.cpu_cores // "n/a"),
+        (if .rss_bytes then .rss_bytes / 1048576 else "n/a" end)
+    ] | @tsv' "$RUN/relay/$label/$workload/summary.json"
+}
+
 run_runtime_comparison() {
     if [[ $(uname -s) != Linux ]]; then
         printf 'runtime comparison needs Linux for io_uring and /proc metrics\n' >&2
@@ -429,7 +468,10 @@ run_runtime_comparison() {
 
 cd "$ROOT"
 
-if [[ $MODE == --runtime ]]; then
+if [[ $MODE == --audio ]]; then
+    run_audio_comparison
+    exit 0
+elif [[ $MODE == --runtime ]]; then
     run_runtime_comparison
     exit 0
 elif [[ -n $BASE ]]; then

bench/workloads/audio-fanout.toml

@@ -0,0 +1,5 @@
+connections = 65
+fanout = "audio"
+fps = 50
+frame_size = 200
+group_size = 0

bench/workloads/audio.toml

@@ -0,0 +1,6 @@
+connections = 16
+broadcasts = 1
+subscribe = 2
+fps = 50
+frame_size = 200
+group_size = 0

doc/lib/js/publish.md

@@ -66,6 +66,19 @@ Disabling a rendition (`muted` on the element) ends the audio timeline with
 a marker, so a viewer that stays subscribed, or joins during the pause, never
 plays the audio before it as live.
 
+Each audio frame is its own group by default, so a relay pays a stream and a
+group's bookkeeping per 20ms frame. `el.audio.groupDuration` sets a minimum
+per group instead, such as `Time.Milli(100)`: the first frame at least 100ms
+after the group's first timestamp opens the next group and closes the previous
+one. If encoding pauses, the current group stays open until the next frame or
+timeline marker. Frames still forward as they are encoded, so grouping does
+not buffer them, but loss gets coarser: a viewer that falls behind skips a
+whole group, and a lost frame holds back the rest of its group until it is retransmitted. A 60ms
+Opus `frameDuration` also cuts the group rate, without code, at the cost of
+encoder latency. The synthetic `just bench-audio` workload reports higher p99
+delivery latency with longer groups; immediate forwarding is not a promise of
+unchanged end-to-end latency.
+
 `el.video.cut()` asks for a keyframe on top of the `keyframeInterval` cadence,
 for a resume, a recording cut, or a known tune-in moment. Requests coalesce into
 the next keyframe, and forced keyframes land at least 500ms apart.

doc/lib/rs/moq-audio.md

@@ -88,6 +88,19 @@ and no platform encoder is wired in yet, so `Codec::Aac` is refused at
 construction on every host for now. Linux has no OS encoder, so it will stay
 that way there.
 
+Each packet is its own group by default, so a relay pays a stream and a
+group's bookkeeping per 20 ms Opus frame. `encode::Options::group_duration`
+sets a minimum per group instead, for every codec: the packet that reaches it
+closes the group. During a pause or after `reset_epoch()`, a partial group
+stays open until the next write; `discontinuity()` closes it immediately.
+Packets forward as they are encoded rather than waiting for the group to fill,
+but loss gets coarser: a subscriber that falls behind skips a whole
+group, and a lost packet holds back the rest of its group until it is
+retransmitted. A 60 ms Opus `Settings::frame_duration` also cuts the group
+rate, without code, at the cost of encoder latency. The synthetic
+`just bench-audio` workload reports higher p99 delivery latency with longer
+groups; immediate forwarding is not a promise of unchanged end-to-end latency.
+
 Highlights:
 
 - **`encode::Control`** advertises the track and opens the microphone only while someone listens. Stop, swap devices, and restart without changing the track subscribers know; read a level meter for the UI.

js/publish/src/audio/encoder.test.ts

@@ -202,20 +202,26 @@ function configured(): Promise<AudioEncoderConfig> {
 	});
 }
 
-// An Encoder wired to a fake capture feed, recording each written frame as [timestamp, payload bytes].
-async function setup(baseline = new Baseline(), codec?: Codec) {
+// An Encoder wired to a fake capture feed, recording each written frame as [timestamp, payload bytes]
+// and how many frames each group carries.
+async function setup(baseline = new Baseline(), codec?: Codec, groupDuration?: Time.Milli) {
 	const configuring = configured();
 
 	const track = new Moq.Track.Producer("audio").accept();
 	const written: [number, number][] = [];
+	const groups: number[] = [];
+	const appended: ReturnType<typeof track.appendGroup>[] = [];
 	const writes = { onWrite: undefined as (() => void) | undefined };
 	const appendGroup = track.appendGroup.bind(track);
 	track.appendGroup = () => {
 		const group = appendGroup();
+		appended.push(group);
+		const index = groups.push(0) - 1;
 		const writeFrame = group.writeFrame.bind(group);
 		group.writeFrame = (frame) => {
 			const [timestamp, payload] = Moq.Varint.decode(frame.payload);
 			written.push([timestamp, payload.byteLength]);
+			groups[index]++;
 			writeFrame(frame);
 			writes.onWrite?.();
 		};
@@ -245,6 +251,7 @@ async function setup(baseline = new Baseline(), codec?: Codec) {
 		capture: capture as never,
 		enabled,
 		codec,
+		groupDuration,
 	});
 
 	const config = await configuring;
@@ -257,6 +264,8 @@ async function setup(baseline = new Baseline(), codec?: Codec) {
 		rendition,
 		feed,
 		written,
+		groups,
+		appended,
 		writes,
 		[Symbol.dispose]() {
 			encoder.close();
@@ -385,6 +394,48 @@ test("a push completing several frames keeps the encoder running", async () => {
 	]);
 });
 
+// The first frame at the minimum opens the next group. A timeline break closes the group early,
+// so the first frame after it opens a fresh one.
+test("a group duration packs frames until the minimum and restarts after a break", async () => {
+	using _webcodecs = installFakeWebCodecs();
+	using env = await setup(new Baseline(), undefined, Time.Milli(100));
+	const { enabled, feed, written, groups, appended, writes } = env;
+
+	let index = 0;
+	const push = async (count: number) => {
+		for (let i = 0; i < count; i++, index++) {
+			await feed.push({ timestamp: Time.Micro(20_000 + index * 20_000), channels: [new Float32Array(960)] });
+		}
+		await feed.drain();
+	};
+
+	await push(7); // five written, two held
+	expect(groups).toEqual([5]);
+	expect(appended[0].closed.peek()).toBeUndefined();
+	await push(2); // the next frame opens a new group and closes the first
+	expect(groups).toEqual([5, 2]);
+	expect(appended[0].closed.peek()).toBeNull();
+
+	const marked = new Promise<void>((resolve) => {
+		writes.onWrite = resolve;
+	});
+	enabled.set(false);
+	await marked;
+	writes.onWrite = undefined;
+
+	const resumed = configured();
+	enabled.set(true);
+	await resumed;
+	await push(4); // a fresh AudioEncoder, so two written and two held
+
+	expect(written.map(([timestamp]) => timestamp)).toEqual([
+		20_000, 40_000, 60_000, 80_000, 100_000, 120_000, 140_000, 200_000, 200_000, 220_000,
+	]);
+	// 120ms opens the next group, closing the five frames at 20-100ms. The break ends the
+	// 120-140ms group early, so resumed frames start a fresh group after the marker.
+	expect(groups).toEqual([5, 2, 1, 2]);
+});
+
 // Chromium stamps Opus output by counting the samples emitted, so every frame DTX suppresses pulls
 // later audio earlier. A plain-JS caller passing the old knob must not reach the encoder.
 test("never enables Opus DTX", async () => {

js/publish/src/audio/encoder.ts

@@ -103,6 +103,9 @@ export type EncoderProps = Inputs<EncoderInput> & {
 
 	// Codec selection plus encoder settings. Defaults to "opus".
 	codec?: Codec | Signal<Codec>;
+
+	// The minimum audio carried by each group. Defaults to 0, a group per frame.
+	groupDuration?: Time.Milli | Signal<Time.Milli>;
 };
 
 type EncoderOutput = {
@@ -143,6 +146,12 @@ export class Encoder {
 	fade: Signal<Time.Milli>;
 	/** The live-editable codec selection plus its encoder settings. */
 	codec: Signal<Codec>;
+	/**
+	 * The minimum timestamp span before a frame opens the next group, closing the previous one. 0 puts every frame in its own group. A longer group costs the
+	 * relay fewer streams but makes loss coarser: a viewer that falls behind skips a whole group. A
+	 * negative or non-finite duration refuses the rendition.
+	 */
+	groupDuration: Signal<Time.Milli>;
 
 	/**
 	 * The capture supplying this rendition, or undefined while none is wired.
@@ -173,7 +182,8 @@ export class Encoder {
 	#pipeline: Pipeline | undefined;
 
 	// The current subscription's track and the producer writing into it, or undefined without demand.
-	#live: { track: Moq.Track.Producer; producer: Container.Legacy.Producer } | undefined;
+	// `start` is the open group's first timestamp, undefined until a frame opens one.
+	#live: { track: Moq.Track.Producer; producer: Container.Legacy.Producer; start?: Time.Micro } | undefined;
 
 	// Where the next frame submitted to the AudioEncoder starts, i.e. the exclusive end of the
 	// newest one, where an epoch's discontinuity marker goes. Cleared once the marker is written.
@@ -188,6 +198,9 @@ export class Encoder {
 	// invalid one, so a bad value never reaches the gain.
 	#fade: Time.Milli = FADE;
 
+	// The last valid group duration, validated by #runConfig the same way as #fade.
+	#groupDuration = Time.Micro(0);
+
 	// The fatal error an AudioEncoder reported, if any. That instance can never encode again and
 	// reconfiguring it would be a retry, so the rendition stays down for the life of this encoder.
 	#fatal = new Signal<Error | undefined>(undefined);
@@ -224,6 +237,7 @@ export class Encoder {
 		this.volume = Signal.from(props?.volume ?? 1);
 		this.fade = Signal.from(props?.fade ?? FADE);
 		this.codec = Signal.from<Codec>(props?.codec ?? "opus");
+		this.groupDuration = Signal.from(props?.groupDuration ?? Time.Milli(0));
 
 		// Only the capture graph has a node to expose.
 		this.#signals.run((effect) => {
@@ -385,6 +399,11 @@ export class Encoder {
 			throw new Error(`audio fade must be a finite, non-negative number of ms: ${fade}`);
 		this.#fade = fade;
 
+		const groupDuration = effect.get(this.groupDuration);
+		if (!Number.isFinite(groupDuration) || groupDuration < 0)
+			throw new Error(`audio group duration must be a finite, non-negative number of ms: ${groupDuration}`);
+		this.#groupDuration = Time.Micro.fromMilli(groupDuration);
+
 		const capture = effect.get(this.in.capture);
 		const captured = capture ? effect.get(capture.out.format) : undefined;
 		if (!effect.get(this.in.enabled) || !captured) {
@@ -436,6 +455,7 @@ export class Encoder {
 		if (end === undefined || !live || live.track.closed.peek() !== undefined) return;
 		this.#floor = end;
 		live.producer.discontinuity(end);
+		live.start = undefined;
 	}
 
 	// Encode captured audio frames into whichever track producer is live. The broadcast owns the
@@ -483,12 +503,18 @@ export class Encoder {
 							bytes: stats.bytes + frame.byteLength,
 						}));
 
-						// Each audio frame is a keyframe, so its own group, which the relay can forward
-						// without waiting for a group boundary. Loss is handled by the codec's PLC.
 						const live = this.#live;
 						if (!live) return;
-						if (this.#floor !== undefined && frame.timestamp < this.#floor) return;
-						live.producer.encode(frame, frame.timestamp as Time.Micro, true);
+						const timestamp = frame.timestamp as Time.Micro;
+						if (this.#floor !== undefined && timestamp < this.#floor) return;
+
+						// Every audio frame decodes on its own, so any of them can open a group: the
+						// first one at or past the minimum after the open group's start. Frames forward
+						// as they are written rather than waiting for the group to fill. A dropped
+						// group leaves a gap for the codec's PLC.
+						const keyframe = live.start === undefined || timestamp - live.start >= this.#groupDuration;
+						if (keyframe) live.start = timestamp;
+						live.producer.encode(frame, timestamp, keyframe);
 						if (this.#estimator.flush(frame.timestamp, baseline)) {
 							const catalog = this.#out.catalog.peek();
 							if (catalog) this.#out.catalog.set({ ...catalog, ...this.#estimator.estimate });

justfile

@@ -44,10 +44,13 @@ dev:
 bench $BASE="":
     bench/run.sh "$BASE"
 
-# Compare one multi-threaded Tokio runtime with the same number of independent
-# Tokio/epoll and io_uring workers; WORKERS defaults to every logical CPU.
+# Measure audio group sizes across relay publisher/subscriber and fanout shapes.
+bench-audio:
+    bench/run.sh --audio
 
-# Compare a shared Tokio runtime with independent workers: `just bench-runtime 5 16`.
+# Compare one multi-threaded Tokio runtime with independent Tokio/epoll and
+# io_uring workers; WORKERS defaults to every logical CPU.
+# Example: `just bench-runtime 5 16`.
 bench-runtime $ROUNDS="3" $WORKERS="":
     MOQ_BENCH_RUNTIME_ROUNDS="$ROUNDS" MOQ_BENCH_RUNTIME_WORKERS="$WORKERS" bench/run.sh --runtime
 

quest/m1/README.md

@@ -134,7 +134,6 @@ blocks. The quests that gated m0 lines moved under them.
 - [Benchmark comparisons](/quest/m1/performance-comparisons.md) - retained evidence, repeated paired runs, and uncertainty for performance claims
 - [Relay profiling](/quest/m1/performance-profiles.md) - reproducible CPU and allocation captures under the existing workloads
 - [Perf](/quest/m1/perf/README.md) - eliminate measured hot-path costs across moq-uring, kio, and the moq-net model
-- [Audio group duration](/quest/m1/audio-group-duration.md) - audio publishers in JS and Rust take a minimum group duration, default one frame
 - [#2924](/quest/m1/2924-moq-relay-tls-rotation-is-not-atomic-across-thread-per.md) - every listener on both runtimes shares one reloadable served identity, so rotation is atomic and generate works with workers
 - [Benchmark regressions in CI](/quest/m1/bench-ci.md) - PRs get a non-blocking comparison of the Criterion benches they affect, and a nightly trend on main alerts on regressions
 - [#3126](/quest/m1/3126-moq-bench-every-readme-example-fails-to-parse-and.md) - moq-bench reports per-interval latency percentiles so the ramp leaves the steady state