From f3a59c787ea920e57ea389b26d0577f5d6c8b019 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 18:13:41 -0800 Subject: [PATCH 1/7] chore(quest): claim quest/m0/pin-audio-capture.md Quest-Claim: 559bffe4e99e0e1b From 41fd483a78bd9b128639ca9435a6561b3e18710c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 18:23:21 -0800 Subject: [PATCH 2/7] chore(moq): pin moq fd07f5b for the audio capture split and mix router Advances the moq gitlink to main and adapts to what it breaks: audio captures open a capture::Producer that feeds the bus, every source with sound is routed into the program bus::Mix while it runs, the audio thread advances the bus Driver, and publisher epochs are now route instances. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 484 +++++++++++++++++-------------- README.md | 6 +- moq | 2 +- quest/m0/README.md | 2 - quest/m0/audio-stall.md | 12 +- quest/m0/compose/sources.md | 12 +- quest/m0/mixes.md | 12 +- quest/m0/pin-audio-capture.md | 11 - quest/m0/pin-audio-mix-router.md | 12 - rs/moq-scene/src/lib.rs | 3 +- rs/moq-scene/src/source.rs | 56 +++- rs/moq-studio/src/engine.rs | 109 +++---- rs/moq-studio/src/health.rs | 10 +- rs/moq-studio/src/main.rs | 14 +- rs/moq-studio/src/ui.rs | 2 +- rs/moq-studio/src/uplink.rs | 2 +- 16 files changed, 404 insertions(+), 345 deletions(-) delete mode 100644 quest/m0/pin-audio-capture.md delete mode 100644 quest/m0/pin-audio-mix-router.md diff --git a/Cargo.lock b/Cargo.lock index d603dd6e..5d5ebb9e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -427,12 +427,6 @@ dependencies = [ "tower-service", ] -[[package]] -name = "base16ct" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd307490d624467aa6f74b0eabb77633d1f758a7b25f12bceb0b22e08d9726f6" - [[package]] name = "base64" version = "0.22.1" @@ -482,6 +476,8 @@ dependencies = [ "cexpr", "clang-sys", "itertools 0.13.0", + "log", + "prettyplease", "proc-macro2", "quote", "regex", @@ -912,6 +908,17 @@ dependencies = [ "libc", ] +[[package]] +name = "core-models" +version = "0.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee3344d349528f449816cfa85e558c439bdca5a6d8a738cfb21e69f0b2b1952f" +dependencies = [ + "hax-lib 0.3.7", + "pastey", + "rand", +] + [[package]] name = "core_affinity" version = "0.8.3" @@ -968,12 +975,6 @@ dependencies = [ "windows-core", ] -[[package]] -name = "cpubits" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "15b85f9c39137c3a891689859392b1bd49812121d0d61c9caf00d46ed5ce06ae" - [[package]] name = "cpufeatures" version = "0.3.1" @@ -983,6 +984,18 @@ dependencies = [ "libc", ] +[[package]] +name = "crabgrind" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "035c801dc4b418e1d857a6536dc88b4d1929c7a1fd163389766e18d322d34a84" +dependencies = [ + "bindgen 0.72.1", + "cc", + "pkg-config", + "rustversion", +] + [[package]] name = "crc" version = "3.4.0" @@ -1037,31 +1050,13 @@ version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" -[[package]] -name = "crypto-bigint" -version = "0.7.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c035cb64cba03dad2fb921f3e218785e8cf9a41790da5eb3a1400f62fa2be11" -dependencies = [ - "cpubits", - "ctutils", - "getrandom 0.4.3", - "hybrid-array", - "num-traits", - "rand_core 0.10.1", - "subtle", - "zeroize", -] - [[package]] name = "crypto-common" version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" dependencies = [ - "getrandom 0.4.3", "hybrid-array", - "rand_core 0.10.1", ] [[package]] @@ -1071,7 +1066,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fe3e9d21d39e6f39ff46e03d5b621424f88ee23d8a1672c6bfc9ccc7f2da4f6a" dependencies = [ "cmov", - "subtle", ] [[package]] @@ -1089,6 +1083,31 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f27ae1dd37df86211c42e150270f82743308803d90a6f6e6651cd730d5e1732f" +[[package]] +name = "curve25519-dalek" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23" +dependencies = [ + "cfg-if", + "cpufeatures", + "curve25519-dalek-derive", + "fiat-crypto", + "rustc_version", + "subtle", +] + +[[package]] +name = "curve25519-dalek-derive" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "darling" version = "0.24.1" @@ -1183,8 +1202,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a878c850e9e421b20262e9b41f9c860e4785fa07541c266b62ff9d1ef998a80a" dependencies = [ "const-oid 0.10.2", - "pem-rfc7468", - "zeroize", ] [[package]] @@ -1375,21 +1392,6 @@ dependencies = [ "syn 1.0.109", ] -[[package]] -name = "ecdsa" -version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0681a4fc24c767085329728d8dfba959af91228aa4610cca4f8ce317ba46ae0" -dependencies = [ - "der 0.8.2", - "digest", - "elliptic-curve", - "rfc6979", - "signature 3.0.0", - "spki 0.8.1", - "zeroize", -] - [[package]] name = "ecolor" version = "0.36.2" @@ -1527,27 +1529,6 @@ version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" -[[package]] -name = "elliptic-curve" -version = "0.14.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d65aa39b3a5c1c9c1b745c9a019234bb7a21b77abcb4f4d266d706e2d577d65" -dependencies = [ - "base16ct", - "crypto-bigint", - "crypto-common", - "digest", - "ff", - "group", - "hybrid-array", - "pem-rfc7468", - "pkcs8 0.11.0", - "rand_core 0.10.1", - "sec1", - "subtle", - "zeroize", -] - [[package]] name = "emath" version = "0.36.2" @@ -1729,16 +1710,6 @@ version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b97b65636e5b9ef369943878ac74335ba1c55c1cb6adbf1e2c293c624248d693" -[[package]] -name = "ff" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1f686ab92a9fb0eaf188f6c6c87b89490baa6fdb0db4544ba4dc47f7942489f" -dependencies = [ - "rand_core 0.10.1", - "subtle", -] - [[package]] name = "fiat-crypto" version = "0.3.0" @@ -2125,17 +2096,6 @@ dependencies = [ "windows", ] -[[package]] -name = "group" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7fd1a1c7a5206c5b7a3f5a0d7ccd3ff85d0c8f5133d62a02680255b0004af5f4" -dependencies = [ - "ff", - "rand_core 0.10.1", - "subtle", -] - [[package]] name = "guillotiere" version = "0.7.0" @@ -2199,7 +2159,6 @@ dependencies = [ "kio", "moq-json", "moq-net", - "regex", "serde", "serde_json", "serde_with", @@ -2254,6 +2213,79 @@ dependencies = [ "hashbrown 0.17.1", ] +[[package]] +name = "hax-lib" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01a33cb227f97ee5c419064c31d7c55a7d895f28d8019ccdadc311cc036500c4" +dependencies = [ + "hax-lib-macros 0.3.7", + "num-bigint", + "num-traits", +] + +[[package]] +name = "hax-lib" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb3bfcfc915e894c331662e4ac76706b777a5956535ea24535f4df30da04b1bf" +dependencies = [ + "hax-lib-macros 0.4.2", + "num-bigint", + "num-traits", +] + +[[package]] +name = "hax-lib-macros" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28da835a5f153e9f3d0cc1f42ee605a1cfa9093c93348193793d60a1b0908b51" +dependencies = [ + "hax-lib-macros-types 0.3.7", + "proc-macro-error2", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "hax-lib-macros" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e9010990fcba1d1494047fa433a433be0a5df23653a9784c739f7bba270272e" +dependencies = [ + "hax-lib-macros-types 0.4.2", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "hax-lib-macros-types" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eed327e9a28d6ee018a4a9ba842d4ee27e378c8a591d2beb6f42f32b24a02fcb" +dependencies = [ + "proc-macro2", + "quote", + "serde", + "serde_json", + "uuid", +] + +[[package]] +name = "hax-lib-macros-types" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e053e3a5aeb92df47b07323180c5088edc488ca49883a234f6c73255c54cc520" +dependencies = [ + "proc-macro2", + "quote", + "serde", + "serde_json", + "uuid", +] + [[package]] name = "heck" version = "0.5.0" @@ -2414,9 +2446,7 @@ version = "0.4.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" dependencies = [ - "subtle", "typenum", - "zeroize", ] [[package]] @@ -2842,14 +2872,13 @@ version = "11.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e75fe14a82d81e5f5af639997db37d8b96045938a7ac6ab18cdbe1c7467e05e1" dependencies = [ - "aws-lc-rs", "base64 0.22.1", "getrandom 0.2.17", "js-sys", "pem", "serde", "serde_json", - "signature 2.2.0", + "signature", "simple_asn1", "zeroize", ] @@ -2937,6 +2966,86 @@ version = "0.2.191" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" +[[package]] +name = "libcrux-aes" +version = "0.0.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd415025511a0bd8731cae34ae2d5e07c1b3f0443d5c12bd199c9e9b2f1fd4d5" +dependencies = [ + "libcrux-intrinsics", + "libcrux-platform", + "libcrux-secrets", + "libcrux-traits", +] + +[[package]] +name = "libcrux-hacl-rs" +version = "0.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27ce64ead37fc67c1039fdbcf1e8c97044e6a644cb3d1f79060b5c9ac8fdfe0a" +dependencies = [ + "libcrux-macros", +] + +[[package]] +name = "libcrux-intrinsics" +version = "0.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de82722c9c7419a311e84ee902e9643a15e39a6fcaade87f21e94f8d0f30fa70" +dependencies = [ + "core-models", + "hax-lib 0.4.2", +] + +[[package]] +name = "libcrux-macros" +version = "0.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2afdd4da30d6e23f76bd6f3ad61be5c773720349eef3e791f8b79cfea3fe1436" +dependencies = [ + "quote", + "syn 3.0.6", +] + +[[package]] +name = "libcrux-platform" +version = "0.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8bf8bee8c35d34867a3762d6ee7b74e8523a882855c3cfce30484e9554342206" +dependencies = [ + "libc", +] + +[[package]] +name = "libcrux-poly1305" +version = "0.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d727ee28ee0d5d2661daa5f2f47965cd3489a0ac9130f3410e95632a98b12e49" +dependencies = [ + "libcrux-hacl-rs", + "libcrux-macros", +] + +[[package]] +name = "libcrux-secrets" +version = "0.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03c4ba47b3596f2771213ff162c792e9b93ddc0cdd178a2e407e00bb8ae4afcd" +dependencies = [ + "crabgrind", + "hax-lib 0.4.2", +] + +[[package]] +name = "libcrux-traits" +version = "0.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8958bb5c3780eb7c3355aeeb051629b830e47bfefd36befed93c81b53366bbc4" +dependencies = [ + "libcrux-secrets", + "rand", +] + [[package]] name = "libloading" version = "0.8.9" @@ -3195,20 +3304,21 @@ dependencies = [ name = "moq-auth" version = "0.2.1" dependencies = [ - "aws-lc-rs", "axum", "base64 0.23.1", - "data-encoding", "jsonwebtoken", "kio", + "moq-crypto", "moq-pattern", - "p256", - "p384", "pkcs1", "pkcs8 0.11.0", "rand", "reqwest", + "rind", + "rind-jsonwebtoken", + "rind-rustls", "rustls", + "rustls-platform-verifier", "serde", "serde_json", "serde_with", @@ -3228,6 +3338,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "moq-crypto" +version = "0.0.11" +dependencies = [ + "data-encoding", + "serde", + "thiserror 2.0.21", +] + [[package]] name = "moq-json" version = "0.5.8" @@ -3315,7 +3434,6 @@ dependencies = [ "thiserror 2.0.21", "tracing", "url", - "uuid", ] [[package]] @@ -3460,6 +3578,7 @@ dependencies = [ "thiserror 2.0.21", "tokio", "tracing", + "uuid", ] [[package]] @@ -3513,7 +3632,6 @@ dependencies = [ "moq-quic", "moq-sock", "notify", - "pin-project-lite", "qmux", "rand", "rcgen", @@ -4527,33 +4645,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "p256" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2c9239b2dbc807adbbe147e8cf72ea7450c3a0aabe62cb8e75ff4ec22e1f72a" -dependencies = [ - "ecdsa", - "elliptic-curve", - "primefield", - "primeorder", - "sha2", -] - -[[package]] -name = "p384" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d17b851e6b3e378ab4ecb07fa2ed23f4d15f075735f8fec9fa1e7bdce5f8301f" -dependencies = [ - "ecdsa", - "elliptic-curve", - "fiat-crypto", - "primefield", - "primeorder", - "sha2", -] - [[package]] name = "parking" version = "2.2.1" @@ -4605,15 +4696,6 @@ dependencies = [ "serde_core", ] -[[package]] -name = "pem-rfc7468" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6305423e0e7738146434843d1694d621cce767262b2a86910beab705e4493d9" -dependencies = [ - "base64ct", -] - [[package]] name = "peniko" version = "0.6.1" @@ -4828,39 +4910,34 @@ dependencies = [ ] [[package]] -name = "primefield" -version = "0.14.0" +name = "proc-macro-crate" +version = "3.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c555a6e4eb7d4e158fcb028c835c3b8642206ddc279b5c6b202ef9a8bdb592f4" +checksum = "e67ba7e9b2b56446f1d419b1d807906278ffa1a658a8a5d8a39dcb1f5a78614f" dependencies = [ - "crypto-bigint", - "crypto-common", - "ff", - "rand_core 0.10.1", - "subtle", - "zeroize", + "toml_edit", ] [[package]] -name = "primeorder" -version = "0.14.0" +name = "proc-macro-error-attr2" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c9f42978c78a00e3d68f69fc03e57a234debae69da4020a4fb588fcdcd07b06" +checksum = "96de42df36bb9bba5542fe9f1a054b8cc87e172759a1868aa05c1f3acc89dfc5" dependencies = [ - "elliptic-curve", - "once_cell", - "primefield", - "serdect", - "wnaf", + "proc-macro2", + "quote", ] [[package]] -name = "proc-macro-crate" -version = "3.5.0" +name = "proc-macro-error2" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e67ba7e9b2b56446f1d419b1d807906278ffa1a658a8a5d8a39dcb1f5a78614f" +checksum = "11ec05c52be0a07b08061f7dd003e7d7092e0472bc731b4af7bb1ef876109802" dependencies = [ - "toml_edit", + "proc-macro-error-attr2", + "proc-macro2", + "quote", + "syn 2.0.119", ] [[package]] @@ -4929,7 +5006,6 @@ dependencies = [ "libc", "moq-net", "moq-quic", - "rustc-hash 2.1.3", "rustls", "thiserror 2.0.21", "tokio", @@ -5274,13 +5350,42 @@ dependencies = [ ] [[package]] -name = "rfc6979" -version = "0.6.0" +name = "rind" +version = "0.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4a459cddafb3fe76b31fd8f1108007566c40301feb64dc7b54656eb7388172b" +checksum = "422500e50252ad3bd319f359c54431441711a8fd55cb228973dde123ff8c2039" dependencies = [ - "crypto-bigint", - "hmac", + "cfg-if", + "curve25519-dalek", + "fiat-crypto", + "getrandom 0.4.3", + "libc", + "libcrux-aes", + "libcrux-hacl-rs", + "libcrux-macros", + "libcrux-poly1305", + "libcrux-traits", +] + +[[package]] +name = "rind-jsonwebtoken" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bf395502633219077e6e360d458ac1c040bf70212d6119a7dabcd123cacf745" +dependencies = [ + "jsonwebtoken", + "pkcs1", + "rind", +] + +[[package]] +name = "rind-rustls" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e67845e30d1c9a194485cdcb585abf7dcc011a4413488d57634ec39d7e832f0" +dependencies = [ + "rind", + "rustls", ] [[package]] @@ -5602,20 +5707,6 @@ dependencies = [ "libc", ] -[[package]] -name = "sec1" -version = "0.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d56d437c2f19203ce5f7122e507831de96f3d2d4d3be5af44a0b0a09d8a80e4d" -dependencies = [ - "base16ct", - "ctutils", - "der 0.8.2", - "hybrid-array", - "subtle", - "zeroize", -] - [[package]] name = "security-framework" version = "3.7.0" @@ -5770,16 +5861,6 @@ dependencies = [ "syn 3.0.6", ] -[[package]] -name = "serdect" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "66cf8fedced2fcf12406bcb34223dffb92eaf34908ede12fed414c82b7f00b3e" -dependencies = [ - "base16ct", - "serde", -] - [[package]] name = "sfv" version = "0.15.0" @@ -5853,16 +5934,6 @@ dependencies = [ "rand_core 0.6.4", ] -[[package]] -name = "signature" -version = "3.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28d567dcbaf0049cb8ac2608a76cd95ff9e4412e1899d389ee400918ca7537f5" -dependencies = [ - "digest", - "rand_core 0.10.1", -] - [[package]] name = "simd-adler32" version = "0.3.10" @@ -6114,7 +6185,6 @@ version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ef958a98b9d5da290cfc78946e9f3e61e1e62a18db0d92cac0b83cc161491a9" dependencies = [ - "base64ct", "der 0.8.2", ] @@ -7108,9 +7178,9 @@ dependencies = [ [[package]] name = "web-transport-proto" -version = "0.6.3" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2baf87f7a4701d964fd23529b440db3da535b1d7539289f7a463c3481dafcff" +checksum = "40d8e5dba9b14dbe25b116ac7e62ca82bf7c328d1c6072238bf0b9b47874c9d1" dependencies = [ "bytes", "http", @@ -7727,18 +7797,6 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" -[[package]] -name = "wnaf" -version = "0.14.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "795ca18b3fdb5e62bf982199278341ddcf7ebf7d32e25e212ad05d496e95f6fa" -dependencies = [ - "ff", - "group", - "hybrid-array", - "primefield", -] - [[package]] name = "writeable" version = "0.6.4" diff --git a/README.md b/README.md index 599e07dc..fc6bbe29 100644 --- a/README.md +++ b/README.md @@ -19,9 +19,9 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ - Without `--connect`, the studio composes and previews but publishes nothing, and broadcast sources stay offline. - `--failover ` publishes the same broadcast to a second relay under the - same publisher epoch, so a viewer linked to both resumes mid-group from it - when the first fails. Both relays must speak the one `--connect-version`, - which must carry publisher epochs (`moq-lite-07-wip` today, which a relay + same broadcast instance, so a viewer linked to both resumes mid-group from + it when the first fails. Both relays must speak the one `--connect-version`, + which must carry instances (`moq-lite-07-wip` today, which a relay accepts only when configured to), or the studio refuses to start; it also refuses a fixed `--connect-bind` port, since each relay's connection binds its own. Its credential rides its URL's `?jwt=`; `--connect-token` goes to diff --git a/moq b/moq index 96244dd8..fd07f5b7 160000 --- a/moq +++ b/moq @@ -1 +1 @@ -Subproject commit 96244dd854e4eeffedfdeaa9ef4a736266eb11b9 +Subproject commit fd07f5b7c93f23b7ed1c4faddbbf4274a7d72e59 diff --git a/quest/m0/README.md b/quest/m0/README.md index 461ee6bb..fb3933d1 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -107,9 +107,7 @@ repository. - [Virtual camera](/quest/m0/vcam/README.md) - a canvas's program exposed as a system camera on Linux, Windows, and macOS - [Signing identities are available](/quest/m0/signing.md) - the Apple and Windows code-signing credentials packaging and the macOS camera need - [Packaging](/quest/m0/packaging.md) - signed installers for macOS and Windows and a Flatpak for Linux -- [One audio capture feeding the bus and an encoder is pinned](/quest/m0/pin-audio-capture.md) - moq-audio's capture split, for audio source tracks - [moq-mux's presentation timeline is pinned](/quest/m0/pin-presentation-timeline.md) - shared playout timing, for broadcast sync -- [moq-audio's mix router is pinned](/quest/m0/pin-audio-mix-router.md) - several mixes and mix-minus, for audio mixes - [The persistable portal restore token is pinned](/quest/m0/pin-portal-token.md) - moq-video's restore token, for Wayland sources - [Platform AAC encoders are pinned](/quest/m0/pin-aac-encode.md) - moq-audio's AAC encoders, for destinations - [moq-audio's filter chain is pinned](/quest/m0/pin-audio-filters.md) - the chain audio filters configure diff --git a/quest/m0/audio-stall.md b/quest/m0/audio-stall.md index f87e964c..6dfff226 100644 --- a/quest/m0/audio-stall.md +++ b/quest/m0/audio-stall.md @@ -8,14 +8,8 @@ after a second without a picture. ## Plan -Audio capture runs inside the audio bus today, which reports no last-delivery -time per source, so the stall check video sources have cannot see it. The -capture Producer/Consumer split that the audio-capture pin brings gives the -studio its own handle on each capture; read delivery from there rather than +The audio bus reports no last-delivery time per source, so the stall check +video sources have cannot see it. The studio opens each capture as a +`capture::Producer`, so read delivery from a consumer of its own rather than adding a side channel in the bus. Raised on [#5](https://github.com/moq-dev/studio/pull/5). - -## Required - -- [moq-audio's capture split is pinned](/quest/m0/pin-audio-capture.md) - the - per-capture handle this reads diff --git a/quest/m0/compose/sources.md b/quest/m0/compose/sources.md index f3c69aae..28963420 100644 --- a/quest/m0/compose/sources.md +++ b/quest/m0/compose/sources.md @@ -17,11 +17,9 @@ catalog keyed by source name, so the root `video` and `audio` sections stay the program's. An audio source adds an `audio` rendition to its own nested catalog the same way. -Blocked upstream: moq-audio's capture stream is crate-private, and the only -public captures feed either the program bus (`bus::Capture`) or an encoder -(`encode::publish_capture`), never both. The -[pin](/quest/m0/pin-audio-capture.md) brings a capture Producer/Consumer split -that feeds both from one device open, rather than opening each device twice. +moq-audio's capture Producer/Consumer split (`capture::Producer`) feeds both +the bus and an encoder from one device open, rather than opening each device +twice. Publish the capture after the source's filters but before its gain and mute (decided 2026-11-02, maintainer): viewers hear filtered audio, as video bakes in its ML effects, and a composing viewer mixes the levels itself. @@ -37,7 +35,3 @@ always-on. Rejected: sources on demand, which would make replaying a source need an armed local subscriber again. Track names stay stable for a source's life, and an edited source that changes what it shows gets a new name. - -## Required - -- [One audio capture feeding the bus and an encoder is pinned](/quest/m0/pin-audio-capture.md) - one device open feeds the mix and the source's own track diff --git a/quest/m0/mixes.md b/quest/m0/mixes.md index 917001b7..9737d146 100644 --- a/quest/m0/mixes.md +++ b/quest/m0/mixes.md @@ -14,17 +14,13 @@ Each guest hears a mix-minus: the program without their own voice. ## Plan Decided 2026-11-02 (maintainer): routing, mix-minus, and sync offsets live in -moq-audio as a multi-mix router rather than one `Bus` per mix in the studio, -so this quest waits on its [pin](/quest/m0/pin-audio-mix-router.md). The -studio adds the routing UI, meters, the headphone monitor, clap-to-measure -offsets, and a mix pick per output and recording. +moq-audio as a multi-mix router rather than one `Bus` per mix in the studio: +`bus::Mix`, which the program audio already is. The studio adds the routing +UI, meters, the headphone monitor, clap-to-measure offsets, and a mix pick per +output and recording. The mixer runs on the audio clock and never blocks on a slow output. An output that falls behind drops its own audio, not the bus's. Benchmark source count against mix count, including sparse routing. A mix must cost only what is routed into it, never a scan of every source. - -## Required - -- [moq-audio's mix router is pinned](/quest/m0/pin-audio-mix-router.md) - routing, mix-minus, and sync offsets diff --git a/quest/m0/pin-audio-capture.md b/quest/m0/pin-audio-capture.md deleted file mode 100644 index e3594903..00000000 --- a/quest/m0/pin-audio-capture.md +++ /dev/null @@ -1,11 +0,0 @@ -# [XS] One audio capture feeding the bus and an encoder is pinned - -## Goal - -The `moq` submodule includes moq-audio's capture Producer/Consumer split, from -[One audio capture feeds the bus and an encoder](https://github.com/moq-dev/moq/blob/main/quest/m1/audio-capture-fanout.md), -which the studio's microphone and system-audio source tracks need. - -Check: that quest file is gone from moq's `main`. Advance it with -`just upgrade-moq` in its own PR, fixing whatever the new API breaks here: -`bus::Capture` and `encode::publish_capture` are replaced by the split. diff --git a/quest/m0/pin-audio-mix-router.md b/quest/m0/pin-audio-mix-router.md deleted file mode 100644 index 467d0d18..00000000 --- a/quest/m0/pin-audio-mix-router.md +++ /dev/null @@ -1,12 +0,0 @@ -# [XS] moq-audio's mix router is pinned - -## Goal - -The `moq` submodule includes moq-audio's multi-mix router (routing, -mix-minus, and per-source sync offsets), from -[moq-audio routes sources into several mixes](https://github.com/moq-dev/moq/blob/main/quest/m1/audio-mix-router.md), -which audio mixes configure. - -Check: that quest file is gone from moq's `main`. Advance it with -`just upgrade-moq` in its own PR, fixing whatever the new API breaks here: -the single-mix `Bus` the program reads changes shape. diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index 99f1be81..b3bb6029 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -13,7 +13,8 @@ //! - [`source::Sources`] runs each source: a camera, display, window, //! microphone, or system-audio capture (with the `capture` feature), a still //! image, a solid color, or a broadcast subscribed over MoQ. Pictures go to the -//! compositor and sound goes to a `moq_audio::bus::Bus`, the program audio. +//! compositor and sound goes to a `moq_audio::bus::Bus`, routed into its +//! program mix. //! - [`Compositor`] draws the scene in preview and the scene on program into //! textures on the caller's `wgpu` device, cutting or fading between them. //! `moq_video::render::export` hands the program to an encoder, on the GPU diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 423f5875..d3820d1f 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -17,7 +17,7 @@ use std::time::Duration; use hang::catalog::Reference; use kio::clock::Instant; -use moq_audio::bus::{Bus, Strip}; +use moq_audio::bus::{Bus, Mix, Strip}; use moq_mux::catalog::Stream; use crate::Error; @@ -68,8 +68,10 @@ pub enum Status { /// What every source shares. #[derive(Clone)] pub struct Context { - /// The program audio that microphones, system audio, and broadcasts feed. + /// Where microphones, system audio, and broadcasts feed their sound. pub bus: Bus, + /// The program audio: every running source with sound is routed into it. + pub program: Mix, /// Where broadcast sources are announced: the studio's relay connection. pub origin: moq_net::origin::Consumer, /// What a relative image or LUT path is relative to: the scene file's directory. @@ -217,6 +219,16 @@ struct Track { status: kio::Producer, } +/// Takes the source out of the program, which [`open`] routed it into, so a +/// removed or edited source leaves nothing behind in the mix. +impl Drop for Driver { + fn drop(&mut self) { + if let Some(strip) = &self.strip { + self.context.program.unroute(strip); + } + } +} + impl Driver { /// The source's name in the collection. pub fn name(&self) -> &str { @@ -403,9 +415,11 @@ impl Driver { let strip = self.strip.clone().expect("audio sources have a strip"); let mut config = moq_audio::capture::Config::default(); config.source = source; - let capture = self.context.bus.capture(config, strip); + let capture = moq_audio::capture::Producer::open(config).await?; + let channel = self.context.bus.capture(capture.consume(), strip)?; set(&self.status, Status::Live); - Ok(capture.run().await?) + tokio::try_join!(capture.run(), channel.run())?; + Ok(()) } #[cfg(not(feature = "capture"))] @@ -583,6 +597,9 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { let picture = Mailbox::default(); let shown = Shown::default(); let strip = source.has_audio().then(Strip::default); + if let Some(strip) = &strip { + context.program.route(strip); + } let status = kio::Producer::new(Status::Starting); // Claimed here, in the order sources open, so an edited source's // replacement owns the listing however their drivers are scheduled. @@ -663,8 +680,10 @@ pub(crate) mod test { pub(crate) fn context(dir: &Path) -> (Context, moq_net::origin::Producer) { let (origin, driver) = moq_net::origin::Producer::new(Default::default()); tokio::task::spawn_local(moq_net::time::run(driver)); + let (bus, _) = Bus::new(Default::default()).unwrap(); let context = Context { - bus: Bus::new(Default::default()).unwrap(), + program: bus.mix(), + bus, origin: origin.consume(), dir: dir.to_path_buf(), output: moq_video::Output::Cpu, @@ -775,7 +794,8 @@ mod tests { } /// Editing a source restarts it, removing one stops it, and an untouched one - /// keeps running. Gain changes reach the strip without restarting anything. + /// keeps running. Gain changes reach the strip without restarting anything, + /// and a source plays in the program until it is removed. #[tokio::test] async fn sync_restarts_only_what_changed() { local(async { @@ -786,6 +806,8 @@ mod tests { let drivers = sources.sync(&collection); start(&sources, drivers).await; let guest = sources.get("guest").unwrap().strip().unwrap().clone(); + let program = sources.context().program.clone(); + assert!(program.routes(&guest)); collection.audio.insert( "guest".into(), @@ -815,6 +837,16 @@ mod tests { collection.sources.remove("red"); assert!(sources.sync(&collection).is_empty()); assert!(sources.get("red").is_none()); + + collection.sources.remove("guest"); + assert!(sources.sync(&collection).is_empty()); + for _ in 0..1000 { + if !program.routes(&guest) { + return; + } + tokio::task::yield_now().await; + } + panic!("a removed source stays in the program"); }) .await; } @@ -845,9 +877,7 @@ mod tests { .unwrap() }; let first = audio("first", None); - publication - .announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) - .unwrap(); + publication.announce(moq_net::origin::Route::default()).unwrap(); let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); @@ -888,9 +918,7 @@ mod tests { .unwrap(); }; point(&mut catalog, "first"); - publication - .announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) - .unwrap(); + publication.announce(moq_net::origin::Route::default()).unwrap(); let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); @@ -925,9 +953,7 @@ mod tests { catalog.audio.renditions.insert("audio".into(), config); }) .unwrap(); - publication - .announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) - .unwrap(); + publication.announce(moq_net::origin::Route::default()).unwrap(); let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 51d49222..6e79b2e7 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -5,7 +5,8 @@ //! Threads: //! - **program** draws preview and program once per canvas frame and hands the //! program to the encoder, stamped with the program clock. -//! - **audio** reads the program bus on the same clock and publishes it. +//! - **audio** advances the audio bus on the same clock and publishes its +//! program mix. //! - **encode** waits for each program copy and encodes it. //! - **source-***: one per running source, since a capture stream may not //! leave the thread that opened it. @@ -249,7 +250,8 @@ pub struct Engine { pub stats: Stats, /// How the stream is delivered, graded every [`health::INTERVAL`]. pub health: Mutex, - pub bus: moq_audio::bus::Bus, + /// The program audio, which every source with sound is routed into. + pub program: moq_audio::bus::Mix, /// `None` when running without a relay. pub output: Option, /// The room guests join, when configured. @@ -261,7 +263,7 @@ pub struct Engine { /// The audio the program publishes. pub const AUDIO_BITRATE: moq_net::bandwidth::Rate = moq_net::bandwidth::Rate::from_kbps(128); -/// How often the audio thread reads the bus. Short enough that a read is never +/// How often the audio thread advances the bus. Short enough that a block is never /// close to the bus's buffering, long enough to cost nothing. const AUDIO_PERIOD: Duration = Duration::from_millis(10); @@ -298,7 +300,8 @@ impl Engine { on_demand, room, } = launch; - let bus = moq_audio::bus::Bus::new(Default::default())?; + let (bus, mixer) = moq_audio::bus::Bus::new(Default::default())?; + let program = bus.mix(); let subscribe = moq_tokio::origin::spawn(); // Every relay feeds `subscribe`, so a guest either one routes is seen. let guests = match &room { @@ -330,6 +333,7 @@ impl Engine { let dir = std::path::absolute(dir.unwrap_or(Path::new(".")))?; let sources = Sources::new(source::Context { bus: bus.clone(), + program: program.clone(), origin: sources, dir, output: moq_video::render::output(&gpu.device), @@ -353,7 +357,7 @@ impl Engine { autosave, stats: Stats::default(), health: Mutex::default(), - bus, + program, output, room, _origin: origin, @@ -399,7 +403,10 @@ impl Engine { program.run_program(start, epoch, video).await })?; let mix = engine.clone(); - clocked("audio".into(), async move || mix.run_audio(start, epoch, audio).await)?; + let config = bus.config().clone(); + clocked("audio".into(), async move || { + mix.run_audio(start, epoch, config, mixer, audio).await + })?; let save = engine.clone(); clocked("autosave".into(), async move || save.autosave.run(&save.state).await)?; tokio::spawn(engine.clone().grade(health::Sampler::new(publisher))); @@ -552,17 +559,22 @@ impl Engine { } } - /// Read the program bus as the program clock advances, and publish it - /// stamped `epoch` past `start`. + /// Advance the bus as the program clock advances, and publish the program + /// mix stamped `epoch` past `start`. /// /// Runs without a relay too, so the meters move and sources never back up. - async fn run_audio(&self, start: Instant, epoch: Duration, mut producer: Option) { - let config = self.bus.config().clone(); + async fn run_audio( + &self, + start: Instant, + epoch: Duration, + config: moq_audio::bus::Config, + mut mixer: moq_audio::bus::Driver, + mut producer: Option, + ) { let rate = u128::from(config.sample_rate); - let channels = config.layout.channels() as usize; let mut clock = kio::clock::interval_at(start, AUDIO_PERIOD); clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); - let mut samples = Vec::new(); + let mut program = self.program.reader(); let mut written: u128 = 0; loop { @@ -574,14 +586,15 @@ impl Engine { if frames == 0 { continue; } - samples.resize(frames * channels, 0.0); - self.bus.read(&mut samples); - let timestamp = moq_net::Timestamp::from_micros((epoch.as_micros() + written * 1_000_000 / rate) as u64); + mixer.advance(frames); written = due; + let block = program.read().await.expect("the mixer is alive"); + let position = u128::from(block.position); + let timestamp = moq_net::Timestamp::from_micros((epoch.as_micros() + position * 1_000_000 / rate) as u64); let Some(publish) = producer.as_mut() else { continue }; let Ok(timestamp) = timestamp else { continue }; - let data: Vec = samples.iter().flat_map(|sample| sample.to_le_bytes()).collect(); + let data: Vec = block.data.iter().flat_map(|sample| sample.to_le_bytes()).collect(); if let Err(err) = publish.write(&moq_audio::Frame::new(Bytes::from(data), timestamp)) { self.stats.fail(format!("audio: {err}")); // Viewers see the audio fail rather than go away unfinished. @@ -608,7 +621,7 @@ struct Publish { sources: moq_scene::publish::Publish, } -/// Announce the program broadcast under one epoch, which also carries each +/// Announce the program broadcast under one instance, which also carries each /// local source's own track, and connect to every relay, each also carrying /// `subscribe`'s broadcast sources. async fn publish( @@ -672,11 +685,11 @@ async fn publish( let stats = catalog.clone(); let audio = moq_audio::encode::Producer::new(&mut broadcast, catalog, input, &options)?; - // One epoch on every relay, so a viewer moving between them resumes + // One instance on every relay, so a viewer moving between them resumes // mid-group rather than restarting. - let epoch = moq_net::Epoch::mint(); + let instance = moq_net::Instance::random(); publication - .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) + .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) .context("cannot announce the broadcast")?; let feedback = health::Feedback::default(); @@ -687,7 +700,7 @@ async fn publish( )); tokio::spawn(log("relay stats", watching.clone().follow(reports, path.clone()))); let health = health::Publisher { - observer: hang::health::Observer::new(path.clone(), Some(epoch)), + observer: hang::health::Observer::new(path.clone(), Some(instance)), catalog: stats, clock: clock.clone(), relays: links, @@ -739,9 +752,9 @@ fn connect( }, None => format!("{}:{}", url.scheme(), url.path()), }; - // Among routes of one instance of a broadcast source, the first - // relay's wins while it has one. A newer instance wins first, - // whichever relay it comes through. + // The first relay's route to a broadcast source wins while it has + // one, whether it carries the same instance as the failover's or + // another. let preference = match rank { 0 => moq_net::origin::Preference::default(), _ => moq_net::origin::Preference::new(-1), @@ -1125,12 +1138,12 @@ mod tests { } /// With a failover relay, both relays route the program under its one - /// epoch over the one protocol version, so a viewer moving between them + /// instance over the one protocol version, so a viewer moving between them /// resumes. A broadcast source only the failover has still reaches the - /// studio, and a newer instance of a source behind the failover replaces - /// the first relay's. + /// studio, and one restarted behind the failover replaces the first + /// relay's once that one ends. #[tokio::test(flavor = "multi_thread")] - async fn every_relay_routes_one_epoch() { + async fn every_relay_routes_one_instance() { let _ = moq_tokio::crypto::install_default(); let version: moq_net::Version = "moq-lite-07-wip".parse().unwrap(); let mut relays = Vec::new(); @@ -1157,25 +1170,22 @@ mod tests { let subscribe = moq_tokio::origin::spawn(); let (_connections, links) = connect(relays, &origin, &subscribe); let publication = origin.create_broadcast("studio").unwrap(); - let epoch = moq_net::Epoch::mint(); + let instance = moq_net::Instance::random(); publication - .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) + .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) .unwrap(); - // Publish `path` under `epoch` through one relay's guest. - let publish = |path: &str, epoch: Option<&moq_net::Epoch>, (client, url): &(moq_tokio::Client, url::Url)| { + // Publish `path` under `instance` through one relay's guest. + let publish = |path: &str, instance: &moq_net::Instance, (client, url): &(moq_tokio::Client, url::Url)| { let origin = moq_tokio::origin::spawn(); let source = origin.create_broadcast(path).unwrap(); - let route = moq_net::origin::Route::default(); - let route = match epoch { - Some(epoch) => route.with_epoch(epoch.clone()), - None => route, - }; - source.announce(route).unwrap(); + source + .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) + .unwrap(); let connection = client.clone().with_publisher(&origin).connect(url.clone()); (origin, source, connection) }; - let _guest = publish("guest", None, &guests[1]); + let _guest = publish("guest", &moq_net::Instance::random(), &guests[1]); let mut sources = subscribe.consume().announced(); let routed = async { @@ -1184,7 +1194,7 @@ mod tests { loop { match announced.next().await.expect("announcements") { moq_net::announce::Event::Start(announce) if announce.prefix.as_str() == "studio" => { - assert_eq!(announce.route.epoch, Some(epoch.clone())); + assert_eq!(announce.route.instance, Some(instance.clone())); break; } _ => continue, @@ -1209,13 +1219,14 @@ mod tests { .await .expect("the source"); - // The newest instance of a source wins before the first relay's rank, - // so one restarted behind the failover replaces the first relay's. - let older: moq_net::Epoch = "01900000-0000-7000-8000-000000000000".parse().unwrap(); - let newer: moq_net::Epoch = "01910000-0000-7000-8000-000000000000".parse().unwrap(); - let _first = publish("show", Some(&older), &guests[0]); + // A source restarted behind the failover takes over once the first + // relay's route to it ends, since that relay ranks first. + let older = moq_net::Instance::random(); + let newer = moq_net::Instance::random(); + let first = publish("show", &older, &guests[0]); routed_at(&mut sources, &older).await; - let _restarted = publish("show", Some(&newer), &guests[1]); + let _restarted = publish("show", &newer, &guests[1]); + drop(first); routed_at(&mut sources, &newer).await; }; kio::clock::timeout(TIMEOUT, routed) @@ -1223,8 +1234,8 @@ mod tests { .expect("both relays route the program, the failover the sources"); } - /// Wait until `announced` routes `show` under `epoch`. - async fn routed_at(announced: &mut moq_net::announce::Consumer, epoch: &moq_net::Epoch) { + /// Wait until `announced` routes `show` under `instance`. + async fn routed_at(announced: &mut moq_net::announce::Consumer, instance: &moq_net::Instance) { use moq_net::announce::Event; loop { let announce = match announced.next().await.expect("announcements") { @@ -1232,7 +1243,7 @@ mod tests { Event::Takeover { announce, .. } => announce, Event::End(_) => continue, }; - if announce.prefix.as_str() == "show" && announce.route.epoch.as_ref() == Some(epoch) { + if announce.prefix.as_str() == "show" && announce.route.instance.as_ref() == Some(instance) { return; } } diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index 5e19011f..55b12354 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -201,7 +201,7 @@ pub struct Local { /// What the studio publishes, for grading its delivery. pub struct Publisher { - /// The broadcast's path and the epoch it was announced with. + /// The broadcast's path and the instance it was announced with. pub observer: Observer, pub catalog: moq_scene::publish::Catalog, /// The catalog's clock, which maps a track's timestamps to the wall time @@ -830,7 +830,11 @@ impl Feedback { false }; if read { - let reader = self.read(origin.clone(), Observer::new(path.clone(), announce.route.epoch), name); + let reader = self.read( + origin.clone(), + Observer::new(path.clone(), announce.route.instance), + name, + ); readers.insert(path, reader); } self.0.lock().unwrap().unread = unread; @@ -892,7 +896,7 @@ fn name(prefix: &moq_net::Path, path: &moq_net::Path) -> Option { /// its broadcast ends or the report cannot be read. async fn read(origin: moq_net::origin::Consumer, observer: Observer, slot: Slot) { let read = async { - let broadcast = origin.request(&observer.path, observer.epoch.clone()).await?; + let broadcast = origin.request(&observer.path, observer.instance.clone()).await?; let track = broadcast.track(hang::echo::TRACK)?.subscribe(None).await?; let mut config = moq_json::snapshot::consumer::Config::default(); config.max_size = Some(MAX_REPORT); diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 4c649f2a..11b85398 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -50,7 +50,7 @@ pub struct Config { /// A second relay to publish the same broadcast to, so a viewer linked to /// both resumes from it when the `--connect` relay fails. Both must speak - /// the one `--connect-version`, which must carry publisher epochs. Its + /// the one `--connect-version`, which must carry broadcast instances. Its /// credential rides its URL's `?jwt=`: `--connect-token` goes to the /// `--connect` relay only. #[usage(long)] @@ -98,15 +98,15 @@ impl Config { } /// Refuse relays a viewer could not move between without a restart: a copy - /// resumes mid-group only under the same epoch, which needs both links on + /// resumes mid-group only under the same instance, which needs both links on /// one protocol version that carries it. fn check_failover(&self) -> anyhow::Result<()> { let [version] = self.client.version.as_slice() else { anyhow::bail!("--failover needs exactly one --connect-version, so both relays negotiate the same protocol"); }; anyhow::ensure!( - matches!(version, moq_net::Version::Lite(lite) if lite.has_epoch()), - "--failover needs a --connect-version with publisher epochs (moq-lite-07-wip today), not {version}, \ + matches!(version, moq_net::Version::Lite(lite) if lite.has_instance()), + "--failover needs a --connect-version with broadcast instances (moq-lite-07-wip today), not {version}, \ or a viewer moving between relays restarts the stream rather than resuming it" ); // Each relay's client binds its own socket, so a fixed port would be @@ -275,10 +275,10 @@ mod tests { } /// A failover relay is accepted only on one protocol version that carries - /// publisher epochs, so a viewer moving between the relays resumes rather + /// broadcast instances, so a viewer moving between the relays resumes rather /// than restarts. #[test] - fn failover_needs_one_version_with_epochs() { + fn failover_needs_one_version_with_instances() { let check = |extra: &[&str]| { let mut args = vec!["show.toml", "--connect", "https://a/", "--failover", "https://b/"]; args.extend(extra); @@ -286,7 +286,7 @@ mod tests { Config::parse_from(&args).unwrap().check_failover() }; let lite07 = ["--connect-version", "moq-lite-07-wip"]; - check(&lite07).expect("one version with epochs"); + check(&lite07).expect("one version with instances"); check(&[&lite07[..], &["--connect-bind", "127.0.0.1:0"]].concat()).expect("an ephemeral port"); for refused in [ &[][..], diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 8a243487..c5b44cb0 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1422,7 +1422,7 @@ impl Panels { fn audio(&mut self, ui: &mut Ui) { ui.horizontal(|ui| { ui.strong("Program"); - meter(ui, self.engine.bus.level()); + meter(ui, self.engine.program.level()); }); ui.separator(); diff --git a/rs/moq-studio/src/uplink.rs b/rs/moq-studio/src/uplink.rs index f054f9e9..b5e42902 100644 --- a/rs/moq-studio/src/uplink.rs +++ b/rs/moq-studio/src/uplink.rs @@ -47,7 +47,7 @@ pub async fn measure( .create_track(TRACK, None) .context("cannot create the probe track")?; publication - .announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default()) .context("cannot announce the probe")?; // One-shot, so a relay that refuses either side ends the probe with its reason From 306559a80e772769dbe87e4451e3cb50e9ddb162 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 18:30:39 -0800 Subject: [PATCH 3/7] docs: broadcast sources prefer the first relay across instances Co-Authored-By: Claude Opus 5.5 --- README.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index fc6bbe29..a3a6d745 100644 --- a/README.md +++ b/README.md @@ -127,10 +127,10 @@ comes along, renamed if its name is taken. Copied filters paste onto the selected source or item, after its own. The clipboard holds text in the scene file's format. -A broadcast source subscribes through the same relay connections. It plays the -newest instance of its broadcast through either relay, preferring the first -relay among routes to the same instance, so a publisher restarted behind only -the failover replaces the first relay's copy. It plays the +A broadcast source subscribes through the same relay connections. It plays its +broadcast through the first relay while that relay has a route to it, and +through the failover otherwise, so a publisher restarted behind only the +failover replaces the first relay's copy once that copy ends. It plays the best video and audio its catalog lists, switches when the catalog changes, and comes back on its own when the publisher restarts. Audio from every source is mixed into the program whether or not its scene is live; mute a source to leave it out. @@ -418,7 +418,7 @@ it is unhealthy. Below that it shows: read them, so most show "unknown". In a cluster, a relay forwarding the stream to another counts as one session. Two relays whose stats use the same node name publish them at the same path, so only one of them counts: - the newer instance, or the first relay's when they are one instance. + the first relay's while it has them. - **Tracks**: each track's bitrate and frame rate, and, from viewers that report, the lowest bitrate any received, its worst grade, and the latency from the program clock to the slowest viewer. The program's tracks come From 9b40167d189be126b0fe3a112c27a48a1d07a5b6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 19:03:49 -0800 Subject: [PATCH 4/7] fix(scene): ask a multichannel input for stereo, and prove the first relay holds A capture device that defaults to more than two channels opens as a discrete layout the stereo bus refuses, so ask it for stereo, then mono. The failover test now checks the first relay's instance holds while the failover routes a newer one. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/source.rs | 27 ++++++++++++++++++++++++--- rs/moq-studio/src/engine.rs | 27 +++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 5 deletions(-) diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index d3820d1f..3918b05e 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -413,9 +413,7 @@ impl Driver { #[cfg(feature = "capture")] async fn listen(&self, source: moq_audio::capture::Source) -> Result<(), Error> { let strip = self.strip.clone().expect("audio sources have a strip"); - let mut config = moq_audio::capture::Config::default(); - config.source = source; - let capture = moq_audio::capture::Producer::open(config).await?; + let capture = capture(source).await?; let channel = self.context.bus.capture(capture.consume(), strip)?; set(&self.status, Status::Live); tokio::try_join!(capture.run(), channel.run())?; @@ -657,6 +655,29 @@ fn set(status: &kio::Producer, value: Status) { } } +/// Open `source` in a layout the bus can mix. Past stereo a device's channels +/// carry no speaker positions to downmix by, such as a microphone array's or an +/// interface's inputs, so a device that defaults to more is asked for stereo, +/// then mono, and fails with the formats it offers if it has neither. +#[cfg(feature = "capture")] +async fn capture(source: moq_audio::capture::Source) -> Result { + let mut config = moq_audio::capture::Config::default(); + config.source = source; + let capture = moq_audio::capture::Producer::open(config.clone()).await?; + if !matches!(capture.consume().input().layout, moq_audio::Layout::Discrete(_)) { + return Ok(capture); + } + drop(capture); + config.channels = Some(2); + match moq_audio::capture::Producer::open(config.clone()).await { + Err(moq_audio::Error::Unsupported(_)) => { + config.channels = Some(1); + moq_audio::capture::Producer::open(config).await + } + result => result, + } +} + /// Read a still image, decoded to straight-alpha RGBA, and the file it came from. fn image(path: &Path) -> Result<(Picture, bytes::Bytes), Error> { let bytes = std::fs::read(path).map_err(|err| Error::Image(format!("{}: {err}", path.display())))?; diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 0210981b..a25834cd 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -1636,13 +1636,36 @@ mod tests { .await .expect("the source"); - // A source restarted behind the failover takes over once the first - // relay's route to it ends, since that relay ranks first. + // A source restarted behind the failover takes over only once the + // first relay's route to it ends, since that relay ranks first. let older = moq_net::Instance::random(); let newer = moq_net::Instance::random(); let first = publish("show", &older, &guests[0]); routed_at(&mut sources, &older).await; let _restarted = publish("show", &newer, &guests[1]); + routed_at(&mut routes[1].announced(), &newer).await; + // Announced after the restart on the failover's one announce + // stream, so once it reaches the studio the restart has too. + let _marker = publish("marker", &moq_net::Instance::random(), &guests[1]); + subscribe + .consume() + .follow("marker", None) + .unwrap() + .available() + .await + .expect("the marker"); + while let Some(event) = sources.try_next() { + if let moq_net::announce::Event::Start(announce) + | moq_net::announce::Event::Update(announce) + | moq_net::announce::Event::Restart(announce) + | moq_net::announce::Event::Takeover { announce, .. } = event + { + assert!( + announce.prefix.as_str() != "show" || announce.route.instance.as_ref() != Some(&newer), + "the failover's instance won while the first relay has one" + ); + } + } drop(first); routed_at(&mut sources, &newer).await; }; From 0b8158d0e34e9eacfd2b8915650c62ac04a756db Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 20:04:07 -0800 Subject: [PATCH 5/7] fix(scene): port replay to moq's track Delay Instant replay, merged from main, still used moq_mux::container::Delay and Subscription::with_max_delay, which the pinned moq replaced with moq_net::track::Delay. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/publish.rs | 4 +++- rs/moq-scene/src/replay.rs | 6 ++++-- rs/moq-scene/src/source.rs | 4 ++-- 3 files changed, 9 insertions(+), 5 deletions(-) diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 35636e81..4d94c2af 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -656,7 +656,9 @@ mod tests { let track = consumer .track(hang::catalog::Reference::track(config, name)) .unwrap() - .subscribe(moq_net::track::Subscription::default().with_max_delay(None)) + .subscribe( + moq_net::track::Subscription::default().with_delay(moq_net::track::Delay::default().with_max(None)), + ) .await .unwrap(); let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); diff --git a/rs/moq-scene/src/replay.rs b/rs/moq-scene/src/replay.rs index 02dc9c7f..1341dd91 100644 --- a/rs/moq-scene/src/replay.rs +++ b/rs/moq-scene/src/replay.rs @@ -202,7 +202,7 @@ fn copy( // stale. let subscription = moq_net::track::Subscription::default() .with_ranges([range]) - .with_max_delay(None) + .with_delay(moq_net::track::Delay::default().with_max(None)) .with_order(moq_net::track::Order::Ascending); let subscribing = target.broadcast.track(name)?.subscribe(subscription); // Never woken, like the reads below: a publisher that never describes the @@ -362,7 +362,9 @@ mod tests { .consume() .track("video") .unwrap() - .subscribe(moq_net::track::Subscription::default().with_max_delay(None)) + .subscribe( + moq_net::track::Subscription::default().with_delay(moq_net::track::Delay::default().with_max(None)), + ) .await .unwrap(); let mut groups = Vec::new(); diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index a34cd286..f52d5a81 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -820,8 +820,8 @@ impl Driver { /// How a replay reads its take: from the first group, where the take's tracks /// declare they start, skipping nothing as late, since a take is history. -const REPLAY: moq_mux::container::Delay = moq_mux::container::Delay { - min: Duration::ZERO, +const REPLAY: moq_net::track::Delay = moq_net::track::Delay { + min: Some(Duration::ZERO), max: None, }; From 4df5425229635817dcb282677be43fff7bb82afd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 22:13:50 -0800 Subject: [PATCH 6/7] fix(scene): port the drain test's subscription to moq's track Delay Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/publish.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index eda48192..90e4aa5f 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -733,8 +733,8 @@ mod tests { let (name, rendition) = listing.catalog.snapshot().video.renditions.pop_first().unwrap(); // Every group, however far behind the newest, since this reads only once // the track has ended. - let mut subscription = moq_net::track::Subscription::default(); - subscription.max_delay = None; + let subscription = + moq_net::track::Subscription::default().with_delay(moq_net::track::Delay::default().with_max(None)); let subscriber = consumer.track(&name).unwrap().subscribe(subscription).await.unwrap(); let container = moq_mux::catalog::hang::Container::try_from(&rendition).unwrap(); let mut reader = moq_mux::container::Consumer::new(subscriber, container); From d1682dab952b32061e22729b5d47174a790dd6f4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 22:16:51 -0800 Subject: [PATCH 7/7] fix(studio): name the program audio mix apart from the program thread #23 added Engine::program for the program clock's thread, so the audio mix is Engine::mix. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/engine.rs | 6 +++--- rs/moq-studio/src/ui.rs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 54b1dd5f..fe9f37f6 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -407,7 +407,7 @@ pub struct Engine { /// How the stream is delivered, graded every [`health::INTERVAL`]. pub health: Mutex, /// The program audio, which every source with sound is routed into. - pub program: moq_audio::bus::Mix, + pub mix: moq_audio::bus::Mix, /// `None` when running without a relay. pub output: Option, /// The room guests join, when configured. @@ -519,7 +519,7 @@ impl Engine { autosave, stats, health: Mutex::default(), - program, + mix: program, output, room, _origin: origin, @@ -805,7 +805,7 @@ impl Engine { let rate = u128::from(config.sample_rate); let mut clock = kio::clock::interval_at(start, AUDIO_PERIOD); clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); - let mut program = self.program.reader(); + let mut program = self.mix.reader(); let mut written: u128 = 0; loop { diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 8603f4f0..7fb1d073 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1634,7 +1634,7 @@ impl Panels { fn audio(&mut self, ui: &mut Ui) { ui.horizontal(|ui| { ui.strong("Program"); - meter(ui, self.engine.program.level()); + meter(ui, self.engine.mix.level()); }); ui.separator();