From 4b535b92a3f5b5149453d0fe05c942b41f30415e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 18:27:51 -0700 Subject: [PATCH 01/14] chore: claim track tail interop quest Co-Authored-By: GPT-6 From f0e73a3d29841b727dedaefe31f4fdacf1375b1d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 18:43:49 -0700 Subject: [PATCH 02/14] test(interop): reproduce finite track loss through a relay Co-Authored-By: GPT-6 --- quest/m1/track-tail-interop.md | 41 ++++++--- rs/moq-cli/Cargo.toml | 4 + test/interop/README.md | 16 ++++ test/interop/clients/js-native/package.json | 3 + test/interop/clients/js-native/tail.ts | 95 ++++++++++++++++++++ test/interop/clients/js-native/tsconfig.json | 7 ++ test/interop/clients/rust/tail.rs | 94 +++++++++++++++++++ test/interop/interop.sh | 93 ++++++++++++++++++- test/justfile | 1 + 9 files changed, 340 insertions(+), 14 deletions(-) create mode 100644 test/interop/clients/js-native/tail.ts create mode 100644 test/interop/clients/js-native/tsconfig.json create mode 100644 test/interop/clients/rust/tail.rs diff --git a/quest/m1/track-tail-interop.md b/quest/m1/track-tail-interop.md index 0078cdae179..e043bf60a32 100644 --- a/quest/m1/track-tail-interop.md +++ b/quest/m1/track-tail-interop.md @@ -14,19 +14,34 @@ Both moq-net and `@moq/net` wait for a track's tail once the publisher ends a subscription, and each is covered by its own in-process tests. Nothing checks that the two agree across a relay on real QUIC. -What stood in the way when the Rust half landed: - -- `moq import` exits the moment stdin ends, closing its session with the tail - still in flight, so a finite CLI publisher cannot end a track cleanly. The - fix is a publisher that waits for its subscriptions to drain before it - closes; `moq export`'s linger ([Export linger](/quest/m1/export-linger.md)) is - the reader-side cousin. -- The native JS subscriber (`test/interop/clients/js-native`) returns on the - first frame. It needs a mode that reads a track to its end and reports how it - ended and which groups it saw. - -QUIC on localhost rarely reorders, so this is a smoke check that the end is -delivered and clean. The ordering race itself stays in the unit tests. +The finite interop clients use raw groups and hold the publishing session open +until the harness acknowledges that the reader verified all payloads and a clean +end. `just test interop --tail` runs Rust-to-Rust and both Rust/JS directions under +Node and Bun; the full interop matrix also runs them. This avoids requiring a new +CLI drain API to exercise the transport. + +The test exposes a remaining relay bug: a JS publisher emits four 256 KiB groups +with an end of 4, but the Rust reader can receive only groups 2 and 3 before a clean +end. `rs/moq-net/src/lite/publisher.rs` resolves SUBSCRIBE_START to the first group +that arrives, then raises its read floor to that sequence. QUIC's newer-group +priority can deliver that stream before older groups still in flight, which the +relay then discards. + +Preserving the source's declared start requires carrying it through the origin's +resume/splice model. `track::Consumer::poll_start` currently returns no declared +start for spliced tracks. Resolve how that start combines segment boundaries and +warm-cache handoffs, retain a private model interface if possible, and add a +source-level regression before changing the relay. Do not replace it with a +constant zero floor or weaken the finite interop assertion. A diagnostic mutation +that advertised zero and kept the read floor at zero made all five lanes pass; +that mutation was reverted. + +`moq import` still exits at stdin EOF before its subscriptions drain. An actual CLI +publisher drain API remains a separate follow-up; model demand ending when tracks +close is not proof of transport completion. + +QUIC ordering races remain covered by unit tests, with this test covering the +observable real-relay failure as well. ## Related diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 032ef228310..304e350e6b8 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -134,3 +134,7 @@ tempfile = { workspace = true } # `test-util` enables `tokio::time::pause` so the TS round-trip test simulates the # exporter drain timeouts instantly instead of waiting on the wall clock. tokio = { workspace = true, features = ["test-util"] } + +[[example]] +name = "interop-tail" +path = "../../test/interop/clients/rust/tail.rs" diff --git a/test/interop/README.md b/test/interop/README.md index e41a0d7179d..a227421142c 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -19,6 +19,19 @@ subscriber checks end-to-end: the browser encodes fake microphone audio, and the Python and Go clients encode a synthetic tone through `moq-ffi` at a 2.5 ms frame duration, so the matrix covers the FFI audio path with a non-default codec config. +The normal run also checks a finite raw track through the relay. Its publisher +announces end 4 before writing four 256 KiB groups; the reader verifies every byte, +exactly groups 0 through 3, and a clean end at 4. Rust-to-Rust always runs. Selecting +native JS adds both Rust-to-JS and JS-to-Rust under that runtime, so `--all` covers +Node and Bun. `--tail` runs just these five lanes. + +The finite clients keep their session alive until the harness acknowledges the +reader's complete clean end over stdin. This avoids using `moq import`, which +currently closes its session at stdin EOF before its subscriptions drain. No +sleep stands in for drain completion. A missing group, error, or stall fails the +lane; the timeout only bounds failure. QUIC on localhost rarely reorders, so the +ordering race remains covered by transport unit tests. + `just test media` is a separate, browser-only run that asks a harder question: is the media a viewer gets actually advancing and in sync, and does the player survive the publication lifecycle. See [Media QA](#media-qa). @@ -76,6 +89,9 @@ just test interop --all # Pick your own axes: just test interop --publishers rust,python --subscribers rust,c,js-native-bun +# Finite track tails only: Rust-to-Rust and Rust/Node/Bun in both directions. +just test interop --tail + # Negative control: no publisher, every subscriber must time out. just test interop-negative diff --git a/test/interop/clients/js-native/package.json b/test/interop/clients/js-native/package.json index 39314080110..e6d2d210893 100644 --- a/test/interop/clients/js-native/package.json +++ b/test/interop/clients/js-native/package.json @@ -2,6 +2,9 @@ "name": "@moq/interop-native", "private": true, "type": "module", + "scripts": { + "check": "tsc --noEmit" + }, "dependencies": { "@moq/hang": "workspace:*", "@moq/json": "workspace:*", diff --git a/test/interop/clients/js-native/tail.ts b/test/interop/clients/js-native/tail.ts new file mode 100644 index 00000000000..529b0005463 --- /dev/null +++ b/test/interop/clients/js-native/tail.ts @@ -0,0 +1,95 @@ +/** Finite raw-track interop: publish until the harness acknowledges, or read through clean EOF. @module */ +import { strict as assert } from "node:assert"; +import { createInterface } from "node:readline"; +import * as Moq from "@moq/net"; +import { install } from "@moq/web-transport"; + +install(); +const GROUPS = 4; +const BYTES = 256 * 1024; +const [role, url, path] = process.argv.slice(2); +if (!url || !path || !["publish", "subscribe"].includes(role)) { + throw new Error("usage: tail.ts publish|subscribe URL BROADCAST"); +} + +const origin = new Moq.Origin.Producer(); +const connection = await Moq.Connection.connect({ + url: new URL(url), + websocket: { enabled: false }, + ...(role === "publish" ? { publish: origin.consume() } : { consume: origin }), +}); +try { + if (role === "publish") { + const broadcast = origin.createBroadcast(Moq.Path.from(path)); + const track = broadcast.createTrack("tail"); + broadcast.announce(); + while (!track.used.peek()) await track.used.changed(); + // The end precedes the last group's data; the transport still has to drain it. + track.finishAt(GROUPS); + for (let sequence = 0; sequence < GROUPS; sequence++) { + const group = track.appendGroup(); + group.writeFrame({ + payload: new Uint8Array(BYTES).fill(sequence), + timestamp: Moq.Time.Timestamp.fromMillis(0), + }); + group.close(); + } + track.close(); + const lines = createInterface({ input: process.stdin }); + try { + const ack = await lines[Symbol.asyncIterator]().next(); + assert.equal(ack.value, "clean end", "reader did not acknowledge a clean end"); + } finally { + lines.close(); + } + broadcast.close(); + console.error("tail acknowledged"); + } else { + const request = origin.request(Moq.Path.from(path), { announced: true }); + try { + let broadcast = request.active.peek(); + while (!broadcast) { + await request.active.changed(); + broadcast = request.active.peek(); + } + const track = broadcast + .track("tail") + .subscribe({ maxAge: Moq.Time.Milli(5000), groups: { start: { included: 0 } } }); + const seen: number[] = []; + try { + for (;;) { + const group = await track.recvGroup(); + if (!group) break; + try { + assert(group.sequence < GROUPS, `unexpected group ${group.sequence}`); + const frame = await group.readFrame(); + assert(frame, "empty group"); + assert.equal(frame.payload.byteLength, BYTES, "truncated group"); + assert( + frame.payload.every((byte) => byte === group.sequence), + "corrupt group", + ); + assert.equal(await group.readFrame(), undefined, "extra frame"); + seen.push(group.sequence); + } finally { + group.close(); + } + } + assert.deepEqual( + seen.sort((a, b) => a - b), + [0, 1, 2, 3], + ); + assert.equal(await track.finished(), GROUPS); + console.error("tail clean end=4 groups=0,1,2,3 bytes=1048576"); + } finally { + track.close(); + } + } finally { + request.close(); + } + } +} finally { + connection.close(); + origin.close(); +} +process.exit(0); diff --git a/test/interop/clients/js-native/tsconfig.json b/test/interop/clients/js-native/tsconfig.json new file mode 100644 index 00000000000..908e9404dc8 --- /dev/null +++ b/test/interop/clients/js-native/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "../../../../js/tsconfig.json", + "compilerOptions": { + "types": ["bun"] + }, + "include": ["*.ts"] +} diff --git a/test/interop/clients/rust/tail.rs b/test/interop/clients/rust/tail.rs new file mode 100644 index 00000000000..765ea1e8eb0 --- /dev/null +++ b/test/interop/clients/rust/tail.rs @@ -0,0 +1,94 @@ +//! Finite raw-track interop client. The harness acknowledges a clean read on stdin. +use std::time::Duration; + +use anyhow::{Context, ensure}; +use moq_tokio::moq_net; +use tokio::io::AsyncBufReadExt; + +const GROUPS: u64 = 4; +const BYTES: usize = 256 * 1024; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let args: Vec<_> = std::env::args().collect(); + ensure!( + args.len() == 4, + "usage: interop-tail publish|subscribe URL BROADCAST" + ); + let url: url::Url = args[2].parse()?; + let mut config = moq_tokio::connect::Config::default(); + config.websocket.enabled = Some(false); + let client = config.init(Default::default())?.with_reconnect(false); + let origin = moq_tokio::origin::spawn(); + + match args[1].as_str() { + "publish" => { + let broadcast = origin.create_broadcast(&args[3])?; + let mut track = broadcast.create_track("tail", None)?; + broadcast.announce(Default::default())?; + let _session = client + .with_publisher(&origin) + .connect(url) + .established() + .await?; + track.demand().used().await?; + // Declare the end before the final group's payload is even available. + track.finish_at(GROUPS)?; + for sequence in 0..GROUPS { + let mut group = track.append_group()?; + group.write_frame(moq_net::Timestamp::ZERO, vec![sequence as u8; BYTES])?; + group.finish()?; + } + // A finished track is not a drained transport. Keep the session alive + // until the harness confirms the subscriber checked every byte and EOF. + let mut ack = String::new(); + tokio::io::BufReader::new(tokio::io::stdin()) + .read_line(&mut ack) + .await?; + ensure!( + ack == "clean end\n", + "reader did not acknowledge a clean end" + ); + broadcast.finish(); + eprintln!("tail acknowledged"); + } + "subscribe" => { + let consumer = origin.consume(); + let _session = client + .with_subscriber(origin) + .connect(url) + .established() + .await?; + let broadcast = consumer.routed_broadcast(&args[3]).await?; + let options = moq_net::track::Subscription::default() + .with_max_age(Duration::from_secs(5)) + .with_groups(0..); + let mut track = broadcast.track("tail")?.subscribe(Some(options)).await?; + let mut seen = Vec::new(); + while let Some(mut group) = track.recv_group().await? { + let sequence = group.sequence; + let frame = group.read_frame().await?.context("empty group")?; + ensure!(sequence < GROUPS, "unexpected group {sequence}"); + ensure!(frame.payload.len() == BYTES, "truncated group {sequence}"); + ensure!( + frame.payload.iter().all(|byte| *byte == sequence as u8), + "corrupt group {sequence}" + ); + ensure!( + group.read_frame().await?.is_none(), + "extra frame in group {sequence}" + ); + seen.push(sequence); + } + seen.sort_unstable(); + ensure!( + seen == (0..GROUPS).collect::>(), + "wrong groups: {seen:?}" + ); + ensure!(track.finished().await? == GROUPS, "wrong declared end"); + eprintln!("tail clean end=4 groups=0,1,2,3 bytes=1048576"); + } + _ => anyhow::bail!("unknown role: {}", args[1]), + } + Ok(()) +} diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 713d4e6865b..81b6f058c98 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -36,6 +36,7 @@ PORT="${INTEROP_PORT:-}" URL="" NEGATIVE=0 MEDIA=0 +TAIL_ONLY=0 # Cargo profile for the relay/cli/libmoq builds. Debug compiles faster, which is # what an interop test wants; the workload (320x240@30) is trivial either way. @@ -76,6 +77,10 @@ while [[ $# -gt 0 ]]; do NEGATIVE=1 shift ;; + --tail) + TAIL_ONLY=1 + shift + ;; --media) MEDIA=1 shift @@ -98,6 +103,15 @@ done exit 2 } +if [[ "$TAIL_ONLY" -eq 1 ]]; then + if [[ "$NEGATIVE" -eq 1 || "$MEDIA" -eq 1 ]]; then + echo "error: --tail, --media, and --negative are separate runs" >&2 + exit 2 + fi + PUBLISHERS="rust,js-native-node,js-native-bun" + SUBSCRIBERS="$PUBLISHERS" +fi + # The media checks drive both roles from the browser client and never touch the matrix, so they # pick their own axes rather than accepting --publishers / --subscribers. if [[ "$MEDIA" -eq 1 ]]; then @@ -183,6 +197,9 @@ build_relay_cli() { echo "error: failed to build moq-relay / moq-cli" >&2 exit 1 } + if [[ "$NEGATIVE" -eq 0 && "$MEDIA" -eq 0 ]]; then + (cd "$WORKSPACE" && cargo build --locked ${flag[@]+"${flag[@]}"} -p moq-cli --example interop-tail) + fi [[ -n "$RELAY" ]] || RELAY="$TARGET_BASE/$PROFILE/moq-relay" # The `moq-cli` crate ships its binary as `moq` (a `[[bin]]` override). [[ -n "$MOQ" ]] || MOQ="$TARGET_BASE/$PROFILE/moq" @@ -235,6 +252,10 @@ prepare_js() { sed 's/^/ /' "$HARNESS_RUN/js-vite.log" >&2 || true fi fi + if (needs js-native-node || needs js-native-bun) && ! (cd "$CLIENTS/js-native" && bun run check) >"$HARNESS_RUN/js-native-check.log" 2>&1; then + for v in js-native-node js-native-bun; do needs "$v" && mark_broken "$v" "type check failed"; done + sed 's/^/ /' "$HARNESS_RUN/js-native-check.log" >&2 || true + fi if needs js-native-node && ! have node; then mark_broken js-native-node "node not found" fi @@ -604,6 +625,65 @@ run_round() { return 0 } +# Raw-track clients have no codecs: every byte, group, and the declared end is checked. +# shellcheck disable=SC2329 # invoked indirectly via harness_spawn +run_tail_client() { + local lang="$1" role="$2" broadcast="$3" + case "$lang" in + rust) timeout -k 3 "$TIMEOUT" "$TARGET_BASE/$PROFILE/examples/interop-tail" "$role" "$URL" "$broadcast" ;; + js-native-node) (cd "$CLIENTS/js-native" && timeout -k 3 "$TIMEOUT" node --import tsx tail.ts "$role" "$URL" "$broadcast") ;; + js-native-bun) (cd "$CLIENTS/js-native" && timeout -k 3 "$TIMEOUT" bun tail.ts "$role" "$URL" "$broadcast") ;; + *) + echo "unknown tail client: $lang" >&2 + return 1 + ;; + esac +} + +# shellcheck disable=SC2329 # invoked indirectly via harness_spawn +run_tail_publisher() { + local fifo="$1" + shift + run_tail_client "$@" <"$fifo" 9>&- +} + +run_tail_pair() { + local pub="$1" sub="$2" name="tail-$1-$2" pid subscriber_pid + local fifo="$HARNESS_RUN/tail-ack" broadcast="tail-$1-$2-$$-$RANDOM" + if is_broken "$pub" || is_broken "$sub"; then + echo " FAIL tail $pub -> $sub (client unavailable)" + overall=1 + return + fi + mkfifo "$fifo" + # Open both ends so neither child startup nor a failed publisher can block the harness. + exec 9<>"$fifo" + harness_spawn "$name-pub" "$HARNESS_RUN/$name-pub.log" run_tail_publisher "$fifo" "$pub" publish "$broadcast" + pid="$HARNESS_PID" + harness_spawn "$name-sub" "$HARNESS_RUN/$name-sub.log" run_tail_client "$sub" subscribe "$broadcast" + subscriber_pid="$HARNESS_PID" + harness_wait "$subscriber_pid" || true + # Like run_native, a verified data result survives the NAPI addon's exit crash. + if grep -qx 'tail clean end=4 groups=0,1,2,3 bytes=1048576' "$HARNESS_RUN/$name-sub.log"; then + printf 'clean end\n' >&9 + harness_wait "$pid" || true + if grep -qx 'tail acknowledged' "$HARNESS_RUN/$name-pub.log"; then + echo " PASS tail $pub -> $sub (groups 0,1,2,3; clean end 4)" + else + echo " FAIL tail $pub -> $sub (publisher did not acknowledge)" + overall=1 + cat "$HARNESS_RUN/$name-pub.log" + fi + else + echo " FAIL tail $pub -> $sub (missing groups or clean end)" + overall=1 + cat "$HARNESS_RUN/$name-pub.log" "$HARNESS_RUN/$name-sub.log" + harness_reap "$pid" + fi + exec 9>&- + rm "$fifo" +} + # One media.ts invocation. It reports its own verdict (a negative control passes by failing on the # assertion it names), so the exit code is the whole answer. run_media() { @@ -644,7 +724,7 @@ elif [[ "$NEGATIVE" -eq 1 ]]; then # no data), proving the harness can actually report failure. echo "=== negative control: subscribers expect NO data ===" run_round "none" "interop-missing-$$-$RANDOM.hang" "" -else +elif [[ "$TAIL_ONLY" -eq 0 ]]; then for pub in "${PUB_LIST[@]}"; do broadcast="interop-${pub}-$$-${RANDOM}.hang" echo "=== publisher: $pub broadcast: $broadcast ===" @@ -660,6 +740,17 @@ else done fi +if [[ "$NEGATIVE" -eq 0 && "$MEDIA" -eq 0 ]]; then + echo "=== finite track tails ===" + run_tail_pair rust rust + for runtime in js-native-node js-native-bun; do + if needs "$runtime"; then + run_tail_pair rust "$runtime" + run_tail_pair "$runtime" rust + fi + done +fi + if [[ "$overall" -eq 0 ]]; then echo "interop: all checks passed" else diff --git a/test/justfile b/test/justfile index 91ba59ba80d..632c43950c7 100644 --- a/test/justfile +++ b/test/justfile @@ -29,6 +29,7 @@ default: # client, and the GStreamer moqsrc plugin). --timeout 30 gives headless Chromium # cold-start headroom. Other flags pass through, e.g. # `just test interop --publishers rust,python --subscribers rust,c`. +# Finite track tails run too; `--tail` selects just Rust/Node/Bun tail checks. interop *args: #!/usr/bin/env bash set -euo pipefail From 298e8826f19cb49a4ac49cc8932338edd39dbed8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 10:01:56 -0700 Subject: [PATCH 03/14] test(interop): follow broadcast close() and track demand() renames Co-Authored-By: Claude Opus 5.5 --- test/interop/clients/js-native/tail.ts | 3 ++- test/interop/clients/rust/tail.rs | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/test/interop/clients/js-native/tail.ts b/test/interop/clients/js-native/tail.ts index 529b0005463..d509e5e50a4 100644 --- a/test/interop/clients/js-native/tail.ts +++ b/test/interop/clients/js-native/tail.ts @@ -23,7 +23,8 @@ try { const broadcast = origin.createBroadcast(Moq.Path.from(path)); const track = broadcast.createTrack("tail"); broadcast.announce(); - while (!track.used.peek()) await track.used.changed(); + const demand = track.demand(); + while (!demand.used.peek()) await demand.used.changed(); // The end precedes the last group's data; the transport still has to drain it. track.finishAt(GROUPS); for (let sequence = 0; sequence < GROUPS; sequence++) { diff --git a/test/interop/clients/rust/tail.rs b/test/interop/clients/rust/tail.rs index 765ea1e8eb0..3990ce33b0a 100644 --- a/test/interop/clients/rust/tail.rs +++ b/test/interop/clients/rust/tail.rs @@ -49,7 +49,7 @@ async fn main() -> anyhow::Result<()> { ack == "clean end\n", "reader did not acknowledge a clean end" ); - broadcast.finish(); + broadcast.close(); eprintln!("tail acknowledged"); } "subscribe" => { From a231756bb2f15e74d8cd478e4af06cca962c637e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 00:34:56 -0700 Subject: [PATCH 04/14] fix(net): keep a lite subscription's demand until its groups drain A relay's moq-lite SubscribeServe dropped its track subscriber once the live edge reached the declared end, while its group streams were still draining. The relay then cancelled its upstream subscription as idle and the publisher reset every group still on the wire, losing a finite track's tail. Hold the subscriber in the Drain state, add a relay regression test over the mock transport, let --tail run without ffmpeg, give the tail publisher a longer deadline, and delete the completed track-tail-interop quest. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/lite-late-lower-group.md | 5 +- quest/m1/subscribe-drop.md | 9 +- quest/m1/track-tail-interop.md | 43 --------- quest/m1/wire-compat.md | 4 - rs/moq-net/src/lite/publisher.rs | 27 +++++- rs/moq-net/tests/finite_relay.rs | 140 ++++++++++++++++++++++++++++++ test/interop/README.md | 9 +- test/interop/interop.sh | 16 ++-- 9 files changed, 181 insertions(+), 73 deletions(-) delete mode 100644 quest/m1/track-tail-interop.md create mode 100644 rs/moq-net/tests/finite_relay.rs diff --git a/quest/m1/README.md b/quest/m1/README.md index 39ceb50da83..efd7c3c7fa4 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -41,7 +41,6 @@ blocks. The quests that gated m0 lines moved under them. - [Cluster routing](/quest/m1/cluster-routing/README.md) - any node routes toward a broadcast's origin over CDN and P2P links alike, with per-origin routes and path-less announces - [One route cost](/quest/m1/route-cost.md) - Warm and Cold collapse to one static route cost - [Delete removed cluster flags](/quest/m1/cluster-shims.md) - `mesh` and `linger` leave `cluster::Config` and their flags become unknown -- [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop` - [Request linger](/quest/m1/request-linger.md) - an upstream FETCH or SUBSCRIBE outlives its last reader by a short linger, so quick re-requests don't churn upstream - [Flat questlines](/quest/m1/quest-flat-lines.md) - moq pins the current quest CLI, lands its questline branches on main, and retires them - [moq-bot may push workflow changes](/quest/m1/bot-workflows-permission.md) - condition: the maintainer grants moq-bot's GitHub App the `workflows` permission, so back-merges carrying workflow changes go through diff --git a/quest/m1/lite-late-lower-group.md b/quest/m1/lite-late-lower-group.md index 451ed6ec016..79f1125a6c5 100644 --- a/quest/m1/lite-late-lower-group.md +++ b/quest/m1/lite-late-lower-group.md @@ -15,9 +15,8 @@ clean end without it. for the first served group and then calls `raise_start_to(start)`, which suppresses every lower group regardless of the subscription's floor and budget. #4387 added it to resolve a relayed subscription's start from its -source; keep that fix (see the note in -[Track tail interop](/quest/m1/track-tail-interop.md)) while honoring an -explicit floor. +source; keep that fix (the `just test interop --tail` lanes cover it through +a relay) while honoring an explicit floor. Decided: diff --git a/quest/m1/subscribe-drop.md b/quest/m1/subscribe-drop.md index 97fc1d714ec..f4f576fe879 100644 --- a/quest/m1/subscribe-drop.md +++ b/quest/m1/subscribe-drop.md @@ -47,14 +47,9 @@ Regression tests: a publisher that expires a group, skips a sequence, and resets a stream before its header; on each version the subscriber settles without waiting out the grace. -Add the lite-07 drop case to the tail interop harness from -[track tail interop](/quest/m1/track-tail-interop.md): Rust and JS +Add the lite-07 drop case to the tail interop harness +(`just test interop --tail`, `test/interop/clients/*/tail.*`): Rust and JS subscribers both settle on SUBSCRIBE_DROP through the relay, so a group the publisher skipped or never opened ends the track without waiting out the grace. Decided in the 2026-09-30 audit: the case moved here so the basic tail interop could land first. - - -## Related - -- [Track tail interop](/quest/m1/track-tail-interop.md) - the Rust-JS tail harness the lite-07 drop case extends diff --git a/quest/m1/track-tail-interop.md b/quest/m1/track-tail-interop.md deleted file mode 100644 index c42b0f3bffa..00000000000 --- a/quest/m1/track-tail-interop.md +++ /dev/null @@ -1,43 +0,0 @@ -# [S] Track tail interop - -## Goal - -`just test interop` covers a publisher that ends a track while its last group -is still in flight: a Rust publisher's track is read to its declared end by the -JS subscriber, and a JS publisher's by the Rust subscriber, through the relay. -Each reader gets every group below the end and then a clean end, never an -error or a stall. - -## Plan - -Both moq-net and `@moq/net` wait for a track's tail once the publisher ends a -subscription, and each is covered by its own in-process tests. Nothing checks -that the two agree across a relay on real QUIC. - -What stood in the way when the Rust half landed: - -- `moq import` exiting the moment stdin ends is cleared: since #4430 a - graceful close delivers finished tracks at stdin EOF. -- The native JS subscriber (`test/interop/clients/js-native`) returns on the - first frame. It needs a mode that reads a track to its end and reports how it - ended and which groups it saw. - -The harness exposed a relay start-floor defect: when a newer group arrives -first, earlier in-flight groups can be lost. #4387 (merged 09-28) fixed that -case, but the tail lanes still fail on this quest's PR #4225 after it: Interop -run 37139033242 (2026-10-03) failed `tail rust -> rust`, `rust -> -js-native-node`, and `rust -> js-native-bun`. Merge `main` (after #4741 and -#4813) into #4225, rerun, and record here whatever defect still fails; the -cause is untracked until then. - -Decided in the 2026-09-30 audit: the lite-07 drop case moved into -[SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md)'s tests, so the basic Rust and -JS tail interop lands now instead of waiting on that [L] quest. - -QUIC on localhost rarely reorders, so this is a smoke check that the end is -delivered and clean. The ordering race itself stays in the unit tests. - -## Related - -- [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - owns the lite-07 drop case on top of this harness -- [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - keeps a reset stream's header, so the reset acts as a one-group drop diff --git a/quest/m1/wire-compat.md b/quest/m1/wire-compat.md index 3d5c5945c66..1e6ae496e54 100644 --- a/quest/m1/wire-compat.md +++ b/quest/m1/wire-compat.md @@ -55,7 +55,3 @@ Guidance: is acknowledged (for example a skip list that the next release clears). - Wire the job into the existing nightly (`interop.yml` already has a schedule) and document the recipe beside `just test interop`. - -## Related - -- [Track tail interop](/quest/m1/track-tail-interop.md) - another cross-language case in the same harness diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 861b82abf13..957ca363ae7 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1223,7 +1223,13 @@ enum SubscribeServe { /// is moved on every transition. Run(Box>), /// The track finished: draining the in-flight group streams before the FIN. - Drain { children: kio::Tasks> }, + Drain { + children: kio::Tasks>, + /// The subscription's demand lasts until its groups drain. A relay cancels its + /// upstream subscription once nobody subscribes, and the publisher then resets + /// every group still on the wire. `None` only in passing between states. + _track: Option, + }, } impl Request for SubscribeServe { @@ -1283,6 +1289,7 @@ impl Request for SubscribeServe { self, Self::Drain { children: Default::default(), + _track: None, }, ) else { @@ -1347,10 +1354,22 @@ impl Request for SubscribeServe { if ready!(run.poll_step(writer, waiter))?.is_continue() { return Poll::Ready(Ok(ControlFlow::Continue(()))); } - let children = std::mem::take(&mut run.children); - *self = Self::Drain { children }; + let Self::Run(run) = std::mem::replace( + self, + Self::Drain { + children: Default::default(), + _track: None, + }, + ) else { + unreachable!() + }; + let TrackRun { children, track, .. } = *run; + *self = Self::Drain { + children, + _track: Some(track), + }; } - Self::Drain { children } => { + Self::Drain { children, .. } => { ready!(children.poll(waiter)); return Poll::Ready(Ok(ControlFlow::Break(()))); } diff --git a/rs/moq-net/tests/finite_relay.rs b/rs/moq-net/tests/finite_relay.rs new file mode 100644 index 00000000000..a29f248cfcd --- /dev/null +++ b/rs/moq-net/tests/finite_relay.rs @@ -0,0 +1,140 @@ +//! A finite track read through a relay: every group below the declared end, then a clean +//! end, even when the relay's subscriber reaches the end before the groups finish. +//! +//! The relay's downstream subscription sees the declared end as soon as every group below +//! it has opened, while their payloads are still crossing the upstream hop. Its demand must +//! last until those groups drain: a relay cancels its upstream subscription once nobody +//! subscribes, and the publisher then resets every group still on the wire. + +mod support; + +use std::time::Duration; + +use moq_net::track::{Position, Subscription}; +use moq_net::{Hop, Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); +const GROUPS: u64 = 4; + +const VERSIONS: &[&str] = &[ + "moq-lite-03", + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-17", + "moq-transport-22", +]; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// Publish groups 0..4 through a relay, each holding its last frame until the end has +/// reached the subscriber, and return the groups the subscriber read whole and how its +/// subscription ended. +async fn round(name: &str) -> (Vec, Result<(), moq_net::Error>) { + let version: Version = name.parse().unwrap(); + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let mut track = broadcast.create_track("tail", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let relay = produce_origin(2); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let upstream = connect_mock(options).await; + + let subscriber = produce_origin(3); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(subscriber.clone()); + let downstream = connect_mock(options).await; + + let consumer = subscriber.consume(); + tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let remote = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + + let reader = tokio::spawn(async move { + let subscription = Subscription::default() + .with_max_age(Duration::from_secs(5)) + .with_start(Position::group(0)); + let mut sub = remote + .track("tail") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + let mut seen = Vec::new(); + let end = loop { + let mut group = match sub.recv_group().await { + Ok(Some(group)) => group, + Ok(None) => break Ok(()), + Err(err) => break Err(err), + }; + let mut frames = 0; + let read = loop { + match group.read_frame().await { + Ok(Some(_)) => frames += 1, + Ok(None) => break Ok(()), + Err(err) => break Err(err), + } + }; + if let Err(err) = read { + break Err(err); + } + if frames == 2 { + seen.push(group.sequence); + } + }; + seen.sort(); + (seen, end) + }); + + tokio::time::timeout(TIMEOUT, track.demand().used()) + .await + .expect("no subscriber appeared") + .unwrap(); + track.finish_at(GROUPS).unwrap(); + let mut groups = Vec::new(); + for _ in 0..GROUPS { + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, &b"head"[..]).unwrap(); + groups.push(group); + } + + // Paused time advances only once every task is idle: each group has reached the + // subscriber, so the relay's downstream subscription has seen the end. + tokio::time::sleep(Duration::from_millis(100)).await; + for mut group in groups { + group.write_frame(Timestamp::ZERO, &b"tail"[..]).unwrap(); + group.finish().unwrap(); + } + + let outcome = tokio::time::timeout(TIMEOUT, reader) + .await + .unwrap_or_else(|_| panic!("{name}: the track never ended")) + .expect("reader panicked"); + drop((track, broadcast, upstream, downstream, publisher, relay, subscriber)); + outcome +} + +#[tokio::test] +async fn relay_keeps_groups_in_flight_past_the_end() { + tokio::time::pause(); + for version in VERSIONS { + let (seen, end) = round(version).await; + assert_eq!(seen, [0, 1, 2, 3], "{version}: end={end:?}"); + assert!(end.is_ok(), "{version}: ended with {end:?}"); + } +} diff --git a/test/interop/README.md b/test/interop/README.md index d9006bc907f..d063ec3ce75 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -26,11 +26,10 @@ native JS adds both Rust-to-JS and JS-to-Rust under that runtime, so `--all` cov Node and Bun. `--tail` runs just these five lanes. The finite clients keep their session alive until the harness acknowledges the -reader's complete clean end over stdin. This avoids using `moq import`, which -currently closes its session at stdin EOF before its subscriptions drain. No -sleep stands in for drain completion. A missing group, error, or stall fails the -lane; the timeout only bounds failure. QUIC on localhost rarely reorders, so the -ordering race remains covered by transport unit tests. +reader's complete clean end over stdin, so a lane tests delivery rather than +shutdown. No sleep stands in for drain completion. A missing group, error, or +stall fails the lane; the timeout only bounds failure. QUIC on localhost rarely +reorders, so the ordering race remains covered by transport unit tests. `just test media` is a separate, browser-only run that asks a harder question: is the media a viewer gets actually advancing and in sync, and does the diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 52c20288476..c7cf6fcff87 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -174,8 +174,10 @@ require_tools() { # The relay, CLI, ffmpeg, and harness essentials are hard requirements. A # missing per-client toolchain (uv / bun / node / cc) just marks that client # broken in prepare, so it fails its own cells instead of the whole run. - local missing=() t - for t in cargo ffmpeg curl timeout; do + # Tail-only runs publish raw tracks, so they never encode with ffmpeg. + local missing=() t tools=(cargo curl timeout) + [[ "$TAIL_ONLY" -eq 1 ]] || tools+=(ffmpeg) + for t in "${tools[@]}"; do have "$t" || missing+=("$t") done if [[ ${#missing[@]} -gt 0 ]]; then @@ -648,11 +650,13 @@ run_round() { # Raw-track clients have no codecs: every byte, group, and the declared end is checked. # shellcheck disable=SC2329 # invoked indirectly via harness_spawn run_tail_client() { - local lang="$1" role="$2" broadcast="$3" + local lang="$1" role="$2" broadcast="$3" limit="$TIMEOUT" + # The publisher outlives the subscriber's whole deadline, waiting for its acknowledgement. + [[ "$role" == publish ]] && limit=$(awk -v t="$TIMEOUT" 'BEGIN { print t * 2 }') case "$lang" in - rust) timeout -k 3 "$TIMEOUT" "$TARGET_BASE/$PROFILE/examples/interop-tail" "$role" "$URL" "$broadcast" ;; - js-native-node) (cd "$CLIENTS/js-native" && timeout -k 3 "$TIMEOUT" node --import tsx tail.ts "$role" "$URL" "$broadcast") ;; - js-native-bun) (cd "$CLIENTS/js-native" && timeout -k 3 "$TIMEOUT" bun tail.ts "$role" "$URL" "$broadcast") ;; + rust) timeout -k 3 "$limit" "$TARGET_BASE/$PROFILE/examples/interop-tail" "$role" "$URL" "$broadcast" ;; + js-native-node) (cd "$CLIENTS/js-native" && timeout -k 3 "$limit" node --import tsx tail.ts "$role" "$URL" "$broadcast") ;; + js-native-bun) (cd "$CLIENTS/js-native" && timeout -k 3 "$limit" bun tail.ts "$role" "$URL" "$broadcast") ;; *) echo "unknown tail client: $lang" >&2 return 1 From 491aef5b0227eeaf308984f7d6c5b64e9da14f7a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 11:02:38 -0700 Subject: [PATCH 05/14] test(net): simulate finite relay group drains Co-Authored-By: GPT-6 --- rs/moq-net/tests/finite_relay.rs | 71 ++++++++++++++++++-------------- 1 file changed, 41 insertions(+), 30 deletions(-) diff --git a/rs/moq-net/tests/finite_relay.rs b/rs/moq-net/tests/finite_relay.rs index a29f248cfcd..891a7ed42ed 100644 --- a/rs/moq-net/tests/finite_relay.rs +++ b/rs/moq-net/tests/finite_relay.rs @@ -10,11 +10,14 @@ mod support; use std::time::Duration; -use moq_net::track::{Position, Subscription}; +use futures::{SinkExt, StreamExt}; + +use moq_net::track::Subscription; use moq_net::{Hop, Timestamp, Version}; use support::harness::{MockConnectOptions, connect_mock}; const TIMEOUT: Duration = Duration::from_secs(10); +const MAX_AGE: Duration = Duration::from_secs(5); const GROUPS: u64 = 4; const VERSIONS: &[&str] = &[ @@ -29,7 +32,7 @@ const VERSIONS: &[&str] = &[ fn produce_origin(hop: u64) -> moq_net::origin::Producer { let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); - tokio::spawn(support::harness::run(driver)); + support::harness::spawn(driver); producer } @@ -56,72 +59,81 @@ async fn round(name: &str) -> (Vec, Result<(), moq_net::Error>) { let downstream = connect_mock(options).await; let consumer = subscriber.consume(); - tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + moq_net_sim::timeout(TIMEOUT, consumer.routed("bcast")) .await .expect("announce timeout") .expect("routed"); - let remote = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + let remote = moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast")) .await .expect("resolve timeout") .expect("broadcast resolves"); - let reader = tokio::spawn(async move { - let subscription = Subscription::default() - .with_max_age(Duration::from_secs(5)) - .with_start(Position::group(0)); + let (mut heads, mut opened) = futures::channel::mpsc::unbounded(); + let reader = moq_net_sim::spawn(async move { + let subscription = Subscription::default().with_max_age(MAX_AGE).with_groups(0..); let mut sub = remote .track("tail") .unwrap() .subscribe(subscription) .await .expect("subscribe"); - let mut seen = Vec::new(); + let mut tails = Vec::new(); let end = loop { let mut group = match sub.recv_group().await { Ok(Some(group)) => group, Ok(None) => break Ok(()), Err(err) => break Err(err), }; - let mut frames = 0; - let read = loop { - match group.read_frame().await { - Ok(Some(_)) => frames += 1, - Ok(None) => break Ok(()), - Err(err) => break Err(err), - } - }; - if let Err(err) = read { - break Err(err); - } - if frames == 2 { - seen.push(group.sequence); - } + let head = group.read_frame().await.unwrap().expect("head frame"); + assert_eq!(&head.payload[..], b"head"); + heads.send(group.sequence).await.unwrap(); + tails.push(moq_net_sim::spawn(async move { + let tail = group.read_frame().await?.expect("tail frame"); + assert_eq!(&tail.payload[..], b"tail"); + assert!(group.read_frame().await?.is_none(), "extra frame"); + Ok::<_, moq_net::Error>(group.sequence) + })); }; + let mut seen = Vec::new(); + for tail in tails { + match tail.await.expect("tail reader panicked") { + Ok(sequence) => seen.push(sequence), + Err(err) => return (seen, Err(err)), + } + } seen.sort(); (seen, end) }); - tokio::time::timeout(TIMEOUT, track.demand().used()) + moq_net_sim::timeout(TIMEOUT, track.demand().used()) .await .expect("no subscriber appeared") .unwrap(); track.finish_at(GROUPS).unwrap(); let mut groups = Vec::new(); - for _ in 0..GROUPS { + for sequence in 0..GROUPS { let mut group = track.append_group().unwrap(); group.write_frame(Timestamp::ZERO, &b"head"[..]).unwrap(); groups.push(group); + // IETF requests the live edge, so observe each header before advancing it. + assert_eq!( + moq_net_sim::timeout(TIMEOUT, opened.next()) + .await + .expect("head timeout"), + Some(sequence), + ); } - // Paused time advances only once every task is idle: each group has reached the + // Simulated time advances only once every task is idle: each group has reached the // subscriber, so the relay's downstream subscription has seen the end. - tokio::time::sleep(Duration::from_millis(100)).await; + moq_net_sim::sleep(Duration::from_millis(100)).await; for mut group in groups { group.write_frame(Timestamp::ZERO, &b"tail"[..]).unwrap(); group.finish().unwrap(); } - let outcome = tokio::time::timeout(TIMEOUT, reader) + // Lite03 carries no declared end: each of the two hops waits one max-age grace. + let outcome = moq_net_sim::timeout(MAX_AGE * 2 + TIMEOUT, reader) .await .unwrap_or_else(|_| panic!("{name}: the track never ended")) .expect("reader panicked"); @@ -129,9 +141,8 @@ async fn round(name: &str) -> (Vec, Result<(), moq_net::Error>) { outcome } -#[tokio::test] +#[moq_net_sim::test] async fn relay_keeps_groups_in_flight_past_the_end() { - tokio::time::pause(); for version in VERSIONS { let (seen, end) = round(version).await; assert_eq!(seen, [0, 1, 2, 3], "{version}: end={end:?}"); From 76a922c5e6a12aba19b772a643e04c39ab690051 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 14:57:05 -0700 Subject: [PATCH 06/14] fix(net): hold readers at a hole until the upstream tail settles A relay learned a finite track's end, then received the last group's stream before the lower ones (QUIC does not order streams). Its readers ended at the boundary with the lower groups still on the wire, so the relay finished the downstream subscription after group 3 alone: the Bun -> Rust finite-tail lane. A wire subscription now marks its copy's tail pending from the declared end (lite SUBSCRIBE_END, IETF END_OF_TRACK) until that tail settles. While pending, a reader at the boundary waits only if a group between the feed's start and the end has not arrived; the last producer going ends it regardless. A local track keeps ending at the boundary with missing lower groups. On lite-05/06 a lost datagram is indistinguishable from a reordered group, so its hole now holds readers for the tail grace; lite-07's stream count keeps that end prompt. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/subscriber.rs | 4 ++ rs/moq-net/src/lite/subscriber.rs | 4 ++ rs/moq-net/src/model/track.rs | 92 ++++++++++++++++++++++++++- rs/moq-net/tests/finite_relay.rs | 102 +++++++++++++++++++++++------- rs/moq-net/tests/support/mock.rs | 11 ++++ rs/moq-net/tests/track_tail.rs | 19 ++++-- 6 files changed, 199 insertions(+), 33 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index f144d58419d..3e31437cc5d 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2146,6 +2146,8 @@ where fill.end(); } } + // The tail settled, so readers may end at an END_OF_TRACK's boundary. + track.set_tail_pending(false); // A no-op once an END_OF_TRACK declared the end. let _ = track.finish(); } @@ -2766,6 +2768,8 @@ fn end_track(track: &mut track::Producer, end: u64) -> Result<(), Error> { let _ = track.clone().abort(Error::ProtocolViolation); return Err(Error::ProtocolViolation); } + // Lower groups may still be on the wire, behind the one that carried the end. + track.set_tail_pending(true); Ok(()) } diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 389aa9ebcf3..e4998737f8e 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -3944,6 +3944,7 @@ impl kio::Task for TrackServeRun { } }, ServeEnd::Finished => { + serve_loop.serving.set_tail_pending(false); let _ = serve_loop.serving.finish(); } ServeEnd::GiveBack(err) => { @@ -4288,6 +4289,9 @@ impl ServeLoop { return Poll::Ready(ServeEnd::GiveBack(Error::ProtocolViolation)); } } + } else { + // Lower groups may still be on the wire, behind a higher one. + self.serving.set_tail_pending(true); } active.end = Some(end.clone()); } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index a57a2bbe029..068e69b4446 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -240,6 +240,11 @@ pub(crate) struct TrackState { // or the receiving session closed locally without declaring an end. sealed: bool, + // A wire subscription declared the end ahead of the group streams below it, which + // QUIC may deliver in any order: readers wait for its tail to settle, or for the + // last producer to go, rather than end at the boundary with groups still to arrive. + tail_pending: bool, + // No producer remains (aborted, sealed, or dropped), so nothing protects the live // edge any longer: the pool's idle expiry reclaims it like every other group, and a // stale consumer cannot pin the cache. @@ -1209,12 +1214,32 @@ impl TrackState { // `sealed` also ends a locally closed receive track without a declared end. // An abort still wins unless that end had already settled: a group below it was still open. let reached = self.sealed - || self - .final_sequence - .is_some_and(|fin| self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin); + || self.final_sequence.is_some_and(|fin| { + self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin && !self.awaits_tail(fin) + }); reached && (self.abort.is_none() || self.settled) } + /// Whether a group below `fin` the wire subscription still owes has yet to arrive: + /// a sequence from where the feed starts up to the end with nothing cached. Only + /// such a hole holds readers, so a tail with nothing missing ends them at once. + fn awaits_tail(&self, fin: u64) -> bool { + if !self.tail_pending { + return false; + } + // Without a declared start, the lowest group that arrived stands in for it. + let floor = match self.start_sequence { + Some(start) => start, + None if self.start_pending => return true, + None => match self.lookup.keys().next() { + Some(&first) => first, + None => return fin > 0, + }, + } + .min(fin); + (self.lookup.range(floor..fin).count() as u64) < fin - floor + } + /// Whether the declared end is reached and every cached group below it finished, /// so nothing the end promised is still in flight. fn is_settled(&self) -> bool { @@ -1671,6 +1696,20 @@ impl Producer { self.modify()?.set_final(final_sequence) } + /// Whether a wire subscription still owes group streams below the declared end. + /// + /// While pending, readers do not end at the boundary: a lower group's stream may + /// arrive after a higher one. The session clears it once its tail settles; the last + /// producer going ends readers regardless. + pub(crate) fn set_tail_pending(&self, pending: bool) { + if self.state.read().tail_pending == pending { + return; + } + if let Ok(mut state) = self.modify() { + state.tail_pending = pending; + } + } + /// Declare the first group the live feed serves (the wire's SUBSCRIBE_START, /// or the start the subscription itself requested): groups below `sequence` /// will never arrive on their own, so a reader waiting for one fails over @@ -7045,6 +7084,53 @@ mod test { )); } + /// A wire subscription's pending tail holds readers at a hole below the end, since a + /// lower group's stream may arrive after a higher one, until the hole fills or the tail + /// settles. The last producer going ends them regardless. + #[moq_net_sim::test] + async fn a_pending_tail_holds_readers_at_a_hole() { + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(None); + // The wire declares where the feed starts. + producer.start_at(0).unwrap(); + producer.finish_at(3).unwrap(); + producer.set_tail_pending(true); + let _high = producer.create_group(group::Info { sequence: 2 }).unwrap(); + assert_eq!(arrival.assert_group().sequence, 2); + assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); + + // The late lower groups are delivered, then the full tail ends the reader. + let _low = producer.create_group(group::Info { sequence: 0 }).unwrap(); + assert_eq!(arrival.assert_group().sequence, 0); + assert!(arrival.recv_group().now_or_never().is_none(), "group 1 is still owed"); + let _mid = producer.create_group(group::Info { sequence: 1 }).unwrap(); + assert_eq!(arrival.assert_group().sequence, 1); + assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); + + // A settled tail ends readers at the boundary, hole and all. + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(None); + producer.start_at(0).unwrap(); + producer.finish_at(2).unwrap(); + producer.set_tail_pending(true); + let _high = producer.create_group(group::Info { sequence: 1 }).unwrap(); + assert_eq!(arrival.assert_group().sequence, 1); + assert!(arrival.recv_group().now_or_never().is_none()); + producer.set_tail_pending(false); + assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); + + // So does the last producer going. + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(None); + producer.start_at(0).unwrap(); + producer.finish_at(2).unwrap(); + producer.set_tail_pending(true); + let _high = producer.create_group(group::Info { sequence: 1 }).unwrap(); + assert_eq!(arrival.assert_group().sequence, 1); + drop(producer); + assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); + } + /// An abort before the declared end settled wins over it: the boundary was reached, /// but a group below it was still open, so the track was cut off rather than ended. #[test] diff --git a/rs/moq-net/tests/finite_relay.rs b/rs/moq-net/tests/finite_relay.rs index 9fa2a5ee487..f4997ade254 100644 --- a/rs/moq-net/tests/finite_relay.rs +++ b/rs/moq-net/tests/finite_relay.rs @@ -36,10 +36,20 @@ fn produce_origin(hop: u64) -> moq_net::origin::Producer { producer } -/// Publish groups 0..4 through a relay, each holding its last frame until the end has -/// reached the subscriber, and return the groups the subscriber read whole and how its -/// subscription ended. -async fn round(name: &str) -> (Vec, Result<(), moq_net::Error>) { +/// How the publisher's groups reach the relay. +#[derive(Clone, Copy)] +enum Arrival { + /// Each group's header arrives in order, but its last frame is held until the end has + /// reached the subscriber. + HeldTails, + /// Every group is written whole, but the newest group's stream reaches the relay + /// before the others, after the end: QUIC does not order streams. + NewestFirst, +} + +/// Publish groups 0..4 through a relay, delivered as `arrival` says, and return the +/// groups the subscriber read whole and how its subscription ended. +async fn round(name: &str, arrival: Arrival) -> (Vec, Result<(), moq_net::Error>) { let version: Version = name.parse().unwrap(); let publisher = produce_origin(1); let broadcast = publisher.create_broadcast("bcast").unwrap(); @@ -110,26 +120,59 @@ async fn round(name: &str) -> (Vec, Result<(), moq_net::Error>) { .expect("no subscriber appeared") .unwrap(); track.finish_at(GROUPS).unwrap(); - let mut groups = Vec::new(); - for sequence in 0..GROUPS { - let mut group = track.append_group().unwrap(); - group.write_frame(Timestamp::ZERO, &b"head"[..]).unwrap(); - groups.push(group); - // IETF requests the live edge, so observe each header before advancing it. - assert_eq!( - moq_net_sim::timeout(TIMEOUT, opened.next()) - .await - .expect("head timeout"), - Some(sequence), - ); - } + match arrival { + Arrival::HeldTails => { + let mut groups = Vec::new(); + for sequence in 0..GROUPS { + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, &b"head"[..]).unwrap(); + groups.push(group); + // IETF requests the live edge, so observe each header before advancing it. + assert_eq!( + moq_net_sim::timeout(TIMEOUT, opened.next()) + .await + .expect("head timeout"), + Some(sequence), + ); + } - // Simulated time advances only once every task is idle: each group has reached the - // subscriber, so the relay's downstream subscription has seen the end. - moq_net_sim::sleep(Duration::from_millis(100)).await; - for mut group in groups { - group.write_frame(Timestamp::ZERO, &b"tail"[..]).unwrap(); - group.finish().unwrap(); + // Simulated time advances only once every task is idle: each group has reached + // the subscriber, so the relay's downstream subscription has seen the end. + moq_net_sim::sleep(Duration::from_millis(100)).await; + for mut group in groups { + group.write_frame(Timestamp::ZERO, &b"tail"[..]).unwrap(); + group.finish().unwrap(); + } + } + Arrival::NewestFirst => { + upstream.server_transport.hold_unis(); + for _ in 0..GROUPS { + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, &b"head"[..]).unwrap(); + group.write_frame(Timestamp::ZERO, &b"tail"[..]).unwrap(); + group.finish().unwrap(); + // IETF serves from the live edge, so let each group open its stream before + // the next one supersedes it. + moq_net_sim::sleep(Duration::from_millis(10)).await; + } + + // Once idle, the publisher has opened every stream and ended the subscription, + // so the relay knows the end before any group arrives. + moq_net_sim::sleep(Duration::from_millis(100)).await; + // Deliver streams newest first (IETF also carries the end on a stream of its + // own) until the last group reaches the subscriber, at the declared boundary. + let newest = loop { + let held = upstream.server_transport.release_newest_uni(); + moq_net_sim::sleep(Duration::from_millis(10)).await; + if let Ok(head) = opened.try_next() { + break head; + } + assert!(held > 0, "{name}: no group reached the subscriber"); + }; + assert_eq!(newest, Some(GROUPS - 1), "{name}"); + moq_net_sim::sleep(Duration::from_millis(100)).await; + upstream.server_transport.release_unis(); + } } // Lite03 carries no declared end: each of the two hops waits one max-delay grace. @@ -144,7 +187,18 @@ async fn round(name: &str) -> (Vec, Result<(), moq_net::Error>) { #[moq_net_sim::test] async fn relay_keeps_groups_in_flight_past_the_end() { for version in VERSIONS { - let (seen, end) = round(version).await; + let (seen, end) = round(version, Arrival::HeldTails).await; + assert_eq!(seen, [0, 1, 2, 3], "{version}: end={end:?}"); + assert!(end.is_ok(), "{version}: ended with {end:?}"); + } +} + +/// The relay learns the end, then receives the last group's stream before the others: it +/// keeps serving until the upstream tail settles instead of ending at the boundary. +#[moq_net_sim::test] +async fn relay_waits_for_reordered_upstream_headers() { + for version in VERSIONS { + let (seen, end) = round(version, Arrival::NewestFirst).await; assert_eq!(seen, [0, 1, 2, 3], "{version}: end={end:?}"); assert!(end.is_ok(), "{version}: ended with {end:?}"); } diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index 427e8df6ad6..9a1ec8166d4 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -610,6 +610,17 @@ impl MockSession { } } + /// Deliver only the newest held uni stream, and keep holding the rest. Returns how + /// many are still held. + pub fn release_newest_uni(&self) -> usize { + let mut held = self.side.held.lock().unwrap(); + let held = held.as_mut().expect("not holding"); + if let Some(stream) = held.pop() { + let _ = self.side.peer_uni.try_push(stream); + } + held.len() + } + /// Deliver the held uni streams newest first, and stop holding. pub fn release_unis_reversed(&self) { for stream in self diff --git a/rs/moq-net/tests/track_tail.rs b/rs/moq-net/tests/track_tail.rs index 03ab5b0a492..cd5ef9d94a3 100644 --- a/rs/moq-net/tests/track_tail.rs +++ b/rs/moq-net/tests/track_tail.rs @@ -304,12 +304,13 @@ async fn ietf_leaving_cancels_a_blocked_end_of_track() { } } -/// A lost datagram is not owed. Nothing tells the subscriber its sequence was a datagram, so -/// on the drafts that account for the owed range its hole keeps the subscription routable for -/// the grace, like a stream reset before its header. A reader never waits on it: the track -/// ends for readers once the live edge reaches the declared end. +/// A lost datagram is not owed, but nothing tells a lite-05 or lite-06 subscriber its +/// sequence was a datagram rather than a group stream still on its way. Its hole holds +/// the subscription, and its readers, for the grace, like a stream reset before its +/// header: ending at once would drop a reordered group. Lite-07 counts the streams, so a +/// lost datagram never delays its end. #[moq_net_sim::test] -async fn a_lost_datagram_never_delays_the_end() { +async fn a_lost_datagram_delays_the_end_only_without_a_stream_count() { for version in ["moq-lite-05", "moq-lite-07-wip"] { let Pair { pair, @@ -357,7 +358,13 @@ async fn a_lost_datagram_never_delays_the_end() { .expect("reader panicked"); assert_eq!(groups, [0, 2], "{version}"); let elapsed = ended - finished; - assert!(elapsed < GRACE / 10, "{version}: ended after {elapsed:?}"); + match version { + "moq-lite-07-wip" => assert!(elapsed < GRACE / 10, "{version}: ended after {elapsed:?}"), + _ => assert!( + (GRACE / 2..GRACE * 2).contains(&elapsed), + "{version}: ended after {elapsed:?}" + ), + } drop((pair, _keep)); } } From b6f7c1879675b849c8e3f62511db702a3a6f52af Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 15:03:28 -0700 Subject: [PATCH 07/14] fix(net): box the drained subscriber and document reader tail holds Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 6 ++++++ rs/moq-net/src/lite/publisher.rs | 4 ++-- rs/moq-net/tests/finite_relay.rs | 4 ++-- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index ff5e6d433c1..7576ed85048 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -57,6 +57,12 @@ datagrams, and `SUBSCRIBE_DROP`, from the start group the subscription last asked for. A lost datagram is not owed, but its hole waits out the grace like a lost stream. +Until the tail settles, a Rust reader at the declared end waits for any group +missing between where the feed starts and the end, so a group whose header +arrives after a higher one is still read. On moq-lite 05 and 06 a lost +datagram's hole therefore delays the end by the grace; moq-lite 07's count +does not. + From moq-lite 06, an end that contradicts the groups received aborts the track: a group at or past `SUBSCRIBE_END`, or a `SUBSCRIBE_END` below a group already received. moq-lite 05 specified an inclusive end, so there it only drops that diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 5d02e17e2cc..bb4f35af612 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1234,7 +1234,7 @@ enum SubscribeServe { /// The subscription's demand lasts until its groups drain. A relay cancels its /// upstream subscription once nobody subscribes, and the publisher then resets /// every group still on the wire. `None` only in passing between states. - _track: Option, + _track: Option>, }, } @@ -1376,7 +1376,7 @@ impl Request for SubscribeServe { let TrackRun { children, track, .. } = *run; *self = Self::Drain { children, - _track: Some(track), + _track: Some(Box::new(track)), }; } Self::Drain { children, .. } => { diff --git a/rs/moq-net/tests/finite_relay.rs b/rs/moq-net/tests/finite_relay.rs index f4997ade254..87a530ca3ee 100644 --- a/rs/moq-net/tests/finite_relay.rs +++ b/rs/moq-net/tests/finite_relay.rs @@ -164,12 +164,12 @@ async fn round(name: &str, arrival: Arrival) -> (Vec, Result<(), moq_net::E let newest = loop { let held = upstream.server_transport.release_newest_uni(); moq_net_sim::sleep(Duration::from_millis(10)).await; - if let Ok(head) = opened.try_next() { + if let Ok(head) = opened.try_recv() { break head; } assert!(held > 0, "{name}: no group reached the subscriber"); }; - assert_eq!(newest, Some(GROUPS - 1), "{name}"); + assert_eq!(newest, GROUPS - 1, "{name}"); moq_net_sim::sleep(Duration::from_millis(100)).await; upstream.server_transport.release_unis(); } From 38bf64e00fc64d4fb6f65362d6d1ad0173b932c6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 16:31:50 -0700 Subject: [PATCH 08/14] fix(net): publish the wire end and its pending tail in one write A reader on another worker could see the end before the pending tail and end at a hole the wire was still filling. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/subscriber.rs | 5 ++-- rs/moq-net/src/lite/subscriber.rs | 9 +++--- rs/moq-net/src/model/track.rs | 46 +++++++++++++++++++++++++++---- 3 files changed, 46 insertions(+), 14 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 3e31437cc5d..c7e95f1364a 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2763,13 +2763,12 @@ fn end_track(track: &mut track::Producer, end: u64) -> Result<(), Error> { if track.final_sequence().is_some() { return Ok(()); } - if let Err(err) = track.finish_at(end) { + // Lower groups may still be on the wire, behind the one that carried the end. + if let Err(err) = track.finish_at_pending(end) { tracing::warn!(%err, end, "invalid END_OF_TRACK"); let _ = track.clone().abort(Error::ProtocolViolation); return Err(Error::ProtocolViolation); } - // Lower groups may still be on the wire, behind the one that carried the end. - track.set_tail_pending(true); Ok(()) } diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index e4998737f8e..0a0f5bc2bd9 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4273,13 +4273,15 @@ impl ServeLoop { // consumers learn the boundary early; the later stream FIN // then finds the track already finished. lite::SubscribeResponse::End(end) => { - // finish_at rejects a boundary at or below a group + // Lower groups may still be on the wire, behind a higher + // one, so the end holds readers at a hole until the tail + // settles. finish_at rejects a boundary at or below a group // already received. lite-05 specified an inclusive end, // and `@moq/net` 0.1.3 to 0.1.9 sent one, so there it // only costs the early boundary: warn, and let the FIN // finish the track. Later drafts made it exclusive, so // the publisher contradicted its own end. - if let Err(err) = self.serving.finish_at(end.group) { + if let Err(err) = self.serving.finish_at_pending(end.group) { match serve.subscriber.version { Version::Lite05 => { tracing::warn!(track = %serve.name, group = end.group, %err, "invalid subscribe end") @@ -4289,9 +4291,6 @@ impl ServeLoop { return Poll::Ready(ServeEnd::GiveBack(Error::ProtocolViolation)); } } - } else { - // Lower groups may still be on the wire, behind a higher one. - self.serving.set_tail_pending(true); } active.end = Some(end.clone()); } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 068e69b4446..55cf5503077 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1696,6 +1696,17 @@ impl Producer { self.modify()?.set_final(final_sequence) } + /// [`Self::finish_at`] for a wire subscription that may still owe groups below the end. + /// + /// The boundary and the pending tail land in one write: a reader that saw the end + /// without the tail would end at a hole the wire is still filling. + pub(crate) fn finish_at_pending(&mut self, final_sequence: u64) -> Result<()> { + let mut state = self.modify()?; + state.set_final(final_sequence)?; + state.tail_pending = true; + Ok(()) + } + /// Whether a wire subscription still owes group streams below the declared end. /// /// While pending, readers do not end at the boundary: a lower group's stream may @@ -7093,8 +7104,7 @@ mod test { let mut arrival = producer.subscribe(None); // The wire declares where the feed starts. producer.start_at(0).unwrap(); - producer.finish_at(3).unwrap(); - producer.set_tail_pending(true); + producer.finish_at_pending(3).unwrap(); let _high = producer.create_group(group::Info { sequence: 2 }).unwrap(); assert_eq!(arrival.assert_group().sequence, 2); assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); @@ -7111,8 +7121,7 @@ mod test { let mut producer = track_producer("test", None); let mut arrival = producer.subscribe(None); producer.start_at(0).unwrap(); - producer.finish_at(2).unwrap(); - producer.set_tail_pending(true); + producer.finish_at_pending(2).unwrap(); let _high = producer.create_group(group::Info { sequence: 1 }).unwrap(); assert_eq!(arrival.assert_group().sequence, 1); assert!(arrival.recv_group().now_or_never().is_none()); @@ -7123,14 +7132,39 @@ mod test { let mut producer = track_producer("test", None); let mut arrival = producer.subscribe(None); producer.start_at(0).unwrap(); - producer.finish_at(2).unwrap(); - producer.set_tail_pending(true); + producer.finish_at_pending(2).unwrap(); let _high = producer.create_group(group::Info { sequence: 1 }).unwrap(); assert_eq!(arrival.assert_group().sequence, 1); drop(producer); assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); } + /// The wire's end can land after the highest group, with lower groups still owed. A + /// reader already past that group stays held at the hole: the end never shows without + /// its pending tail. + #[moq_net_sim::test] + async fn an_end_after_the_highest_group_holds_the_reader() { + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(None); + producer.start_at(0).unwrap(); + let _high = producer.create_group(group::Info { sequence: 2 }).unwrap(); + assert_eq!(arrival.assert_group().sequence, 2); + assert!(arrival.recv_group().now_or_never().is_none()); + + producer.finish_at_pending(3).unwrap(); + assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); + assert!( + matches!(producer.finish_at_pending(2), Err(Error::Closed)), + "a second end is refused without touching the first" + ); + + for sequence in [0, 1] { + let _low = producer.create_group(group::Info { sequence }).unwrap(); + assert_eq!(arrival.assert_group().sequence, sequence); + } + assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); + } + /// An abort before the declared end settled wins over it: the boundary was reached, /// but a group below it was still open, so the track was cut off rather than ended. #[test] From 270332814eabd6de5d1e1b1dc390ad1a93a0d464 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 21:20:28 -0700 Subject: [PATCH 09/14] test(net): check the first end stands after a refused second end Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index c7981110d14..fe70dd9743b 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -7419,8 +7419,9 @@ mod test { assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); assert!( matches!(producer.finish_at_pending(2), Err(Error::Closed)), - "a second end is refused without touching the first" + "a second end is refused" ); + assert_eq!(producer.final_sequence(), Some(3), "the first end stands"); for sequence in [0, 1] { let _low = producer.create_group(group::Info { sequence }).unwrap(); From 41f98d3918fc236326915bfb98db01a6ab88574b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 21:54:06 -0700 Subject: [PATCH 10/14] test(interop): publish the JS tail frames untimed The tail track has no timescale, so #4968 refuses a timestamped frame on it. Co-Authored-By: Claude Opus 5.5 --- test/interop/clients/js-native/tail.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/test/interop/clients/js-native/tail.ts b/test/interop/clients/js-native/tail.ts index a15159a73d8..0a2cd33c8eb 100644 --- a/test/interop/clients/js-native/tail.ts +++ b/test/interop/clients/js-native/tail.ts @@ -29,10 +29,7 @@ try { track.finishAt(GROUPS); for (let sequence = 0; sequence < GROUPS; sequence++) { const group = track.appendGroup(); - group.writeFrame({ - payload: new Uint8Array(BYTES).fill(sequence), - timestamp: Moq.Time.Timestamp.fromMillis(0), - }); + group.writeFrame({ payload: new Uint8Array(BYTES).fill(sequence) }); group.close(); } track.close(); From 7538bf299f8b44a62dd47821a9425850cbcccdc7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:20:25 -0700 Subject: [PATCH 11/14] fix(net): an abort ends the wait for a pending tail The pending-tail hole count judges by the cache, so an evicted group or a datagram reads as a hole. An abort during the tail window then counted the end as unsettled, and a reader that had every group got the abort instead of a clean end. An abort now ends the wait: whatever reached the end and finished is a clean end, as before the tail hold. The finite_relay reader reports a failed head instead of a head timeout. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 42 ++++++++++++++++++++++++++++---- rs/moq-net/tests/finite_relay.rs | 26 +++++++++++++------- 2 files changed, 54 insertions(+), 14 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 2b8382adfe1..fb47c37d025 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1353,17 +1353,26 @@ impl TrackState { // `sealed` also ends a locally closed receive track without a declared end. // An abort still wins unless that end had already settled: a group below it was still open. let reached = self.sealed - || self.final_sequence.is_some_and(|fin| { - self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin && !self.awaits_tail(fin) - }); + || self + .final_sequence + .is_some_and(|fin| self.edge() >= fin && !self.awaits_tail(fin)); reached && (self.abort.is_none() || self.settled) } + /// One past the highest sequence produced: where the live edge stands against the end. + fn edge(&self) -> u64 { + self.max_sequence.map_or(0, |max| max.saturating_add(1)) + } + /// Whether a group below `fin` the wire subscription still owes has yet to arrive: /// a sequence from where the feed starts up to the end with nothing cached. Only /// such a hole holds readers, so a tail with nothing missing ends them at once. + /// + /// Judged by the cache, so an evicted group or a datagram also reads as a hole until + /// the session settles the tail. That only delays the end: an abort stops the wait. fn awaits_tail(&self, fin: u64) -> bool { - if !self.tail_pending { + // Nothing arrives after an abort, so a hole left then is final. + if !self.tail_pending || self.abort.is_some() { return false; } // Without a declared start, the lowest group that arrived stands in for it. @@ -1381,11 +1390,17 @@ impl TrackState { /// Whether the declared end is reached and every cached group below it finished, /// so nothing the end promised is still in flight. + /// + /// Decided as an abort lands, which ends any wait for the tail: a hole the cache + /// cannot tell from an evicted group does not turn a delivered track into a failure. fn is_settled(&self) -> bool { let Some(fin) = self.final_sequence else { return false; }; - self.is_complete() && self.lookup.range(..fin).all(|(_, slot)| slot.group.is_finished()) + let reached = self.sealed || self.edge() >= fin; + reached + && (self.abort.is_none() || self.settled) + && self.lookup.range(..fin).all(|(_, slot)| slot.group.is_finished()) } /// Where a replacement route should pick this track up: one past the last frame @@ -7663,6 +7678,23 @@ mod test { assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); } + /// An abort ends the wait for a pending tail. Whatever reached the end finished, so + /// readers end cleanly: the cache cannot tell a hole from a group it already evicted. + #[moq_net_sim::test] + async fn an_abort_ends_a_pending_tail_cleanly() { + let mut producer = track_producer("test", None); + let mut arrival = producer.subscribe(None); + producer.start_at(0).unwrap(); + producer.finish_at_pending(2).unwrap(); + let mut high = producer.create_group(group::Info { sequence: 1 }).unwrap(); + high.finish().unwrap(); + assert_eq!(arrival.assert_group().sequence, 1); + assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); + + producer.abort(Error::Dropped).unwrap(); + assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); + } + /// The wire's end can land after the highest group, with lower groups still owed. A /// reader already past that group stays held at the hole: the end never shows without /// its pending tail. diff --git a/rs/moq-net/tests/finite_relay.rs b/rs/moq-net/tests/finite_relay.rs index 67a20576bc0..c42cbf20033 100644 --- a/rs/moq-net/tests/finite_relay.rs +++ b/rs/moq-net/tests/finite_relay.rs @@ -10,6 +10,7 @@ mod support; use std::time::Duration; +use futures::channel::mpsc::TryRecvError; use futures::{SinkExt, StreamExt}; use moq_net::track::Subscription; @@ -94,7 +95,11 @@ async fn round(name: &str, arrival: Arrival) -> (Vec, Result<(), moq_net::E Ok(None) => break Ok(()), Err(err) => break Err(err), }; - let head = group.read_frame().await.unwrap().expect("head frame"); + // A failed head ends the read with its error, which the main task reports. + let head = match group.read_frame().await { + Ok(head) => head.expect("head frame"), + Err(err) => break Err(err), + }; assert_eq!(&head.payload[..], b"head"); heads.send(group.sequence).await.unwrap(); tails.push(moq_net_sim::spawn(async move { @@ -128,12 +133,13 @@ async fn round(name: &str, arrival: Arrival) -> (Vec, Result<(), moq_net::E group.write_frame(Timestamp::ZERO, &b"head"[..]).unwrap(); groups.push(group); // IETF requests the live edge, so observe each header before advancing it. - assert_eq!( - moq_net_sim::timeout(TIMEOUT, opened.next()) - .await - .expect("head timeout"), - Some(sequence), - ); + match moq_net_sim::timeout(TIMEOUT, opened.next()) + .await + .expect("head timeout") + { + Some(head) => assert_eq!(head, sequence, "{name}"), + None => panic!("{name}: the reader stopped: {:?}", reader.await), + } } // Simulated time advances only once every task is idle: each group has reached @@ -164,8 +170,10 @@ async fn round(name: &str, arrival: Arrival) -> (Vec, Result<(), moq_net::E let newest = loop { let held = upstream.server_transport.release_newest_uni(); moq_net_sim::sleep(Duration::from_millis(10)).await; - if let Ok(head) = opened.try_recv() { - break head; + match opened.try_recv() { + Ok(head) => break head, + Err(TryRecvError::Closed) => panic!("{name}: the reader stopped: {:?}", reader.await), + Err(TryRecvError::Empty) => {} } assert!(held > 0, "{name}: no group reached the subscriber"); }; From e7e3c91fd03f2da9566d81293441bbf93728276f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:55:13 -0700 Subject: [PATCH 12/14] test(net): drop an unneeded mut Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 69dee268060..23df935bcc9 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -7805,7 +7805,7 @@ mod test { let mut arrival = producer.subscribe(None); producer.start_at(0).unwrap(); producer.finish_at_pending(2).unwrap(); - let mut high = producer.create_group(group::Info { sequence: 1 }).unwrap(); + let high = producer.create_group(group::Info { sequence: 1 }).unwrap(); high.finish().unwrap(); assert_eq!(arrival.assert_group().sequence, 1); assert!(arrival.recv_group().now_or_never().is_none(), "held at the hole"); From 5726689722eb345c8b4ae78b8fb9a3e44af62a81 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 09:24:04 -0700 Subject: [PATCH 13/14] fix(net): fetched backfill does not fill a pending tail; tail lanes skip wire-compat A fetched copy of a group the live feed still owes never reaches an arrival-order reader, so it no longer counts toward the pending tail. The finite-tail lanes build their clients from this checkout, so a run that swaps in released relays or clients (the wire-compat lanes) skips them. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 32 ++++++++++++++++++++++++++++---- test/interop/README.md | 2 +- test/interop/interop.sh | 16 ++++++++-------- 3 files changed, 37 insertions(+), 13 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 6ea0c04e916..7e7d047309a 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1432,17 +1432,19 @@ impl TrackState { if !self.tail_pending || self.abort.is_some() { return false; } - // Without a declared start, the lowest group that arrived stands in for it. + // Fetched backfill never reaches a reader in arrival order, so only the live feed's + // groups count. Without a declared start, the lowest one stands in for it. let floor = match self.start_sequence { Some(start) => start, None if self.start_pending => return true, - None => match self.lookup.keys().next() { - Some(&first) => first, + None => match self.lookup.iter().find(|(_, slot)| slot.visible) { + Some((&first, _)) => first, None => return fin > 0, }, } .min(fin); - (self.lookup.range(floor..fin).count() as u64) < fin - floor + let arrived = self.lookup.range(floor..fin).filter(|(_, slot)| slot.visible).count(); + (arrived as u64) < fin - floor } /// Whether the declared end is reached and every cached group below it finished, @@ -7849,6 +7851,28 @@ mod test { assert!(arrival.recv_group().now_or_never().unwrap().unwrap().is_none()); } + /// A fetched copy of a group the live feed still owes never reaches an arrival-order + /// reader, so it does not fill the hole: the reader waits for the live group. + #[moq_net_sim::test] + async fn a_fetched_group_does_not_fill_a_pending_tail() { + let mut producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + let mut arrival = producer.subscribe(None); + producer.start_at(0).unwrap(); + producer.finish_at_pending(2).unwrap(); + let _high = producer.create_group(group::Info { sequence: 1 }).unwrap(); + assert_eq!(arrival.assert_group().sequence, 1); + + let _pending = consumer.fetch_group(0, None); + let req = dynamic.requested_group().now_or_never().unwrap().unwrap(); + req.accept(None).unwrap().finish().unwrap(); + assert!( + arrival.recv_group().now_or_never().is_none(), + "the live group 0 is still owed" + ); + } + /// An abort ends the wait for a pending tail. Whatever reached the end finished, so /// readers end cleanly: the cache cannot tell a hole from a group it already evicted. #[moq_net_sim::test] diff --git a/test/interop/README.md b/test/interop/README.md index 11488f8a07a..373c0f2984a 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -23,7 +23,7 @@ The normal run also checks a finite raw track through the relay. Its publisher announces end 4 before writing four 256 KiB groups; the reader verifies every byte, exactly groups 0 through 3, and a clean end at 4. Rust-to-Rust always runs. Selecting native JS adds both Rust-to-JS and JS-to-Rust under that runtime, so `--all` covers -Node and Bun. `--tail` runs just these five lanes. +Node and Bun. `--tail` runs just these five lanes. The wire-compat lanes, which swap in released relays and clients, skip them. The finite clients keep their session alive until the harness acknowledges the reader's complete clean end over stdin, so a lane tests delivery rather than diff --git a/test/interop/interop.sh b/test/interop/interop.sh index ee3adeb8c04..340795e66bd 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -30,13 +30,13 @@ RERUN="$(harness_env INTEROP_TIMEOUT INTEROP_FPS INTEROP_SIZE INTEROP_PORT INTER PUBLISHERS="rust" SUBSCRIBERS="rust" -# The idle-out check guards this checkout's relay and clients. A run that swaps in other -# binaries or JS clients (the wire-compat lanes run released ones) can't be held to it: a -# released client may not close cleanly, and a released relay names no connection when one -# closes. -IDLE_CHECK=1 +# The idle-out check and the finite-tail lanes guard this checkout's relay and clients. A +# run that swaps in other binaries or JS clients (the wire-compat lanes run released ones) +# can't be held to them: a released client may not close cleanly, a released relay names no +# connection when one closes, and the tail clients are built from this checkout only. +IN_TREE=1 if [[ -n "${RELAY_BIN:-}${MOQ_BIN:-}${INTEROP_SUB_MOQ:-}${INTEROP_NATIVE_CLIENT:-}${INTEROP_JS_PUBLISH_CLIENT:-}" ]]; then - IDLE_CHECK=0 + IN_TREE=0 fi TIMEOUT="${INTEROP_TIMEOUT:-20}" FPS="${INTEROP_FPS:-30}" @@ -865,7 +865,7 @@ run_round() { stop_publisher "$pub_pid" "$pub" fi # The negative control's subscribers end by timing out, which is its point. - [[ "$NEGATIVE" -eq 1 || "$IDLE_CHECK" -eq 0 ]] || check_idle_outs "$pub" "$from" + [[ "$NEGATIVE" -eq 1 || "$IN_TREE" -eq 0 ]] || check_idle_outs "$pub" "$from" return 0 } @@ -1007,7 +1007,7 @@ elif [[ "$TAIL_ONLY" -eq 0 ]]; then fi fi -if [[ "$NEGATIVE" -eq 0 && "$MEDIA" -eq 0 ]]; then +if [[ "$NEGATIVE" -eq 0 && "$MEDIA" -eq 0 && "$IN_TREE" -eq 1 ]]; then echo "=== finite track tails ===" run_tail_pair rust rust for runtime in js-native-node js-native-bun; do From 3ab018e1da9c94d12be0b097a0d03bb59ab3edfa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 09:45:23 -0700 Subject: [PATCH 14/14] test(interop): refuse --tail with released binaries or clients A tail-only run with overrides skipped every lane and reported success. Co-Authored-By: Claude Opus 5.5 --- test/interop/interop.sh | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 340795e66bd..06b0bc9611a 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -126,6 +126,10 @@ if [[ "$TAIL_ONLY" -eq 1 ]]; then echo "error: --tail, --media, and --negative are separate runs" >&2 exit 2 fi + if [[ "$IN_TREE" -eq 0 ]]; then + echo "error: --tail builds its clients from this checkout; drop the binary and client overrides" >&2 + exit 2 + fi PUBLISHERS="rust,js-native-node,js-native-bun" SUBSCRIBERS="$PUBLISHERS" fi