diff --git a/Cargo.lock b/Cargo.lock index 9ec7955..68e54ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -471,12 +471,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" @@ -526,6 +520,8 @@ dependencies = [ "cexpr", "clang-sys", "itertools 0.13.0", + "log", + "prettyplease", "proc-macro2", "quote", "regex", @@ -1086,6 +1082,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" @@ -1142,12 +1149,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" @@ -1157,6 +1158,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" @@ -1211,31 +1224,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]] @@ -1245,7 +1240,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fe3e9d21d39e6f39ff46e03d5b621424f88ee23d8a1672c6bfc9ccc7f2da4f6a" dependencies = [ "cmov", - "subtle", ] [[package]] @@ -1263,6 +1257,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" @@ -1357,8 +1376,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a878c850e9e421b20262e9b41f9c860e4785fa07541c266b62ff9d1ef998a80a" dependencies = [ "const-oid 0.10.2", - "pem-rfc7468", - "zeroize", ] [[package]] @@ -1569,21 +1586,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" @@ -1721,27 +1723,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" @@ -1929,16 +1910,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" @@ -2335,17 +2306,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" @@ -2409,7 +2369,6 @@ dependencies = [ "kio", "moq-json", "moq-net", - "regex", "serde", "serde_json", "serde_with", @@ -2464,6 +2423,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" @@ -2624,9 +2656,7 @@ version = "0.4.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" dependencies = [ - "subtle", "typenum", - "zeroize", ] [[package]] @@ -3071,14 +3101,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", ] @@ -3172,6 +3201,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" @@ -3430,20 +3539,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", @@ -3463,6 +3573,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" @@ -3550,7 +3669,6 @@ dependencies = [ "thiserror 2.0.21", "tracing", "url", - "uuid", ] [[package]] @@ -3698,6 +3816,7 @@ dependencies = [ "thiserror 2.0.21", "tokio", "tracing", + "uuid", ] [[package]] @@ -3752,7 +3871,6 @@ dependencies = [ "moq-quic", "moq-sock", "notify", - "pin-project-lite", "qmux", "rand", "rcgen", @@ -4775,33 +4893,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" @@ -4853,15 +4944,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" @@ -5076,39 +5158,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]] @@ -5177,7 +5254,6 @@ dependencies = [ "libc", "moq-net", "moq-quic", - "rustc-hash 2.1.3", "rustls", "thiserror 2.0.21", "tokio", @@ -5522,13 +5598,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]] @@ -5851,20 +5956,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" @@ -6019,16 +6110,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" @@ -6108,16 +6189,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" @@ -6369,7 +6440,6 @@ version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ef958a98b9d5da290cfc78946e9f3e61e1e62a18db0d92cac0b83cc161491a9" dependencies = [ - "base64ct", "der 0.8.2", ] @@ -7424,9 +7494,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", @@ -8043,18 +8113,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 7bba27a..0c6c3e4 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 @@ -130,10 +130,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. @@ -479,7 +479,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 diff --git a/moq b/moq index 96244dd..fd07f5b 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 40c732b..7ae2afb 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -104,9 +104,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 f87e964..6dfff22 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 fa7727b..77dd74c 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 917001b..9737d14 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 e359490..0000000 --- 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 467d0d1..0000000 --- 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 88d6b4d..6da4f34 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -14,7 +14,8 @@ //! microphone, or system-audio capture (with the `capture` feature), a still //! image, a solid color, a broadcast subscribed over MoQ, or a web page //! rendered by a browser engine (the [`browser`] module). 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/publish.rs b/rs/moq-scene/src/publish.rs index c368f85..90e4aa5 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); @@ -884,7 +884,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 02dc9c7..1341dd9 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 c6c1773..c131239 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::{AudioConfig, Reference, VideoConfig}; use kio::clock::Instant; -use moq_audio::bus::{Bus, Strip}; +use moq_audio::bus::{Bus, Mix, Strip}; use moq_mux::catalog::Stream; use tokio::sync::watch; @@ -109,8 +109,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. @@ -321,6 +323,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 { @@ -510,11 +522,11 @@ 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 = self.context.bus.capture(config, strip); + let capture = capture(source).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"))] @@ -868,8 +880,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, }; @@ -902,6 +914,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); let page = matches!(source, Source::Browser { .. }).then(Page::default); // Claimed here, in the order sources open, so an edited source's @@ -981,6 +996,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())))?; @@ -1004,8 +1042,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, @@ -1187,7 +1227,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 { @@ -1198,6 +1239,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(), @@ -1227,6 +1270,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; } @@ -1257,9 +1310,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()); @@ -1300,9 +1351,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()); @@ -1337,9 +1386,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 b501842..fe9f37f 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**: one per program rendition, which waits for each program copy //! and encodes it. A new canvas size or frame rate starts a new rendition. //! - **source-***: one per running source, since a capture stream may not @@ -405,7 +406,8 @@ pub struct Engine { pub stats: Arc, /// 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 mix: moq_audio::bus::Mix, /// `None` when running without a relay. pub output: Option, /// The room guests join, when configured. @@ -421,7 +423,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); @@ -458,7 +460,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 { @@ -491,6 +494,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), @@ -515,7 +519,7 @@ impl Engine { autosave, stats, health: Mutex::default(), - bus, + mix: program, output, room, _origin: origin, @@ -547,7 +551,10 @@ impl Engine { })?; *engine.program.lock().unwrap() = Some(thread); 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))); @@ -783,17 +790,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.mix.reader(); let mut written: u128 = 0; loop { @@ -805,14 +817,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_audio(err); // Viewers see the audio fail rather than go away unfinished. @@ -840,7 +853,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( @@ -915,11 +928,11 @@ async fn publish( let audio = moq_audio::encode::Producer::new(&mut broadcast, catalog, input, &options)?; drop(reserved); - // 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(); @@ -930,7 +943,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: graded, clock: clock.clone(), relays: links, @@ -1114,9 +1127,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), @@ -1663,12 +1676,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(); @@ -1695,25 +1708,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 { @@ -1722,7 +1732,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, @@ -1747,13 +1757,37 @@ 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 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", Some(&newer), &guests[1]); + 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; }; kio::clock::timeout(TIMEOUT, routed) @@ -1761,8 +1795,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") { @@ -1770,7 +1804,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 5e19011..55b1235 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 fb191cc..9e3defd 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -53,7 +53,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)] @@ -101,15 +101,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 @@ -288,10 +288,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); @@ -299,7 +299,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 038c0ed..7fb1d07 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.bus.level()); + meter(ui, self.engine.mix.level()); }); ui.separator(); diff --git a/rs/moq-studio/src/uplink.rs b/rs/moq-studio/src/uplink.rs index f054f9e..b5e4290 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 diff --git a/rs/moq-studio/tests/browser.rs b/rs/moq-studio/tests/browser.rs index aede51b..f46a7f4 100644 --- a/rs/moq-studio/tests/browser.rs +++ b/rs/moq-studio/tests/browser.rs @@ -23,7 +23,8 @@ struct Studio { compositor: Compositor, device: wgpu::Device, queue: wgpu::Queue, - bus: moq_audio::bus::Bus, + /// Advances the bus, which meters each source's strip. + mixer: moq_audio::bus::Driver, _origin: moq_net::origin::Producer, _dir: tempfile::TempDir, } @@ -58,9 +59,10 @@ impl Studio { let dir = tempfile::tempdir().unwrap(); let (origin, driver) = moq_net::origin::Producer::new(Default::default()); tokio::spawn(moq_net::time::run(driver)); - let bus = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let (bus, mixer) = moq_audio::bus::Bus::new(Default::default()).unwrap(); let mut sources = Sources::new(Context { - bus: bus.clone(), + program: bus.mix(), + bus, origin: origin.consume(), dir: dir.path().to_path_buf(), output: moq_video::render::output(&device), @@ -79,7 +81,7 @@ impl Studio { compositor, device, queue, - bus, + mixer, _origin: origin, _dir: dir, } @@ -199,10 +201,9 @@ async fn a_page_draws_and_plays() { let strip = feed.strip().unwrap().clone(); let deadline = Instant::now() + TIMEOUT; - let mut samples = vec![0.0; 960]; while strip.level().peak() <= 0.0 { assert!(Instant::now() < deadline, "the page's tone never reached the bus"); - studio.bus.read(&mut samples); + studio.mixer.advance(480); kio::clock::sleep(Duration::from_millis(10)).await; } })