cat / goq
clone

turns a linux box into a private steam streaming appliance: plug in, tap your key, play from another computer

Use upstream Iroh/MoQ for authenticated media (#15)

FelineStateMachine committed e30e069
* build: pin upstream iroh-moq transport

Bind host and Portal to one reviewed iroh-live revision and the matching moq-net model so transport behavior cannot drift with upstream main.

* feat: frame media for upstream MoQ objects

Encode only bounded Sigil codec metadata inside opaque MoQ frames, leaving group ordering, cancellation, and delivery semantics to the upstream transport.

* fix: align upstream MoQ crypto and model versions

Use one exact Iroh AWS-LC backend and the upstream-locked moq-net and WebTransport versions so host and Portal share a single audited transport graph.

* feat: name session-scoped MoQ media

* feat: gate upstream MoQ behind control auth

* feat: receive authenticated upstream MoQ media

* test: prove authenticated upstream MoQ sessions

* docs: describe authenticated upstream MoQ

* fix: bound upstream MoQ reader teardown
commit
e30e0693e1b036126c4d1c87a0e45c567c2534f9
parent
3d4d6d095f5b2505360300741b753e3d588f29fa
author
FelineStateMachine <dami.etoile@gmail.com>
13 files changed, +2705 -128
MAGENTS.md23+-
MCargo.lock228+++++++-
MCargo.toml7+
MREADME.md28+-
Mcrates/sigil-host/Cargo.toml6+-
Mcrates/sigil-host/src/bin/sigil-probe.rs581+++++++++++++++++++-
Mcrates/sigil-host/src/main.rs26+-
Mcrates/sigil-host/src/server.rs866++++++++++++++++++++++++++++++-
Mcrates/sigil-protocol/src/lib.rs32+-
Mcrates/sigil-protocol/src/media.rs66++
Mscripts/loopback-proof.sh113+++-
Msrc-tauri/Cargo.toml5+-
Msrc-tauri/src/commands/network.rs852+++++++++++++++++++++++++++++-
M AGENTS.md +11 -12
@@ -52,12 +52,12 @@ Sigil host.
5252 - Gamescope's exact PipeWire node is resolved from strict configured5353 properties. AMD GstVA H.264 sustains the fixed 1280x800/60 first target on5454 the GPD Pocket 4 test host with bounded post-encode delivery.55-- Media v3 uses one bounded, prioritized QUIC object stream per encoded frame56- with explicit GOP/object identity, publisher priority, delivery deadlines,57- and a persistent keyframe-request leg. Stale dependent objects are cancelled;58- discontinuity keyframes provide a latest-frame barrier. Media v2 and v159- remain compatibility fallbacks. This is a custom MoQ-style protocol, not a60- claim of IETF MoQ Transport compatibility.55+- The preferred media path authenticates and claims the single client over56+ `sigil/control/1`, then admits that exact peer to a session-scoped upstream57+ `iroh-moq` broadcast and static H.264 track. One configured GOP maps to one58+ bounded native MoQ group; a newer independently decodable group cancels its59+ predecessor and provides the latest-frame barrier. The custom media v3, v2,60+ and v1 protocols remain explicit compatibility fallbacks.6161 - Portal crosses Rust-to-webview video and audio through bounded binary Tauri6262 channels. It reports transport, frontend, decoder, presentation, and audio6363 queue/drop timing separately.
@@ -137,13 +137,12 @@ Sigil host.
137137 and resource percentiles without latency growth, difficult-NAT relay138138 diagnostics, and reconnect/second-client rejection using the exact release139139 candidates.140-- Issues #7-#10 are streaming hardening: full Iroh/MoQ semantics, adaptive140+- Issues #7-#10 are streaming hardening: Iroh/MoQ recovery hardening, adaptive141141 bitrate, motion-sensitive resolution, and automatic low-latency codec142- selection. Media v3 adds grouped object identity, expiry, cancellation, and143- explicit recovery, but the external `gst-launch` encoder can only honor a144- keyframe request at its next natural configured IDR. An in-process GStreamer145- control path plus hardware loss/relay proof remain before claiming the full146- MoQ milestone complete.142+ selection. The preferred path now uses pinned upstream `iroh-moq`; the143+ external `gst-launch` encoder can still honor a keyframe request only at its144+ next natural configured IDR. An in-process GStreamer control path plus145+ hardware loss/relay proof remain before claiming issue #7 complete.147146 148147 ## Working rules149148
M Cargo.lock +201 -27
@@ -146,7 +146,7 @@ version = "1.1.5"
146146 source = "registry+https://github.com/rust-lang/crates.io-index"147147 checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"148148 dependencies = [149- "windows-sys 0.61.2",149+ "windows-sys 0.60.2",150150 ]151151 152152 [[package]]
@@ -157,7 +157,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
157157 dependencies = [158158 "anstyle",159159 "once_cell_polyfill",160- "windows-sys 0.61.2",160+ "windows-sys 0.60.2",161161 ]162162 163163 [[package]]
@@ -415,6 +415,29 @@ version = "1.5.1"
415415 source = "registry+https://github.com/rust-lang/crates.io-index"416416 checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"417417 418+[[package]]419+name = "aws-lc-rs"420+version = "1.17.3"421+source = "registry+https://github.com/rust-lang/crates.io-index"422+checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1"423+dependencies = [424+ "aws-lc-sys",425+ "zeroize",426+]427+428+[[package]]429+name = "aws-lc-sys"430+version = "0.43.0"431+source = "registry+https://github.com/rust-lang/crates.io-index"432+checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c"433+dependencies = [434+ "cc",435+ "cmake",436+ "dunce",437+ "fs_extra",438+ "pkg-config",439+]440+418441 [[package]]419442 name = "backon"420443 version = "1.6.0"
@@ -843,6 +866,15 @@ version = "1.1.0"
843866 source = "registry+https://github.com/rust-lang/crates.io-index"844867 checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9"845868 869+[[package]]870+name = "cmake"871+version = "0.1.58"872+source = "registry+https://github.com/rust-lang/crates.io-index"873+checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678"874+dependencies = [875+ "cc",876+]877+846878 [[package]]847879 name = "cmov"848880 version = "0.5.4"
@@ -1248,7 +1280,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
12481280 checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090"12491281 dependencies = [12501282 "data-encoding",1251- "syn 2.0.119",1283+ "syn 1.0.109",12521284 ]12531285 12541286 [[package]]
@@ -1363,7 +1395,7 @@ dependencies = [
13631395 "libc",13641396 "option-ext",13651397 "redox_users",1366- "windows-sys 0.61.2",1398+ "windows-sys 0.59.0",13671399 ]13681400 13691401 [[package]]
@@ -1624,7 +1656,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
16241656 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"16251657 dependencies = [16261658 "libc",1627- "windows-sys 0.61.2",1659+ "windows-sys 0.59.0",16281660 ]16291661 16301662 [[package]]
@@ -1752,6 +1784,12 @@ dependencies = [
17521784 "percent-encoding",17531785 ]17541786 1787+[[package]]1788+name = "fs_extra"1789+version = "1.3.0"1790+source = "registry+https://github.com/rust-lang/crates.io-index"1791+checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"1792+17551793 [[package]]17561794 name = "futures"17571795 version = "0.3.33"
@@ -1976,8 +2014,8 @@ dependencies = [
19762014 "libc",19772015 "log",19782016 "rustversion",1979- "windows-link 0.2.1",1980- "windows-result 0.4.1",2017+ "windows-link 0.1.3",2018+ "windows-result 0.3.4",19812019 ]19822020 19832021 [[package]]
@@ -2007,10 +2045,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
20072045 checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0"20082046 dependencies = [20092047 "cfg-if",2010- "js-sys",20112048 "libc",20122049 "wasi",2013- "wasm-bindgen",20142050 ]20152051 20162052 [[package]]
@@ -2512,7 +2548,7 @@ dependencies = [
25122548 "js-sys",25132549 "log",25142550 "wasm-bindgen",2515- "windows-core 0.62.2",2551+ "windows-core 0.61.2",25162552 ]25172553 25182554 [[package]]
@@ -2897,6 +2933,22 @@ dependencies = [
28972933 "syn 2.0.119",28982934 ]28992935 2936+[[package]]2937+name = "iroh-moq"2938+version = "0.1.0"2939+source = "git+https://github.com/n0-computer/iroh-live?rev=5f95758fcd1450e443a9134c9d9342bcc3957b85#5f95758fcd1450e443a9134c9d9342bcc3957b85"2940+dependencies = [2941+ "iroh",2942+ "moq-net",2943+ "n0-error",2944+ "n0-future",2945+ "tokio",2946+ "tokio-util",2947+ "tracing",2948+ "url",2949+ "web-transport-iroh",2950+]2951+29002952 [[package]]29012953 name = "iroh-relay"29022954 version = "1.0.3"
@@ -3105,6 +3157,15 @@ dependencies = [
31053157 "unicode-segmentation",31063158 ]31073159 3160+[[package]]3161+name = "kio"3162+version = "0.4.4"3163+source = "registry+https://github.com/rust-lang/crates.io-index"3164+checksum = "b31d86314938a5b2c2bea39f271424a3a2432d5b01412555a02181ecbb656af3"3165+dependencies = [3166+ "smallvec",3167+]3168+31083169 [[package]]31093170 name = "lazy_static"31103171 version = "1.5.0"
@@ -3315,6 +3376,26 @@ dependencies = [
33153376 "uuid",33163377 ]33173378 3379+[[package]]3380+name = "moq-net"3381+version = "0.1.11"3382+source = "registry+https://github.com/rust-lang/crates.io-index"3383+checksum = "72e78427132ce874978ef554d9cab419762abd115ba5c97f4bea4ea0aa523d3f"3384+dependencies = [3385+ "bytes",3386+ "futures",3387+ "kio",3388+ "num_enum",3389+ "rand",3390+ "serde",3391+ "serde_json",3392+ "thiserror 2.0.19",3393+ "tokio",3394+ "tracing",3395+ "web-async",3396+ "web-transport-trait",3397+]3398+33183399 [[package]]33193400 name = "moxcms"33203401 version = "0.8.1"
@@ -3343,7 +3424,7 @@ dependencies = [
33433424 "png 0.18.1",33443425 "serde",33453426 "thiserror 2.0.19",3346- "windows-sys 0.61.2",3427+ "windows-sys 0.60.2",33473428 ]33483429 33493430 [[package]]
@@ -3352,6 +3433,7 @@ version = "1.0.0"
33523433 source = "registry+https://github.com/rust-lang/crates.io-index"33533434 checksum = "c37e81176a83a77d2514528b91bdafc70ef88aab428f0e1b91aebb8d99888895"33543435 dependencies = [3436+ "anyhow",33553437 "n0-error-macros",33563438 "spez",33573439 ]
@@ -3607,6 +3689,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
36073689 checksum = "334c3c9833f7b2c573cceb9896ddc7aaeb58c8807cbb63211b24d1fe88bf866e"36083690 dependencies = [36093691 "aes-gcm",3692+ "aws-lc-rs",36103693 "bytes",36113694 "derive_more",36123695 "enum-assoc",
@@ -3615,7 +3698,6 @@ dependencies = [
36153698 "lru-slab",36163699 "rand",36173700 "rand_pcg",3618- "ring",36193701 "rustc-hash",36203702 "rustls",36213703 "rustls-pki-types",
@@ -3646,7 +3728,7 @@ version = "0.50.3"
36463728 source = "registry+https://github.com/rust-lang/crates.io-index"36473729 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"36483730 dependencies = [3649- "windows-sys 0.61.2",3731+ "windows-sys 0.59.0",36503732 ]36513733 36523734 [[package]]
@@ -3743,7 +3825,7 @@ version = "0.7.6"
37433825 source = "registry+https://github.com/rust-lang/crates.io-index"37443826 checksum = "680998035259dcfcafe653688bf2aa6d3e2dc05e98be6ab46afb089dc84f1df8"37453827 dependencies = [3746- "proc-macro-crate 3.5.0",3828+ "proc-macro-crate 1.3.1",37473829 "proc-macro2",37483830 "quote",37493831 "syn 2.0.119",
@@ -4272,6 +4354,12 @@ version = "0.2.17"
42724354 source = "registry+https://github.com/rust-lang/crates.io-index"42734355 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"42744356 4357+[[package]]4358+name = "pin-utils"4359+version = "0.1.0"4360+source = "registry+https://github.com/rust-lang/crates.io-index"4361+checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"4362+42754363 [[package]]42764364 name = "piper"42774365 version = "0.2.5"
@@ -4379,13 +4467,16 @@ version = "0.1.0"
43794467 dependencies = [43804468 "anyhow",43814469 "base64 0.22.1",4470+ "bytes",43824471 "ctap-hid-fido2",43834472 "ed25519-dalek",43844473 "getrandom 0.4.3",43854474 "image",43864475 "iroh",4476+ "iroh-moq",43874477 "libc",43884478 "log",4479+ "moq-net",43894480 "openh264",43904481 "serde",43914482 "serde_json",
@@ -4813,7 +4904,7 @@ dependencies = [
48134904 "errno",48144905 "libc",48154906 "linux-raw-sys",4816- "windows-sys 0.61.2",4907+ "windows-sys 0.59.0",48174908 ]48184909 48194910 [[package]]
@@ -4822,9 +4913,9 @@ version = "0.23.42"
48224913 source = "registry+https://github.com/rust-lang/crates.io-index"48234914 checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"48244915 dependencies = [4916+ "aws-lc-rs",48254917 "log",48264918 "once_cell",4827- "ring",48284919 "rustls-pki-types",48294920 "rustls-webpki",48304921 "subtle",
@@ -4871,7 +4962,7 @@ dependencies = [
48714962 "security-framework",48724963 "security-framework-sys",48734964 "webpki-root-certs",4874- "windows-sys 0.61.2",4965+ "windows-sys 0.59.0",48754966 ]48764967 48774968 [[package]]
@@ -4886,6 +4977,7 @@ version = "0.103.13"
48864977 source = "registry+https://github.com/rust-lang/crates.io-index"48874978 checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"48884979 dependencies = [4980+ "aws-lc-rs",48894981 "ring",48904982 "rustls-pki-types",48914983 "untrusted",
@@ -5023,7 +5115,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
50235115 checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521"50245116 dependencies = [50255117 "libc",5026- "windows-sys 0.61.2",5118+ "windows-sys 0.59.0",50275119 ]50285120 50295121 [[package]]
@@ -5239,6 +5331,17 @@ dependencies = [
52395331 "stable_deref_trait",52405332 ]52415333 5334+[[package]]5335+name = "sfv"5336+version = "0.14.0"5337+source = "registry+https://github.com/rust-lang/crates.io-index"5338+checksum = "0d471eaefb14f4b30032525bdb124b36e55ba9cb1292080e06f1a236cd10fe87"5339+dependencies = [5340+ "base64 0.22.1",5341+ "indexmap 2.14.0",5342+ "ref-cast",5343+]5344+52425345 [[package]]52435346 name = "sha1_smol"52445347 version = "1.0.1"
@@ -5287,11 +5390,14 @@ name = "sigil-host"
52875390 version = "0.1.0"52885391 dependencies = [52895392 "anyhow",5393+ "bytes",52905394 "clap",52915395 "getrandom 0.4.3",52925396 "input-linux",52935397 "iroh",5398+ "iroh-moq",52945399 "libc",5400+ "moq-net",52955401 "serde",52965402 "serde_json",52975403 "sigil-protocol",
@@ -5300,6 +5406,7 @@ dependencies = [
53005406 "toml 0.9.12+spec-1.1.0",53015407 "tracing",53025408 "tracing-subscriber",5409+ "web-transport-iroh",53035410 "x11rb",53045411 ]53055412
@@ -5390,7 +5497,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
53905497 checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"53915498 dependencies = [53925499 "libc",5393- "windows-sys 0.61.2",5500+ "windows-sys 0.60.2",53945501 ]53955502 53965503 [[package]]
@@ -5555,6 +5662,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
55555662 checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"55565663 dependencies = [55575664 "proc-macro2",5665+ "quote",55585666 "unicode-ident",55595667 ]55605668
@@ -6034,10 +6142,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
60346142 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"60356143 dependencies = [60366144 "fastrand",6037- "getrandom 0.4.3",6145+ "getrandom 0.3.4",60386146 "once_cell",60396147 "rustix",6040- "windows-sys 0.61.2",6148+ "windows-sys 0.59.0",60416149 ]60426150 60436151 [[package]]
@@ -6234,6 +6342,7 @@ version = "0.13.3"
62346342 source = "registry+https://github.com/rust-lang/crates.io-index"62356343 checksum = "d52efb639344a7c6adb8e62c6f3d2c19c001ff1b79a5041ba1c6ed42e19c6aa5"62366344 dependencies = [6345+ "aws-lc-rs",62376346 "base64 0.22.1",62386347 "bytes",62396348 "futures-core",
@@ -6242,7 +6351,6 @@ dependencies = [
62426351 "http",62436352 "httparse",62446353 "rand",6245- "ring",62466354 "rustls-pki-types",62476355 "sha1_smol",62486356 "simdutf8",
@@ -6497,7 +6605,7 @@ dependencies = [
64976605 "png 0.18.1",64986606 "serde",64996607 "thiserror 2.0.19",6500- "windows-sys 0.61.2",6608+ "windows-sys 0.60.2",65016609 ]65026610 65036611 [[package]]
@@ -6526,7 +6634,7 @@ checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e"
65266634 dependencies = [65276635 "memoffset",65286636 "tempfile",6529- "windows-sys 0.61.2",6637+ "windows-sys 0.60.2",65306638 ]65316639 65326640 [[package]]
@@ -6799,6 +6907,31 @@ dependencies = [
67996907 "web-sys",68006908 ]68016909 6910+[[package]]6911+name = "wasmtimer"6912+version = "0.4.3"6913+source = "registry+https://github.com/rust-lang/crates.io-index"6914+checksum = "1c598d6b99ea013e35844697fc4670d08339d5cda15588f193c6beedd12f644b"6915+dependencies = [6916+ "futures",6917+ "js-sys",6918+ "parking_lot",6919+ "pin-utils",6920+ "wasm-bindgen",6921+]6922+6923+[[package]]6924+name = "web-async"6925+version = "0.1.5"6926+source = "registry+https://github.com/rust-lang/crates.io-index"6927+checksum = "1ba1426dcf56093a94d9415ba4c016e45c1323679501e7309f1919ab32f08206"6928+dependencies = [6929+ "tokio",6930+ "tracing",6931+ "wasm-bindgen-futures",6932+ "wasmtimer",6933+]6934+68026935 [[package]]68036936 name = "web-sys"68046937 version = "0.3.103"
@@ -6819,6 +6952,47 @@ dependencies = [
68196952 "wasm-bindgen",68206953 ]68216954 6955+[[package]]6956+name = "web-transport-iroh"6957+version = "0.6.0"6958+source = "registry+https://github.com/rust-lang/crates.io-index"6959+checksum = "d34a91cf241951b8fe02ff216fdf16fc0a52b910568053652d546ded52399f77"6960+dependencies = [6961+ "bytes",6962+ "http",6963+ "iroh",6964+ "n0-error",6965+ "n0-future",6966+ "tokio",6967+ "tracing",6968+ "url",6969+ "web-transport-proto",6970+ "web-transport-trait",6971+]6972+6973+[[package]]6974+name = "web-transport-proto"6975+version = "0.6.0"6976+source = "registry+https://github.com/rust-lang/crates.io-index"6977+checksum = "0225d295c8ac00a2e9a498aefeaf3f3c6186da12a251c938189b15b82ea22808"6978+dependencies = [6979+ "bytes",6980+ "http",6981+ "sfv",6982+ "thiserror 2.0.19",6983+ "tokio",6984+ "url",6985+]6986+6987+[[package]]6988+name = "web-transport-trait"6989+version = "0.3.7"6990+source = "registry+https://github.com/rust-lang/crates.io-index"6991+checksum = "6394e09415ddc1e7c88d605443a3d2f333bf278f74c17020a2b6c44e97507883"6992+dependencies = [6993+ "bytes",6994+]6995+68226996 [[package]]68236997 name = "web_atoms"68246998 version = "0.2.5"
@@ -6967,7 +7141,7 @@ version = "0.1.11"
69677141 source = "registry+https://github.com/rust-lang/crates.io-index"69687142 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"69697143 dependencies = [6970- "windows-sys 0.61.2",7144+ "windows-sys 0.59.0",69717145 ]69727146 69737147 [[package]]
@@ -7503,8 +7677,8 @@ dependencies = [
75037677 "log",75047678 "serde",75057679 "thiserror 2.0.19",7506- "windows 0.62.2",7507- "windows-core 0.62.2",7680+ "windows 0.61.3",7681+ "windows-core 0.61.2",75087682 ]75097683 75107684 [[package]]
M Cargo.toml +7 -0
@@ -2,6 +2,13 @@
22 members = ["crates/sigil-host", "crates/sigil-protocol", "src-tauri"]33 resolver = "2"44 5+[workspace.dependencies]6+bytes = "1.11"7+iroh = { version = "=1.0.3", default-features = false, features = ["metrics", "portmapper", "fast-apple-datapath", "tls-aws-lc-rs"] }8+iroh-moq = { version = "=0.1.0", git = "https://github.com/n0-computer/iroh-live", rev = "5f95758fcd1450e443a9134c9d9342bcc3957b85" }9+moq-net = "=0.1.11"10+web-transport-iroh = "=0.6.0"11+512 [profile.release]613 codegen-units = 1714 lto = "thin"
M README.md +15 -13
@@ -62,13 +62,15 @@ Steam and games.
6262 rollback, and package-owned user assets that follow the active release.6363 6464 The software JPEG client decode fallback and legacy wire compatibility remain.65-Media v3 adds bounded QUIC objects with explicit GOP/object identity, publisher66-priority, delivery expiry, dependent-object cancellation, and client-requested67-discontinuity recovery. It is deliberately MoQ-style rather than an IETF MoQ68-Transport compatibility claim; v2 and v1 remain fallbacks. A dedicated persistent69-PipeWire sink, bounded Opus datagrams, WebCodecs decode, and AudioWorklet70-playback are implemented; longer-run audio/video synchronization measurement71-remains. The debug direct-node bypass is routing, not authentication.65+The preferred video path authenticates the client over a dedicated control66+connection, then admits that exact peer to a session-scoped upstream67+`iroh-moq` broadcast and H.264 track. Configured GOPs become bounded native MoQ68+groups, and a new independently decodable group cancels the stale predecessor.69+The custom media v3, v2, and v1 transports remain explicit compatibility modes.70+A dedicated persistent PipeWire sink, bounded Opus datagrams, WebCodecs decode,71+and AudioWorklet playback are implemented; longer-run audio/video72+synchronization measurement remains. The debug direct-node bypass is routing,73+not authentication.7274 7375 ## Immediate milestones7476
@@ -78,12 +80,12 @@ remains. The debug direct-node bypass is routing, not authentication.
7880 and service startup remain public-alpha acceptance gates.7981 3. **Hardware-proven:** encode H.264 at 1280x800/60 with bounded buffers and no B-frames.8082 4. **Done:** replace base64 WebCodecs delivery with a bounded binary Tauri channel.81-5. **Grouped object transport implemented; hardware proof pending:** bounded82- QUIC media objects carry GOP/object identity and delivery deadlines, cancel83- stale dependent frames, and recover through explicit keyframe requests.84- The external encoder currently responds at its next natural configured IDR;85- in-process force-key-unit control and induced-loss/relay proof remain in86- issue #7.83+5. **Upstream Iroh/MoQ transport implemented; hardware proof pending:** an84+ authenticated control lease gates a session-scoped upstream MoQ broadcast;85+ bounded native groups cancel stale GOPs and recover through explicit86+ keyframe requests. The external encoder currently responds at its next87+ natural configured IDR; in-process force-key-unit control and induced-loss88+ relay proof remain in issue #7.8789 6. **Host hardware-proven; physical client controller pending:** the virtual8890 Xbox-style controller negotiated over Iroh and produced the expected8991 button, stick, trigger, D-pad, and neutral-release events on the Bazzite
M crates/sigil-host/Cargo.toml +5 -1
@@ -19,10 +19,13 @@ path = "src/bin/sigil-probe.rs"
1919 2020 [dependencies]2121 anyhow = "1"22+bytes.workspace = true2223 clap = { version = "4", features = ["derive"] }2324 getrandom = "0.4"24-iroh = "1.0"25+iroh.workspace = true26+iroh-moq.workspace = true2527 libc = "0.2"28+moq-net.workspace = true2629 serde = { version = "1", features = ["derive"] }2730 serde_json = "1"2831 sigil-protocol = { path = "../sigil-protocol" }
@@ -30,6 +33,7 @@ tokio = { version = "1", features = ["io-util", "macros", "net", "process", "rt-
3033 toml = "0.9"3134 tracing = "0.1"3235 tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }36+web-transport-iroh.workspace = true3337 3438 [target.'cfg(target_os = "linux")'.dependencies]3539 input-linux = "0.7.1"
M crates/sigil-host/src/bin/sigil-probe.rs +549 -32
@@ -6,13 +6,16 @@ use std::{
66 use anyhow::{Context, Result, bail, ensure};77 use clap::Parser;88 use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, endpoint::presets};9+use iroh_moq::{Moq, MoqSession};10+use moq_net::{BroadcastConsumer, GroupConsumer, Track, TrackConsumer};911 use sigil_protocol::{10- Capability, ClientHello, FrameFlags, GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN, GAMEPAD_TRIGGER_MAX,11- GamepadState, INPUT_ALPN_V1, InputEvent, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,12- MEDIA_ALPN_V1, MEDIA_ALPN_V2, MEDIA_ALPN_V3, MediaControlRequestV3, MediaFrame, MediaObjectV3,13- PointerSurfaceDimensions, ProtocolError, read_host_hello, read_input_ack, read_media_frame,14- read_media_object, read_media_object_v3, write_client_hello, write_input_event,15- write_media_control_request_v3,12+ CONTROL_ALPN_V1, Capability, ClientHello, FrameFlags, GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN,13+ GAMEPAD_TRIGGER_MAX, GamepadState, INPUT_ALPN_V1, InputEvent, KeyframeRequestReasonV3,14+ MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_ID_V3, MEDIA_ALPN_V1, MEDIA_ALPN_V2, MEDIA_ALPN_V3,15+ MOQ_VIDEO_H264_TRACK, MediaCodec, MediaControlRequestV3, MediaFrame, MediaObjectV3,16+ PointerSurfaceDimensions, ProtocolError, decode_media_frame_object, media_moq_broadcast_name,17+ read_host_hello, read_input_ack, read_media_frame, read_media_object, read_media_object_v3,18+ write_client_hello, write_input_event, write_media_control_request_v3,1619 };1720 1821 const MEDIA_OBJECT_CAPACITY: usize = 4;
@@ -32,16 +35,20 @@ struct Args {
3235 minimum_fps: Option<f64>,3336 #[arg(long, default_value = "1280x800", value_parser = parse_size)]3437 expect_size: (u16, u16),35- /// Exercise independent v2 media objects instead of grouped v3 media36- /// objects. Intended only for compatibility validation.37- #[arg(long, conflicts_with = "media_v1")]38+ /// Exercise custom grouped v3 media instead of upstream MoQ. Intended only39+ /// for compatibility validation.40+ #[arg(long, conflicts_with_all = ["media_v2", "media_v1"])]41+ media_v3: bool,42+ /// Exercise independent v2 media objects instead of upstream MoQ.43+ /// Intended only for compatibility validation.44+ #[arg(long, conflicts_with_all = ["media_v3", "media_v1"])]3845 media_v2: bool,39- /// Exercise the reliable ordered v1 media stream instead of grouped v340- /// media objects. Intended only for compatibility validation.41- #[arg(long, conflicts_with = "media_v2")]46+ /// Exercise the reliable ordered v1 media stream instead of upstream MoQ.47+ /// Intended only for compatibility validation.48+ #[arg(long, conflicts_with_all = ["media_v3", "media_v2"])]4249 media_v1: bool,43- /// Request a configured recovery keyframe after three accepted v3 frames,44- /// then prove no delta history is delivered before the discontinuity.50+ /// Request a configured recovery keyframe after three accepted frames,51+ /// then prove no delta history is delivered before the recovery barrier.4552 #[arg(long)]4653 keyframe_smoke: bool,4754 /// Correlation identifier for `--keyframe-smoke` host evidence.
@@ -59,6 +66,7 @@ struct Args {
5966 6067 #[derive(Clone, Copy, Debug, PartialEq, Eq)]6168 enum MediaTransport {69+ UpstreamMoq,6270 GroupedV3,6371 IndependentV2,6472 ReliableV1,
@@ -67,6 +75,7 @@ enum MediaTransport {
6775 impl MediaTransport {6876 fn alpn(self) -> &'static [u8] {6977 match self {78+ Self::UpstreamMoq => CONTROL_ALPN_V1,7079 Self::GroupedV3 => MEDIA_ALPN_V3,7180 Self::IndependentV2 => MEDIA_ALPN_V2,7281 Self::ReliableV1 => MEDIA_ALPN_V1,
@@ -75,11 +84,22 @@ impl MediaTransport {
7584 7685 fn label(self) -> &'static str {7786 match self {87+ Self::UpstreamMoq => "iroh-moq",7888 Self::GroupedV3 => "grouped-v3",7989 Self::IndependentV2 => "independent-v2",8090 Self::ReliableV1 => "reliable-v1",8191 }8292 }93+94+ fn media_alpn_label(self) -> &'static str {95+ match self {96+ Self::UpstreamMoq => std::str::from_utf8(iroh_moq::ALPN)97+ .expect("pinned iroh-moq ALPN must be printable UTF-8"),98+ Self::GroupedV3 => "sigil/media/3",99+ Self::IndependentV2 => "sigil/media/2",100+ Self::ReliableV1 => "sigil/media/1",101+ }102+ }83103 }84104 85105 const KEYFRAME_SMOKE_REQUEST_AFTER_FRAMES: u32 = 3;
@@ -590,6 +610,292 @@ impl MediaObjectSequenceV3 {
590610 }591611 }592612 613+#[derive(Debug)]614+enum MoqProbeOutcome {615+ Frame {616+ frame: MediaFrame,617+ group_sequence: u64,618+ recovery: bool,619+ },620+ Dropped,621+}622+623+struct MoqProbeGroupCursor {624+ group: GroupConsumer,625+ sequence: u64,626+ object_count: usize,627+ object_bytes: usize,628+ recovery: bool,629+}630+631+/// Own all upstream handles for the complete proof session. Dropping `Moq`632+/// stops its actor, while dropping the consumers cancels their subscriptions.633+struct MoqProbeLifetime {634+ _moq: Moq,635+ session: MoqSession,636+ _broadcast: BroadcastConsumer,637+}638+639+struct MoqProbeReceiver {640+ lifetime: MoqProbeLifetime,641+ track: TrackConsumer,642+ current_group: Option<MoqProbeGroupCursor>,643+ last_group_sequence: Option<u64>,644+ last_frame_sequence: Option<u64>,645+ waiting_for_keyframe: bool,646+ pending_recovery: bool,647+ cancelled_groups: u64,648+ group_gaps: u64,649+ maximum_group_objects: usize,650+ maximum_group_bytes: usize,651+ historical_suffix_frames: u64,652+}653+654+impl MoqProbeReceiver {655+ async fn connect(656+ endpoint: &Endpoint,657+ address: EndpointAddr,658+ session_id: u64,659+ timeout: Duration,660+ ) -> Result<Self> {661+ let broadcast_name = media_moq_broadcast_name(session_id)662+ .context("deriving session-scoped MoQ broadcast name")?;663+ let moq = Moq::new(endpoint.clone());664+ let mut session = tokio::time::timeout(timeout, moq.connect(address.clone()))665+ .await666+ .context("timed out connecting upstream MoQ media session")?667+ .context("connecting upstream MoQ media session")?;668+ ensure!(669+ session.remote_id() == address.id,670+ "upstream MoQ connected to unexpected peer {} instead of {}",671+ session.remote_id(),672+ address.id673+ );674+ let broadcast = tokio::time::timeout(timeout, session.subscribe(&broadcast_name))675+ .await676+ .with_context(|| {677+ format!("timed out subscribing to upstream MoQ broadcast {broadcast_name}")678+ })?679+ .with_context(|| format!("subscribing to upstream MoQ broadcast {broadcast_name}"))?;680+ let track = broadcast681+ .subscribe_track(&Track::new(MOQ_VIDEO_H264_TRACK))682+ .with_context(|| format!("subscribing to MoQ track {MOQ_VIDEO_H264_TRACK}"))?;683+ Ok(Self {684+ lifetime: MoqProbeLifetime {685+ _moq: moq,686+ session,687+ _broadcast: broadcast,688+ },689+ track,690+ current_group: None,691+ last_group_sequence: None,692+ last_frame_sequence: None,693+ waiting_for_keyframe: true,694+ pending_recovery: false,695+ cancelled_groups: 0,696+ group_gaps: 0,697+ maximum_group_objects: 0,698+ maximum_group_bytes: 0,699+ historical_suffix_frames: 0,700+ })701+ }702+703+ fn connection(&self) -> &iroh::endpoint::Connection {704+ self.lifetime.session.conn()705+ }706+707+ /// Stop consuming the current group immediately. The track cursor already708+ /// advanced when the group was selected, so a subsequent read can only709+ /// enter a newer native group; no cancelled delta suffix can be accepted.710+ fn request_recovery(&mut self) -> Option<u64> {711+ self.waiting_for_keyframe = true;712+ self.pending_recovery = true;713+ self.current_group.take().map(|cursor| {714+ self.cancelled_groups = self.cancelled_groups.saturating_add(1);715+ self.last_group_sequence = Some(cursor.sequence);716+ cursor.sequence717+ })718+ }719+720+ async fn next(&mut self) -> Result<Option<MoqProbeOutcome>> {721+ loop {722+ if self.current_group.is_none() {723+ let group = self724+ .track725+ .next_group()726+ .await727+ .context("reading the next sequential upstream MoQ group")?;728+ let Some(group) = group else {729+ return Ok(None);730+ };731+ let sequence = group.sequence;732+ let group_gap = classify_moq_probe_group(self.last_group_sequence, sequence)?;733+ if group_gap {734+ self.group_gaps = self.group_gaps.saturating_add(1);735+ self.waiting_for_keyframe = true;736+ self.pending_recovery = true;737+ }738+ self.current_group = Some(MoqProbeGroupCursor {739+ group,740+ sequence,741+ object_count: 0,742+ object_bytes: 0,743+ recovery: self.pending_recovery,744+ });745+ self.pending_recovery = false;746+ }747+748+ let cursor = self749+ .current_group750+ .as_mut()751+ .expect("MoQ group cursor was initialized");752+ let object = match cursor.group.read_frame().await {753+ Ok(object) => object,754+ Err(error) if moq_probe_error_is_recoverable(&error) => {755+ self.cancelled_groups = self.cancelled_groups.saturating_add(1);756+ self.last_group_sequence = Some(cursor.sequence);757+ self.current_group = None;758+ self.waiting_for_keyframe = true;759+ self.pending_recovery = true;760+ return Ok(Some(MoqProbeOutcome::Dropped));761+ }762+ Err(error) => {763+ bail!(764+ "upstream MoQ group {} failed terminally: {error}",765+ cursor.sequence766+ );767+ }768+ };769+ let Some(object) = object else {770+ ensure!(771+ cursor.object_count > 0,772+ "upstream MoQ group {} was empty",773+ cursor.sequence774+ );775+ self.last_group_sequence = Some(cursor.sequence);776+ self.current_group = None;777+ continue;778+ };779+780+ let next_bytes = validate_moq_probe_object_bounds(781+ cursor.sequence,782+ cursor.object_count,783+ cursor.object_bytes,784+ object.len(),785+ )?;786+ let frame = decode_media_frame_object(&object).with_context(|| {787+ format!(788+ "decoding upstream MoQ group {} object {}",789+ cursor.sequence, cursor.object_count790+ )791+ })?;792+ let first_object = cursor.object_count == 0;793+ let contiguous = validate_moq_probe_frame(794+ cursor.sequence,795+ first_object,796+ self.last_frame_sequence,797+ &frame,798+ )?;799+ let recovery = cursor.recovery800+ || self.waiting_for_keyframe801+ || (first_object && !contiguous)802+ || frame.header.flags.contains(FrameFlags::DISCONTINUITY);803+ if self.waiting_for_keyframe && !first_object {804+ self.historical_suffix_frames = self.historical_suffix_frames.saturating_add(1);805+ bail!(806+ "upstream MoQ delivered a historical delta suffix while recovery was pending"807+ );808+ }809+ cursor.object_count += 1;810+ cursor.object_bytes = next_bytes;811+ self.maximum_group_objects = self.maximum_group_objects.max(cursor.object_count);812+ self.maximum_group_bytes = self.maximum_group_bytes.max(next_bytes);813+ self.last_frame_sequence = Some(frame.header.sequence);814+ self.waiting_for_keyframe = false;815+ return Ok(Some(MoqProbeOutcome::Frame {816+ frame,817+ group_sequence: cursor.sequence,818+ recovery,819+ }));820+ }821+ }822+}823+824+fn classify_moq_probe_group(previous: Option<u64>, current: u64) -> Result<bool> {825+ let Some(previous) = previous else {826+ return Ok(false);827+ };828+ ensure!(829+ current > previous,830+ "upstream MoQ group sequence did not increase: previous={previous}, current={current}"831+ );832+ Ok(previous.checked_add(1) != Some(current))833+}834+835+fn validate_moq_probe_object_bounds(836+ group_sequence: u64,837+ object_count: usize,838+ object_bytes: usize,839+ object_len: usize,840+) -> Result<usize> {841+ let maximum_objects = MAX_MEDIA_OBJECT_ID_V3 as usize + 1;842+ ensure!(843+ object_count < maximum_objects,844+ "upstream MoQ group {group_sequence} exceeded {maximum_objects} objects"845+ );846+ let next_bytes = object_bytes847+ .checked_add(object_len)848+ .context("upstream MoQ group byte count overflowed")?;849+ ensure!(850+ next_bytes <= MAX_MEDIA_GROUP_BYTES_V3,851+ "upstream MoQ group {group_sequence} exceeded {MAX_MEDIA_GROUP_BYTES_V3} bytes"852+ );853+ Ok(next_bytes)854+}855+856+fn validate_moq_probe_frame(857+ group_sequence: u64,858+ first_object: bool,859+ last_frame_sequence: Option<u64>,860+ frame: &MediaFrame,861+) -> Result<bool> {862+ if first_object {863+ ensure!(864+ frame.header.codec == MediaCodec::H264865+ && frame.header.flags.contains(FrameFlags::KEYFRAME)866+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),867+ "upstream MoQ group {group_sequence} did not begin with a configured H.264 keyframe"868+ );869+ // A native group cancellation is itself the transport discontinuity.870+ // Portal marks the first configured IDR in the replacement group as a871+ // decoder discontinuity even when the embedded compatibility header872+ // does not redundantly carry the v3 DISCONTINUITY flag.873+ }874+ let contiguous = last_frame_sequence875+ .is_none_or(|previous| previous.checked_add(1) == Some(frame.header.sequence));876+ ensure!(877+ last_frame_sequence.is_none_or(|previous| frame.header.sequence > previous),878+ "upstream MoQ group {group_sequence} contains a non-monotonic frame sequence"879+ );880+ ensure!(881+ first_object || contiguous,882+ "upstream MoQ group {group_sequence} contains a non-contiguous frame sequence"883+ );884+ Ok(contiguous)885+}886+887+fn moq_probe_error_is_recoverable(error: &moq_net::Error) -> bool {888+ matches!(889+ error,890+ moq_net::Error::Cancel891+ | moq_net::Error::Old892+ | moq_net::Error::Timeout893+ | moq_net::Error::Dropped894+ | moq_net::Error::CacheFull895+ | moq_net::Error::Remote(0 | 2 | 3 | 24 | 26)896+ )897+}898+593899 struct AcceptedMedia {594900 flags: FrameFlags,595901 width: u16,
@@ -634,7 +940,7 @@ async fn main() -> Result<()> {
634940 );635941 ensure!(636942 !args.keyframe_smoke || (!args.media_v1 && !args.media_v2),637- "--keyframe-smoke requires grouped v3 media"943+ "--keyframe-smoke requires upstream MoQ or grouped v3 media"638944 );639945 ensure!(640946 !args.keyframe_smoke || args.frames >= KEYFRAME_SMOKE_MINIMUM_FRAMES,
@@ -655,8 +961,10 @@ async fn main() -> Result<()> {
655961 MediaTransport::ReliableV1656962 } else if args.media_v2 {657963 MediaTransport::IndependentV2658- } else {964+ } else if args.media_v3 {659965 MediaTransport::GroupedV3966+ } else {967+ MediaTransport::UpstreamMoq660968 };661969 662970 let media_connection = endpoint
@@ -667,22 +975,27 @@ async fn main() -> Result<()> {
667975 .open_bi()668976 .await669977 .context("opening media stream")?;978+ let handshake_name = if media_transport == MediaTransport::UpstreamMoq {979+ "control"980+ } else {981+ "media"982+ };670983 let media_negotiation = negotiate(671984 &mut media_send,672985 &mut media_recv,673986 nonce,674987 vec![Capability::VideoH264],675988 Capability::VideoH264,676- "media",989+ handshake_name,677990 )678991 .await?;679992 let session_id = media_negotiation.session_id;680993 let mut media_send = Some(media_send);681994 let mut media_recv = Some(media_recv);682995 match media_transport {683- MediaTransport::GroupedV3 => {996+ MediaTransport::UpstreamMoq | MediaTransport::GroupedV3 => {684997 // Sigil finishes its response half after HostHello. Keep our send685- // half alive as the bounded v3 keyframe-control stream.998+ // half alive as the bounded keyframe-control stream.686999 drop(media_recv.take());6871000 }6881001 MediaTransport::IndependentV2 => {
@@ -696,8 +1009,22 @@ async fn main() -> Result<()> {
6961009 MediaTransport::ReliableV1 => {}6971010 }6981011 1012+ let mut moq_receiver = if media_transport == MediaTransport::UpstreamMoq {1013+ Some(1014+ MoqProbeReceiver::connect(1015+ &endpoint,1016+ address.clone(),1017+ session_id,1018+ Duration::from_secs(args.timeout_seconds),1019+ )1020+ .await?,1021+ )1022+ } else {1023+ None1024+ };1025+6991026 let input_connection = endpoint700- .connect(address, INPUT_ALPN_V1)1027+ .connect(address.clone(), INPUT_ALPN_V1)7011028 .await7021029 .context("connecting input protocol")?;7031030 let (mut input_send, mut input_recv) = input_connection
@@ -829,6 +1156,62 @@ async fn main() -> Result<()> {
8291156 8301157 while received < args.frames {8311158 let (frame, recovery_frame): (AcceptedMedia, bool) = match media_transport {1159+ MediaTransport::UpstreamMoq => loop {1160+ let outcome = tokio::time::timeout(1161+ Duration::from_secs(args.timeout_seconds)1162+ .saturating_add(Duration::from_secs(1)),1163+ moq_receiver1164+ .as_mut()1165+ .expect("upstream MoQ receiver is present")1166+ .next(),1167+ )1168+ .await1169+ .context("timed out waiting for an upstream MoQ media object")??1170+ .context("host closed the upstream MoQ video track")?;1171+ match outcome {1172+ MoqProbeOutcome::Dropped => {1173+ media_objects_dropped = media_objects_dropped.saturating_add(1);1174+ }1175+ MoqProbeOutcome::Frame {1176+ frame,1177+ group_sequence,1178+ recovery,1179+ } => {1180+ let requested_recovery =1181+ keyframe_request_sent && !keyframe_recovery_verified;1182+ if requested_recovery {1183+ ensure!(1184+ recovery1185+ && frame.header.flags.contains(FrameFlags::KEYFRAME)1186+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),1187+ "MoQ keyframe request recovery did not begin at a configured IDR barrier"1188+ );1189+ ensure!(1190+ keyframe_request_group_id1191+ .is_some_and(|prior| group_sequence > prior),1192+ "MoQ keyframe request recovery did not advance the native group"1193+ );1194+ ensure!(1195+ keyframe_request_last_sequence1196+ .is_some_and(|prior| frame.header.sequence > prior),1197+ "MoQ keyframe request recovery did not advance media sequence"1198+ );1199+ keyframe_recovery_verified = true;1200+ }1201+ break (1202+ AcceptedMedia {1203+ flags: frame.header.flags,1204+ width: frame.header.width,1205+ height: frame.header.height,1206+ sequence: frame.header.sequence,1207+ payload_len: frame.payload.len(),1208+ v3_group_id: Some(group_sequence),1209+ },1210+ requested_recovery,1211+ );1212+ }1213+ }1214+ },8321215 MediaTransport::ReliableV1 => (8331216 tokio::time::timeout(8341217 Duration::from_secs(args.timeout_seconds),
@@ -1008,7 +1391,22 @@ async fn main() -> Result<()> {
10081391 keyframe_request_group_id = frame.v3_group_id;10091392 keyframe_request_last_sequence = Some(frame.sequence);10101393 keyframe_request_sent = true;1011- object_sequence_v3.request_recovery();1394+ match media_transport {1395+ MediaTransport::UpstreamMoq => {1396+ let cancelled_group = moq_receiver1397+ .as_mut()1398+ .expect("upstream MoQ receiver is present")1399+ .request_recovery();1400+ ensure!(1401+ cancelled_group == keyframe_request_group_id,1402+ "MoQ recovery did not cancel the currently accepted native group"1403+ );1404+ }1405+ MediaTransport::GroupedV3 => object_sequence_v3.request_recovery(),1406+ MediaTransport::IndependentV2 | MediaTransport::ReliableV1 => {1407+ unreachable!("keyframe smoke excludes v2 and v1")1408+ }1409+ }10121410 }10131411 }10141412
@@ -1042,7 +1440,25 @@ async fn main() -> Result<()> {
10421440 "probe sustained only {accepted_fps:.3} accepted fps; required at least {minimum_fps:.3} fps"10431441 );10441442 }1045- let (path_mode, path_rtt_ms) = selected_path_diagnostics(&media_connection);1443+ let diagnostics_connection = moq_receiver1444+ .as_ref()1445+ .map_or(&media_connection, MoqProbeReceiver::connection);1446+ let (path_mode, path_rtt_ms) = selected_path_diagnostics(diagnostics_connection);1447+ let (1448+ moq_cancelled_groups,1449+ moq_group_gaps,1450+ moq_maximum_group_objects,1451+ moq_maximum_group_bytes,1452+ moq_historical_suffix_frames,1453+ ) = moq_receiver.as_ref().map_or((0, 0, 0, 0, 0), |receiver| {1454+ (1455+ receiver.cancelled_groups,1456+ receiver.group_gaps,1457+ receiver.maximum_group_objects,1458+ receiver.maximum_group_bytes,1459+ receiver.historical_suffix_frames,1460+ )1461+ });10461462 input_send.finish().context("finishing input stream")?;10471463 if let Some(mut media_send) = media_send {10481464 media_send
@@ -1054,6 +1470,9 @@ async fn main() -> Result<()> {
10541470 // short-lived probes, which obscures whether the host released its10551471 // one-client lease deterministically.10561472 input_connection.close(0_u32.into(), b"probe complete");1473+ if let Some(receiver) = moq_receiver.as_ref() {1474+ receiver.connection().close(0_u32.into(), b"probe complete");1475+ }10571476 media_connection.close(0_u32.into(), b"probe complete");10581477 tokio::time::sleep(Duration::from_millis(50)).await;10591478 endpoint.close().await;
@@ -1067,6 +1486,54 @@ async fn main() -> Result<()> {
10671486 println!("media_objects_dropped={media_objects_dropped}");10681487 println!("media_objects_late={media_objects_late}");10691488 println!("transport={}", media_transport.label());1489+ println!(1490+ "control_alpn={}",1491+ if media_transport == MediaTransport::UpstreamMoq {1492+ "sigil/control/1"1493+ } else {1494+ "not-used"1495+ }1496+ );1497+ println!("transport_alpn={}", media_transport.media_alpn_label());1498+ println!("first_configured_idr=ok");1499+ println!("frame_sequence=monotonic");1500+ println!(1501+ "group_sequence={}",1502+ if media_transport == MediaTransport::UpstreamMoq {1503+ "monotonic"1504+ } else {1505+ "not-applicable"1506+ }1507+ );1508+ if media_transport == MediaTransport::UpstreamMoq {1509+ println!("moq_group_capacity=1");1510+ println!("moq_cancelled_groups={moq_cancelled_groups}");1511+ println!("moq_group_gaps={moq_group_gaps}");1512+ println!("moq_unrecovered_group_gaps=0");1513+ println!("moq_maximum_group_objects={moq_maximum_group_objects}");1514+ println!("moq_maximum_group_bytes={moq_maximum_group_bytes}");1515+ println!("moq_historical_suffix_frames={moq_historical_suffix_frames}");1516+ } else {1517+ for field in [1518+ "moq_group_capacity",1519+ "moq_cancelled_groups",1520+ "moq_group_gaps",1521+ "moq_unrecovered_group_gaps",1522+ "moq_maximum_group_objects",1523+ "moq_maximum_group_bytes",1524+ "moq_historical_suffix_frames",1525+ ] {1526+ println!("{field}=not-applicable");1527+ }1528+ }1529+ println!(1530+ "recovery_barrier={}",1531+ if args.keyframe_smoke {1532+ "configured-idr"1533+ } else {1534+ "not-requested"1535+ }1536+ );10701537 println!(10711538 "keyframe_recovery={}",10721539 if args.keyframe_smoke {
@@ -1625,21 +2092,71 @@ mod tests {
16252092 );16262093 }16272094 2095+ #[test]2096+ fn upstream_moq_group_and_object_limits_fail_closed() {2097+ assert!(!classify_moq_probe_group(None, 8).unwrap());2098+ assert!(!classify_moq_probe_group(Some(8), 9).unwrap());2099+ assert!(classify_moq_probe_group(Some(8), 10).unwrap());2100+ assert!(classify_moq_probe_group(Some(8), 8).is_err());2101+ assert!(classify_moq_probe_group(Some(8), 7).is_err());2102+2103+ assert_eq!(validate_moq_probe_object_bounds(4, 0, 0, 42).unwrap(), 42);2104+ assert!(2105+ validate_moq_probe_object_bounds(4, MAX_MEDIA_OBJECT_ID_V3 as usize + 1, 0, 1,)2106+ .is_err()2107+ );2108+ assert!(validate_moq_probe_object_bounds(4, 0, MAX_MEDIA_GROUP_BYTES_V3, 1).is_err());2109+ assert!(validate_moq_probe_object_bounds(4, 0, usize::MAX, 1).is_err());2110+ }2111+2112+ #[test]2113+ fn upstream_moq_requires_configured_group_zero_and_idr_recovery() {2114+ let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2115+ let recovery = keyframe.union(FrameFlags::DISCONTINUITY);2116+ let delta = media_frame(11, FrameFlags::NONE);2117+ let configured = media_frame(10, keyframe);2118+ let configured_next = media_frame(12, keyframe);2119+ let recovered = media_frame(20, recovery);2120+2121+ assert!(validate_moq_probe_frame(3, true, None, &delta).is_err());2122+ assert!(validate_moq_probe_frame(3, true, None, &configured).is_ok());2123+ assert!(validate_moq_probe_frame(4, true, Some(11), &configured).is_err());2124+ assert!(validate_moq_probe_frame(4, true, Some(11), &configured_next).is_ok());2125+ assert!(validate_moq_probe_frame(4, true, Some(11), &recovered).is_ok());2126+ assert!(validate_moq_probe_frame(4, false, Some(10), &delta).is_ok());2127+ assert!(validate_moq_probe_frame(4, false, Some(9), &delta).is_err());2128+ }2129+2130+ #[test]2131+ fn upstream_moq_transport_reports_the_pinned_native_alpn() {2132+ assert_eq!(MediaTransport::UpstreamMoq.label(), "iroh-moq");2133+ assert_eq!(MediaTransport::UpstreamMoq.alpn(), CONTROL_ALPN_V1);2134+ assert_eq!(2135+ MediaTransport::UpstreamMoq.media_alpn_label(),2136+ "moq-lite-04"2137+ );2138+ assert_eq!(2139+ MediaTransport::UpstreamMoq.media_alpn_label().as_bytes(),2140+ iroh_moq::ALPN2141+ );2142+ }2143+16282144 #[test]16292145 fn media_compatibility_flags_are_mutually_exclusive() {16302146 let node_id = SecretKey::from_bytes(&[0x2a; 32]).public().to_string();16312147 let default = Args::try_parse_from(["sigil-probe", "--node-id", &node_id]).unwrap();16322148 assert!(!default.media_v1);16332149 assert!(!default.media_v2);1634- assert!(1635- Args::try_parse_from([1636- "sigil-probe",1637- "--node-id",1638- &node_id,1639- "--media-v1",1640- "--media-v2",1641- ])1642- .is_err()1643- );2150+ assert!(!default.media_v3);2151+ for flags in [2152+ ["--media-v1", "--media-v2"],2153+ ["--media-v1", "--media-v3"],2154+ ["--media-v2", "--media-v3"],2155+ ] {2156+ assert!(2157+ Args::try_parse_from(["sigil-probe", "--node-id", &node_id, flags[0], flags[1],])2158+ .is_err()2159+ );2160+ }16442161 }16452162 }
M crates/sigil-host/src/main.rs +22 -4
@@ -22,9 +22,10 @@ use clap::{Args, Parser, Subcommand, ValueEnum};
2222 use iroh::endpoint::{QuicTransportConfig, presets};2323 use iroh::protocol::Router;2424 use iroh::{Endpoint, EndpointId};25+use moq_net::Origin;2526 use sigil_protocol::{26- AUDIO_ALPN_V1, INPUT_ALPN_V1, InvitationGrants, MAX_INVITATION_TTL_SECS, MEDIA_ALPN_V1,27- MEDIA_ALPN_V2, MEDIA_ALPN_V3, SignedInvitation,27+ AUDIO_ALPN_V1, CONTROL_ALPN_V1, INPUT_ALPN_V1, InvitationGrants, MAX_INVITATION_TTL_SECS,28+ MEDIA_ALPN_V1, MEDIA_ALPN_V2, MEDIA_ALPN_V3, SignedInvitation,2829 };2930 use tracing::{info, warn};3031 use tracing_subscriber::EnvFilter;
@@ -36,7 +37,8 @@ use crate::config::{HostConfig, InputMode, VideoSource};
3637 use crate::cursor::PointerPositionTracker;3738 use crate::input::InputBackend;3839 use crate::server::{39- AudioHandler, InputHandler, MediaHandler, MediaV2Handler, MediaV3Handler, SessionRegistry,40+ AudioHandler, AuthorizedMoqHandler, ControlHandler, InputHandler, MediaHandler, MediaV2Handler,41+ MediaV3Handler, SessionRegistry,4042 };4143 4244 const CONNECTION_IDLE_TIMEOUT: Duration = Duration::from_secs(5);
@@ -638,7 +640,23 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
638640 }639641 640642 let sessions = Arc::new(SessionRegistry::default());643+ let moq_origin = Origin::random();641644 let router = Router::builder(endpoint)645+ .accept(646+ CONTROL_ALPN_V1,647+ ControlHandler {648+ config: config.clone(),649+ sessions: Arc::clone(&sessions),650+ authorization: authorization.clone(),651+ },652+ )653+ .accept(654+ iroh_moq::ALPN,655+ AuthorizedMoqHandler {656+ sessions: Arc::clone(&sessions),657+ origin: moq_origin,658+ },659+ )642660 .accept(643661 MEDIA_ALPN_V3,644662 MediaV3Handler {
@@ -660,7 +678,7 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
660678 MediaHandler {661679 config: config.clone(),662680 sessions: Arc::clone(&sessions),663- authorization,681+ authorization: authorization.clone(),664682 },665683 )666684 .accept(
M crates/sigil-host/src/server.rs +862 -4
@@ -8,14 +8,19 @@ use anyhow::{Context, Result, bail, ensure};
88 use iroh::EndpointId;99 use iroh::endpoint::{Connection, SendStream};1010 use iroh::protocol::ProtocolHandler;11+use moq_net::{12+ Broadcast, BroadcastConsumer, BroadcastProducer, Error as MoqError, GroupProducer, Origin,13+ Track, TrackProducer,14+};1115 use sigil_protocol::{1216 AUDIO_HEADER_LEN, AudioFlags, AudioPacket, AudioPacketHeader, Capability, ClientHello,1317 FrameFlags, HostHello, InputAck, InvitationGrants, KeyframeRequestReasonV3,1418 MAX_AUDIO_PAYLOAD_LEN, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,15- MAX_MEDIA_OBJECT_ID_V3, MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MediaControlRequestV3,16- MediaFrame, MediaFrameHeader, MediaObjectHeaderV3, MediaObjectV3, read_client_hello,17- read_input_event, read_media_control_request_v3, write_host_hello, write_input_ack,18- write_media_frame, write_media_object_v3,19+ MAX_MEDIA_OBJECT_ID_V3, MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MOQ_VIDEO_H264_TRACK,20+ MediaControlRequestV3, MediaFrame, MediaFrameHeader, MediaObjectHeaderV3, MediaObjectV3,21+ encode_media_frame_object, media_moq_broadcast_name, read_client_hello, read_input_event,22+ read_media_control_request_v3, write_host_hello, write_input_ack, write_media_frame,23+ write_media_object_v3,1924 };2025 use tracing::{debug, error, info, warn};2126
@@ -61,15 +66,52 @@ const GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN: Duration = Duration::from_millis(1
6166 const GAMESCOPE_STARTUP_TARGET_MIN_FRAMES: u64 = 8;6267 const GAMESCOPE_STARTUP_TARGET_MIN_FPS: f64 = 45.0;6368 const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);69+const MOQ_ATTACHMENT_TIMEOUT: Duration = Duration::from_secs(10);70+const MOQ_VIDEO_TRACK_PRIORITY: u8 = u8::MAX;71+const MOQ_REJECT_CODE: u32 = 0x534d;6472 6573 #[derive(Debug)]6674 pub struct SessionRegistry {6775 active: Mutex<Option<ActiveSession>>,76+ pending_moq: Mutex<Option<PendingMoqAttachment>>,6877 next_session_id: AtomicU64,6978 session_changed: tokio::sync::Notify,7079 pending_handshakes: tokio::sync::Semaphore,7180 }7281 82+struct PendingMoqAttachment {83+ remote: EndpointId,84+ session_id: u64,85+ broadcast_name: String,86+ broadcast: BroadcastConsumer,87+ attached: tokio::sync::oneshot::Sender<()>,88+ closed: tokio::sync::oneshot::Sender<()>,89+}90+91+impl std::fmt::Debug for PendingMoqAttachment {92+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {93+ formatter94+ .debug_struct("PendingMoqAttachment")95+ .field("remote", &self.remote)96+ .field("session_id", &self.session_id)97+ .field("broadcast_name", &self.broadcast_name)98+ .finish_non_exhaustive()99+ }100+}101+102+struct ClaimedMoqAttachment {103+ session_id: u64,104+ broadcast_name: String,105+ broadcast: BroadcastConsumer,106+ attached: tokio::sync::oneshot::Sender<()>,107+ closed: tokio::sync::oneshot::Sender<()>,108+}109+110+struct MoqAttachmentWait {111+ attached: tokio::sync::oneshot::Receiver<()>,112+ closed: tokio::sync::oneshot::Receiver<()>,113+}114+73115 #[derive(Clone, Copy, Debug)]74116 struct ActiveSession {75117 remote: EndpointId,
@@ -86,6 +128,7 @@ impl Default for SessionRegistry {
86128 fn default() -> Self {87129 Self {88130 active: Mutex::new(None),131+ pending_moq: Mutex::new(None),89132 next_session_id: AtomicU64::new(0),90133 session_changed: tokio::sync::Notify::new(),91134 pending_handshakes: tokio::sync::Semaphore::new(MAX_PENDING_HANDSHAKES),
@@ -167,12 +210,79 @@ impl SessionRegistry {
167210 })168211 }169212 213+ fn expect_moq(214+ &self,215+ remote: EndpointId,216+ session_id: u64,217+ broadcast_name: String,218+ broadcast: BroadcastConsumer,219+ ) -> Result<MoqAttachmentWait> {220+ let active = self.active.lock().expect("session registry poisoned");221+ ensure!(222+ active.is_some_and(|session| {223+ session.media_active && session.remote == remote && session.session_id == session_id224+ }),225+ "MoQ expectation does not match the active control session"226+ );227+ let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");228+ ensure!(229+ pending.is_none(),230+ "active control session already expects MoQ"231+ );232+ let (attached, attached_rx) = tokio::sync::oneshot::channel();233+ let (closed, closed_rx) = tokio::sync::oneshot::channel();234+ *pending = Some(PendingMoqAttachment {235+ remote,236+ session_id,237+ broadcast_name,238+ broadcast,239+ attached,240+ closed,241+ });242+ Ok(MoqAttachmentWait {243+ attached: attached_rx,244+ closed: closed_rx,245+ })246+ }247+248+ fn claim_moq(&self, remote: EndpointId) -> Result<ClaimedMoqAttachment> {249+ let active = self.active.lock().expect("session registry poisoned");250+ let session = active251+ .as_ref()252+ .filter(|session| session.media_active && session.remote == remote)253+ .context("MoQ connection does not match the active control session")?;254+ let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");255+ let attachment = pending256+ .as_ref()257+ .filter(|attachment| {258+ attachment.remote == remote && attachment.session_id == session.session_id259+ })260+ .context("active control session is not expecting a MoQ connection")?;261+ debug_assert_eq!(attachment.remote, session.remote);262+ let attachment = pending263+ .take()264+ .expect("validated pending MoQ attachment disappeared");265+ Ok(ClaimedMoqAttachment {266+ session_id: attachment.session_id,267+ broadcast_name: attachment.broadcast_name,268+ broadcast: attachment.broadcast,269+ attached: attachment.attached,270+ closed: attachment.closed,271+ })272+ }273+170274 fn release(&self, remote: EndpointId, session_id: u64) {171275 let mut active = self.active.lock().expect("session registry poisoned");172276 if let Some(session) = active.as_mut()173277 && session.remote == remote174278 && session.session_id == session_id175279 {280+ let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");281+ if pending.as_ref().is_some_and(|attachment| {282+ attachment.remote == remote && attachment.session_id == session_id283+ }) {284+ *pending = None;285+ }176286 // Keep the registry occupied until the input handler has observed177287 // media shutdown and released all held uinput transitions. This178288 // prevents a reconnect from sharing the device with a draining
@@ -701,6 +811,128 @@ impl MediaV3GroupCursor {
701811 }702812 }703813 814+#[derive(Clone, Copy, Debug, PartialEq, Eq)]815+enum MoqGroupDecision {816+ Published {817+ group_id: u64,818+ frame_id: u32,819+ cancelled_previous: bool,820+ },821+ SkipUntilKeyframe,822+ EnterResync,823+}824+825+/// Owns the single bounded live MoQ track. One configured H.264 GOP maps to826+/// one native MoQ group; its application frame sequence remains inside the827+/// encoded object envelope and is never reused as the transport group id.828+struct MoqGroupPublisher {829+ track: TrackProducer,830+ current_group: Option<GroupProducer>,831+ cursor: MediaV3GroupCursor,832+ object_bytes: usize,833+}834+835+impl MoqGroupPublisher {836+ fn new(track: TrackProducer) -> Self {837+ Self {838+ track,839+ current_group: None,840+ cursor: MediaV3GroupCursor::default(),841+ object_bytes: 0,842+ }843+ }844+845+ fn publish(846+ &mut self,847+ config: &HostConfig,848+ frame: &EncodedFrame,849+ replay_discontinuity: bool,850+ ) -> Result<MoqGroupDecision> {851+ let position = match self.cursor.classify(frame) {852+ MediaV3GroupDecision::Send(position) => position,853+ MediaV3GroupDecision::SkipUntilKeyframe => {854+ return Ok(MoqGroupDecision::SkipUntilKeyframe);855+ }856+ MediaV3GroupDecision::EnterResync => {857+ self.abort_current();858+ return Ok(MoqGroupDecision::EnterResync);859+ }860+ };861+ let object = encode_media_frame_object(&media_frame_for_encoded(862+ config,863+ frame,864+ replay_discontinuity || position.discontinuity,865+ )?)?;866+ let next_object_bytes = if position.object_id == 0 {867+ Some(object.len())868+ } else {869+ self.object_bytes.checked_add(object.len())870+ };871+ let Some(next_object_bytes) =872+ next_object_bytes.filter(|bytes| *bytes <= MAX_MEDIA_GROUP_BYTES_V3)873+ else {874+ self.cursor.request_keyframe();875+ self.abort_current();876+ return Ok(MoqGroupDecision::EnterResync);877+ };878+879+ if position.object_id == 0 {880+ // A new independently-decodable GOP supersedes the previous one.881+ // Actively aborting it cancels a slow subscriber rather than882+ // retaining a playable history behind the live edge.883+ let cancelled_previous = self.abort_current().is_some();884+ let mut group = self885+ .track886+ .append_group()887+ .context("creating sequential MoQ video group")?;888+ let group_id = group.sequence;889+ group890+ .write_frame(object)891+ .context("writing configured keyframe to MoQ group")?;892+ self.object_bytes = next_object_bytes;893+ self.current_group = Some(group);894+ return Ok(MoqGroupDecision::Published {895+ group_id,896+ frame_id: 0,897+ cancelled_previous,898+ });899+ }900+901+ let group = self902+ .current_group903+ .as_mut()904+ .context("MoQ delta frame has no active configured-keyframe group")?;905+ let group_id = group.sequence;906+ group907+ .write_frame(object)908+ .context("writing delta access unit to MoQ group")?;909+ self.object_bytes = next_object_bytes;910+ Ok(MoqGroupDecision::Published {911+ group_id,912+ frame_id: position.object_id,913+ cancelled_previous: false,914+ })915+ }916+917+ fn request_keyframe(&mut self) -> Option<u64> {918+ self.cursor.request_keyframe();919+ self.abort_current()920+ }921+922+ fn abort_current(&mut self) -> Option<u64> {923+ self.object_bytes = 0;924+ let mut group = self.current_group.take()?;925+ let group_id = group.sequence;926+ let _ = group.abort(MoqError::Cancel);927+ Some(group_id)928+ }929+930+ fn abort(mut self) {931+ self.abort_current();932+ let _ = self.track.abort(MoqError::Cancel);933+ }934+}935+704936 #[derive(Debug, PartialEq, Eq)]705937 enum MediaV3ScheduleDecision {706938 Send {
@@ -1096,6 +1328,25 @@ pub struct MediaV3Handler {
10961328 pub authorization: AuthorizationPolicy,10971329 }10981330 1331+#[derive(Clone, Debug)]1332+pub struct ControlHandler {1333+ pub config: HostConfig,1334+ pub sessions: Arc<SessionRegistry>,1335+ pub authorization: AuthorizationPolicy,1336+}1337+1338+/// Upstream MoQ admission guarded by an already-authenticated control lease.1339+///1340+/// This deliberately does not use `iroh_moq::Moq::protocol_handler`: that1341+/// actor makes a completed session globally visible before application-level1342+/// acceptance. Consuming the exact pending attachment first prevents MoQ from1343+/// bypassing Sigil's invitation, enrollment, and one-client gate.1344+#[derive(Clone, Debug)]1345+pub struct AuthorizedMoqHandler {1346+ pub sessions: Arc<SessionRegistry>,1347+ pub origin: Origin,1348+}1349+10991350 impl ProtocolHandler for MediaHandler {11001351 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {11011352 let remote = connection.remote_id();
@@ -1147,6 +1398,41 @@ impl ProtocolHandler for MediaV3Handler {
11471398 }11481399 }11491400 1401+impl ProtocolHandler for ControlHandler {1402+ async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {1403+ let remote = connection.remote_id();1404+ if let Err(error) = serve_control_moq(1405+ connection,1406+ self.config.clone(),1407+ &self.sessions,1408+ &self.authorization,1409+ )1410+ .await1411+ {1412+ warn!(%remote, %error, "MoQ control connection ended");1413+ }1414+ Ok(())1415+ }1416+}1417+1418+impl ProtocolHandler for AuthorizedMoqHandler {1419+ async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {1420+ let remote = connection.remote_id();1421+ let attachment = match self.sessions.claim_moq(remote) {1422+ Ok(attachment) => attachment,1423+ Err(error) => {1424+ connection.close(MOQ_REJECT_CODE.into(), b"unauthorized MoQ attachment");1425+ warn!(%remote, %error, "rejected unsolicited MoQ connection");1426+ return Ok(());1427+ }1428+ };1429+ if let Err(error) = serve_authorized_moq(connection, self.origin, attachment).await {1430+ warn!(%remote, %error, "authorized MoQ connection ended");1431+ }1432+ Ok(())1433+ }1434+}1435+11501436 #[derive(Clone, Debug)]11511437 pub struct InputHandler {11521438 pub backend: InputBackend,
@@ -1187,6 +1473,344 @@ impl ProtocolHandler for InputHandler {
11871473 }11881474 }11891475 1476+async fn serve_authorized_moq(1477+ connection: Connection,1478+ origin: Origin,1479+ attachment: ClaimedMoqAttachment,1480+) -> Result<()> {1481+ let ClaimedMoqAttachment {1482+ session_id,1483+ broadcast_name,1484+ broadcast,1485+ attached,1486+ closed,1487+ } = attachment;1488+ let result: Result<()> = async {1489+ let web_transport = web_transport_iroh::Session::raw(connection);1490+ let session = tokio::time::timeout(1491+ MOQ_ATTACHMENT_TIMEOUT,1492+ iroh_moq::MoqSession::session_accept(web_transport, origin),1493+ )1494+ .await1495+ .context("timed out completing authorized MoQ handshake")?1496+ .context("completing authorized MoQ handshake")?;1497+ let broadcast_closed = broadcast.clone();1498+ session.publish(&broadcast_name, broadcast);1499+ ensure!(1500+ attached.send(()).is_ok(),1501+ "control session ended before MoQ attachment completed"1502+ );1503+ info!(1504+ remote = %session.remote_id(),1505+ session_id,1506+ %broadcast_name,1507+ track = MOQ_VIDEO_H264_TRACK,1508+ "authorized MoQ media attachment accepted"1509+ );1510+ tokio::select! {1511+ reason = session.closed() => {1512+ debug!(remote = %session.remote_id(), ?reason, "MoQ media session closed");1513+ }1514+ reason = broadcast_closed.closed() => {1515+ debug!(remote = %session.remote_id(), ?reason, "control-owned MoQ broadcast closed");1516+ session.close(0, b"control session ended");1517+ }1518+ }1519+ Ok(())1520+ }1521+ .await;1522+ let _ = closed.send(());1523+ result1524+}1525+1526+async fn serve_control_moq(1527+ connection: Connection,1528+ config: HostConfig,1529+ sessions: &Arc<SessionRegistry>,1530+ authorization: &AuthorizationPolicy,1531+) -> Result<()> {1532+ let remote = connection.remote_id();1533+ let handshake_permit = sessions1534+ .pending_handshakes1535+ .try_acquire()1536+ .context("too many pending handshakes")?;1537+ let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1538+ .await1539+ .context("timed out accepting MoQ control stream")?1540+ .context("accepting MoQ control stream")?;1541+ let hello = receive_hello(&mut recv, Capability::VideoH264).await?;1542+ drop(handshake_permit);1543+ debug!(%remote, agent = %hello.agent, "MoQ control hello received");1544+1545+ let grants = match authorization.authorize_or_redeem(1546+ remote,1547+ hello.invitation.as_deref(),1548+ unix_timestamp_now()?,1549+ ) {1550+ Ok(grants) => grants,1551+ Err(error) => {1552+ send_rejection(&mut send, "Portal peer is not authorized").await?;1553+ return Err(error.context("authorizing MoQ control peer"));1554+ }1555+ };1556+ ensure!(1557+ grants.contains(InvitationGrants::VIEW),1558+ "authorized MoQ control peer lacks view permission"1559+ );1560+ let lease = match sessions.claim(remote, hello.nonce, grants) {1561+ Ok(lease) => lease,1562+ Err(error) => {1563+ send_rejection(&mut send, "host already has an active client").await?;1564+ return Err(error);1565+ }1566+ };1567+1568+ let source = match config.source {1569+ VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),1570+ VideoSource::GamescopePipewire => {1571+ let primary = spawn_gamescope_pipewire_after_static_preflight(1572+ config.clone(),1573+ lease.session_clock,1574+ )1575+ .await?;1576+ select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await1577+ }1578+ };1579+ let EncodedSource {1580+ frames: frame_receiver,1581+ current_gop: mut current_gop_receiver,1582+ task: source_task,1583+ pointer_surface_dimensions,1584+ } = match source {1585+ Ok(source) => source,1586+ Err(error) => {1587+ send_rejection(&mut send, "video source is unavailable").await?;1588+ return Err(error);1589+ }1590+ };1591+ let source_task = SourceTaskGuard::new(source_task);1592+1593+ let mut broadcast = Broadcast::new().produce();1594+ let track = broadcast1595+ .create_track(Track {1596+ name: MOQ_VIDEO_H264_TRACK.to_owned(),1597+ priority: MOQ_VIDEO_TRACK_PRIORITY,1598+ })1599+ .context("creating static MoQ H.264 track")?;1600+ let broadcast_name = media_moq_broadcast_name(lease.session_id)?;1601+ let attachment = sessions.expect_moq(1602+ remote,1603+ lease.session_id,1604+ broadcast_name.clone(),1605+ broadcast.consume(),1606+ )?;1607+1608+ let mut control_hello = HostHello::accepted(1609+ lease.session_id,1610+ negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1611+ );1612+ if let Some(dimensions) = pointer_surface_dimensions {1613+ control_hello = control_hello.with_pointer_surface_dimensions(dimensions);1614+ }1615+ write_host_hello(&mut send, &control_hello).await?;1616+ send.finish().context("finishing MoQ control response")?;1617+ drop(send);1618+ info!(1619+ %remote,1620+ session_id = lease.session_id,1621+ %broadcast_name,1622+ "MoQ control client accepted; awaiting authorized media attachment"1623+ );1624+1625+ let MoqAttachmentWait {1626+ mut attached,1627+ closed,1628+ } = attachment;1629+ tokio::time::timeout(MOQ_ATTACHMENT_TIMEOUT, async {1630+ tokio::select! {1631+ result = &mut attached => {1632+ result.context("authorized MoQ handler ended before attachment")1633+ }1634+ reason = connection.closed() => {1635+ Err(anyhow::anyhow!("control connection closed before MoQ attachment: {reason:?}"))1636+ }1637+ }1638+ })1639+ .await1640+ .context("timed out waiting for authorized MoQ attachment")??;1641+1642+ let session_result = run_control_moq_session(1643+ &connection,1644+ &config,1645+ &mut current_gop_receiver,1646+ recv,1647+ remote,1648+ closed,1649+ track,1650+ &mut broadcast,1651+ )1652+ .await;1653+1654+ drop(current_gop_receiver);1655+ drop(frame_receiver);1656+ source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;1657+ drop(lease);1658+ info!(%remote, "MoQ control client released");1659+ session_result1660+}1661+1662+#[allow(clippy::too_many_arguments)]1663+async fn run_control_moq_session(1664+ connection: &Connection,1665+ config: &HostConfig,1666+ current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,1667+ control_recv: iroh::endpoint::RecvStream,1668+ remote: EndpointId,1669+ mut moq_closed: tokio::sync::oneshot::Receiver<()>,1670+ track: TrackProducer,1671+ broadcast: &mut BroadcastProducer,1672+) -> Result<()> {1673+ let maximum_replay_age = maximum_media_replay_age(config.framerate);1674+ let mut replay_cursor = MediaReplayCursor::default();1675+ let mut publisher = MoqGroupPublisher::new(track);1676+ let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);1677+ let mut control_task = tokio::spawn(forward_media_v3_control_requests(1678+ control_recv,1679+ control_sender,1680+ ));1681+ let mut control_task_finished = false;1682+ let mut control_receiver_open = true;1683+1684+ let result = async {1685+ loop {1686+ tokio::select! {1687+ biased;1688+ control_result = &mut control_task, if !control_task_finished => {1689+ control_task_finished = true;1690+ match control_result {1691+ Ok(Ok(())) => {1692+ debug!(%remote, "MoQ keyframe-control stream closed");1693+ }1694+ Ok(Err(error)) => {1695+ return Err(error).context("reading MoQ keyframe-control stream");1696+ }1697+ Err(error) => {1698+ return Err(error).context("MoQ keyframe-control task failed");1699+ }1700+ }1701+ }1702+ changed = control_requests.changed(), if control_receiver_open => {1703+ if changed.is_err() {1704+ control_receiver_open = false;1705+ continue;1706+ }1707+ let Some(request) = *control_requests.borrow_and_update() else {1708+ continue;1709+ };1710+ let through_sequence = current_gop_receiver1711+ .borrow()1712+ .as_ref()1713+ .and_then(|gop| gop.frames.last())1714+ .map(|frame| frame.sequence);1715+ let cancelled_group = publisher.request_keyframe();1716+ replay_cursor.enter_resync_through(through_sequence);1717+ debug!(1718+ %remote,1719+ request_id = request.request_id,1720+ ?request.reason,1721+ advisory_last_sequence = ?request.last_sequence,1722+ coalesced = cancelled_group.is_none(),1723+ ?cancelled_group,1724+ "accepted MoQ keyframe request"1725+ );1726+ }1727+ reason = connection.closed() => {1728+ debug!(%remote, ?reason, "MoQ control connection closed");1729+ return Ok(());1730+ }1731+ result = &mut moq_closed => {1732+ debug!(%remote, ?result, "authorized MoQ media attachment closed");1733+ return Ok(());1734+ }1735+ changed = current_gop_receiver.changed() => {1736+ if let Err(error) = changed {1737+ return Err(error).context("encoded source stopped");1738+ }1739+ let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {1740+ continue;1741+ };1742+ let initial_replay_started_at =1743+ replay_cursor.last_sequence.is_none().then(Instant::now);1744+ let replay_through_sequence = current_gop1745+ .frames1746+ .last()1747+ .map(|frame| frame.sequence)1748+ .context("current GOP snapshot is empty")?;1749+ for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {1750+ let replay_discontinuity = match replay_cursor.classify(1751+ &frame,1752+ replay_through_sequence,1753+ initial_replay_started_at,1754+ Instant::now(),1755+ maximum_replay_age,1756+ ) {1757+ MediaReplayDecision::Send { discontinuity } => discontinuity,1758+ MediaReplayDecision::SkipUntilKeyframe => {1759+ publisher.request_keyframe();1760+ replay_cursor.enter_resync_through(Some(replay_through_sequence));1761+ break;1762+ }1763+ MediaReplayDecision::DiscardStaleSuffix { through_sequence } => {1764+ let cancelled_group = publisher.request_keyframe();1765+ debug!(1766+ %remote,1767+ through_sequence,1768+ ?cancelled_group,1769+ "cancelled stale MoQ media suffix"1770+ );1771+ break;1772+ }1773+ };1774+ match publisher.publish(config, &frame, replay_discontinuity)? {1775+ MoqGroupDecision::Published {1776+ group_id,1777+ frame_id,1778+ cancelled_previous,1779+ } => {1780+ debug!(1781+ sequence = frame.sequence,1782+ group_id,1783+ frame_id,1784+ cancelled_previous,1785+ "published upstream MoQ video frame"1786+ );1787+ replay_cursor.commit_sent(&frame);1788+ }1789+ MoqGroupDecision::SkipUntilKeyframe => {1790+ replay_cursor.enter_resync_through(Some(replay_through_sequence));1791+ break;1792+ }1793+ MoqGroupDecision::EnterResync => {1794+ replay_cursor.enter_resync_through(Some(replay_through_sequence));1795+ break;1796+ }1797+ }1798+ }1799+ }1800+ }1801+ }1802+ }1803+ .await;1804+1805+ publisher.abort();1806+ let _ = broadcast.abort(MoqError::Cancel);1807+ if !control_task_finished {1808+ control_task.abort();1809+ let _ = control_task.await;1810+ }1811+ result1812+}1813+11901814 async fn serve_media(11911815 connection: Connection,11921816 config: HostConfig,
@@ -2746,6 +3370,150 @@ mod tests {
27463370 );27473371 }27483372 3373+ fn moq_test_config() -> HostConfig {3374+ HostConfig {3375+ identity_path: "identity".into(),3376+ state_path: "state".into(),3377+ source: VideoSource::TestPattern,3378+ width: 1280,3379+ height: 800,3380+ framerate: 60,3381+ codec: "h264".to_owned(),3382+ input_mode: crate::config::InputMode::Disabled,3383+ uinput: None,3384+ ffmpeg_path: "ffmpeg".into(),3385+ gamescope_pipewire: None,3386+ audio: None,3387+ }3388+ }3389+3390+ #[tokio::test]3391+ async fn upstream_moq_groups_are_sequential_and_cancel_the_superseded_gop() {3392+ let track_info = Track {3393+ name: MOQ_VIDEO_H264_TRACK.to_owned(),3394+ priority: MOQ_VIDEO_TRACK_PRIORITY,3395+ };3396+ let mut broadcast = Broadcast::new().produce();3397+ let track = broadcast.create_track(track_info.clone()).unwrap();3398+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();3399+ let mut publisher = MoqGroupPublisher::new(track);3400+ let config = moq_test_config();3401+3402+ assert_eq!(3403+ publisher3404+ .publish(&config, &media_v3_encoded_frame(100, true, true, 4), false)3405+ .unwrap(),3406+ MoqGroupDecision::Published {3407+ group_id: 0,3408+ frame_id: 0,3409+ cancelled_previous: false,3410+ }3411+ );3412+ let mut first_group = consumer.recv_group().await.unwrap().unwrap();3413+ assert_eq!(first_group.sequence, 0);3414+ assert!(first_group.read_frame().await.unwrap().is_some());3415+3416+ assert_eq!(3417+ publisher3418+ .publish(3419+ &config,3420+ &media_v3_encoded_frame(101, false, false, 4),3421+ false,3422+ )3423+ .unwrap(),3424+ MoqGroupDecision::Published {3425+ group_id: 0,3426+ frame_id: 1,3427+ cancelled_previous: false,3428+ }3429+ );3430+ assert!(first_group.read_frame().await.unwrap().is_some());3431+3432+ assert_eq!(3433+ publisher3434+ .publish(&config, &media_v3_encoded_frame(200, true, true, 4), false)3435+ .unwrap(),3436+ MoqGroupDecision::Published {3437+ group_id: 1,3438+ frame_id: 0,3439+ cancelled_previous: true,3440+ }3441+ );3442+ assert!(first_group.finished().await.is_err());3443+ let mut second_group = consumer.recv_group().await.unwrap().unwrap();3444+ assert_eq!(second_group.sequence, 1);3445+ let object = second_group.read_frame().await.unwrap().unwrap();3446+ let frame = sigil_protocol::decode_media_frame_object(&object).unwrap();3447+ assert_eq!(frame.header.sequence, 200);3448+ assert!(frame.header.flags.contains(FrameFlags::DISCONTINUITY));3449+ }3450+3451+ #[tokio::test]3452+ async fn upstream_moq_resync_aborts_current_group_and_waits_for_configured_idr() {3453+ let track_info = Track {3454+ name: MOQ_VIDEO_H264_TRACK.to_owned(),3455+ priority: MOQ_VIDEO_TRACK_PRIORITY,3456+ };3457+ let mut broadcast = Broadcast::new().produce();3458+ let track = broadcast.create_track(track_info.clone()).unwrap();3459+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();3460+ let mut publisher = MoqGroupPublisher::new(track);3461+ let config = moq_test_config();3462+3463+ publisher3464+ .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)3465+ .unwrap();3466+ let mut cancelled = consumer.recv_group().await.unwrap().unwrap();3467+ assert_eq!(publisher.request_keyframe(), Some(0));3468+ assert!(cancelled.finished().await.is_err());3469+ assert_eq!(3470+ publisher3471+ .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)3472+ .unwrap(),3473+ MoqGroupDecision::SkipUntilKeyframe3474+ );3475+ assert_eq!(3476+ publisher3477+ .publish(&config, &media_v3_encoded_frame(20, true, true, 1), false)3478+ .unwrap(),3479+ MoqGroupDecision::Published {3480+ group_id: 1,3481+ frame_id: 0,3482+ cancelled_previous: false,3483+ }3484+ );3485+ }3486+3487+ #[tokio::test]3488+ async fn upstream_moq_group_counts_envelope_bytes_before_upstream_cache_eviction() {3489+ let track_info = Track {3490+ name: MOQ_VIDEO_H264_TRACK.to_owned(),3491+ priority: MOQ_VIDEO_TRACK_PRIORITY,3492+ };3493+ let mut broadcast = Broadcast::new().produce();3494+ let track = broadcast.create_track(track_info.clone()).unwrap();3495+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();3496+ let mut publisher = MoqGroupPublisher::new(track);3497+ let config = moq_test_config();3498+ publisher3499+ .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)3500+ .unwrap();3501+ let mut cancelled = consumer.recv_group().await.unwrap().unwrap();3502+3503+ // Payload-only accounting would accept this next one-byte access unit,3504+ // but its fixed application envelope would overflow moq-net's 32 MiB3505+ // group cache and silently evict the keyframe.3506+ publisher.object_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;3507+ assert_eq!(3508+ publisher3509+ .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)3510+ .unwrap(),3511+ MoqGroupDecision::EnterResync3512+ );3513+ assert!(cancelled.finished().await.is_err());3514+ assert_eq!(publisher.object_bytes, 0);3515+ }3516+27493517 #[test]27503518 fn media_v3_group_rejects_noncontiguous_or_unconfigured_frames() {27513519 let mut cursor = MediaV3GroupCursor::default();
@@ -3285,6 +4053,96 @@ mod tests {
32854053 );32864054 }32874055 4056+ fn test_moq_broadcast() -> (BroadcastProducer, BroadcastConsumer) {4057+ let producer = Broadcast::new().produce();4058+ let consumer = producer.consume();4059+ (producer, consumer)4060+ }4061+4062+ #[test]4063+ fn moq_attachment_requires_exact_active_control_remote_and_is_single_use() {4064+ let sessions = Arc::new(SessionRegistry::default());4065+ assert!(sessions.claim_moq(endpoint(1)).is_err());4066+ let lease = sessions4067+ .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)4068+ .unwrap();4069+ let (_producer, consumer) = test_moq_broadcast();4070+ let _wait = sessions4071+ .expect_moq(4072+ endpoint(1),4073+ lease.session_id,4074+ media_moq_broadcast_name(lease.session_id).unwrap(),4075+ consumer,4076+ )4077+ .unwrap();4078+4079+ // A wrong peer cannot consume the exact pending attachment.4080+ assert!(sessions.claim_moq(endpoint(2)).is_err());4081+ let attachment = sessions.claim_moq(endpoint(1)).unwrap();4082+ assert_eq!(attachment.session_id, lease.session_id);4083+ assert_eq!(4084+ attachment.broadcast_name,4085+ media_moq_broadcast_name(lease.session_id).unwrap()4086+ );4087+ // The pending token was atomically consumed before the MoQ handshake.4088+ assert!(sessions.claim_moq(endpoint(1)).is_err());4089+ }4090+4091+ #[test]4092+ fn competing_moq_connections_cannot_both_claim_one_control_attachment() {4093+ let sessions = Arc::new(SessionRegistry::default());4094+ let lease = sessions4095+ .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)4096+ .unwrap();4097+ let (_producer, consumer) = test_moq_broadcast();4098+ let _wait = sessions4099+ .expect_moq(4100+ endpoint(1),4101+ lease.session_id,4102+ media_moq_broadcast_name(lease.session_id).unwrap(),4103+ consumer,4104+ )4105+ .unwrap();4106+ let barrier = Arc::new(std::sync::Barrier::new(3));4107+ let contenders = (0..2)4108+ .map(|_| {4109+ let sessions = Arc::clone(&sessions);4110+ let barrier = Arc::clone(&barrier);4111+ std::thread::spawn(move || {4112+ barrier.wait();4113+ sessions.claim_moq(endpoint(1)).is_ok()4114+ })4115+ })4116+ .collect::<Vec<_>>();4117+ barrier.wait();4118+ let claimed = contenders4119+ .into_iter()4120+ .map(|thread| usize::from(thread.join().unwrap()))4121+ .sum::<usize>();4122+ assert_eq!(claimed, 1);4123+ }4124+4125+ #[tokio::test]4126+ async fn releasing_control_clears_an_unclaimed_moq_attachment() {4127+ let sessions = Arc::new(SessionRegistry::default());4128+ let lease = sessions4129+ .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)4130+ .unwrap();4131+ let (_producer, consumer) = test_moq_broadcast();4132+ let wait = sessions4133+ .expect_moq(4134+ endpoint(1),4135+ lease.session_id,4136+ media_moq_broadcast_name(lease.session_id).unwrap(),4137+ consumer,4138+ )4139+ .unwrap();4140+ drop(lease);4141+ assert!(sessions.claim_moq(endpoint(1)).is_err());4142+ assert!(wait.attached.await.is_err());4143+ assert!(wait.closed.await.is_err());4144+ }4145+32884146 #[test]32894147 fn pending_handshakes_are_bounded() {32904148 let sessions = SessionRegistry::default();
M crates/sigil-protocol/src/lib.rs +30 -2
@@ -32,8 +32,9 @@ pub use invitation::{
3232 MAX_INVITATION_TOKEN_LEN, MAX_INVITATION_TTL_SECS, SignedInvitation,3333 };3434 pub use media::{35- FrameFlags, MEDIA_HEADER_LEN, MediaCodec, MediaFrame, MediaFrameHeader, read_media_frame,36- read_media_object, write_media_frame,35+ FrameFlags, MEDIA_HEADER_LEN, MediaCodec, MediaFrame, MediaFrameHeader,36+ decode_media_frame_object, encode_media_frame_object, read_media_frame, read_media_object,37+ write_media_frame,3738 };3839 pub use media_v3::{3940 KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,
@@ -58,6 +59,8 @@ pub const MEDIA_ALPN_V3: &[u8] = b"sigil/media/3";
5859 pub const INPUT_ALPN_V1: &[u8] = b"sigil/input/1";5960 /// ALPN for the v1 session-control stream.6061 pub const CONTROL_ALPN_V1: &[u8] = b"sigil/control/1";62+/// Static upstream MoQ track carrying bounded H.264 access-unit objects.63+pub const MOQ_VIDEO_H264_TRACK: &str = "video/h264";6164 /// ALPN for the v1 low-latency Opus datagram connection.6265 pub const AUDIO_ALPN_V1: &[u8] = b"sigil/audio/1";6366
@@ -85,6 +88,21 @@ pub const OPUS_SAMPLE_RATE: u32 = 48_000;
8588 pub const OPUS_CHANNELS: u8 = 2;8689 pub const OPUS_FRAME_SAMPLES: u16 = 960;8790 91+/// Derive the session-scoped upstream MoQ broadcast name advertised by Sigil.92+///93+/// The authenticated control handshake supplies the non-zero session id, so94+/// Portal can derive this application namespace without accepting a95+/// peer-controlled path.96+pub fn media_moq_broadcast_name(session_id: u64) -> Result<String> {97+ if session_id == 0 {98+ return Err(ProtocolError::InvalidMessage {99+ message_type: "MoQ broadcast name",100+ reason: "session id must be non-zero",101+ });102+ }103+ Ok(format!("sigil/session/{session_id}/video"))104+}105+88106 #[cfg(test)]89107 mod tests {90108 use super::*;
@@ -100,4 +118,14 @@ mod tests {
100118 assert_eq!(LEGACY_FRAME_ALPN_V0, b"sigil/frame-stream/0");101119 assert_eq!(LEGACY_INPUT_ALPN_V0, b"sigil/input-stream/0");102120 }121+122+ #[test]123+ fn moq_media_namespace_is_session_scoped_and_stable() {124+ assert_eq!(MOQ_VIDEO_H264_TRACK, "video/h264");125+ assert_eq!(126+ media_moq_broadcast_name(42).unwrap(),127+ "sigil/session/42/video"128+ );129+ assert!(media_moq_broadcast_name(0).is_err());130+ }103131 }
M crates/sigil-protocol/src/media.rs +66 -0
@@ -208,6 +208,59 @@ pub struct MediaFrame {
208208 pub payload: Vec<u8>,209209 }210210 211+/// Encode one complete application media frame for an object transport such212+/// as MoQ. The MoQ group and frame boundaries carry delivery semantics; this213+/// envelope carries only Sigil's bounded codec metadata and compressed access214+/// unit.215+pub fn encode_media_frame_object(frame: &MediaFrame) -> Result<Vec<u8>> {216+ frame.header.validate()?;217+ if frame.payload.len() != frame.header.payload_len as usize {218+ return Err(ProtocolError::InvalidMessage {219+ message_type: "media frame object",220+ reason: "payload does not match declared length",221+ });222+ }223+ let object_len = MEDIA_HEADER_LEN.checked_add(frame.payload.len()).ok_or(224+ ProtocolError::InvalidMediaPayloadLength {225+ actual: frame.payload.len(),226+ maximum: MAX_MEDIA_PAYLOAD_LEN,227+ },228+ )?;229+ let mut object = Vec::with_capacity(object_len);230+ object.extend_from_slice(&frame.header.encode()?);231+ object.extend_from_slice(&frame.payload);232+ Ok(object)233+}234+235+/// Decode exactly one application media frame from an object transport.236+/// Header validation, including the declared payload bound, occurs before the237+/// payload is copied into its owned buffer.238+pub fn decode_media_frame_object(object: &[u8]) -> Result<MediaFrame> {239+ if object.len() < MEDIA_HEADER_LEN {240+ return Err(ProtocolError::InvalidMessage {241+ message_type: "media frame object",242+ reason: "object ended before the media header",243+ });244+ }245+ let wire: &[u8; MEDIA_HEADER_LEN] = object[..MEDIA_HEADER_LEN]246+ .try_into()247+ .expect("fixed media header slice");248+ let header = MediaFrameHeader::decode(wire)?;249+ let expected_len = MEDIA_HEADER_LEN250+ .checked_add(header.payload_len as usize)251+ .ok_or(ProtocolError::InvalidMediaPayloadLength {252+ actual: header.payload_len as usize,253+ maximum: MAX_MEDIA_PAYLOAD_LEN,254+ })?;255+ if object.len() != expected_len {256+ return Err(ProtocolError::InvalidMessage {257+ message_type: "media frame object",258+ reason: "object length does not match declared payload",259+ });260+ }261+ MediaFrame::new(header, object[MEDIA_HEADER_LEN..].to_vec())262+}263+211264 impl MediaFrame {212265 pub fn new(header: MediaFrameHeader, payload: Vec<u8>) -> Result<Self> {213266 header.validate()?;
@@ -386,6 +439,19 @@ mod tests {
386439 assert_eq!(read_media_object(&mut receiver).await.unwrap(), expected);387440 }388441 442+ #[test]443+ fn object_transport_envelope_round_trips_exactly() {444+ let frame = MediaFrame::new(header(), vec![0, 0, 0, 1]).unwrap();445+ let object = encode_media_frame_object(&frame).unwrap();446+ assert_eq!(object.len(), MEDIA_HEADER_LEN + frame.payload.len());447+ assert_eq!(decode_media_frame_object(&object).unwrap(), frame);448+449+ let mut trailing = object.clone();450+ trailing.push(0);451+ assert!(decode_media_frame_object(&trailing).is_err());452+ assert!(decode_media_frame_object(&object[..MEDIA_HEADER_LEN - 1]).is_err());453+ }454+389455 #[tokio::test]390456 async fn media_object_rejects_empty_and_trailing_streams() {391457 let (mut empty_sender, mut empty_receiver) = duplex(16);
M scripts/loopback-proof.sh +92 -21
@@ -13,10 +13,14 @@ probe_timeout_seconds=15
1313 command_timeout_seconds=451414 readiness_timeout_seconds=301515 host_runtime_seconds=12016-media_transport="grouped-v3"17-media_accept_log="media v3 client accepted"18-media_release_log="media v3 client released"16+media_transport="iroh-moq"17+media_alpn="moq-lite-04"18+media_accept_log="authorized MoQ media attachment accepted"19+media_release_log="MoQ control client released"20+keyframe_request_log="accepted MoQ keyframe request"21+session_gate_name="control"1922 keyframe_recovery="ok"23+media_v3=false2024 media_v2=false2125 media_v1=false2226
@@ -24,9 +28,10 @@ usage() {
2428 cat <<'EOF'2529 Usage: scripts/loopback-proof.sh [options]2630 27-Build and exercise the exact sigil and sigil-probe binaries over a real28-loopback Iroh connection. The proof covers the decodable first media frame,29-input acknowledgment, single-active-client rejection, and clean reconnects.31+Build and exercise the exact sigil and sigil-probe binaries over real loopback32+Iroh and upstream MoQ connections. The ordinary proof covers a configured IDR,33+native group bounds and recovery, input acknowledgment, single-active-client34+rejection, and clean reconnects.3035 3136 Options:3237 --profile debug|release Cargo profile to build (default: debug)
@@ -34,10 +39,12 @@ Options:
3439 --reconnect-cycles COUNT Fresh sessions after the primary (default: 3)3540 --reconnect-frames COUNT Frames per reconnect session (default: 30)3641 --probe-timeout-seconds SEC Probe per-operation timeout (default: 15)42+ --media-v3 Exercise custom grouped media v3 compatibility43+ instead of upstream MoQ3744 --media-v2 Exercise independent media v2 compatibility38- instead of grouped media v345+ instead of upstream MoQ3946 --media-v1 Exercise ordered media v1 compatibility40- instead of grouped media v347+ instead of upstream MoQ4148 --help Show this help4249 EOF4350 }
@@ -86,23 +93,41 @@ while [[ $# -gt 0 ]]; do
8693 shift 28794 ;;8895 --media-v1)89- [[ "$media_v2" == false ]] || die "--media-v1 and --media-v2 cannot be combined"96+ [[ "$media_v2" == false && "$media_v3" == false ]] \97+ || die "media compatibility flags cannot be combined"9098 media_transport="reliable-v1"99+ media_alpn="sigil/media/1"91100 media_accept_log="media client accepted"92101 media_release_log="media client released"102+ session_gate_name="media"93103 keyframe_recovery="not-requested"94104 media_v1=true95105 shift96106 ;;97107 --media-v2)98- [[ "$media_v1" == false ]] || die "--media-v1 and --media-v2 cannot be combined"108+ [[ "$media_v1" == false && "$media_v3" == false ]] \109+ || die "media compatibility flags cannot be combined"99110 media_transport="independent-v2"111+ media_alpn="sigil/media/2"100112 media_accept_log="media v2 client accepted"101113 media_release_log="media v2 client released"114+ session_gate_name="media"102115 keyframe_recovery="not-requested"103116 media_v2=true104117 shift105118 ;;119+ --media-v3)120+ [[ "$media_v1" == false && "$media_v2" == false ]] \121+ || die "media compatibility flags cannot be combined"122+ media_transport="grouped-v3"123+ media_alpn="sigil/media/3"124+ media_accept_log="media v3 client accepted"125+ media_release_log="media v3 client released"126+ keyframe_request_log="accepted media v3 keyframe request"127+ session_gate_name="media"128+ media_v3=true129+ shift130+ ;;106131 --help|-h)107132 usage108133 exit 0
@@ -128,9 +153,9 @@ require_positive_integer "--reconnect-frames" "$reconnect_frames"
128153 require_positive_integer "--probe-timeout-seconds" "$probe_timeout_seconds"129154 if [[ "$media_v1" == false && "$media_v2" == false ]]; then130155 [[ "$primary_frames" -ge 4 ]] \131- || die "--primary-frames must be at least 4 for grouped v3 keyframe recovery"156+ || die "--primary-frames must be at least 4 for keyframe recovery"132157 [[ "$reconnect_frames" -ge 4 ]] \133- || die "--reconnect-frames must be at least 4 for grouped v3 keyframe recovery"158+ || die "--reconnect-frames must be at least 4 for keyframe recovery"134159 fi135160 136161 # A fresh Iroh endpoint performs relay discovery and path establishment on
@@ -279,12 +304,24 @@ host_accepted_keyframe_request() {
279304 local log_path="$1"280305 local request_id="$2"281306 sed $'s/\033\\[[0-9;]*m//g' "$log_path" 2>/dev/null \282- | awk -v request_id="$request_id" '283- index($0, "accepted media v3 keyframe request") \284- && $0 ~ ("request_id=" request_id "([^[:digit:]]|$)") \285- && index($0, "coalesced=false") { found=1 }286- END { exit !found }287- '307+ | awk -v request_id="$request_id" -v request_log="$keyframe_request_log" '308+ index($0, request_log) \309+ && $0 ~ ("request_id=" request_id "([^[:digit:]]|$)") \310+ && index($0, "coalesced=false") { found=1 }311+ END { exit !found }312+ '313+}314+315+host_keyframe_request_count() {316+ local log_path="$1"317+ local request_id="$2"318+ sed $'s/\033\\[[0-9;]*m//g' "$log_path" 2>/dev/null \319+ | awk -v request_id="$request_id" -v request_log="$keyframe_request_log" '320+ index($0, request_log) \321+ && $0 ~ ("request_id=" request_id "([^[:digit:]]|$)") \322+ && index($0, "coalesced=false") { count++ }323+ END { print count + 0 }324+ '288325 }289326 290327 wait_for_keyframe_request() {
@@ -374,6 +411,9 @@ validate_probe_log() {
374411 grep -Fxq 'dimensions=1280x800' "$log_path" || return 1375412 grep -Fxq 'sequence_gaps=0' "$log_path" || return 1376413 grep -Fxq "transport=${media_transport}" "$log_path" || return 1414+ grep -Fxq "transport_alpn=${media_alpn}" "$log_path" || return 1415+ grep -Fxq 'first_configured_idr=ok' "$log_path" || return 1416+ grep -Fxq 'frame_sequence=monotonic' "$log_path" || return 1377417 grep -Fxq "keyframe_recovery=${keyframe_recovery}" "$log_path" || return 1378418 grep -Fxq "keyframe_request_id=${expected_request_id}" "$log_path" || return 1379419 awk -F= '$1 == "media_objects_dropped" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \
@@ -383,7 +423,24 @@ validate_probe_log() {
383423 awk -F= '$1 == "keyframes" && $2 ~ /^[0-9]+$/ && $2 > 0 { found=1 } END { exit !found }' \384424 "$log_path" || return 1385425 awk -F= '$1 == "input_ack_micros" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \386- "$log_path"426+ "$log_path" || return 1427+ grep -Fxq 'path_mode=direct' "$log_path" || return 1428+ if [[ "$media_transport" == "iroh-moq" ]]; then429+ grep -Fxq 'control_alpn=sigil/control/1' "$log_path" || return 1430+ grep -Fxq 'group_sequence=monotonic' "$log_path" || return 1431+ grep -Fxq 'moq_group_capacity=1' "$log_path" || return 1432+ grep -Fxq 'moq_unrecovered_group_gaps=0' "$log_path" || return 1433+ grep -Fxq 'moq_historical_suffix_frames=0' "$log_path" || return 1434+ grep -Fxq 'recovery_barrier=configured-idr' "$log_path" || return 1435+ awk -F= '$1 == "moq_group_gaps" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \436+ "$log_path" || return 1437+ awk -F= '$1 == "moq_cancelled_groups" && $2 ~ /^[0-9]+$/ && $2 > 0 { found=1 } END { exit !found }' \438+ "$log_path" || return 1439+ awk -F= '$1 == "moq_maximum_group_objects" && $2 ~ /^[0-9]+$/ && $2 > 0 && $2 <= 256 { found=1 } END { exit !found }' \440+ "$log_path" || return 1441+ awk -F= '$1 == "moq_maximum_group_bytes" && $2 ~ /^[0-9]+$/ && $2 > 0 && $2 <= 33554432 { found=1 } END { exit !found }' \442+ "$log_path" || return 1443+ fi387444 }388445 389446 run_probe() {
@@ -391,6 +448,8 @@ run_probe() {
391448 "$probe_bin" --media-v1 "$@"392449 elif [[ "$media_v2" == true ]]; then393450 "$probe_bin" --media-v2 "$@"451+ elif [[ "$media_v3" == true ]]; then452+ "$probe_bin" --media-v3 "$@"394453 else395454 "$probe_bin" "$@"396455 fi
@@ -471,7 +530,7 @@ fi
471530 if [[ -e "${secondary_log}.timeout" ]]; then472531 die "secondary client timed out instead of receiving the active-client rejection"473532 fi474-if ! grep -Fxq 'Error: host rejected media stream: host already has an active client' "$secondary_log"; then533+if ! grep -Fxq "Error: host rejected ${session_gate_name} stream: host already has an active client" "$secondary_log"; then475534 printf 'secondary probe did not return the expected rejection. Log follows:\n' >&2476535 sed -n '1,120p' "$secondary_log" >&2 || true477536 die "unexpected secondary-client failure"
@@ -529,6 +588,16 @@ while [[ "$cycle" -le "$reconnect_cycles" ]]; do
529588 cycle=$((cycle + 1))530589 done531590 591+if [[ "$media_v1" == false && "$media_v2" == false ]]; then592+ request_id=1593+ while [[ "$request_id" -le $((reconnect_cycles + 1)) ]]; do594+ request_count="$(host_keyframe_request_count "$host_log" "$request_id")"595+ [[ "$request_count" -eq 1 ]] \596+ || die "keyframe request $request_id was accepted $request_count times; expected exactly once"597+ request_id=$((request_id + 1))598+ done599+fi600+532601 kill -TERM "$host_pid" 2>/dev/null || true533602 if wait "$host_pid"; then534603 host_status=0
@@ -553,7 +622,9 @@ printf 'probe_binary=%s\n' "$probe_bin"
553622 printf 'probe_sha256=%s\n' "$(sha256_file "$probe_bin")"554623 printf 'node_id=%s\n' "$node_id"555624 printf 'primary_frames=%s\n' "$primary_frames"556-grep -E '^(transport|keyframe_recovery|keyframe_request_id|keyframes|sequence_gaps|media_objects_dropped|media_objects_late|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"625+grep -E '^(transport|control_alpn|transport_alpn|first_configured_idr|frame_sequence|group_sequence|keyframe_recovery|keyframe_request_id|keyframes|sequence_gaps|media_objects_dropped|media_objects_late|moq_group_capacity|moq_cancelled_groups|moq_group_gaps|moq_unrecovered_group_gaps|moq_maximum_group_objects|moq_maximum_group_bytes|moq_historical_suffix_frames|recovery_barrier|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"626+printf 'keyframe_request_correlation=%s\n' \627+ "$([[ "$media_v1" == false && "$media_v2" == false ]] && printf unique || printf not-requested)"557628 printf 'active_client_rejection=ok\n'558629 printf 'reconnect_cycles=%s\n' "$reconnect_cycles"559630 printf 'cleanup=ok\n'
M src-tauri/Cargo.toml +4 -1
@@ -16,6 +16,7 @@ demo-direct-node = []
1616 tauri-build = { version = "2", features = [] }1717 1818 [dependencies]19+bytes.workspace = true1920 serde_json = "1"2021 serde = { version = "1", features = ["derive"] }2122 log = "0.4"
@@ -27,7 +28,9 @@ tauri-plugin-dialog = "2"
2728 tauri-plugin-single-instance = { version = "2", features = ["deep-link"] }2829 ctap-hid-fido2 = "3.5"2930 getrandom = "0.4"30-iroh = "1.0"31+iroh.workspace = true32+iroh-moq.workspace = true33+moq-net.workspace = true3134 anyhow = "1"3235 tokio = { version = "1", features = ["macros", "rt-multi-thread", "io-util", "time"] }3336 image = { version = "0.25", default-features = false, features = ["jpeg"] }
M src-tauri/src/commands/network.rs +841 -11
@@ -6,17 +6,20 @@ use super::state::{
66 };77 use base64::Engine;88 use iroh::{Endpoint, SecretKey, endpoint::presets};9+use iroh_moq::{Moq, MoqSession};10+use moq_net::{BroadcastConsumer, GroupConsumer, Track, TrackConsumer};911 use openh264::{formats::YUVSource, nal_units};1012 use serde::Serialize;1113 use sigil_protocol::{12- AUDIO_ALPN_V1, AudioFlags, AudioPacket, AudioPacketHeader, Capability, ClientHello, FrameFlags,13- INPUT_ALPN_V1, InputEvent, InvitationGrants, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,14+ AUDIO_ALPN_V1, AudioFlags, AudioPacket, AudioPacketHeader, CONTROL_ALPN_V1, Capability,15+ ClientHello, FrameFlags, INPUT_ALPN_V1, InputEvent, InvitationGrants, KeyframeRequestReasonV3,16+ MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3,1417 MAX_MEDIA_PAYLOAD_LEN, MAX_VIDEO_DIMENSION, MAX_VIDEO_PIXELS, MEDIA_ALPN_V1, MEDIA_ALPN_V2,15- MEDIA_ALPN_V3, MediaCodec, MediaControlRequestV3, MediaFrame, MediaObjectV3, PointerPosition,16- PointerSurfaceDimensions, ProtocolError, RELATIVE_POINTER_DELTA_MAX,17- RELATIVE_POINTER_DELTA_MIN, read_host_hello, read_input_ack, read_media_frame,18- read_media_object, read_media_object_v3, write_client_hello, write_input_event,19- write_media_control_request_v3,18+ MEDIA_ALPN_V3, MOQ_VIDEO_H264_TRACK, MediaCodec, MediaControlRequestV3, MediaFrame,19+ MediaObjectV3, PointerPosition, PointerSurfaceDimensions, ProtocolError,20+ RELATIVE_POINTER_DELTA_MAX, RELATIVE_POINTER_DELTA_MIN, decode_media_frame_object,21+ media_moq_broadcast_name, read_host_hello, read_input_ack, read_media_frame, read_media_object,22+ read_media_object_v3, write_client_hello, write_input_event, write_media_control_request_v3,2023 };2124 use std::collections::{BTreeMap, VecDeque};2225 use std::io::Cursor;
@@ -24,7 +27,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
2427 use std::sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard};2528 use std::time::{Duration, Instant};2629 use tauri::{27- AppHandle, Emitter, State,30+ AppHandle, Emitter, Manager, State,2831 ipc::{Channel, Response},2932 };3033
@@ -71,6 +74,15 @@ fn emit_frame_error(app: &AppHandle, generation: u64, error: impl Into<String>)
7174 const LEGACY_MEDIA_IDLE_TIMEOUT: Duration = Duration::from_secs(10);7275 const CLIENT_MEDIA_OBJECT_CAPACITY: usize = 4;7376 const CLIENT_MEDIA_OBJECT_READ_TIMEOUT: Duration = Duration::from_secs(1);77+const CLIENT_MOQ_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);78+const CLIENT_MOQ_SUBSCRIBE_TIMEOUT: Duration = Duration::from_secs(10);79+// Match the protocol's maximum single-object delivery horizon. A publisher may80+// never hold a partially delivered object open indefinitely.81+const CLIENT_MOQ_OBJECT_READ_TIMEOUT: Duration =82+ Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64);83+// Sigil's external encoder can take 500 ms to reach its next configured IDR.84+// Allow another 500 ms for a relay path to deliver the superseding group.85+const CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT: Duration = Duration::from_secs(1);7486 // Absorb brief webview→Rust acknowledgment jitter without allowing Tauri IPC7587 // to grow without bound. Four 60 fps frames cap this handoff at about 67 ms;7688 // WebCodecs has a separate, stricter decode-queue bound in the frontend.
@@ -390,6 +402,83 @@ fn close_generation_connection<T>(
390402 })391403 }392404 405+fn take_generation_owned<T>(slot: &mut Option<(u64, T)>, expected_generation: u64) -> Option<T> {406+ if slot407+ .as_ref()408+ .is_some_and(|(generation, _)| *generation == expected_generation)409+ {410+ slot.take().map(|(_, value)| value)411+ } else {412+ None413+ }414+}415+416+async fn retire_upstream_moq_generation(417+ app: &AppHandle,418+ media_generation: u64,419+ audio_generation: Option<u64>,420+) -> bool {421+ let state = app.state::<AppState>();422+ let endpoint = {423+ // Selection and retirement are one generation-checked transaction.424+ // A stale reader must never close a replacement session whose task was425+ // installed after an explicit disconnect/reconnect.426+ let _connection_serial = state.client_connection_serial.lock().await;427+ let media_connection = {428+ let mut slot = state.media_connection.lock().await;429+ take_generation_owned(&mut slot, media_generation)430+ };431+ let Some(media_connection) = media_connection else {432+ return false;433+ };434+435+ {436+ let mut control = state.media_control.lock().await;437+ let _ = take_generation_owned(&mut control, media_generation);438+ }439+ {440+ let mut delivery = state.frame_delivery.lock().await;441+ let _ = take_generation_owned(&mut delivery, media_generation);442+ }443+ *state.input_send.lock().await = None;444+445+ let audio_connection = if let Some(audio_generation) = audio_generation {446+ if let Err(error) = cancel_audio_generation(447+ &state.audio_connection_generation,448+ &state.audio_deliveries,449+ audio_generation,450+ ) {451+ eprintln!(452+ "[client] failed to retire audio generation after upstream MoQ ended: {error}"453+ );454+ }455+ let mut slot = state.audio_connection.lock().await;456+ take_generation_owned(&mut slot, audio_generation)457+ } else {458+ None459+ };460+461+ let endpoint = state.client_endpoint.lock().await.take();462+ media_connection.close(0_u32.into(), b"upstream MoQ media ended");463+ if let Some(audio_connection) = audio_connection {464+ audio_connection.close(0_u32.into(), b"upstream MoQ media ended");465+ }466+ state467+ .client_connection_active468+ .store(false, Ordering::SeqCst);469+ endpoint470+ };471+472+ if let Some(endpoint) = endpoint473+ && tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close())474+ .await475+ .is_err()476+ {477+ eprintln!("[client] timed out retiring endpoint after upstream MoQ media ended");478+ }479+ true480+}481+393482 #[derive(Clone, Debug, Serialize)]394483 struct FrameStatsPayload {395484 generation: u64,
@@ -655,6 +744,7 @@ struct NegotiatedV1Stream {
655744 656745 #[derive(Clone, Copy, Debug, PartialEq, Eq)]657746 enum MediaTransport {747+ UpstreamMoq,658748 LegacyV0,659749 ReliableStreamV1,660750 IndependentObjectsV2,
@@ -664,6 +754,7 @@ enum MediaTransport {
664754 impl MediaTransport {665755 const fn diagnostic_name(self) -> &'static str {666756 match self {757+ Self::UpstreamMoq => "iroh-moq",667758 Self::LegacyV0 => "reliable-v0",668759 Self::ReliableStreamV1 => "reliable-v1",669760 Self::IndependentObjectsV2 => "independent-v2",
@@ -672,6 +763,355 @@ impl MediaTransport {
672763 }673764 }674765 766+#[derive(Debug)]767+enum MoqMediaReadOutcome {768+ Frame {769+ frame: MediaFrame,770+ discontinuity: bool,771+ },772+ Dropped {773+ reason: KeyframeRequestReasonV3,774+ },775+ Malformed(String),776+}777+778+#[derive(Clone, Copy, Debug, PartialEq, Eq)]779+enum MoqGroupRecovery {780+ /// Sigil deliberately cancels the old GOP when a configured IDR starts a781+ /// replacement. This is normal live-edge supersession, not evidence that782+ /// another keyframe request is needed.783+ ExpectedSupersession,784+ RecoverableGap(KeyframeRequestReasonV3),785+}786+787+struct MoqGroupCursor {788+ group: GroupConsumer,789+ sequence: u64,790+ object_count: usize,791+ object_bytes: usize,792+ group_gap: bool,793+ replacement_for_cancelled_group: bool,794+}795+796+/// Own every upstream handle for as long as the Portal frame task is alive.797+/// In particular, dropping `Moq` aborts its actor and dropping the consumers798+/// cancels their subscriptions, so none of these may be scoped only to setup.799+struct MoqMediaLifetime {800+ _moq: Moq,801+ _session: MoqSession,802+ _broadcast: BroadcastConsumer,803+}804+805+struct MoqMediaReceiver {806+ _lifetime: Option<MoqMediaLifetime>,807+ track: TrackConsumer,808+ current_group: Option<MoqGroupCursor>,809+ last_group_sequence: Option<u64>,810+ last_frame_sequence: Option<u64>,811+ waiting_for_keyframe: bool,812+ pending_group_recovery: Option<MoqGroupRecovery>,813+}814+815+impl MoqMediaReceiver {816+ fn new(817+ moq: Moq,818+ session: MoqSession,819+ broadcast: BroadcastConsumer,820+ track: TrackConsumer,821+ ) -> Self {822+ Self {823+ _lifetime: Some(MoqMediaLifetime {824+ _moq: moq,825+ _session: session,826+ _broadcast: broadcast,827+ }),828+ track,829+ current_group: None,830+ last_group_sequence: None,831+ last_frame_sequence: None,832+ waiting_for_keyframe: true,833+ pending_group_recovery: None,834+ }835+ }836+837+ #[cfg(test)]838+ fn for_test(track: TrackConsumer) -> Self {839+ Self {840+ _lifetime: None,841+ track,842+ current_group: None,843+ last_group_sequence: None,844+ last_frame_sequence: None,845+ waiting_for_keyframe: true,846+ pending_group_recovery: None,847+ }848+ }849+850+ async fn next(&mut self) -> Result<Option<MoqMediaReadOutcome>, String> {851+ self.next_with_timeouts(852+ CLIENT_MOQ_OBJECT_READ_TIMEOUT,853+ CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT,854+ )855+ .await856+ }857+858+ async fn next_with_timeouts(859+ &mut self,860+ object_read_timeout: Duration,861+ group_recovery_timeout: Duration,862+ ) -> Result<Option<MoqMediaReadOutcome>, String> {863+ loop {864+ if self.current_group.is_none() {865+ let replacement_for_cancelled_group = self.pending_group_recovery.is_some();866+ let group = if replacement_for_cancelled_group {867+ match tokio::time::timeout(868+ group_recovery_timeout,869+ self.track.next_group(),870+ )871+ .await872+ {873+ Ok(group) => group,874+ Err(_) => {875+ let recovery = self876+ .pending_group_recovery877+ .take()878+ .expect("pending MoQ recovery reason was present");879+ return match recovery {880+ MoqGroupRecovery::ExpectedSupersession => Err(format!(881+ "Timed out after {} ms waiting for the MoQ group that supersedes an expected GOP cancellation",882+ group_recovery_timeout.as_millis()883+ )),884+ MoqGroupRecovery::RecoverableGap(reason) => {885+ Ok(Some(MoqMediaReadOutcome::Dropped { reason }))886+ }887+ };888+ }889+ }890+ } else {891+ self.track.next_group().await892+ }893+ .map_err(|error| format!("Upstream MoQ video track failed: {error}"))?;894+ let Some(group) = group else {895+ return Ok(None);896+ };897+ self.pending_group_recovery = None;898+ let sequence = group.sequence;899+ let group_gap =900+ match classify_moq_group_sequence(self.last_group_sequence, sequence) {901+ Ok(group_gap) => group_gap,902+ Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),903+ };904+ if group_gap {905+ self.waiting_for_keyframe = true;906+ }907+ self.current_group = Some(MoqGroupCursor {908+ group,909+ sequence,910+ object_count: 0,911+ object_bytes: 0,912+ group_gap,913+ replacement_for_cancelled_group,914+ });915+ }916+917+ let cursor = self918+ .current_group919+ .as_mut()920+ .expect("MoQ group cursor was initialized");921+ let object =922+ match tokio::time::timeout(object_read_timeout, cursor.group.read_frame()).await {923+ Err(_) => {924+ let sequence = cursor.sequence;925+ self.last_group_sequence = Some(sequence);926+ self.current_group = None;927+ self.waiting_for_keyframe = true;928+ return Ok(Some(MoqMediaReadOutcome::Dropped {929+ reason: KeyframeRequestReasonV3::DeliveryTimeout,930+ }));931+ }932+ Ok(Ok(object)) => object,933+ Ok(Err(error)) if moq_group_error_is_recoverable(&error) => {934+ let sequence = cursor.sequence;935+ self.last_group_sequence = Some(sequence);936+ self.current_group = None;937+ self.waiting_for_keyframe = true;938+ let reason = moq_group_error_reason(&error);939+ self.pending_group_recovery =940+ Some(if matches!(error, moq_net::Error::Cancel) {941+ MoqGroupRecovery::ExpectedSupersession942+ } else {943+ MoqGroupRecovery::RecoverableGap(reason)944+ });945+ continue;946+ }947+ Ok(Err(error)) => {948+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(949+ "Upstream MoQ group {} failed: {error}",950+ cursor.sequence951+ ))));952+ }953+ };954+ let Some(object) = object else {955+ if cursor.object_count == 0 {956+ if cursor.replacement_for_cancelled_group {957+ self.last_group_sequence = Some(cursor.sequence);958+ self.current_group = None;959+ self.waiting_for_keyframe = true;960+ return Ok(Some(MoqMediaReadOutcome::Dropped {961+ reason: KeyframeRequestReasonV3::TransportGap,962+ }));963+ }964+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(965+ "Upstream MoQ group {} was empty",966+ cursor.sequence967+ ))));968+ }969+ self.last_group_sequence = Some(cursor.sequence);970+ self.current_group = None;971+ continue;972+ };973+974+ let next_group_bytes = match validate_moq_object_bounds(975+ cursor.sequence,976+ cursor.object_count,977+ cursor.object_bytes,978+ object.len(),979+ ) {980+ Ok(next_group_bytes) => next_group_bytes,981+ Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),982+ };983+ let frame = match decode_media_frame_object(&object) {984+ Ok(frame) => frame,985+ Err(error) => {986+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(987+ "Invalid upstream MoQ media object in group {} object {}: {error}",988+ cursor.sequence, cursor.object_count989+ ))));990+ }991+ };992+993+ let first_object = cursor.object_count == 0;994+ let frame_contiguous = match validate_moq_group_frame(995+ cursor.sequence,996+ first_object,997+ self.last_frame_sequence,998+ &frame,999+ ) {1000+ Ok(frame_contiguous) => frame_contiguous,1001+ Err(_error) if first_object && cursor.replacement_for_cancelled_group => {1002+ self.last_group_sequence = Some(cursor.sequence);1003+ self.current_group = None;1004+ self.waiting_for_keyframe = true;1005+ return Ok(Some(MoqMediaReadOutcome::Dropped {1006+ reason: KeyframeRequestReasonV3::TransportGap,1007+ }));1008+ }1009+ Err(_error) if !first_object => {1010+ self.last_group_sequence = Some(cursor.sequence);1011+ self.current_group = None;1012+ self.waiting_for_keyframe = true;1013+ return Ok(Some(MoqMediaReadOutcome::Dropped {1014+ reason: KeyframeRequestReasonV3::TransportGap,1015+ }));1016+ }1017+ Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),1018+ };1019+ let discontinuity = self.waiting_for_keyframe1020+ || (first_object && cursor.group_gap)1021+ || (first_object && !frame_contiguous)1022+ || frame.header.flags.contains(FrameFlags::DISCONTINUITY);1023+ cursor.object_count += 1;1024+ cursor.object_bytes = next_group_bytes;1025+ self.last_frame_sequence = Some(frame.header.sequence);1026+ self.waiting_for_keyframe = false;1027+ return Ok(Some(MoqMediaReadOutcome::Frame {1028+ frame,1029+ discontinuity,1030+ }));1031+ }1032+ }1033+}1034+1035+fn classify_moq_group_sequence(previous: Option<u64>, current: u64) -> Result<bool, String> {1036+ let Some(previous) = previous else {1037+ return Ok(false);1038+ };1039+ if current <= previous {1040+ return Err(format!(1041+ "Upstream MoQ group sequence did not increase: previous={previous}, current={current}"1042+ ));1043+ }1044+ Ok(previous.checked_add(1) != Some(current))1045+}1046+1047+fn validate_moq_object_bounds(1048+ group_sequence: u64,1049+ object_count: usize,1050+ object_bytes: usize,1051+ object_len: usize,1052+) -> Result<usize, String> {1053+ let max_objects = MAX_MEDIA_OBJECT_ID_V3 as usize + 1;1054+ if object_count >= max_objects {1055+ return Err(format!(1056+ "Upstream MoQ group {group_sequence} exceeded {max_objects} media objects"1057+ ));1058+ }1059+ let next_group_bytes = object_bytes1060+ .checked_add(object_len)1061+ .ok_or_else(|| "Upstream MoQ group byte count overflowed".to_string())?;1062+ if next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3 {1063+ return Err(format!(1064+ "Upstream MoQ group {group_sequence} exceeded the {MAX_MEDIA_GROUP_BYTES_V3} byte limit"1065+ ));1066+ }1067+ Ok(next_group_bytes)1068+}1069+1070+fn validate_moq_group_frame(1071+ group_sequence: u64,1072+ first_object: bool,1073+ last_frame_sequence: Option<u64>,1074+ frame: &MediaFrame,1075+) -> Result<bool, String> {1076+ if first_object1077+ && !(frame.header.codec == MediaCodec::H2641078+ && frame.header.flags.contains(FrameFlags::KEYFRAME)1079+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG))1080+ {1081+ return Err(format!(1082+ "Upstream MoQ group {group_sequence} did not begin with a configured H.264 keyframe"1083+ ));1084+ }1085+ let contiguous = last_frame_sequence1086+ .is_none_or(|previous| previous.checked_add(1) == Some(frame.header.sequence));1087+ if !first_object && !contiguous {1088+ return Err(format!(1089+ "Upstream MoQ group {group_sequence} contains a non-contiguous access-unit sequence"1090+ ));1091+ }1092+ Ok(contiguous)1093+}1094+1095+fn moq_group_error_is_recoverable(error: &moq_net::Error) -> bool {1096+ matches!(1097+ error,1098+ moq_net::Error::Cancel1099+ | moq_net::Error::Old1100+ | moq_net::Error::Timeout1101+ | moq_net::Error::Dropped1102+ | moq_net::Error::CacheFull1103+ | moq_net::Error::Remote(0 | 2 | 3 | 24 | 26)1104+ )1105+}1106+1107+fn moq_group_error_reason(error: &moq_net::Error) -> KeyframeRequestReasonV3 {1108+ if matches!(error, moq_net::Error::Timeout | moq_net::Error::Remote(3)) {1109+ KeyframeRequestReasonV3::DeliveryTimeout1110+ } else {1111+ KeyframeRequestReasonV3::TransportGap1112+ }1113+}1114+6751115 #[derive(Debug)]6761116 enum MediaObjectReadOutcome {6771117 Frame {
@@ -1386,7 +1826,7 @@ fn connect_error_is_unsupported_alpn(error: &iroh::endpoint::ConnectError) -> bo
13861826 }13871827 }13881828 1389-async fn open_negotiated_media_stream(1829+async fn open_legacy_negotiated_media_stream(13901830 endpoint: &Endpoint,13911831 address: &iroh::EndpointAddr,13921832 nonce: [u8; 16],
@@ -1496,6 +1936,99 @@ async fn open_negotiated_media_stream(
14961936 }14971937 }14981938 1939+async fn open_negotiated_media_stream(1940+ endpoint: &Endpoint,1941+ address: &iroh::EndpointAddr,1942+ nonce: [u8; 16],1943+ invitation: Option<&str>,1944+) -> Result<1945+ (1946+ iroh::endpoint::Connection,1947+ iroh::endpoint::RecvStream,1948+ Option<iroh::endpoint::SendStream>,1949+ NegotiatedV1Stream,1950+ MediaTransport,1951+ ),1952+ String,1953+> {1954+ match endpoint.connect(address.clone(), CONTROL_ALPN_V1).await {1955+ Ok(connection) => {1956+ let (mut send, mut recv) = connection1957+ .open_bi()1958+ .await1959+ .map_err(|error| format!("Failed to open control handshake: {error}"))?;1960+ let negotiation = negotiate_v1(1961+ &mut send,1962+ &mut recv,1963+ nonce,1964+ vec![Capability::VideoH264],1965+ Some(Capability::VideoH264),1966+ "control",1967+ invitation,1968+ )1969+ .await?;1970+ // CONTROL owns the authenticated host lease. Keep both the1971+ // connection and the client->host send leg alive for keyframe1972+ // requests while media uses a separate upstream MoQ session.1973+ Ok((1974+ connection,1975+ recv,1976+ Some(send),1977+ negotiation,1978+ MediaTransport::UpstreamMoq,1979+ ))1980+ }1981+ Err(control_error) if connect_error_is_unsupported_alpn(&control_error) => {1982+ open_legacy_negotiated_media_stream(endpoint, address, nonce, invitation).await1983+ }1984+ Err(control_error) => Err(format!(1985+ "Control connection failed without an explicit unsupported-ALPN signal; refusing an unsafe media downgrade: {control_error}"1986+ )),1987+ }1988+}1989+1990+async fn open_upstream_moq_media(1991+ endpoint: &Endpoint,1992+ address: &iroh::EndpointAddr,1993+ session_id: u64,1994+) -> Result<(MoqMediaReceiver, iroh::endpoint::Connection), String> {1995+ let broadcast_name = media_moq_broadcast_name(session_id)1996+ .map_err(|error| format!("Invalid MoQ media session name: {error}"))?;1997+ let moq = Moq::new(endpoint.clone());1998+ let mut session =1999+ tokio::time::timeout(CLIENT_MOQ_CONNECT_TIMEOUT, moq.connect(address.clone()))2000+ .await2001+ .map_err(|_| "Timed out connecting upstream MoQ media session".to_string())?2002+ .map_err(|error| format!("Failed to connect upstream MoQ media session: {error:#}"))?;2003+ if session.remote_id() != address.id {2004+ session.close(1, b"remote identity mismatch");2005+ return Err(format!(2006+ "Upstream MoQ connected to unexpected peer {}; expected {}",2007+ session.remote_id(),2008+ address.id2009+ ));2010+ }2011+ let diagnostics_connection = session.conn().clone();2012+ let broadcast = tokio::time::timeout(2013+ CLIENT_MOQ_SUBSCRIBE_TIMEOUT,2014+ session.subscribe(&broadcast_name),2015+ )2016+ .await2017+ .map_err(|_| format!("Timed out waiting for upstream MoQ broadcast {broadcast_name}"))?2018+ .map_err(|error| {2019+ format!("Failed to subscribe to upstream MoQ broadcast {broadcast_name}: {error}")2020+ })?;2021+ let track = broadcast2022+ .subscribe_track(&Track::new(MOQ_VIDEO_H264_TRACK))2023+ .map_err(|error| {2024+ format!("Failed to subscribe to upstream MoQ track {MOQ_VIDEO_H264_TRACK}: {error}")2025+ })?;2026+ Ok((2027+ MoqMediaReceiver::new(moq, session, broadcast, track),2028+ diagnostics_connection,2029+ ))2030+}2031+14992032 async fn run_media_control_writer_v3(15002033 mut stream: iroh::endpoint::SendStream,15012034 mut requests: tokio::sync::mpsc::Receiver<(KeyframeRequestReasonV3, Option<u64>)>,
@@ -2082,10 +2615,32 @@ pub async fn iroh_client_connect(
20822615 .map_err(|e| format!("Failed to finish frame start stream: {e}"))?;20832616 (connection, recv, None, None, MediaTransport::LegacyV0)20842617 };2085- let frame_connection_for_stats = frame_conn.clone();20862618 let media_session_id = media_negotiation20872619 .as_ref()20882620 .map(|negotiation| negotiation.session_id);2621+ let (upstream_moq_media, frame_connection_for_stats) = if media_transport2622+ == MediaTransport::UpstreamMoq2623+ {2624+ let session_id = media_session_id2625+ .ok_or_else(|| "Host omitted the control session ID required by MoQ".to_string())?;2626+ let (receiver, diagnostics_connection) =2627+ match open_upstream_moq_media(&endpoint, &addr, session_id).await {2628+ Ok(media) => media,2629+ Err(error) => {2630+ // CONTROL already authenticated and owns the host's2631+ // one-client lease. A post-auth MoQ failure is2632+ // terminal, and must explicitly release that lease;2633+ // it must never fall through to a legacy media ALPN.2634+ frame_conn.close(1_u32.into(), b"upstream MoQ setup failed");2635+ let _ =2636+ tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close()).await;2637+ return Err(error);2638+ }2639+ };2640+ (Some(receiver), diagnostics_connection)2641+ } else {2642+ (None, frame_conn.clone())2643+ };20892644 let pointer_surface_dimensions = media_negotiation20902645 .as_ref()20912646 .and_then(|negotiation| negotiation.pointer_surface_dimensions);
@@ -2264,7 +2819,9 @@ pub async fn iroh_client_connect(
22642819 let (control_tx, control_rx) = tokio::sync::mpsc::channel(1);22652820 *state.media_control.lock().await = Some((media_generation, control_tx.clone()));22662821 tokio::spawn(run_media_control_writer_v3(control_stream, control_rx));2267- let _ = control_tx.try_send((KeyframeRequestReasonV3::Join, None));2822+ if media_transport == MediaTransport::GroupedObjectsV3 {2823+ let _ = control_tx.try_send((KeyframeRequestReasonV3::Join, None));2824+ }22682825 Some(control_tx)22692826 } else {22702827 *state.media_control.lock().await = None;
@@ -2390,6 +2947,7 @@ pub async fn iroh_client_connect(
23902947 let mut media_objects_v3 = (media_transport == MediaTransport::GroupedObjectsV3)23912948 .then(|| MediaObjectReceiverV3::new(frame_connection_for_stats.clone()));23922949 let mut media_object_sequence_v3 = MediaObjectSequenceV3::new();2950+ let mut upstream_moq_media = upstream_moq_media;23932951 23942952 let mut decoder = if use_webcodecs {23952953 None
@@ -2398,6 +2956,14 @@ pub async fn iroh_client_connect(
23982956 Ok(d) => Some(d),23992957 Err(e) => {24002958 emit_frame_error(&app, media_generation, format!("Decoder init failed: {e}"));2959+ if media_transport == MediaTransport::UpstreamMoq {2960+ retire_upstream_moq_generation(2961+ &app,2962+ media_generation,2963+ connected_audio_generation,2964+ )2965+ .await;2966+ }24012967 return;24022968 }24032969 }
@@ -2456,6 +3022,64 @@ pub async fn iroh_client_connect(
24563022 pts_micros,24573023 discontinuity,24583024 ) = match media_transport {3025+ MediaTransport::UpstreamMoq => {3026+ let receiver = upstream_moq_media3027+ .as_mut()3028+ .expect("upstream MoQ receiver must exist for MoQ transport");3029+ loop {3030+ let outcome = match receiver.next().await {3031+ Ok(Some(outcome)) => outcome,3032+ Ok(None) => {3033+ emit_frame_error(3034+ &app,3035+ media_generation,3036+ "Upstream MoQ video track closed",3037+ );3038+ break 'frames;3039+ }3040+ Err(error) => {3041+ emit_frame_error(&app, media_generation, error);3042+ break 'frames;3043+ }3044+ };3045+ match outcome {3046+ MoqMediaReadOutcome::Dropped { reason } => {3047+ frontend_waiting_for_keyframe = true;3048+ let mut metrics = lock_client_media_metrics(&metrics);3049+ metrics.observe_transport_object_drop(false);3050+ metrics.begin_frontend_resync(metrics_started.elapsed());3051+ try_queue_media_keyframe_request(3052+ media_control_requests.as_ref(),3053+ reason,3054+ receiver.last_frame_sequence,3055+ );3056+ }3057+ MoqMediaReadOutcome::Malformed(error) => {3058+ emit_frame_error(&app, media_generation, error);3059+ break 'frames;3060+ }3061+ MoqMediaReadOutcome::Frame {3062+ frame,3063+ discontinuity,3064+ } => {3065+ let codec = match frame.header.codec {3066+ MediaCodec::H264 => "h264".to_string(),3067+ };3068+ break (3069+ u32::from(frame.header.width),3070+ u32::from(frame.header.height),3071+ frame.payload,3072+ frame.header.flags.contains(FrameFlags::KEYFRAME),3073+ codec,3074+ Some(frame.header.sequence),3075+ Some(frame.header.capture_timestamp_us),3076+ Some(frame.header.pts_us),3077+ discontinuity,3078+ );3079+ }3080+ }3081+ }3082+ }24593083 MediaTransport::GroupedObjectsV3 => {24603084 let receiver = media_objects_v324613085 .as_mut()
@@ -2840,6 +3464,10 @@ pub async fn iroh_client_connect(
28403464 28413465 let _ = stats_stop.send(true);28423466 let _ = stats_task.await;3467+ if media_transport == MediaTransport::UpstreamMoq {3468+ retire_upstream_moq_generation(&app, media_generation, connected_audio_generation)3469+ .await;3470+ }28433471 drop(endpoint);28443472 });28453473
@@ -2979,6 +3607,199 @@ mod tests {
29793607 }29803608 }29813609 3610+ #[test]3611+ fn upstream_moq_group_ids_detect_only_real_transport_gaps() {3612+ assert!(!classify_moq_group_sequence(None, 41).unwrap());3613+ assert!(!classify_moq_group_sequence(Some(41), 42).unwrap());3614+ assert!(classify_moq_group_sequence(Some(42), 44).unwrap());3615+ assert!(classify_moq_group_sequence(Some(44), 44).is_err());3616+ assert!(classify_moq_group_sequence(Some(44), 43).is_err());3617+ }3618+3619+ #[test]3620+ fn upstream_moq_group_gap_keyframe_is_the_recovery_barrier() {3621+ let keyframe = media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));3622+ let contiguous = validate_moq_group_frame(9, true, Some(42), &keyframe).unwrap();3623+ assert!(!contiguous);3624+3625+ // A native group-id gap puts the decoder into resync, but this exact3626+ // configured frame 0 exits it immediately. There is deliberately no3627+ // keyframe-request action on MoqMediaReadOutcome::Frame: requesting a3628+ // replacement here would recreate the grouped-v3 feedback loop.3629+ let group_gap = classify_moq_group_sequence(Some(7), 9).unwrap();3630+ let discontinuity = group_gap || !contiguous;3631+ assert!(discontinuity);3632+ }3633+3634+ #[tokio::test]3635+ async fn upstream_moq_idr_abort_to_next_frame_zero_has_no_feedback_loop() {3636+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();3637+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());3638+3639+ let mut prior_group = producer.append_group().unwrap();3640+ let prior_keyframe =3641+ media_object_frame(40, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));3642+ prior_group3643+ .write_frame(sigil_protocol::encode_media_frame_object(&prior_keyframe).unwrap())3644+ .unwrap();3645+ assert!(matches!(3646+ receiver.next().await.unwrap(),3647+ Some(MoqMediaReadOutcome::Frame { .. })3648+ ));3649+3650+ prior_group.abort(moq_net::Error::Cancel).unwrap();3651+ let mut replacement = producer.append_group().unwrap();3652+ let replacement_keyframe =3653+ media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));3654+ replacement3655+ .write_frame(sigil_protocol::encode_media_frame_object(&replacement_keyframe).unwrap())3656+ .unwrap();3657+ replacement.finish().unwrap();3658+3659+ // `next` consumes the expected Cancel internally and returns the3660+ // replacement frame directly. A Dropped outcome here would enqueue a3661+ // needless keyframe request and recreate the recovery feedback loop.3662+ assert!(matches!(3663+ receiver.next().await.unwrap(),3664+ Some(MoqMediaReadOutcome::Frame {3665+ discontinuity: true,3666+ ..3667+ })3668+ ));3669+ }3670+3671+ #[tokio::test]3672+ async fn upstream_moq_expected_cancel_without_replacement_is_terminal_without_request() {3673+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();3674+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());3675+ let mut group = producer.append_group().unwrap();3676+ let keyframe = media_object_frame(1, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));3677+ group3678+ .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())3679+ .unwrap();3680+ assert!(matches!(3681+ receiver.next().await.unwrap(),3682+ Some(MoqMediaReadOutcome::Frame { .. })3683+ ));3684+ group.abort(moq_net::Error::Cancel).unwrap();3685+3686+ let error = receiver3687+ .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))3688+ .await3689+ .unwrap_err();3690+ assert!(error.contains("expected GOP cancellation"));3691+ }3692+3693+ #[tokio::test]3694+ async fn upstream_moq_partial_object_read_has_protocol_bounded_deadline() {3695+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();3696+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());3697+ let _stalled_group = producer.append_group().unwrap();3698+3699+ assert!(matches!(3700+ receiver3701+ .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))3702+ .await3703+ .unwrap(),3704+ Some(MoqMediaReadOutcome::Dropped {3705+ reason: KeyframeRequestReasonV3::DeliveryTimeout3706+ })3707+ ));3708+ assert_eq!(3709+ CLIENT_MOQ_OBJECT_READ_TIMEOUT,3710+ Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64)3711+ );3712+ assert!(CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT >= Duration::from_millis(1_000));3713+ }3714+3715+ #[tokio::test]3716+ async fn upstream_moq_invalid_cancel_replacement_requests_recovery() {3717+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();3718+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());3719+ let mut prior = producer.append_group().unwrap();3720+ let keyframe = media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));3721+ prior3722+ .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())3723+ .unwrap();3724+ assert!(matches!(3725+ receiver.next().await.unwrap(),3726+ Some(MoqMediaReadOutcome::Frame { .. })3727+ ));3728+ prior.abort(moq_net::Error::Cancel).unwrap();3729+3730+ let mut invalid_replacement = producer.append_group().unwrap();3731+ let delta = media_object_frame(11, FrameFlags::NONE);3732+ invalid_replacement3733+ .write_frame(sigil_protocol::encode_media_frame_object(&delta).unwrap())3734+ .unwrap();3735+ invalid_replacement.finish().unwrap();3736+ assert!(matches!(3737+ receiver.next().await.unwrap(),3738+ Some(MoqMediaReadOutcome::Dropped {3739+ reason: KeyframeRequestReasonV3::TransportGap3740+ })3741+ ));3742+ }3743+3744+ #[test]3745+ fn upstream_moq_requires_configured_keyframe_zero_and_contiguous_deltas() {3746+ let delta = media_object_frame(10, FrameFlags::NONE);3747+ assert!(validate_moq_group_frame(3, true, None, &delta).is_err());3748+3749+ let unconfigured_keyframe = media_object_frame(10, FrameFlags::KEYFRAME);3750+ assert!(validate_moq_group_frame(3, true, None, &unconfigured_keyframe).is_err());3751+3752+ let configured =3753+ media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));3754+ assert!(validate_moq_group_frame(3, true, None, &configured).is_ok());3755+ assert!(validate_moq_group_frame(3, false, Some(8), &delta).is_err());3756+ assert!(validate_moq_group_frame(3, false, Some(9), &delta).is_ok());3757+ }3758+3759+ #[test]3760+ fn upstream_moq_bounds_objects_and_group_bytes_before_growth() {3761+ assert_eq!(validate_moq_object_bounds(1, 0, 0, 40).unwrap(), 40);3762+ assert!(validate_moq_object_bounds(1, MAX_MEDIA_OBJECT_ID_V3 as usize + 1, 0, 1,).is_err());3763+ assert!(validate_moq_object_bounds(1, 0, MAX_MEDIA_GROUP_BYTES_V3, 1).is_err());3764+ assert!(validate_moq_object_bounds(1, 0, usize::MAX, 1).is_err());3765+ }3766+3767+ #[test]3768+ fn upstream_moq_cancellation_is_resync_but_protocol_errors_are_terminal() {3769+ for recoverable in [3770+ moq_net::Error::Cancel,3771+ moq_net::Error::Old,3772+ moq_net::Error::Timeout,3773+ moq_net::Error::Dropped,3774+ moq_net::Error::CacheFull,3775+ moq_net::Error::Remote(3),3776+ ] {3777+ assert!(moq_group_error_is_recoverable(&recoverable));3778+ }3779+ assert!(!moq_group_error_is_recoverable(3780+ &moq_net::Error::ProtocolViolation3781+ ));3782+ assert!(!moq_group_error_is_recoverable(&moq_net::Error::WrongSize));3783+ assert_eq!(3784+ moq_group_error_reason(&moq_net::Error::Timeout),3785+ KeyframeRequestReasonV3::DeliveryTimeout3786+ );3787+ assert_eq!(3788+ moq_group_error_reason(&moq_net::Error::Cancel),3789+ KeyframeRequestReasonV3::TransportGap3790+ );3791+ }3792+3793+ #[test]3794+ fn upstream_moq_transport_is_distinct_from_legacy_compatibility() {3795+ assert_eq!(MediaTransport::UpstreamMoq.diagnostic_name(), "iroh-moq");3796+ assert_ne!(3797+ MediaTransport::UpstreamMoq,3798+ MediaTransport::GroupedObjectsV33799+ );3800+ assert!(decode_media_frame_object(&[0_u8; 8]).is_err());3801+ }3802+29823803 fn completed_object_index(outcome: &MediaObjectReadOutcome) -> Option<u64> {29833804 outcome.object_index()29843805 }
@@ -3573,6 +4394,15 @@ mod tests {
35734394 assert_eq!(closes.load(Ordering::SeqCst), 1);35744395 }35754396 4397+ #[test]4398+ fn generation_owned_retirement_cannot_take_a_replacement_session() {4399+ let mut slot = Some((12, "replacement"));4400+ assert_eq!(take_generation_owned(&mut slot, 11), None);4401+ assert_eq!(slot, Some((12, "replacement")));4402+ assert_eq!(take_generation_owned(&mut slot, 12), Some("replacement"));4403+ assert_eq!(slot, None);4404+ }4405+35764406 #[test]35774407 fn pointer_feedback_channel_has_explicit_bounded_terminal_envelopes() {35784408 let position = serde_json::to_value(PointerFeedbackPayload::Position {