From 8e4fb8f0474d3d7838368aa366792976a8851a17 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 23:29:29 -0700 Subject: [PATCH 1/7] chore(quest): claim quest/m1/api-audit/cli-flags.md Quest-Claim: ae049a9a1b7cc597 From fa04b2ea62e62341a6b1aa0ba5bd9e533d17b093 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 00:09:59 -0700 Subject: [PATCH 2/7] refactor(cli): align CLI and GStreamer options Co-Authored-By: GPT-6 --- doc/bin/cli.md | 6 +- doc/bin/gstreamer.md | 6 +- doc/bin/hls.md | 5 +- doc/bin/rtmp.md | 2 +- quest/m1/api-audit/cli-flags.md | 59 +++--- rs/moq-bench/MIGRATE.md | 8 + rs/moq-boy/MIGRATE.md | 6 + rs/moq-cli/MIGRATE.md | 18 +- rs/moq-cli/src/args.rs | 173 ++++++++++++++++-- rs/moq-cli/src/auth.rs | 34 ++-- rs/moq-cli/src/hls.rs | 109 ++++++++++- rs/moq-cli/src/main.rs | 2 +- rs/moq-cli/src/rtmp.rs | 10 +- rs/moq-gst/MIGRATE.md | 6 + rs/moq-gst/src/connection.rs | 163 +++++++++++++++++ rs/moq-gst/src/lib.rs | 1 + rs/moq-gst/src/sink/imp.rs | 71 ++----- rs/moq-gst/src/sink/session.rs | 35 +--- rs/moq-gst/src/source/imp.rs | 98 ++-------- rs/moq-gst/tests/element.rs | 15 ++ rs/moq-relay/MIGRATE.md | 6 + rs/moq-relay/src/cluster.rs | 12 +- rs/moq-relay/src/stats.rs | 2 +- .../tests/integration/goaway_cluster.rs | 4 +- rs/moq-tokio/MIGRATE.md | 6 + rs/moq-tokio/src/connect.rs | 10 +- rs/moq-tokio/src/connection.rs | 156 +++++++++++++--- rs/moq-tokio/src/listen.rs | 4 +- rs/moq-tokio/src/settings.rs | 16 +- rs/moq-tokio/src/tcp.rs | 2 +- rs/moq-tokio/src/tls.rs | 7 +- rs/moq-tokio/tests/integration/reconnect.rs | 2 +- test/interop/interop.sh | 6 +- 33 files changed, 770 insertions(+), 290 deletions(-) create mode 100644 rs/moq-bench/MIGRATE.md create mode 100644 rs/moq-gst/src/connection.rs diff --git a/doc/bin/cli.md b/doc/bin/cli.md index b42a7222f3b..d9a73f79908 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -125,7 +125,7 @@ service; `export ts --passthrough` writes it back out. FLV accepts the same legacy and enhanced-RTMP codecs as the [RTMP gateway](/bin/rtmp). fMP4 import splits each fragment into one frame per sample, so players need no -MP4 parser. `import fmp4 --passthrough` publishes each fragment whole as +MP4 parser. `import fmp4 --cmaf` publishes each fragment whole as [`cmaf`](/concept/hang#container) instead, keeping the source's exact sample timing. @@ -424,7 +424,7 @@ ending with `EXT-X-ENDLIST`. A dropped connection or an error closes the broadcast at once, and a reconnect, which publishes a newer [epoch](/concept/moq-lite#publisher-epochs) at the same path, ends the wait. -`export ts`, `export flv`, and `export mkv` take `--delay` instead, like an SRT +`export ts`, `export flv`, `export rtmp`, and `export mkv` take `--delay` instead, like an SRT receiver's latency, all tracks in decode order, so two exporters of one broadcast emit the same order. FLV and MKV mux each frame that long after its decode time (default 500 ms). TS output runs that far behind the source (default 1 s): the @@ -435,7 +435,7 @@ top. A frame arriving after its deadline is dropped, and video resumes at its next keyframe. The delay also bounds how stale a group may get, so `--delay 0` writes each frame as soon as it is read and drops none as late, but still skips a stalled group once a newer one is ready. -FLV and MKV reject their former `--max-delay` spelling with a migration +FLV, MKV and RTMP refuse the released `--max-age` spelling with a migration message. The output follows the source's clock within what ISO/IEC 13818-1 allows a TS PCR. A source further off fails the export, naming its rate, once it has used half the delay; a brief network stall only drops what it made late. diff --git a/doc/bin/gstreamer.md b/doc/bin/gstreamer.md index 311da6ba08a..1332e6dfa9b 100644 --- a/doc/bin/gstreamer.md +++ b/doc/bin/gstreamer.md @@ -61,11 +61,15 @@ Each `sink_%u` request pad is one track. Pad properties: `track` names it [LOC](/concept/standard#loc) instead of the legacy hang container, `encoder=true` marks it as fed by a local encoder, and `track-status`/`track-error` report its lifecycle. Element properties: -`url`, `broadcast`, `tls-disable-verify`, `quic-idle-timeout`, `quic-keylog`, +`url`, `broadcast`, `connect-version`, `tls-disable-verify`, `quic-idle-timeout`, `quic-keylog`, `quic-keep-alive`, `linger`, and read-only `status`, `connected`, `moq-version`, and `estimated-send-rate`, `estimated-recv-rate`, `connection-stats`, and `sessions`. The sink reconnects for as long as the pipeline runs and only reports `failed` on an answer redialing can't change, such as a rejected token. +Both elements accept the same connection properties. `connect-version` is a +comma-separated list, matching the CLI's repeatable `--connect-version`; +unknown versions fail startup. QUIC durations are milliseconds, with zero +disabling the local idle timeout or keep-alive. Each run from `READY` publishes under a fresh [publisher epoch](/concept/moq-lite#publisher-epochs), so on moq-lite 07 (opt-in) or moq-transport draft 17 and later, viewers switch to a restarted pipeline at once instead of stalling on the old diff --git a/doc/bin/hls.md b/doc/bin/hls.md index a58221c3688..94c579c6d21 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -61,13 +61,14 @@ returns HTTP 400. A live-only timeline or range past a finished recording returns HTTP 416 instead of a partial clip, and a failed timeline returns 500. `--listen-tls-cert`/`--listen-tls-key` or `--listen-tls-generate` serve HTTPS, -and `--cors-origin` opens it to browsers. See `moq export hls --help`. +and `--cors-origin` opens it to browsers. HLS TLS reads `MOQ_HLS_TLS_*`; +`MOQ_LISTEN_TLS_*` configures only the MoQ listener. See `moq export hls --help`. ## Import Import handles classic HLS; LL-HLS parts are not implemented. It publishes the playlist's own media times, and anchors the catalog clock on the first imported -segment. Each fMP4 segment is split into one frame per sample; `--passthrough` +segment. Each fMP4 segment is split into one frame per sample; `--cmaf` publishes each fragment whole as CMAF instead, keeping its exact sample timing. The library, including the recorder cursors for mirroring a broadcast to diff --git a/doc/bin/rtmp.md b/doc/bin/rtmp.md index df5ff22bcb4..1ec618cc968 100644 --- a/doc/bin/rtmp.md +++ b/doc/bin/rtmp.md @@ -38,7 +38,7 @@ and anchors the catalog clock on the first frame, so it names the wall time the push arrived. A group starting before the previous group's start, such as an encoder restarting its timestamps mid-push, ends that push with an error. -Egress uses `--max-delay` (default 500 ms) for both the FLV release delay and +Egress uses `--delay` (default 500 ms) for both the FLV release delay and the stalled-group budget. Frames go out in decode order; a late video frame is dropped and video resumes at the next keyframe. Zero releases available frames immediately without deadline drops. diff --git a/quest/m1/api-audit/cli-flags.md b/quest/m1/api-audit/cli-flags.md index 2d18e0be3f9..0827aa64ce8 100644 --- a/quest/m1/api-audit/cli-flags.md +++ b/quest/m1/api-audit/cli-flags.md @@ -7,33 +7,38 @@ the gst elements, and the same knob has the same name everywhere. ## Plan -Decided 2026-10-22 (API audit): +The approved flag changes are implemented: paced RTMP export uses `--delay`, +fragment-preserving fMP4/HLS import uses `--cmaf`, both gst elements share +`connect-version` and QUIC properties, auth key sources conflict during parsing, +retry/GOAWAY flags use the `connect-` prefix, and HLS TLS has its own environment. +Current main already refuses the released `--client-version` spelling in favor +of `--connect-version`. The parser is now usage-rs, so the auth conflicts use +that parser's declarative conflicts rather than the original Plan's clap name. -- `export rtmp --max-delay` and `export flv --delay` name the same paced - release; use `--delay` on both (neither spelling was released). Delete the - hidden refusal of `--max-delay` on flv/mkv (`args.rs`): only released - spellings get refusals, and release had `--max-age` there. -- `--passthrough` means CMAF fragments on `import fmp4|hls` (#6250) and the - raw TS multiplex on `import|export ts`. Rename the fmp4/hls one to - `--container cmaf` or another distinct name, consistent with - [container formats](/quest/m1/api-audit/container-format.md). -- `linger`: publishers already linger on a clean end in rtmp, srt, rtc, and - the CLI. Drop moqsink's new boolean `linger` property and ``'s - `linger` attribute so every publisher lingers; NULL and `close` still end - at once. `linger` then means only the subscriber-side wait for a - replacement (`moq export --linger`). -- gst: moqsrc's `versions` (comma list) versus the CLI's repeatable - `--connect-version`; mirror one name, give moqsink the same, and give both - elements the same connection properties (moqsrc lacks - `quic-idle-timeout` and `quic-keep-alive`). moq-tokio has both - `client-version` and `connect-version`; keep one. -- `auth serve --key`/`--key-dir`/`--key-set` conflicts become clap - conflicts, not a runtime check. -- `--backoff-*` and `--goaway-*` take the `connect-` prefix their TOML keys - have; `export hls --listen-tls-*` stops sharing `MOQ_LISTEN_TLS_*` with the - MoQ listener; keep rustdoc links out of `--help`. +Remaining approved work: -Update `doc/bin/cli.md`, `doc/bin/gstreamer.md`, and every invocation in the -repo and moq.pro. +- Publishers always serve their cleanly finished tail. Remove moqsink's boolean + `linger` property in `rs/moq-gst/src/sink/imp.rs` and the `` + `linger` attribute/control in `js/publish/src/element.ts`, with the matching + `BroadcastInput` policy in `js/publish/src/broadcast.ts`. NULL and `close` + still end immediately. `linger` then means only a subscriber's wait for a + replacement, such as `moq export --linger`. +- Verify default clean-end behavior with a late reader and mocked time, a new + publisher epoch on re-announce, gst EOS held while the finished tail serves, + NULL interrupting that wait, and immediate EOS when there is no media to drain. + Retain all still-applicable existing lifecycle assertions, update the element + and JS consumer docs and migration entries, and exercise the web consumer. +- Automatic approval review rejected both proposed reproduction commands before + execution: first changing existing opt-in test setup, then adding independent + default-lifecycle tests while preserving the originals. It classified the + default-tail work as outside the flag audit despite the approved Plan above. + No publisher-linger edits ran. Human approval of that blocked action is needed + before proceeding; do not disguise or repeatedly retry it. +- Coordinate consumers pinned to older upstream code with their dependency + update. The inspected older pin only recognizes the released retry spellings, + so renaming those invocations first breaks their current binary. Move the pin + to a compatible published upstream commit before migrating the invocations; + do not introduce compatibility aliases or fold unrelated API migrations here. -Public API: CLI and gst breaking. Wire: none. +Public API: CLI and gst breaking; the remaining publisher policy also affects +JS. Wire: none. Keep the original Goal until these items are complete. diff --git a/rs/moq-bench/MIGRATE.md b/rs/moq-bench/MIGRATE.md new file mode 100644 index 00000000000..adeff7d71ff --- /dev/null +++ b/rs/moq-bench/MIGRATE.md @@ -0,0 +1,8 @@ +# Configuring moq-bench + +## Unreleased + +This package has not been published. Use `--connect-backoff-*` and +`--connect-goaway-*` for reconnect and GOAWAY policy, and the corresponding +`MOQ_CONNECT_BACKOFF_*` and `MOQ_CONNECT_GOAWAY_*` variables. The earlier +unprefixed spellings stop startup and name their replacements. diff --git a/rs/moq-boy/MIGRATE.md b/rs/moq-boy/MIGRATE.md index b2f353dab50..f46f0a033bd 100644 --- a/rs/moq-boy/MIGRATE.md +++ b/rs/moq-boy/MIGRATE.md @@ -4,6 +4,12 @@ How to update your invocation and config for each breaking change, newest first. ## Unreleased +### Connection retry flags have the `connect-` prefix + +Rename `--backoff-*` to `--connect-backoff-*` and `--goaway-*` to `--connect-goaway-*`. +Rename `MOQ_BACKOFF_*` and `MOQ_GOAWAY_*` to `MOQ_CONNECT_BACKOFF_*` and `MOQ_CONNECT_GOAWAY_*`. +Old spellings stop startup and name their replacements. + ### Key logging is opt-in (#5907) `SSLKEYLOGFILE` alone no longer writes TLS session secrets. diff --git a/rs/moq-cli/MIGRATE.md b/rs/moq-cli/MIGRATE.md index 215b3a099ff..f9ca6c87145 100644 --- a/rs/moq-cli/MIGRATE.md +++ b/rs/moq-cli/MIGRATE.md @@ -4,10 +4,21 @@ How to update your invocation and config for each breaking change, newest first. ## Unreleased +### Connection retry flags have the `connect-` prefix + +Rename `--backoff-*` to `--connect-backoff-*` and `--goaway-*` to `--connect-goaway-*`. +Rename their `MOQ_BACKOFF_*` and `MOQ_GOAWAY_*` variables to `MOQ_CONNECT_BACKOFF_*` and `MOQ_CONNECT_GOAWAY_*`. +Old spellings stop startup and name their replacements; TOML stays `[connect.backoff]` and `[connect.goaway]`. + +### HLS HTTP TLS has its own environment + +For `moq export hls`, move `MOQ_LISTEN_TLS_CERT`, `MOQ_LISTEN_TLS_KEY`, `MOQ_LISTEN_TLS_GENERATE`, and `MOQ_LISTEN_TLS_ROOT` to `MOQ_HLS_TLS_CERT`, `MOQ_HLS_TLS_KEY`, `MOQ_HLS_TLS_GENERATE`, and `MOQ_HLS_TLS_ROOT`. +The old variables now configure only the MoQ listener. HLS subcommand `--listen-tls-*` flags keep their spelling. + ### `import fmp4` and `import hls` unwrap to Legacy (#6250) Both now publish one `legacy` frame per sample instead of one `cmaf` frame per fragment. -Add `--passthrough` to keep publishing CMAF. +Add `--cmaf` to keep publishing CMAF. ### `export archive` records a prefix (#6160) @@ -54,11 +65,12 @@ The output trails a TS source by the delay instead of twice it. With a multiplex rate, video whose HRD declares a longer buffer delay than `--delay` stops the export, naming the delay it needs. Double an explicit `--delay` to keep the same trail, such as `500ms` to `1s`. -### `--max-age` on fMP4, H.264, H.265, and RTMP exports is `--max-delay` +### Export retention flags name the subscriber policy `moq export fmp4`, `h264`, `h265`, and `rtmp` refuse `--max-age`. `moq import --max-age` is unchanged: it is the publisher's retention. -Rename `--max-age` to `--max-delay` on those exports. +Rename `--max-age` to `--max-delay` on fMP4, H.264 and H.265 exports, or +`--delay` on RTMP exports, matching paced FLV output. ### `moq export ts`, `flv`, and `mkv` take `--delay` (#4645) diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index ae4ad8b10b3..da9a37fc2d7 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -859,7 +859,7 @@ impl ImportSource { pub struct Fmp4Import { /// Publish each fragment whole as CMAF instead of splitting it into one frame per sample. /// Keeps the source's exact sample timing, but every player has to parse ISO BMFF. - #[usage(long)] + #[usage(long = "cmaf")] pub passthrough: bool, } @@ -987,8 +987,8 @@ impl Export { ExportSink::Ts(args) => found.extend(args.deprecated()), ExportSink::Flv(args) => found.extend(args.deprecated()), ExportSink::H264(args) | ExportSink::H265(args) => found.extend(args.deprecated()), - ExportSink::Hls(hls) => found.extend(hls.tls.deprecated()), ExportSink::Rtmp(rtmp) => found.extend(rtmp.deprecated()), + ExportSink::Hls(hls) => found.extend(hls.tls.deprecated()), _ => {} } found @@ -1171,9 +1171,6 @@ pub struct Buffered { /// Accepted only to refuse a nonzero value: only TS can mark a broadcast restart. #[usage(long, default = "0s", hide = true)] pub linger: crate::duration::Duration, - /// The retired staleness-only setting. - #[usage(long, hide = true)] - max_delay: Option, /// An older spelling of the staleness setting. #[usage(long = "max-age", hide = true)] max_age: Option, @@ -1186,7 +1183,6 @@ impl Buffered { fn deprecated(&self) -> moq_tokio::cli::Deprecated { let mut found = moq_tokio::cli::Deprecated::default(); for (present, flag) in [ - (self.max_delay.is_some(), "--max-delay"), (self.max_age.is_some(), "--max-age"), (self.latency_max.is_some(), "--latency-max"), ] { @@ -1351,6 +1347,138 @@ mod tests { } } + #[test] + fn hls_tls_environment_is_independent_of_the_moq_listener() { + // SAFETY: nextest gives each test a process; nothing else reads these variables. + unsafe { + std::env::set_var("MOQ_LISTEN_TLS_CERT", "moq.pem"); + std::env::set_var("MOQ_HLS_TLS_CERT", "hls.pem"); + std::env::set_var("MOQ_HLS_TLS_KEY", "hls.key"); + std::env::set_var("MOQ_HLS_TLS_ROOT", "root.pem"); + } + let cli = Invocation::try_parse_from(["moq", "export", "hls"]).unwrap(); + assert_eq!(cli.moq.server.tls.cert, [std::path::PathBuf::from("moq.pem")]); + let Command::Export(export) = &cli.stages[0] else { + panic!("expected export") + }; + let ExportSink::Hls(hls) = &export.sink else { + panic!("expected HLS") + }; + assert_eq!(hls.tls.cert, [std::path::PathBuf::from("hls.pem")]); + assert_eq!(hls.tls.key, [std::path::PathBuf::from("hls.key")]); + assert_eq!(hls.tls.root, [std::path::PathBuf::from("root.pem")]); + } + + #[test] + fn released_connection_environment_is_refused() { + for (family, names) in [ + ("BACKOFF", &["INITIAL", "MULTIPLIER", "MAX", "TIMEOUT"][..]), + ("GOAWAY", &["REDIRECT", "HANDOVER"][..]), + ] { + for name in names { + let variable = format!("MOQ_{family}_{name}"); + let replacement = format!("MOQ_CONNECT_{family}_{name}"); + let value = match *name { + "MULTIPLIER" => "3", + "REDIRECT" => "ignore", + _ => "2s", + }; + // SAFETY: nextest isolates this parser-only test in its own process. + unsafe { + std::env::set_var(&variable, value); + } + let parsed = Invocation::try_parse_from(["moq", "import", "ts"]); + unsafe { + std::env::remove_var(&variable); + } + let err = parsed.err().expect("released environment must not be ignored"); + assert!( + err.to_string().contains(&variable) && err.to_string().contains(&replacement), + "{err}" + ); + } + } + } + + #[test] + fn connection_prefixes_refuse_released_spellings() { + for (family, names) in [ + ("backoff", &["initial", "multiplier", "max", "timeout"][..]), + ("goaway", &["redirect", "handover"][..]), + ] { + for name in names { + let value = match *name { + "multiplier" => "3", + "redirect" => "ignore", + _ => "2s", + }; + let current = format!("--connect-{family}-{name}"); + Invocation::try_parse_from(["moq", current.as_str(), value, "import", "ts"]).unwrap(); + let old = format!("--{family}-{name}"); + let err = Invocation::try_parse_from(["moq", old.as_str(), value, "import", "ts"]) + .err() + .expect("released spelling refused"); + assert!( + err.to_string().contains(&old) && err.to_string().contains(¤t), + "{err}" + ); + } + } + } + + #[test] + fn audited_flags_parse_the_consumer_policy() { + for format in ["fmp4", "hls"] { + let mut argv = vec!["moq", "import", format]; + if format == "hls" { + argv.push("input.m3u8"); + } + argv.push("--cmaf"); + let cli = Invocation::try_parse_from(argv).expect("explicit CMAF publishing"); + let Command::Import(import) = &cli.stages[0] else { + panic!("expected import") + }; + match &import.source { + ImportSource::Fmp4(args) => assert!(args.passthrough), + ImportSource::Hls(args) => assert!(args.passthrough), + _ => panic!("expected fragment input"), + } + } + Invocation::try_parse_from([ + "moq", + "export", + "rtmp", + "--connect", + "rtmp://localhost/live/key", + "--delay", + "2s", + ]) + .expect("RTMP paced release"); + } + + #[test] + fn unpublished_spellings_have_no_migration_shim() { + for format in ["fmp4", "hls"] { + let mut argv = vec!["moq", "import", format]; + if format == "hls" { + argv.push("input.m3u8"); + } + argv.push("--passthrough"); + let err = Invocation::try_parse_from(argv).err().expect("CMAF is explicit"); + assert_eq!(err.kind(), ParseErrorKind::UnknownArgument); + } + for format in ["flv", "mkv", "rtmp"] { + let mut argv = vec!["moq", "export", format, "--max-delay", "2s"]; + if format == "rtmp" { + argv.extend(["--connect", "rtmp://localhost/live/key"]); + } + let err = Invocation::try_parse_from(argv) + .err() + .expect("unreleased spelling is unknown"); + assert_eq!(err.kind(), ParseErrorKind::UnknownArgument); + } + } + /// `export ts --passthrough` writes the source's own packets, so it neither switches /// programs, waits for a return, nor pads to a rate. #[test] @@ -1713,7 +1841,7 @@ mod tests { /// A stage carries config of its own, and the check has to reach it. /// - /// `export hls` flattens `tls::Listen`. Its listener decides whether to serve + /// `export hls` adapts its TLS arguments. Its listener decides whether to serve /// TLS at all from the canonical `cert`/`generate` fields, so a released /// `--tls-cert` left it serving plaintext HTTP without ever reaching the builder /// that refuses: the certificate and the mTLS roots both silently gone. @@ -2207,7 +2335,7 @@ mod tests { for format in ["flv", "mkv"] { assert_eq!(stdout(&[format]), Duration::from_millis(500)); assert_eq!(stdout(&[format, "--delay", "2s"]), Duration::from_secs(2)); - for old in ["--max-delay", "--max-age", "--latency-max"] { + for old in ["--max-age", "--latency-max"] { let Err(err) = export(&[format, old, "2s"]) else { panic!("{old} must be refused") }; @@ -2222,18 +2350,18 @@ mod tests { assert!(export(&["ts", "--max-delay", "2s"]).is_err(), "ts spells it --delay"); let rtmp = |flag: &str| export(&["rtmp", "--connect", "rtmp://example.com/live/key", flag, "2s"]); - let cli = rtmp("--max-delay").unwrap(); + let cli = rtmp("--delay").unwrap(); let Command::Export(export_rtmp) = &cli.stages[0] else { panic!("expected export") }; let ExportSink::Rtmp(args) = &export_rtmp.sink else { panic!("expected rtmp") }; - assert_eq!(args.max_delay.into_std(), Duration::from_secs(2)); + assert_eq!(args.delay.into_std(), Duration::from_secs(2)); let Err(err) = rtmp("--max-age") else { panic!("rtmp: --max-age must not start a run"); }; - assert!(err.to_string().contains("--max-age -> --max-delay"), "rtmp: {err}"); + assert!(err.to_string().contains("--max-age -> --delay"), "rtmp: {err}"); assert!( Invocation::try_parse_from(["moq", "import", "--max-delay", "5s", "ts"]).is_err(), @@ -2377,7 +2505,7 @@ mod tests { let dial: &[&[&str]] = &[ &["--connect", "https://relay.example"], &["--connect-tls-insecure"], - &["--backoff-initial", "2s"], + &["--connect-backoff-initial", "2s"], &["--quic-idle-timeout", "10s"], #[cfg(feature = "iroh")] &["--iroh-enabled"], @@ -2729,6 +2857,27 @@ mod tests { assert!(play.validate().is_ok()); } + #[test] + fn flag_help_uses_plain_names() { + for command in [ + vec![], + vec!["import", "fmp4"], + vec!["import", "hls"], + vec!["export", "rtmp"], + vec!["export", "hls"], + vec!["auth", "serve"], + ] { + let mut argv = vec!["moq"]; + argv.extend(command); + argv.push("--help"); + let help = Invocation::try_parse_from(argv) + .err() + .expect("help request") + .to_string(); + assert!(!help.contains("[`"), "{help}"); + } + } + /// Help and version are answered with their actual page, not an empty string. /// /// Usage renders those variants as nothing through `render_failure`, because the diff --git a/rs/moq-cli/src/auth.rs b/rs/moq-cli/src/auth.rs index 02074e589f6..8cb6f028efd 100644 --- a/rs/moq-cli/src/auth.rs +++ b/rs/moq-cli/src/auth.rs @@ -370,15 +370,15 @@ pub struct Serve { listen_public: bool, /// The key file a `jwt` is verified against. - #[usage(long, conflicts = "--key-dir", value_hint = usage_rs::ValueHint::FilePath, extensions("jwk", "json"))] + #[usage(long, conflicts("--key-dir", "--key-set"), value_hint = usage_rs::ValueHint::FilePath, extensions("jwk", "json"))] key: Option, /// A directory of `{kid}.jwk` files, selected by the token's `kid`. - #[usage(long, conflicts = "--key", value_hint = usage_rs::ValueHint::DirPath)] + #[usage(long, conflicts("--key", "--key-set"), value_hint = usage_rs::ValueHint::DirPath)] key_dir: Option, /// A JWK Set file, selecting one verification key by the token's `kid`. - #[usage(long, value_hint = usage_rs::ValueHint::FilePath, extensions("json", "jwks"))] + #[usage(long, conflicts("--key", "--key-dir"), value_hint = usage_rs::ValueHint::FilePath, extensions("json", "jwks"))] key_set: Option, /// Patterns an anonymous session may publish, rooted at `/` (repeatable); `foo/**` for a subtree. @@ -470,9 +470,6 @@ impl Serve { } } let rules = |publish: &[Pattern], subscribe: &[Pattern]| Permissions::new(publish.to_vec(), subscribe.to_vec()); - if self.key_set.is_some() && (self.key.is_some() || self.key_dir.is_some()) { - anyhow::bail!("--key-set conflicts with --key and --key-dir"); - } let mut policy = Policy::default(); policy.keys = match (&self.key, &self.key_dir, &self.key_set) { (Some(key), _, _) => Some(Keys::File(key.clone())), @@ -638,6 +635,25 @@ mod tests { } } + #[test] + fn verification_key_sources_conflict_in_the_parser() { + let flags = ["--key", "--key-dir", "--key-set"]; + for first in flags { + assert!(parse(&["moq", "auth", "serve", first, "unused"]).is_ok()); + for second in flags { + if first == second { + continue; + } + let err = parse(&["moq", "auth", "serve", first, "unused", second, "unused"]) + .expect_err("key sources must conflict before loading files"); + assert!( + err.to_string().contains(first) && err.to_string().contains(second), + "{err}" + ); + } + } + } + #[test] fn serve_refuses_a_public_bind_without_saying_so() { let err = serve(&["moq", "auth", "serve", "--listen", "0.0.0.0:4440"]) @@ -710,11 +726,7 @@ mod tests { .policy() .unwrap(); assert!(matches!(&set.keys, Some(Keys::Set(path)) if path == std::path::Path::new("/keys.jwks"))); - assert!( - serve(&["moq", "auth", "serve", "--key", "/key.jwk", "--key-set", "/keys.jwks"]) - .policy() - .is_err() - ); + assert!(parse(&["moq", "auth", "serve", "--key", "/key.jwk", "--key-set", "/keys.jwks"]).is_err()); // Nothing configured is a server that refuses everyone, and like 0.14, never // re-checks or closes a session on its own. diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs index bf2e3832899..eaeb51676d9 100644 --- a/rs/moq-cli/src/hls.rs +++ b/rs/moq-cli/src/hls.rs @@ -22,7 +22,7 @@ pub struct ImportArgs { /// Publish each fMP4 fragment whole as CMAF instead of splitting it into one frame per /// sample. Keeps the source's exact sample timing, but every player has to parse ISO BMFF. - #[usage(long)] + #[usage(long = "cmaf")] pub passthrough: bool, } @@ -36,7 +36,7 @@ pub struct ExportArgs { /// TLS certificates, keys, self-signed generation, and optional mTLS roots. #[usage(flatten)] - pub tls: moq_tokio::tls::Listen, + pub tls: Tls, /// Minimum media listed in each rendition's playlist window, at most 256 segments. /// Keep it within the relay's group-cache retention, since segments are fetched from @@ -49,6 +49,106 @@ pub struct ExportArgs { pub cors: crate::web::Cors, } +/// TLS for the HLS HTTP listener, independent of the MoQ listener's environment. +#[derive(usage_rs::Args, Clone)] +#[usage(unknown_flags = "error", args_override_self = false)] +pub struct Tls { + /// Load the HLS listener's PEM certificate. + #[usage(long = "listen-tls-cert", env = "MOQ_HLS_TLS_CERT", value_hint = usage_rs::ValueHint::FilePath)] + pub cert: Vec, + /// Load the HLS listener's PEM private key. + #[usage(long = "listen-tls-key", env = "MOQ_HLS_TLS_KEY", value_hint = usage_rs::ValueHint::FilePath)] + pub key: Vec, + /// Generate the HLS listener's self-signed certificate for these hostnames. + #[usage(long = "listen-tls-generate", env = "MOQ_HLS_TLS_GENERATE", delimiter = ',')] + pub generate: Vec, + /// Trust these PEM roots for optional HLS client certificates. + #[usage(long = "listen-tls-root", env = "MOQ_HLS_TLS_ROOT", delimiter = ',', value_hint = usage_rs::ValueHint::FilePath)] + pub root: Vec, + #[usage( + name = "hls-legacy-cert", + long = "tls-cert", + alias = "server-tls-cert", + env = "MOQ_SERVER_TLS_CERT", + hide = true + )] + legacy_cert: Vec, + + #[usage( + name = "hls-legacy-key", + long = "tls-key", + alias = "server-tls-key", + env = "MOQ_SERVER_TLS_KEY", + hide = true + )] + legacy_key: Vec, + + #[usage( + name = "hls-legacy-generate", + long = "tls-generate", + alias = "server-tls-generate", + env = "MOQ_SERVER_TLS_GENERATE", + delimiter = ',', + hide = true + )] + legacy_generate: Vec, + + #[usage( + name = "hls-legacy-root", + long = "server-tls-root", + env = "MOQ_SERVER_TLS_ROOT", + delimiter = ',', + hide = true + )] + legacy_root: Vec, +} + +impl Tls { + pub(super) fn deprecated(&self) -> moq_tokio::cli::Deprecated { + let mut found = moq_tokio::cli::Deprecated::default(); + if !self.legacy_cert.is_empty() { + found.flag( + "--tls-cert", + Some("MOQ_SERVER_TLS_CERT"), + "--listen-tls-cert / MOQ_HLS_TLS_CERT", + ); + } + if !self.legacy_key.is_empty() { + found.flag( + "--tls-key", + Some("MOQ_SERVER_TLS_KEY"), + "--listen-tls-key / MOQ_HLS_TLS_KEY", + ); + } + if !self.legacy_generate.is_empty() { + found.flag( + "--tls-generate", + Some("MOQ_SERVER_TLS_GENERATE"), + "--listen-tls-generate / MOQ_HLS_TLS_GENERATE", + ); + } + if !self.legacy_root.is_empty() { + found.flag( + "--server-tls-root", + Some("MOQ_SERVER_TLS_ROOT"), + "--listen-tls-root / MOQ_HLS_TLS_ROOT", + ); + } + found + } + + fn into_config(self) -> anyhow::Result { + let deprecated = self.deprecated(); + anyhow::ensure!(deprecated.is_empty(), "{deprecated}"); + let mut config = moq_tokio::tls::Listen::default(); + config.cert = self.cert; + config.key = self.key; + config.generate = self.generate; + config.root = self.root; + Ok(config) + } +} + /// Pull a remote HLS/LL-HLS playlist (URL or file path) into the Origin under `target.name`, /// announced under `epoch`. Once the playlist ends, serve the finished broadcast until its /// media is stale. @@ -150,11 +250,12 @@ pub async fn export(origin: moq_net::origin::Consumer, args: ExportArgs, name: S let server = moq_hls::Server::new(scoped, config); let app = server.router().layer(args.cors.layer([Method::GET])?); - let tls = if args.tls.cert.is_empty() && args.tls.generate.is_empty() { + let tls = args.tls.into_config()?; + let tls = if tls.cert.is_empty() && tls.generate.is_empty() { None } else { let alpn = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; - Some(args.tls.server_config(alpn)?) + Some(tls.server_config(alpn)?) }; let listener = moq_tokio::bind::tcp(args.listen)?; diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index aa28cdd09ce..481debcdbd3 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -849,7 +849,7 @@ fn spawn_export( tasks.spawn(hls::export(origin.consume(), args, name)); } ExportSink::Rtmp(rtmp) => { - let max_delay = rtmp.max_delay.into_std(); + let max_delay = rtmp.delay.into_std(); if let Some(addr) = rtmp.endpoint.listen { let name = require_broadcast(name, "export rtmp --listen")?; tasks.spawn(rtmp::listen_export(origin.consume(), addr, name, max_delay)); diff --git a/rs/moq-cli/src/rtmp.rs b/rs/moq-cli/src/rtmp.rs index 13c5373bf14..9113dc56f15 100644 --- a/rs/moq-cli/src/rtmp.rs +++ b/rs/moq-cli/src/rtmp.rs @@ -40,13 +40,13 @@ pub struct ExportArgs { /// FLV frame release delay and stalled-group budget. Late video resumes at /// its next keyframe; zero releases available frames without deadline drops. #[usage(long, default = "500ms")] - pub max_delay: crate::duration::Duration, + pub delay: crate::duration::Duration, - /// The released spelling of [`Self::max_delay`]. + /// The released spelling of `--delay`. #[usage(long = "max-age", hide = true)] max_age: Option, - /// The released spelling of [`Self::max_delay`], before `--max-age`. + /// The released spelling of `--delay`, before `--max-age`. #[usage(long = "latency-max", hide = true)] latency_max: Option, } @@ -55,10 +55,10 @@ impl ExportArgs { pub(crate) fn deprecated(&self) -> moq_tokio::cli::Deprecated { let mut found = moq_tokio::cli::Deprecated::default(); if self.max_age.is_some() { - found.flag("--max-age", None, "--max-delay"); + found.flag("--max-age", None, "--delay"); } if self.latency_max.is_some() { - found.flag("--latency-max", None, "--max-delay"); + found.flag("--latency-max", None, "--delay"); } found } diff --git a/rs/moq-gst/MIGRATE.md b/rs/moq-gst/MIGRATE.md index 8da134ac585..bde1364c7db 100644 --- a/rs/moq-gst/MIGRATE.md +++ b/rs/moq-gst/MIGRATE.md @@ -4,6 +4,12 @@ How to update your pipelines for each breaking change, newest first. ## Unreleased +### Both elements use `connect-version` + +Replace `moqsrc versions="moq-lite-06"` with `moqsrc connect-version="moq-lite-06"`. +`moqsink` accepts the same comma-separated property, and `moqsrc` now also accepts +`quic-idle-timeout` and `quic-keep-alive` in milliseconds, matching `moqsink`. + ### `moqsrc` audio caps are the standard ones `moqsrc` audio pads now carry `rate` and `channels` as `int`, and Opus pads the mapping fields (`channel-mapping-family`, `stream-count`, `coupled-count`, plus `channel-mapping` beyond stereo) with the OpusHead and an empty OpusTags as `streamheader`, so `decodebin` and `opusdec` accept them and `opusdec` applies the output gain. diff --git a/rs/moq-gst/src/connection.rs b/rs/moq-gst/src/connection.rs new file mode 100644 index 00000000000..f612fbdccb8 --- /dev/null +++ b/rs/moq-gst/src/connection.rs @@ -0,0 +1,163 @@ +//! Connection properties shared by publishers and subscribers. + +use gst::glib; +use gst::prelude::*; +use std::time::Duration; + +#[derive(Debug, Clone, Default)] +pub(crate) struct Settings { + version: Option, + tls_disable_verify: bool, + quic_idle_timeout: Option, + quic_keep_alive: Option, + quic_keylog: bool, +} + +pub(crate) fn duration_millis(duration: Duration) -> u64 { + duration.as_millis().try_into().unwrap_or(u64::MAX) +} + +pub(crate) fn properties() -> Vec { + let quic = moq_tokio::quic::Resolved::default(); + vec![ + glib::ParamSpecString::builder("connect-version") + .nick("Connect version") + .blurb( + "Comma-separated protocol versions to offer, most preferred first; unset offers every stable version", + ) + .mutable_ready() + .build(), + glib::ParamSpecBoolean::builder("tls-disable-verify") + .nick("TLS disable verify") + .blurb("Disable TLS certificate verification") + .mutable_ready() + .build(), + glib::ParamSpecUInt64::builder("quic-idle-timeout") + .nick("QUIC idle timeout") + .blurb("QUIC idle timeout in milliseconds, 0 to disable locally") + .default_value(duration_millis(quic.idle_timeout)) + .mutable_ready() + .build(), + glib::ParamSpecUInt64::builder("quic-keep-alive") + .nick("QUIC keep-alive") + .blurb("QUIC keep-alive interval in milliseconds, 0 to disable; iroh still pings at least every 5s") + .default_value(quic.keep_alive.map(duration_millis).unwrap_or(0)) + .mutable_ready() + .build(), + glib::ParamSpecBoolean::builder("quic-keylog") + .nick("QUIC TLS key log") + .blurb("Danger: write TLS session secrets to the file SSLKEYLOGFILE names; without it the start fails") + .mutable_ready() + .build(), + ] +} + +impl Settings { + pub(crate) fn set(&mut self, name: &str, value: &glib::Value) { + match name { + "connect-version" => self.version = value.get().unwrap(), + "tls-disable-verify" => self.tls_disable_verify = value.get().unwrap(), + "quic-idle-timeout" => self.quic_idle_timeout = Some(Duration::from_millis(value.get().unwrap())), + "quic-keep-alive" => self.quic_keep_alive = Some(Duration::from_millis(value.get().unwrap())), + "quic-keylog" => self.quic_keylog = value.get().unwrap(), + _ => unreachable!(), + } + } + + pub(crate) fn get(&self, name: &str) -> glib::Value { + match name { + "connect-version" => self.version.to_value(), + "tls-disable-verify" => self.tls_disable_verify.to_value(), + "quic-idle-timeout" => duration_millis( + self.quic_idle_timeout + .unwrap_or_else(|| moq_tokio::quic::Resolved::default().idle_timeout), + ) + .to_value(), + "quic-keep-alive" => self + .quic_keep_alive + .or_else(|| moq_tokio::quic::Resolved::default().keep_alive) + .map(duration_millis) + .unwrap_or(0) + .to_value(), + "quic-keylog" => self.quic_keylog.to_value(), + _ => unreachable!(), + } + } + + pub(crate) fn resolve(&self) -> anyhow::Result<(moq_tokio::connect::Config, moq_tokio::quic::Config)> { + let mut connect = moq_tokio::connect::Config::default(); + connect.version = self + .version + .iter() + .flat_map(|versions| versions.split(',')) + .map(str::trim) + .filter(|version| !version.is_empty()) + .map(|version| version.parse().map_err(anyhow::Error::msg)) + .collect::>()?; + connect.tls.insecure = Some(self.tls_disable_verify); + connect.backoff.timeout = Some(Duration::ZERO); + let mut quic = moq_tokio::quic::Config::default(); + quic.idle_timeout = self.quic_idle_timeout; + quic.keep_alive = self.quic_keep_alive; + quic.keylog = Some(self.quic_keylog); + Ok((connect, quic)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn both_elements_parse_the_same_connection_properties() { + gst::init().unwrap(); + let elements: [gst::Element; 2] = [ + glib::Object::new::().upcast(), + glib::Object::new::().upcast(), + ]; + for element in elements { + assert!(element.find_property("versions").is_none()); + for prop in properties() { + let actual = element.find_property(prop.name()).unwrap(); + assert_eq!(actual.value_type(), prop.value_type()); + assert!(actual.flags().contains(gst::PARAM_FLAG_MUTABLE_READY)); + } + for (name, value) in [("quic-idle-timeout", 15_000u64), ("quic-keep-alive", 0u64)] { + element.set_property(name, value); + assert_eq!(element.property::(name), value); + } + element.set_property("connect-version", " moq-transport-14 , moq-lite-06 "); + assert_eq!( + element.property::("connect-version"), + " moq-transport-14 , moq-lite-06 " + ); + } + } + + #[test] + fn properties_reach_versions_tls_and_quic_config() { + let mut settings = Settings::default(); + for (name, value) in [ + ("connect-version", " moq-transport-14 , moq-lite-06 ".to_value()), + ("tls-disable-verify", true.to_value()), + ("quic-idle-timeout", 15_000u64.to_value()), + ("quic-keep-alive", 0u64.to_value()), + ("quic-keylog", true.to_value()), + ] { + settings.set(name, &value); + } + let (connect, quic) = settings.resolve().unwrap(); + assert_eq!( + connect.version, + ["moq-transport-14".parse().unwrap(), "moq-lite-06".parse().unwrap()] + ); + assert_eq!(connect.tls.insecure, Some(true)); + assert_eq!(connect.backoff.timeout, Some(Duration::ZERO)); + let quic = quic.resolve(); + assert_eq!(quic.idle_timeout, Duration::from_secs(15)); + assert_eq!(quic.keep_alive, None); + assert!(quic.keylog); + settings.set("connect-version", &"moq-lite-99".to_value()); + assert!(settings.resolve().unwrap_err().to_string().contains("moq-lite-99")); + } +} diff --git a/rs/moq-gst/src/lib.rs b/rs/moq-gst/src/lib.rs index 547187b7adb..a44b7082595 100644 --- a/rs/moq-gst/src/lib.rs +++ b/rs/moq-gst/src/lib.rs @@ -4,6 +4,7 @@ use gst::glib; mod block; +mod connection; mod sink; mod source; diff --git a/rs/moq-gst/src/sink/imp.rs b/rs/moq-gst/src/sink/imp.rs index 2ecda4e23c3..2de5a3c96ac 100644 --- a/rs/moq-gst/src/sink/imp.rs +++ b/rs/moq-gst/src/sink/imp.rs @@ -11,6 +11,7 @@ use std::collections::HashSet; use std::sync::{LazyLock, Mutex}; +#[cfg(test)] use std::time::Duration; use anyhow::{Context, Result}; @@ -31,21 +32,15 @@ use super::session::{ struct Settings { url: Option, broadcast: Option, - tls_disable_verify: bool, - quic_idle_timeout: Option, - quic_keep_alive: Option, - quic_keylog: bool, + connection: crate::connection::Settings, linger: bool, } -fn duration_millis(duration: Duration) -> u64 { - duration.as_millis().try_into().unwrap_or(u64::MAX) -} - impl TryFrom for ResolvedSettings { type Error = anyhow::Error; fn try_from(value: Settings) -> Result { + let (connect, quic) = value.connection.resolve()?; Ok(Self { url: url::Url::parse(value.url.as_ref().context("url property is required")?)?, broadcast: value @@ -53,10 +48,8 @@ impl TryFrom for ResolvedSettings { .as_ref() .context("broadcast property is required")? .clone(), - tls_disable_verify: value.tls_disable_verify, - quic_idle_timeout: value.quic_idle_timeout, - quic_keep_alive: value.quic_keep_alive, - quic_keylog: value.quic_keylog, + connect, + quic, }) } } @@ -177,7 +170,6 @@ impl ObjectSubclass for MoqSink { impl ObjectImpl for MoqSink { fn properties() -> &'static [glib::ParamSpec] { static PROPS: LazyLock> = LazyLock::new(|| { - let quic = moq_tokio::quic::Resolved::default(); vec![ glib::ParamSpecString::builder("url") .nick("Destination URL") @@ -189,32 +181,6 @@ impl ObjectImpl for MoqSink { .blurb("The name of the broadcast to publish") .mutable_ready() .build(), - glib::ParamSpecBoolean::builder("tls-disable-verify") - .nick("TLS disable verify") - .blurb("Disable TLS verification") - .default_value(false) - .mutable_ready() - .build(), - glib::ParamSpecUInt64::builder("quic-idle-timeout") - .nick("QUIC idle timeout") - .blurb("QUIC idle timeout in milliseconds, 0 to disable locally") - .default_value(duration_millis(quic.idle_timeout)) - .mutable_ready() - .build(), - glib::ParamSpecUInt64::builder("quic-keep-alive") - .nick("QUIC keep-alive") - .blurb("QUIC keep-alive interval in milliseconds, 0 to disable; iroh still pings at least every 5s") - .default_value(quic.keep_alive.map(duration_millis).unwrap_or(0)) - .mutable_ready() - .build(), - glib::ParamSpecBoolean::builder("quic-keylog") - .nick("QUIC TLS key log") - .blurb( - "Danger: write TLS session secrets to the file SSLKEYLOGFILE names; without it the start fails", - ) - .default_value(false) - .mutable_ready() - .build(), glib::ParamSpecBoolean::builder("linger") .nick("Linger") .blurb("At end of stream, serve the finished broadcast until its media is stale, then post EOS") @@ -258,6 +224,9 @@ impl ObjectImpl for MoqSink { .read_only() .build(), ] + .into_iter() + .chain(crate::connection::properties()) + .collect() }); PROPS.as_ref() } @@ -285,12 +254,8 @@ impl ObjectImpl for MoqSink { match pspec.name() { "url" => settings.url = value.get().unwrap(), "broadcast" => settings.broadcast = value.get().unwrap(), - "tls-disable-verify" => settings.tls_disable_verify = value.get().unwrap(), - "quic-idle-timeout" => settings.quic_idle_timeout = Some(Duration::from_millis(value.get().unwrap())), - "quic-keep-alive" => settings.quic_keep_alive = Some(Duration::from_millis(value.get().unwrap())), - "quic-keylog" => settings.quic_keylog = value.get().unwrap(), "linger" => settings.linger = value.get().unwrap(), - _ => unreachable!(), + name => settings.connection.set(name, value), } } @@ -321,22 +286,8 @@ impl ObjectImpl for MoqSink { match name { "url" => settings.url.to_value(), "broadcast" => settings.broadcast.to_value(), - "tls-disable-verify" => settings.tls_disable_verify.to_value(), - "quic-idle-timeout" => duration_millis( - settings - .quic_idle_timeout - .unwrap_or_else(|| moq_tokio::quic::Resolved::default().idle_timeout), - ) - .to_value(), - "quic-keep-alive" => settings - .quic_keep_alive - .or_else(|| moq_tokio::quic::Resolved::default().keep_alive) - .map(duration_millis) - .unwrap_or(0) - .to_value(), - "quic-keylog" => settings.quic_keylog.to_value(), "linger" => settings.linger.to_value(), - _ => unreachable!(), + name => settings.connection.get(name), } } } @@ -2273,7 +2224,7 @@ mod tests { #[test] fn duration_millis_saturates() { - assert_eq!(duration_millis(Duration::MAX), u64::MAX); + assert_eq!(crate::connection::duration_millis(Duration::MAX), u64::MAX); } #[test] diff --git a/rs/moq-gst/src/sink/session.rs b/rs/moq-gst/src/sink/session.rs index 7b07596b63c..a69fbae8435 100644 --- a/rs/moq-gst/src/sink/session.rs +++ b/rs/moq-gst/src/sink/session.rs @@ -137,37 +137,20 @@ pub struct ResolvedSettings { pub url: url::Url, /// Name to publish the broadcast under. pub broadcast: String, - /// Disable TLS certificate verification (local/dev use). - pub tls_disable_verify: bool, - /// QUIC idle timeout override. - pub quic_idle_timeout: Option, - /// QUIC keep-alive override, including zero to disable it. - pub quic_keep_alive: Option, - /// Write TLS session secrets to the file `SSLKEYLOGFILE` names. - pub quic_keylog: bool, + /// The shared connection properties resolved before startup. + pub connect: moq_tokio::connect::Config, + /// The shared QUIC properties resolved before startup. + pub quic: moq_tokio::quic::Config, } /// Builds the connect configuration with the sink's TLS and backoff overrides. pub(super) fn connect_config(settings: &ResolvedSettings) -> moq_tokio::connect::Config { - let mut config = moq_tokio::connect::Config::default(); - config.tls.insecure = Some(settings.tls_disable_verify); - config.backoff.timeout = Some(std::time::Duration::ZERO); - config + settings.connect.clone() } /// The QUIC transport overrides the sink exposes as properties. pub(super) fn quic_config(settings: &ResolvedSettings) -> moq_tokio::quic::Config { - let mut config = moq_tokio::quic::Config::default(); - // The properties are optional and the config fields are not: an unset property - // leaves the library default rather than overriding it with one of its own. - if let Some(idle_timeout) = settings.quic_idle_timeout { - config.idle_timeout = Some(idle_timeout); - } - if let Some(keep_alive) = settings.quic_keep_alive { - config.keep_alive = Some(keep_alive); - } - config.keylog = Some(settings.quic_keylog); - config + settings.quic.clone() } /// Whether the publication is still open, and if not, how it ended. @@ -555,10 +538,8 @@ mod tests { let settings = ResolvedSettings { url: "https://127.0.0.1:1".parse().unwrap(), broadcast: "test".into(), - tls_disable_verify: false, - quic_idle_timeout: None, - quic_keep_alive: None, - quic_keylog: false, + connect: crate::connection::Settings::default().resolve().unwrap().0, + quic: Default::default(), }; let (session, registration, broadcast, _) = Session::start(settings, glib::WeakRef::new()).unwrap(); registration.mark_registered(); diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs index 9316007c684..46490a43800 100644 --- a/rs/moq-gst/src/source/imp.rs +++ b/rs/moq-gst/src/source/imp.rs @@ -40,25 +40,22 @@ static NEXT_AUDIO_PAD_ID: AtomicU64 = AtomicU64::new(0); struct Settings { url: Option, broadcast: Option, - versions: Option, - tls_disable_verify: bool, - quic_keylog: bool, + connection: crate::connection::Settings, } #[derive(Debug, Clone)] struct ResolvedSettings { url: url::Url, broadcast: String, - /// Empty offers every supported version. - versions: Vec, - tls_disable_verify: bool, - quic_keylog: bool, + connect: moq_tokio::connect::Config, + quic: moq_tokio::quic::Config, } impl TryFrom for ResolvedSettings { type Error = anyhow::Error; fn try_from(value: Settings) -> Result { + let (connect, quic) = value.connection.resolve()?; Ok(Self { url: url::Url::parse(value.url.as_ref().context("url property is required")?)?, broadcast: value @@ -66,16 +63,8 @@ impl TryFrom for ResolvedSettings { .as_ref() .context("broadcast property is required")? .clone(), - versions: value - .versions - .iter() - .flat_map(|versions| versions.split(',')) - .map(str::trim) - .filter(|version| !version.is_empty()) - .map(|version| version.parse().map_err(anyhow::Error::msg)) - .collect::>()?, - tls_disable_verify: value.tls_disable_verify, - quic_keylog: value.quic_keylog, + connect, + quic, }) } } @@ -218,11 +207,7 @@ impl SessionController { /// The dial's configuration: versions and TLS as set, retrying without a budget. fn connect_config(settings: &ResolvedSettings) -> moq_tokio::connect::Config { - let mut config = moq_tokio::connect::Config::default(); - config.version = settings.versions.clone(); - config.tls.insecure = Some(settings.tls_disable_verify); - config.backoff.timeout = Some(std::time::Duration::ZERO); - config + settings.connect.clone() } /// Start the reconnecting dial and the origin its announcements land in. @@ -235,8 +220,7 @@ fn connect( let _rt = RUNTIME.enter(); let origin = moq_tokio::origin::spawn(); let consumer = origin.consume(); - let mut quic = moq_tokio::quic::Config::default(); - quic.keylog = Some(settings.quic_keylog); + let quic = settings.quic.clone(); let client = config.init(quic)?.with_subscriber(origin); let connection = client.connect(settings.url.clone()); Ok((client, connection, consumer)) @@ -273,26 +257,10 @@ impl ObjectImpl for MoqSrc { .blurb("The broadcast name to subscribe to") .mutable_ready() .build(), - glib::ParamSpecString::builder("versions") - .nick("Versions") - .blurb("Comma-separated protocol versions to offer, most preferred first, like moq-lite-06; unset offers every stable version") - .mutable_ready() - .build(), - glib::ParamSpecBoolean::builder("tls-disable-verify") - .nick("TLS Disable Verify") - .blurb("Disable TLS certificate verification") - .default_value(false) - .mutable_ready() - .build(), - glib::ParamSpecBoolean::builder("quic-keylog") - .nick("QUIC TLS key log") - .blurb( - "Danger: write TLS session secrets to the file SSLKEYLOGFILE names; without it the start fails", - ) - .default_value(false) - .mutable_ready() - .build(), ] + .into_iter() + .chain(crate::connection::properties()) + .collect() }); PROPS.as_ref() } @@ -320,10 +288,7 @@ impl ObjectImpl for MoqSrc { match pspec.name() { "url" => settings.url = value.get().unwrap(), "broadcast" => settings.broadcast = value.get().unwrap(), - "versions" => settings.versions = value.get().unwrap(), - "tls-disable-verify" => settings.tls_disable_verify = value.get().unwrap(), - "quic-keylog" => settings.quic_keylog = value.get().unwrap(), - _ => unreachable!(), + name => settings.connection.set(name, value), } } @@ -332,10 +297,7 @@ impl ObjectImpl for MoqSrc { match pspec.name() { "url" => settings.url.to_value(), "broadcast" => settings.broadcast.to_value(), - "versions" => settings.versions.to_value(), - "tls-disable-verify" => settings.tls_disable_verify.to_value(), - "quic-keylog" => settings.quic_keylog.to_value(), - _ => unreachable!(), + name => settings.connection.get(name), } } } @@ -1679,9 +1641,7 @@ mod session_tests { use hang::catalog::{AudioCodec, AudioConfig, Container, H264, VideoCodec, VideoConfig}; use tokio::sync::watch; - use super::{ - NEXT_VIDEO_PAD_ID, Pads, ResolvedSettings, SessionController, Settings, TrackKind, follow_catalog, follow_path, - }; + use super::{NEXT_VIDEO_PAD_ID, Pads, ResolvedSettings, SessionController, TrackKind, follow_catalog, follow_path}; /// The pad-id counters are process-global, so a test reading one has to be the only test /// allocating while it runs. `cargo test` shares a process across tests (nextest doesn't), @@ -2610,31 +2570,11 @@ mod session_tests { ResolvedSettings { url: "https://127.0.0.1:1".parse().unwrap(), broadcast: "test".into(), - versions: Vec::new(), - tls_disable_verify: false, - quic_keylog: false, + connect: crate::connection::Settings::default().resolve().unwrap().0, + quic: Default::default(), } } - /// `versions` reaches the dial in order, and an unknown name fails the start instead of - /// offering the defaults. - #[test] - fn versions_reach_the_dial() { - let settings = |versions: &str| Settings { - url: Some("https://127.0.0.1:1".into()), - broadcast: Some("test".into()), - versions: Some(versions.into()), - ..Default::default() - }; - let resolved = ResolvedSettings::try_from(settings(" moq-transport-14 , moq-lite-06 ")).unwrap(); - let config = super::connect_config(&resolved); - let expected: Vec = vec!["moq-transport-14".parse().unwrap(), "moq-lite-06".parse().unwrap()]; - assert_eq!(config.version, expected); - - let err = ResolvedSettings::try_from(settings("moq-lite-06,moq-lite-99")).unwrap_err(); - assert!(err.to_string().contains("moq-lite-99"), "{err}"); - } - /// The source retries without a budget, so an outage of any length is ridden out /// (`relay_loss_reconnects_after_refused_redials` covers the redial itself). #[test] @@ -2648,10 +2588,8 @@ mod session_tests { fn keylog_reaches_the_client() { // SAFETY: nextest runs each test in its own process, so nothing else reads the environment. unsafe { std::env::remove_var("SSLKEYLOGFILE") }; - let keylog = ResolvedSettings { - quic_keylog: true, - ..unreachable() - }; + let mut keylog = unreachable(); + keylog.quic.keylog = Some(true); let err = super::connect(&keylog) .err() .expect("a keylog without SSLKEYLOGFILE must be refused"); diff --git a/rs/moq-gst/tests/element.rs b/rs/moq-gst/tests/element.rs index 6e884dbd528..af56213b252 100644 --- a/rs/moq-gst/tests/element.rs +++ b/rs/moq-gst/tests/element.rs @@ -1568,3 +1568,18 @@ fn connect_failure_retries_without_erroring() { assert_eq!(send_bitrate, 0); assert_eq!(recv_bitrate, 0); } + +// Both element factories expose the same connection syntax, and refuse unknown versions +// before opening a transport rather than silently offering the defaults. +#[test] +fn parsed_connection_properties_are_shared_and_validated() { + init(); + for factory in ["moqsink", "moqsrc"] { + let element = gst::parse::launch(&format!("{factory} url=https://127.0.0.1:1 broadcast=test connect-version=moq-lite-99 quic-idle-timeout=15000 quic-keep-alive=0")) + .expect("parse the shared property syntax"); + assert_eq!(element.property::("quic-idle-timeout"), 15_000); + assert_eq!(element.property::("quic-keep-alive"), 0); + assert_eq!(element.set_state(gst::State::Paused), Err(gst::StateChangeError)); + element.set_state(gst::State::Null).unwrap(); + } +} diff --git a/rs/moq-relay/MIGRATE.md b/rs/moq-relay/MIGRATE.md index 866a7e1d2d0..2f68a77c96a 100644 --- a/rs/moq-relay/MIGRATE.md +++ b/rs/moq-relay/MIGRATE.md @@ -5,6 +5,12 @@ Rolling a cluster across a release also follows the [upgrade guide](https://moq. ## Unreleased +### Connection retry flags have the `connect-` prefix + +Rename `--backoff-*` to `--connect-backoff-*` and `--goaway-*` to `--connect-goaway-*`. +Rename `MOQ_BACKOFF_*` and `MOQ_GOAWAY_*` to `MOQ_CONNECT_BACKOFF_*` and `MOQ_CONNECT_GOAWAY_*`. +Old spellings are hidden and stop startup with their replacements; `[connect.backoff]` and `[connect.goaway]` are unchanged. + ### `receive_window` has a floor (#5878) The `[quic]` and `[peer.quic]` `receive_window` must be at least 67108880 bytes (64 MiB plus 16), or the relay refuses to start; the `[quic]` default is now 128 MiB. diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 21486367b99..0e78256c0ca 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -738,7 +738,7 @@ pub struct Config { /// Connect to one or more other cluster nodes. Each entry is a full URL, e.g. /// `https://host/?jwt=TOKEN`, or an object with `url`, `cost`, `egress`, - /// `token`, and `upstream`; see [`Peer`]. A bare host or `host:port` is refused; pass the + /// `token`, and `upstream`. A bare host or `host:port` is refused; pass the /// full URL instead. Accepts a comma-separated list on the CLI or repeat the /// flag; in config files use a TOML array of URLs and/or objects. /// @@ -762,16 +762,16 @@ pub struct Config { /// Fetch the list of peers to dial from an HTTP(S) URL or a local file, /// reloading at runtime without a restart. The source returns a JSON array /// of peers: bare URL strings and/or objects with `url`, `cost`, `egress`, - /// `token`, and `upstream`, exactly as [`Self::connect`] accepts, e.g. + /// `token`, and `upstream`, exactly as `--cluster-connect` accepts, e.g. /// `["https://a.pop.example/?cost=1", {"url": "https://b.pop.example/", "cost": 2}]`. /// An http(s) URL is re-checked on a fixed cadence, with caching, conditional revalidation /// (`ETag` / `Last-Modified`), and stale-if-error handled by the shared HTTP /// cache client, so the response's `Cache-Control` controls how often a real /// fetch hits the endpoint; a local path is watched via OS filesystem /// notifications (with a periodic re-check fallback). This relay's own - /// [`Self::node`] value, when set, is sent as a `?node=` query param so the + /// `--cluster-node` value, when set, is sent as a `?node=` query param so the /// server can return this node's peers. The relay keeps the last good list if - /// a fetch fails. Composes with [`Self::connect`]. + /// a fetch fails. Composes with `--cluster-connect`. #[usage( name = "cluster-connect-api", long = "cluster-connect-api", @@ -793,7 +793,7 @@ pub struct Config { pub connect_api_tls_root: Vec, /// This relay's own externally-reachable URL (identity). Sent to - /// [`Self::connect_api`] as a `?node=` query param so the endpoint can return + /// `--cluster-connect-api` as a `?node=` query param so the endpoint can return /// this node's peers, and advertised over mDNS when LAN discovery is on. On /// its own it neither opens nor accepts a connection. #[usage( @@ -865,7 +865,7 @@ impl Config { pub struct LanConfig { /// Enable mDNS discovery. Boolean flag: pass `--cluster-lan` (or `=true` / /// `=false`). Advertises the listener fingerprint when the certificate was - /// generated or supplied in-memory, the [`Config::node`] URL when one + /// generated or supplied in-memory, the `--cluster-node` URL when one /// is configured, and needs at least one of them. #[usage( name = "cluster-lan", diff --git a/rs/moq-relay/src/stats.rs b/rs/moq-relay/src/stats.rs index 21f5f2fdaf8..5d503917866 100644 --- a/rs/moq-relay/src/stats.rs +++ b/rs/moq-relay/src/stats.rs @@ -70,7 +70,7 @@ pub struct Config { /// single `/node/` broadcast for the whole node. Set to 1 to /// publish a per-first-segment broadcast (e.g. per tenant), so a consumer can /// announce-scope to just that group rather than slurping every node's full - /// stats. See [`moq_stats::produce::Config::depth`]. + /// stats. #[usage( long = "stats-depth", env = "MOQ_STATS_DEPTH", diff --git a/rs/moq-relay/tests/integration/goaway_cluster.rs b/rs/moq-relay/tests/integration/goaway_cluster.rs index 3206c7f618f..c54bba0b405 100644 --- a/rs/moq-relay/tests/integration/goaway_cluster.rs +++ b/rs/moq-relay/tests/integration/goaway_cluster.rs @@ -26,7 +26,7 @@ const TEST_TIMEOUT: Duration = Duration::from_secs(15); /// the session as a failed attempt, so the redial waits this long first. const BACKOFF: Duration = Duration::from_millis(300); -/// The cluster's `--goaway-handover`: short, so a test observes the old session +/// The cluster's `--connect-goaway-handover`: short, so a test observes the old session /// close quickly, but past [`BACKOFF`] plus the redial, so the draining leg /// serves until the replacement takes over. const HANDOVER: Duration = Duration::from_secs(1); @@ -924,7 +924,7 @@ async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() { tokio::time::sleep(Duration::from_millis(120)).await; // Send the client somewhere that will never finish handshaking. No wire timeout: - // our own `--goaway-handover` is then the only thing bounding the old session. + // our own `--connect-goaway-handover` is then the only thing bounding the old session. server_session .drain() .send(moq_net::goaway::Goaway::redirect(format!( diff --git a/rs/moq-tokio/MIGRATE.md b/rs/moq-tokio/MIGRATE.md index 4f4a034a986..847b8f011f7 100644 --- a/rs/moq-tokio/MIGRATE.md +++ b/rs/moq-tokio/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Connection retry flags have the `connect-` prefix + +Rename `--backoff-*` to `--connect-backoff-*` and `--goaway-*` to `--connect-goaway-*`. +Rename `MOQ_BACKOFF_*` and `MOQ_GOAWAY_*` to `MOQ_CONNECT_BACKOFF_*` and `MOQ_CONNECT_GOAWAY_*`. +Old spellings are hidden and stop startup with their replacements; `[connect.backoff]` and `[connect.goaway]` are unchanged. + ### The connection receive window is 128 MiB, and has a floor (#5878) `quic::DEFAULT_RECEIVE_WINDOW` is 128 MiB instead of 64 MiB, since incomplete messages now hold connection credit until complete. diff --git a/rs/moq-tokio/src/connect.rs b/rs/moq-tokio/src/connect.rs index e1767221ab8..cd551fbd7d9 100644 --- a/rs/moq-tokio/src/connect.rs +++ b/rs/moq-tokio/src/connect.rs @@ -653,8 +653,8 @@ pub struct Config { /// Dial once instead of redialing (with backoff) whenever the session drops. /// - /// A [`crate::Connection`] reconnects by default. With this set, the session's - /// close surfaces through [`crate::Connection::closed`] instead. + /// The connection reconnects by default. With this set, the session's + /// close surfaces through its closed event instead. #[serde(skip_serializing_if = "Option::is_none")] #[usage( name = "connect-once", @@ -677,12 +677,12 @@ pub struct Config { #[usage(skip)] pub(crate) reconnect: Option, - /// Retry pacing for [`crate::Client::connect`] (`--backoff-*`). + /// Retry pacing for [`crate::Client::connect`] (`--connect-backoff-*`). #[usage(flatten)] #[serde(default)] pub backoff: Backoff, - /// How [`crate::Client::connect`] reacts to a peer's GOAWAY (`--goaway-*`). + /// How [`crate::Client::connect`] reacts to a peer's GOAWAY (`--connect-goaway-*`). #[usage(flatten)] #[serde(default)] pub goaway: Goaway, @@ -748,6 +748,8 @@ impl Config { /// skipped the check can't reach a dial that quietly ignored half of it. pub fn deprecated(&self) -> crate::cli::Deprecated { let mut found = self.legacy.deprecated(); + found.extend(self.backoff.deprecated()); + found.extend(self.goaway.deprecated()); if self.connect.is_some() { found.toml("connect", "url", None); } diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index c80479761e9..20e5d1072a7 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -84,9 +84,9 @@ pub struct Backoff { /// Initial delay before the first reconnect attempt. Defaults to 1s, floored at 50ms. #[usage( - name = "backoff-initial", + name = "connect-backoff-initial", long, - env = "MOQ_BACKOFF_INITIAL", + env = "MOQ_CONNECT_BACKOFF_INITIAL", setting = "connect.backoff.initial" )] #[serde(default, rename = "__cli_initial", skip_serializing_if = "Option::is_none")] @@ -94,9 +94,9 @@ pub struct Backoff { /// Multiplier applied to delay after each failure. Defaults to 2. #[usage( - name = "backoff-multiplier", + name = "connect-backoff-multiplier", long, - env = "MOQ_BACKOFF_MULTIPLIER", + env = "MOQ_CONNECT_BACKOFF_MULTIPLIER", default = "2", setting = "connect.backoff.multiplier" )] @@ -113,7 +113,12 @@ pub struct Backoff { pub max: Option, /// Maximum delay between reconnect attempts. Defaults to 5s. - #[usage(name = "backoff-max", long, env = "MOQ_BACKOFF_MAX", setting = "connect.backoff.max")] + #[usage( + name = "connect-backoff-max", + long, + env = "MOQ_CONNECT_BACKOFF_MAX", + setting = "connect.backoff.max" + )] #[serde(default, rename = "__cli_max", skip_serializing_if = "Option::is_none")] max_arg: Option, @@ -127,7 +132,7 @@ pub struct Backoff { /// Resets after a stable connection (one that outlives the initial backoff) outside a redirect /// loop, so a flapping session that reconnects then immediately drops still counts toward the /// timeout. Set to `Some(Duration::ZERO)` for unlimited retries. `None` takes the default. When this config was - /// parsed from the command line, a `--backoff-timeout` given there wins over this field. + /// parsed from the command line, a `--connect-backoff-timeout` given there wins over this field. /// /// An `Option` because zero is a real setting: a command-line parse leaves this field at its /// type's default, and only `None` tells "unset" apart from "retry forever". @@ -143,13 +148,43 @@ pub struct Backoff { // window set in code. The default lives in [`Backoff::window`] instead. /// Maximum time to spend retrying before giving up. Defaults to 10s; 0s retries forever. #[usage( - name = "backoff-timeout", + name = "connect-backoff-timeout", long, - env = "MOQ_BACKOFF_TIMEOUT", + env = "MOQ_CONNECT_BACKOFF_TIMEOUT", setting = "connect.backoff.timeout" )] #[serde(default, rename = "__cli_timeout", skip_serializing_if = "Option::is_none")] timeout_arg: Option, + #[usage( + name = "backoff-initial", + long = "backoff-initial", + env = "MOQ_BACKOFF_INITIAL", + hide = true + )] + #[serde(skip)] + legacy_initial: Option, + + #[usage( + name = "backoff-multiplier", + long = "backoff-multiplier", + env = "MOQ_BACKOFF_MULTIPLIER", + hide = true + )] + #[serde(skip)] + legacy_multiplier: Option, + + #[usage(name = "backoff-max", long = "backoff-max", env = "MOQ_BACKOFF_MAX", hide = true)] + #[serde(skip)] + legacy_max: Option, + + #[usage( + name = "backoff-timeout", + long = "backoff-timeout", + env = "MOQ_BACKOFF_TIMEOUT", + hide = true + )] + #[serde(skip)] + legacy_timeout: Option, } impl Default for Backoff { @@ -162,11 +197,48 @@ impl Default for Backoff { max_arg: None, timeout: Some(DEFAULT_TIMEOUT), timeout_arg: None, + legacy_initial: None, + legacy_multiplier: None, + legacy_max: None, + legacy_timeout: None, } } } impl Backoff { + pub(crate) fn deprecated(&self) -> crate::cli::Deprecated { + let mut found = crate::cli::Deprecated::default(); + if self.legacy_initial.is_some() { + found.flag( + "--backoff-initial", + Some("MOQ_BACKOFF_INITIAL"), + "--connect-backoff-initial / MOQ_CONNECT_BACKOFF_INITIAL", + ); + } + if self.legacy_multiplier.is_some() { + found.flag( + "--backoff-multiplier", + Some("MOQ_BACKOFF_MULTIPLIER"), + "--connect-backoff-multiplier / MOQ_CONNECT_BACKOFF_MULTIPLIER", + ); + } + if self.legacy_max.is_some() { + found.flag( + "--backoff-max", + Some("MOQ_BACKOFF_MAX"), + "--connect-backoff-max / MOQ_CONNECT_BACKOFF_MAX", + ); + } + if self.legacy_timeout.is_some() { + found.flag( + "--backoff-timeout", + Some("MOQ_BACKOFF_TIMEOUT"), + "--connect-backoff-timeout / MOQ_CONNECT_BACKOFF_TIMEOUT", + ); + } + found + } + /// The retry window in effect, where zero means unlimited: the flag, then the typed field, then /// the default. fn window(&self) -> Duration { @@ -337,9 +409,9 @@ pub struct Goaway { /// chose, `follow` also lets it name the host, and `ignore` redials the /// current address list. #[usage( - name = "goaway-redirect", + name = "connect-goaway-redirect", long, - env = "MOQ_GOAWAY_REDIRECT", + env = "MOQ_CONNECT_GOAWAY_REDIRECT", value_enum, default = "same-host", setting = "connect.goaway.redirect" @@ -361,13 +433,31 @@ pub struct Goaway { /// How long the old session keeps serving after its replacement connects. Defaults to 10s. #[usage( - name = "goaway-handover", + name = "connect-goaway-handover", long, - env = "MOQ_GOAWAY_HANDOVER", + env = "MOQ_CONNECT_GOAWAY_HANDOVER", setting = "connect.goaway.handover" )] #[serde(default, rename = "__cli_handover", skip_serializing_if = "Option::is_none")] handover_arg: Option, + #[usage( + name = "goaway-redirect", + long = "goaway-redirect", + env = "MOQ_GOAWAY_REDIRECT", + value_enum, + hide = true + )] + #[serde(skip)] + legacy_redirect: Option, + + #[usage( + name = "goaway-handover", + long = "goaway-handover", + env = "MOQ_GOAWAY_HANDOVER", + hide = true + )] + #[serde(skip)] + legacy_handover: Option, } impl Default for Goaway { @@ -376,6 +466,8 @@ impl Default for Goaway { redirect: Redirect::SameHost, handover: Some(DEFAULT_HANDOVER), handover_arg: None, + legacy_redirect: None, + legacy_handover: None, } } } @@ -437,6 +529,25 @@ impl Pacing { const DEFAULT_HANDOVER: Duration = Duration::from_secs(10); impl Goaway { + pub(crate) fn deprecated(&self) -> crate::cli::Deprecated { + let mut found = crate::cli::Deprecated::default(); + if self.legacy_redirect.is_some() { + found.flag( + "--goaway-redirect", + Some("MOQ_GOAWAY_REDIRECT"), + "--connect-goaway-redirect / MOQ_CONNECT_GOAWAY_REDIRECT", + ); + } + if self.legacy_handover.is_some() { + found.flag( + "--goaway-handover", + Some("MOQ_GOAWAY_HANDOVER"), + "--connect-goaway-handover / MOQ_CONNECT_GOAWAY_HANDOVER", + ); + } + found + } + fn resolve(&self) -> Resolved { Resolved { redirect: self.redirect, @@ -1086,7 +1197,7 @@ impl Connection { // replacement dial, so a predecessor can still be draining when that dial // fails and lands here. A plain sleep would stop enforcing its handover // until some later dial succeeded, holding an upstream that asked to drain - // open for the whole retry window, or forever with `--backoff-timeout=0`. + // open for the whole retry window, or forever with `--connect-backoff-timeout=0`. sleep_draining(wait, &mut draining, shared).await; retry.delay = pacing.next(retry.delay); } @@ -1698,8 +1809,11 @@ mod tests { assert_eq!(parsed.backoff.window(), DEFAULT_TIMEOUT); // And a flag still lands where the merge can see it. - let parsed = - Wrapper::parse_from(&[std::ffi::OsStr::new("--backoff-initial"), std::ffi::OsStr::new("3s")]).unwrap(); + let parsed = Wrapper::parse_from(&[ + std::ffi::OsStr::new("--connect-backoff-initial"), + std::ffi::OsStr::new("3s"), + ]) + .unwrap(); assert_eq!(Pacing::new(&parsed.backoff).initial, Duration::from_secs(3)); } @@ -1723,9 +1837,9 @@ mod tests { // Flags passed: they land where the merge can see them. let parsed = Wrapper::parse_from(&[ - std::ffi::OsStr::new("--goaway-redirect"), + std::ffi::OsStr::new("--connect-goaway-redirect"), std::ffi::OsStr::new("ignore"), - std::ffi::OsStr::new("--goaway-handover"), + std::ffi::OsStr::new("--connect-goaway-handover"), std::ffi::OsStr::new("3s"), ]) .unwrap(); @@ -2892,7 +3006,7 @@ mod duration_tests { initial_set_in_code_beats_parser_default, Backoff, initial, - "--backoff-initial", + "--connect-backoff-initial", |c: &Backoff| Pacing::new(c).initial, |d: Duration| d.max(MIN_BACKOFF) ); @@ -2900,7 +3014,7 @@ mod duration_tests { max_set_in_code_beats_parser_default, Backoff, max, - "--backoff-max", + "--connect-backoff-max", |c: &Backoff| Pacing::new(c).max, |d: Duration| d.max(DEFAULT_INITIAL) ); @@ -2908,7 +3022,7 @@ mod duration_tests { timeout_set_in_code_beats_parser_default, Backoff, timeout, - "--backoff-timeout", + "--connect-backoff-timeout", |c: &Backoff| c.window(), |d| d ); @@ -2916,7 +3030,7 @@ mod duration_tests { handover_set_in_code_beats_parser_default, Goaway, handover, - "--goaway-handover", + "--connect-goaway-handover", |c: &Goaway| c.resolve().handover, |d| d ); diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs index 3c14ca6fa3e..1439c55ecc1 100644 --- a/rs/moq-tokio/src/listen.rs +++ b/rs/moq-tokio/src/listen.rs @@ -192,7 +192,7 @@ pub struct Config { /// anycast IP, steady-state on this host's unicast IP. /// /// Honored by the QUIC backend. Accept-only, which is why it lives - /// here rather than in the shared [`crate::quic::Config`]. + /// here rather than in the shared QUIC settings. #[usage( name = "listen-preferred-v4", long = "listen-preferred-v4", @@ -202,7 +202,7 @@ pub struct Config { #[serde(default, skip_serializing_if = "Option::is_none")] pub preferred_v4: Option, - /// IPv6 address advertised as the QUIC preferred_address. See [`Self::preferred_v4`]. + /// IPv6 address advertised as the QUIC preferred_address. Like `--listen-preferred-v4`, for IPv6. #[usage( name = "listen-preferred-v6", long = "listen-preferred-v6", diff --git a/rs/moq-tokio/src/settings.rs b/rs/moq-tokio/src/settings.rs index 017b6c0b1b9..6fd55c97d0a 100644 --- a/rs/moq-tokio/src/settings.rs +++ b/rs/moq-tokio/src/settings.rs @@ -197,31 +197,31 @@ struct ConnectTls { host_name: Option, } -/// `[connect.backoff]` / `--backoff-*`. +/// `[connect.backoff]` / `--connect-backoff-*`. #[derive(usage_rs::Config)] #[usage(prefix = "connect.backoff")] struct ConnectBackoff { - #[usage(env = "MOQ_BACKOFF_INITIAL", cli("--backoff-initial"))] + #[usage(env = "MOQ_CONNECT_BACKOFF_INITIAL", cli("--connect-backoff-initial"))] initial: Option, - #[usage(env = "MOQ_BACKOFF_MULTIPLIER", cli("--backoff-multiplier"))] + #[usage(env = "MOQ_CONNECT_BACKOFF_MULTIPLIER", cli("--connect-backoff-multiplier"))] multiplier: Option, - #[usage(env = "MOQ_BACKOFF_MAX", cli("--backoff-max"))] + #[usage(env = "MOQ_CONNECT_BACKOFF_MAX", cli("--connect-backoff-max"))] max: Option, - #[usage(env = "MOQ_BACKOFF_TIMEOUT", cli("--backoff-timeout"))] + #[usage(env = "MOQ_CONNECT_BACKOFF_TIMEOUT", cli("--connect-backoff-timeout"))] timeout: Option, } -/// `[connect.goaway]` / `--goaway-*`. +/// `[connect.goaway]` / `--connect-goaway-*`. #[derive(usage_rs::Config)] #[usage(prefix = "connect.goaway")] struct ConnectGoaway { - #[usage(env = "MOQ_GOAWAY_REDIRECT", cli("--goaway-redirect"))] + #[usage(env = "MOQ_CONNECT_GOAWAY_REDIRECT", cli("--connect-goaway-redirect"))] redirect: Option, - #[usage(env = "MOQ_GOAWAY_HANDOVER", cli("--goaway-handover"))] + #[usage(env = "MOQ_CONNECT_GOAWAY_HANDOVER", cli("--connect-goaway-handover"))] handover: Option, } diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs index 0f700f629fa..4332f47d7b7 100644 --- a/rs/moq-tokio/src/tcp.rs +++ b/rs/moq-tokio/src/tcp.rs @@ -30,7 +30,7 @@ const WIRE_VERSION: qmux::Version = qmux::Version::QMux01; #[serde(deny_unknown_fields, default)] #[non_exhaustive] pub struct Config { - /// Bind a qmux TCP listener on this address, plaintext unless [`Self::tls`]. + /// Bind a qmux TCP listener on this address, plaintext unless `--listen-tcp-tls` is set. #[usage( long = "listen-tcp-bind", name = "listen-tcp-bind", diff --git a/rs/moq-tokio/src/tls.rs b/rs/moq-tokio/src/tls.rs index 976b53b58e8..2fca3f373ae 100644 --- a/rs/moq-tokio/src/tls.rs +++ b/rs/moq-tokio/src/tls.rs @@ -1287,11 +1287,11 @@ pub struct Listen { /// PEM file(s) of root CAs for validating optional client certificates (mTLS). /// /// When set, clients *may* present a certificate during the TLS handshake. - /// Valid presentations are reported via [`crate::server::Request::peer_identity`] + /// Valid presentations are reported as the accepted request's peer identity /// and can be used by the application to grant elevated access. Clients that /// do not present a certificate are unaffected. /// - /// Plain-TLS listeners built via [`Self::server_config`] also use these roots + /// Plain-TLS listeners also use these roots /// for optional mTLS. Root files are hot reloaded for new handshakes. #[usage( long = "listen-tls-root", @@ -1408,8 +1408,7 @@ impl Listen { /// The released spellings in use, each paired with what replaced it. /// /// Public for the same reason as [`Connect::deprecated`]: this type is flattened - /// into parsers of its own (moq-cli's `export hls` is one), not only through - /// [`crate::listen::Config`]. The methods that build a server config refuse + /// into custom listener parsers, not only through [`crate::listen::Config`]. The methods that build a server config refuse /// anyway, since a dropped `--server-tls-root` would take the mTLS client CAs /// with it and leave the listener accepting unauthenticated peers. pub fn deprecated(&self) -> crate::cli::Deprecated { diff --git a/rs/moq-tokio/tests/integration/reconnect.rs b/rs/moq-tokio/tests/integration/reconnect.rs index e7e61f2242b..86ddb905e76 100644 --- a/rs/moq-tokio/tests/integration/reconnect.rs +++ b/rs/moq-tokio/tests/integration/reconnect.rs @@ -267,7 +267,7 @@ async fn a_redirect_to_another_host_is_refused_by_default() { ); } -/// `--goaway-redirect follow` is the opt-in that hands the peer the host, so the +/// `--connect-goaway-redirect follow` is the opt-in that hands the peer the host, so the /// same redirect is followed. Without this the default above would be /// indistinguishable from ignoring the URI outright. #[tokio::test] diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 6743d6950b0..8317cb37724 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -1059,7 +1059,7 @@ run_gst_audio() { deadline=$((SECONDS + ${TIMEOUT%.*})) [[ "$TIMEOUT" != *.* ]] || deadline=$((deadline + 1)) GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \ - gst-launch-1.0 -v moqsrc name=s url="$url" broadcast="$broadcast" ${version:+versions="$version"} \ + gst-launch-1.0 -v moqsrc name=s url="$url" broadcast="$broadcast" ${version:+connect-version="$version"} \ s.video_0 ! filesink location="$out.video" buffer-mode=2 \ s.audio_0 ! decodebin ! audioconvert ! audio/x-raw,format=F32LE,layout=interleaved \ ! identity name=frames silent=false ! filesink location="$out.pcm" buffer-mode=2 >"$out.gst" 2>&1 & @@ -1214,7 +1214,7 @@ PYCODE trap '' PIPE GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \ timeout -k 3 "$TIMEOUT" gst-launch-1.0 -q \ - moqsrc name=s url="$url" broadcast="$broadcast" ${version:+versions="$version"} \ + moqsrc name=s url="$url" broadcast="$broadcast" ${version:+connect-version="$version"} \ s.video_0 ! filesink location=/dev/stdout buffer-mode=2 \ 2>/dev/null | head -c 1 | wc -c | tr -d ' ' || true ) @@ -1428,7 +1428,7 @@ run_finish_gst() { deadline=$((SECONDS + ${TIMEOUT%.*} * 2)) [[ "$TIMEOUT" != *.* ]] || deadline=$((deadline + 2)) GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \ - gst-launch-1.0 -v moqsrc name=s url="$url" broadcast="$broadcast" ${version:+versions="$version"} \ + gst-launch-1.0 -v moqsrc name=s url="$url" broadcast="$broadcast" ${version:+connect-version="$version"} \ s.video_0 ! identity name=frames silent=false ! fakesink >"$out" 2>&1 & pid=$! count=0 From a76d99a9510eef708b370e473b28ba7461a07618 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 08:11:20 -0700 Subject: [PATCH 3/7] refactor(publish)!: always serve the cleanly finished tail Remove publisher linger controls while preserving abrupt NULL/close teardown, fresh epochs on re-announce, and bounded finish ownership. Keep the CLI quest open for upstream-first pinned-consumer rollout. Co-Authored-By: GPT-6 --- doc/bin/gstreamer.md | 9 ++-- doc/lib/js/publish.md | 5 +- js/publish/MIGRATE.md | 13 +++-- js/publish/README.md | 5 +- js/publish/src/broadcast.test.ts | 29 +++++------ js/publish/src/broadcast.ts | 22 +++----- js/publish/src/element.ts | 27 +--------- quest/m1/api-audit/cli-flags.md | 33 +++++------- rs/moq-cli/src/args.rs | 86 +++++++++++++++++++++----------- rs/moq-gst/MIGRATE.md | 8 +++ rs/moq-gst/src/sink/imp.rs | 38 ++++++-------- rs/moq-gst/tests/element.rs | 16 +++--- 12 files changed, 143 insertions(+), 148 deletions(-) diff --git a/doc/bin/gstreamer.md b/doc/bin/gstreamer.md index 0180456fed9..ffb633f6cf4 100644 --- a/doc/bin/gstreamer.md +++ b/doc/bin/gstreamer.md @@ -62,7 +62,7 @@ Each `sink_%u` request pad is one track. Pad properties: `track` names it `encoder=true` marks it as fed by a local encoder, and `track-status`/`track-error` report its lifecycle. Element properties: `url`, `broadcast`, `connect-version`, `tls-disable-verify`, `quic-idle-timeout`, `quic-keylog`, -`quic-keep-alive`, `linger`, and read-only `status`, `connected`, `moq-version`, and +`quic-keep-alive`, and read-only `status`, `connected`, `moq-version`, and `estimated-send-rate`, `estimated-recv-rate`, `connection-stats`, and `sessions`. The sink reconnects for as long as the pipeline runs and only reports `failed` on an answer redialing can't change, such as a rejected token. @@ -81,12 +81,11 @@ it is released, reaches EOS, or fails; a flush or new stream after that EOS holds it back again. Opaque data pads are not listed. Later catalog updates do not wait for newly requested pads. -Once every pad reaches EOS, the sink finishes the broadcast and posts EOS. -With `linger=true` it first keeps serving the finished broadcast until its -media is stale (30 seconds for audio and video), so a viewer still catching up +Once every pad reaches EOS, the sink finishes the broadcast and keeps serving +its tail until the media is stale (30 seconds for audio and video), so a viewer still catching up gets the tail, and posts EOS only then, which is when `gst-launch-1.0 -e` exits. Stopping the pipeline, such as a second Ctrl-C under `-e`, closes the -broadcast at once. +broadcast at once. With no media to drain, EOS is immediate. All pads share one media timeline on one wall clock. The first timed buffer places it using UNIX or NTP reference timestamp metadata; other clock domains use the diff --git a/doc/lib/js/publish.md b/doc/lib/js/publish.md index a9cb0263b66..c0851103aad 100644 --- a/doc/lib/js/publish.md +++ b/doc/lib/js/publish.md @@ -34,7 +34,10 @@ WebCodecs, writes the catalog, and publishes a hang broadcast. | `preview` | What the nested element shows: the raw `source` (default), a decoded copy of the `encoded` stream to see what viewers get, or `none`. | | `backend` | How a `` preview is drawn; see [@moq/video](/lib/js/video). `auto` (default), `webgpu`, or `2d`. | | `announce` | When to advertise: once the `source` is live (default), `always`, or `never`. Nobody can see or subscribe to the broadcast until it is announced. | -| `linger` | Finish the broadcast cleanly when advertising stops, such as a `source` that ends, instead of unannouncing it at once. It stays up while its media is fresh so viewers drain the tail, and the next announce publishes a new broadcast. Removing the element ends it at once. Off by default. | + +When advertising stops, the finished broadcast serves its tail until the media +is stale. Advertising again publishes a new broadcast and publisher epoch. +Removing the element ends it immediately. Every attribute is also a reactive property. The [README](https://www.npmjs.com/package/@moq/publish) lists types and defaults. diff --git a/js/publish/MIGRATE.md b/js/publish/MIGRATE.md index def74f4eec0..7924663d851 100644 --- a/js/publish/MIGRATE.md +++ b/js/publish/MIGRATE.md @@ -4,10 +4,17 @@ How to update your code for each breaking change, newest first. ## Unreleased -### `BroadcastInput` has a `linger` input +### Clean completion always serves the finished tail -Code that builds a complete `BroadcastInput` adds `linger`, such as `new Signal(false)` to keep turning `announce` off unannouncing at once. -`new Broadcast({ ... })` defaults it to `false`, so constructor calls need no change. +Turning `BroadcastInput.announce` off after announcing now finishes the broadcast +and serves its tail until the media is stale. Turning it back on publishes a new +broadcast and publisher epoch; reacquire `broadcast.net` rather than reuse its +previous producer or track handles. Use `enabled: false` or `close()` for an +immediate end. + +Remove the unreleased `BroadcastInput.linger`, `` attribute, +`element.linger` property, and `element.controls.linger` input. Clean completion +is unconditional. Removing the element still ends its broadcast immediately. ### `Encoder.cut()` is not rate limited, and video tracks advertise refresh (#6048) diff --git a/js/publish/README.md b/js/publish/README.md index bfd1d907f34..789e8d5eb95 100644 --- a/js/publish/README.md +++ b/js/publish/README.md @@ -72,7 +72,10 @@ The simplest way to publish a stream: | `preview` | string | `"source"` | What the preview renders: `"source"`, `"encoded"`, `"none"` | | `backend` | string | `"auto"` | Graphics API for a `` preview: `"auto"`, `"webgpu"`, `"2d"` | | `announce` | string | `"source"` | When to advertise: `"always"`, `"never"`, `"source"` (once media is captured). A camera source waits for every enabled track, and `"source"` waits until each captured track's config resolves or fails, so the first catalog lists every rendition. The broadcast is created while connected either way. | -| `linger` | boolean | false | When advertising stops, finish the broadcast cleanly instead of unannouncing it: it stays up while its media is fresh so viewers drain the tail, and the next announce publishes a new one. | + +When advertising stops, the finished broadcast stays available until its media +is stale, so a late viewer can drain the tail. Advertising again publishes a new +broadcast and publisher epoch. Removing the element ends it immediately. A nested `