From 0608f287b4b4c977a80c5239978d585f5996d9de Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 20 Jul 2026 13:10:03 -0700 Subject: [PATCH 1/6] refactor(net)!: pre-bump API polish for kio, moq-net, moq-token, moq-native Land breaking API cleanups so they ride the pending semver release instead of costing another major later. kio: restore Sync/RefUnwindSafe on Waiter/Pending/Shared by switching the lazy waker cell from OnceCell to OnceLock (the lazy fast path is preserved); add a compile-time Sync assertion so it cannot silently regress. Drop the vestigial R: Unpin bound on Consumer::wait. moq-net: mark Role #[non_exhaustive] (the wire already tolerates unknown role codes); delete the deprecated with_publish/with_consume builder aliases; add Route::announced() so the advertised publish path is one call; rename bandwidth::Producer::close to abort for crate-wide idiom consistency; move serde_json to dev-dependencies; document every public item and turn on warn(missing_docs); remove em dashes from prose. moq-token: mark Error and KeyError #[non_exhaustive]. moq-native: mark the Backoff config and Status enum #[non_exhaustive]. Co-Authored-By: Claude Fable 5 --- rs/CLAUDE.md | 18 +++++++++-- rs/kio/src/consumer.rs | 1 - rs/kio/src/waiter.rs | 22 +++++++++---- rs/moq-bench/src/connection.rs | 1 + rs/moq-gst/src/sink/session.rs | 1 + rs/moq-native/src/client.rs | 12 -------- rs/moq-native/src/reconnect.rs | 4 ++- rs/moq-native/src/server.rs | 24 --------------- rs/moq-net/Cargo.toml | 2 +- rs/moq-net/src/client.rs | 15 ++------- rs/moq-net/src/coding/decode.rs | 15 +++++++++ rs/moq-net/src/coding/encode.rs | 6 ++++ rs/moq-net/src/coding/varint.rs | 36 ++++++++++++---------- rs/moq-net/src/error.rs | 16 ++++++++++ rs/moq-net/src/ietf/adapter.rs | 4 +-- rs/moq-net/src/ietf/properties.rs | 4 +-- rs/moq-net/src/ietf/subscribe_namespace.rs | 12 ++++---- rs/moq-net/src/lib.rs | 2 ++ rs/moq-net/src/lite/connecting.rs | 4 +-- rs/moq-net/src/lite/priority.rs | 2 +- rs/moq-net/src/lite/setup.rs | 1 + rs/moq-net/src/model/bandwidth.rs | 10 +++--- rs/moq-net/src/model/broadcast.rs | 30 ++++++++++++++++++ rs/moq-net/src/model/origin.rs | 1 + rs/moq-net/src/model/time.rs | 3 +- rs/moq-net/src/model/track.rs | 28 +++++++++++++++++ rs/moq-net/src/path.rs | 14 +++++++++ rs/moq-net/src/server.rs | 15 ++------- rs/moq-net/src/setup.rs | 2 +- rs/moq-net/src/version.rs | 4 +++ rs/moq-relay/src/connection.rs | 7 +++-- rs/moq-token/src/error.rs | 2 ++ 32 files changed, 208 insertions(+), 110 deletions(-) diff --git a/rs/CLAUDE.md b/rs/CLAUDE.md index 60a6bee9a6..d7bd26129e 100644 --- a/rs/CLAUDE.md +++ b/rs/CLAUDE.md @@ -76,11 +76,25 @@ Follow the root `poll_*` conventions: collapse `Poll::Pending => Poll::Pending` ## Version matching -`moq_net::Version` is `#[non_exhaustive]`, splitting `Lite(lite::Version)` and `Ietf(ietf::Version)` (`version.rs`). When matching on a `Version` (or the inner draft enums), default to the **newest** draft so future versions fall forward; list older versions explicitly: +`moq_net::Version` is `#[non_exhaustive]`, splitting `Lite(lite::Version)` and `Ietf(ietf::Version)` (`version.rs`). The inner `lite::Version` / `ietf::Version` payloads are crate-private, so outside `moq-net` you branch on the accessors rather than on variants: `is_lite()` / `is_ietf()` for the protocol family, and `alpn()` / `code()` for the specific draft. + +```rust +// Outside the crate: family first, then the ALPN string for a specific draft. +if version.is_lite() { + // moq-lite behavior +} else { + match version.alpn() { + "moqt-15" | "moqt-16" => { /* old behavior */ } + _ => { /* newest / draft-17+ behavior */ } + } +} +``` + +Inside `moq-net`, match the inner draft enums directly. Either way, default to the **newest** draft so future versions fall forward, and list older versions explicitly: ```rust match version { - Version::Draft14 | Version::Draft15 | Version::Draft16 => { /* old behavior */ } + ietf::Version::Draft14 | ietf::Version::Draft15 | ietf::Version::Draft16 => { /* old behavior */ } _ => { /* newest / draft-17+ behavior */ } } ``` diff --git a/rs/kio/src/consumer.rs b/rs/kio/src/consumer.rs index 8e86cd2100..2f62d529c0 100644 --- a/rs/kio/src/consumer.rs +++ b/rs/kio/src/consumer.rs @@ -66,7 +66,6 @@ impl Consumer { pub async fn wait(&self, mut f: F) -> Result where F: FnMut(&Ref<'_, T>) -> Poll + Unpin, - R: Unpin, { // The `Ref` is dropped here inside the closure, releasing the lock before the // caller ever sees the result. diff --git a/rs/kio/src/waiter.rs b/rs/kio/src/waiter.rs index fd7a438548..8b7d5f24c4 100644 --- a/rs/kio/src/waiter.rs +++ b/rs/kio/src/waiter.rs @@ -1,10 +1,9 @@ use std::{ - cell::OnceCell, fmt, future::Future, marker::PhantomData, pin::Pin, - sync::{Arc, Weak}, + sync::{Arc, OnceLock, Weak}, task::{Context, Poll, Waker}, }; @@ -27,7 +26,7 @@ pub struct Waiter { // The shared handle downgraded into every list this waiter registers with. Created on the // first `register` (a poll that never parks never allocates it), then reused so multiple // lists in one poll share a single allocation whose `Weak`s die together when the waiter drops. - shared: OnceCell>, + shared: OnceLock>, } impl Waiter { @@ -35,7 +34,7 @@ impl Waiter { pub fn new(waker: Waker) -> Self { Self { waker, - shared: OnceCell::new(), + shared: OnceLock::new(), } } @@ -102,8 +101,8 @@ impl WaiterList { if self.entries[self.cursor].strong_count() == 0 { // Reuse the dead slot in place. Each Waiter owns a // unique Arc, so strong_count == 0 uniquely - // identifies a slot whose owner has been dropped — - // no will_wake / pointer comparison needed. + // identifies a slot whose owner has been dropped. + // No will_wake / pointer comparison needed. self.entries[self.cursor] = new_weak; return; } @@ -203,6 +202,17 @@ mod tests { assert_eq!(waiter.poll_future(boxed.as_mut()), Poll::Ready(9)); } + // `Waiter` is shared behind `&self` across threads, so the lazily allocated + // `shared` handle must use a thread-safe cell. A `!Sync` waiter silently + // infects `Pending` and `Shared`, and through them every moq-net consumer. + const fn assert_sync() {} + + const _: () = { + assert_sync::(); + assert_sync::>>(); + assert_sync::>(); + }; + #[test] fn wait_output_need_not_be_unpin() { struct NotUnpin(#[allow(dead_code)] std::marker::PhantomPinned); diff --git a/rs/moq-bench/src/connection.rs b/rs/moq-bench/src/connection.rs index 3e8e633526..cb25336f22 100644 --- a/rs/moq-bench/src/connection.rs +++ b/rs/moq-bench/src/connection.rs @@ -147,6 +147,7 @@ pub async fn run(ctx: Connection) { stats.connections.fetch_sub(1, Ordering::Relaxed); } } + Ok(_) => {} Err(err) => { tracing::warn!(connection, %err, "connection gave up"); break; diff --git a/rs/moq-gst/src/sink/session.rs b/rs/moq-gst/src/sink/session.rs index 0a517cb97a..d2eeeafa13 100644 --- a/rs/moq-gst/src/sink/session.rs +++ b/rs/moq-gst/src/sink/session.rs @@ -240,6 +240,7 @@ async fn forward( match state { moq_native::Status::Connected => gst::info!(CAT, "session connected"), moq_native::Status::Disconnected => gst::warning!(CAT, "session disconnected, reconnecting"), + _ => {} } notify(&element, &["status", "connected", "moq-version"]); } diff --git a/rs/moq-native/src/client.rs b/rs/moq-native/src/client.rs index 0608a3edb6..4d88302370 100644 --- a/rs/moq-native/src/client.rs +++ b/rs/moq-native/src/client.rs @@ -236,18 +236,6 @@ impl Client { self } - #[doc(hidden)] - #[deprecated(note = "renamed to `with_publisher`")] - pub fn with_publish(self, publish: moq_net::origin::Consumer) -> Self { - self.with_publisher(publish) - } - - #[doc(hidden)] - #[deprecated(note = "renamed to `with_subscriber`")] - pub fn with_consume(self, subscribe: moq_net::origin::Producer) -> Self { - self.with_subscriber(subscribe) - } - /// Attach a tier-scoped [`moq_net::stats::Handle`] to all sessions opened by this client. pub fn with_stats(mut self, stats: moq_net::stats::Handle) -> Self { self.moq = self.moq.with_stats(stats); diff --git a/rs/moq-native/src/reconnect.rs b/rs/moq-native/src/reconnect.rs index 885796d1e0..8bddb6a8bd 100644 --- a/rs/moq-native/src/reconnect.rs +++ b/rs/moq-native/src/reconnect.rs @@ -11,6 +11,7 @@ use crate::{Client, Error}; /// Exponential backoff configuration for reconnection attempts. #[derive(Clone, Debug, clap::Args, serde::Serialize, serde::Deserialize)] #[serde(default, deny_unknown_fields)] +#[non_exhaustive] pub struct Backoff { /// Initial delay before first reconnect attempt. #[arg( @@ -66,6 +67,7 @@ impl Default for Backoff { /// A connection lifecycle transition reported by [`Reconnect::status`]. #[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] pub enum Status { /// A session connected (the first connect, or a reconnect after a drop). Connected, @@ -436,7 +438,7 @@ mod tests { assert_eq!(out_rx.peek(), Some(9_000)); // Closing the source is what retires the arm, so we stop polling a dead one. - src.close(moq_net::Error::Cancel).unwrap(); + src.abort(moq_net::Error::Cancel).unwrap(); poll_forward(&mut bw, &out, &waiter); assert!(bw.is_none()); } diff --git a/rs/moq-native/src/server.rs b/rs/moq-native/src/server.rs index 4e5404989f..69e280c130 100644 --- a/rs/moq-native/src/server.rs +++ b/rs/moq-native/src/server.rs @@ -252,18 +252,6 @@ impl Server { self } - #[doc(hidden)] - #[deprecated(note = "renamed to `with_publisher`")] - pub fn with_publish(self, publish: moq_net::origin::Consumer) -> Self { - self.with_publisher(publish) - } - - #[doc(hidden)] - #[deprecated(note = "renamed to `with_subscriber`")] - pub fn with_consume(self, subscribe: moq_net::origin::Producer) -> Self { - self.with_subscriber(subscribe) - } - /// Attach a tier-scoped [`moq_net::stats::Handle`] to all sessions accepted by this server. pub fn with_stats(mut self, stats: moq_net::stats::Handle) -> Self { self.moq = self.moq.with_stats(stats); @@ -944,18 +932,6 @@ impl Request { } } - #[doc(hidden)] - #[deprecated(note = "renamed to `with_publisher`")] - pub fn with_publish(self, publish: moq_net::origin::Consumer) -> Self { - self.with_publisher(publish) - } - - #[doc(hidden)] - #[deprecated(note = "renamed to `with_subscriber`")] - pub fn with_consume(self, subscribe: moq_net::origin::Producer) -> Self { - self.with_subscriber(subscribe) - } - /// Attach a tier-scoped [`moq_net::stats::Handle`] to this session. pub fn with_stats(self, stats: moq_net::stats::Handle) -> Self { let Request { diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index dc791abd54..96f4f25ef8 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -22,7 +22,6 @@ kio = { workspace = true } num_enum = "0.7" rand = "0.10.1" serde = { workspace = true } -serde_json = "1" thiserror = "2" tracing = "0.1" web-async = { workspace = true } @@ -30,6 +29,7 @@ web-transport-trait = { workspace = true } [dev-dependencies] criterion = "0.8" +serde_json = "1" # test-util (tokio::time::pause/advance) is test-only and is NOT supported on # wasm, so it must not leak into the normal dependency feature set. tokio = { workspace = true, features = ["macros", "io-util", "sync", "test-util", "time", "rt"] } diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index e4bd3affc9..d7c140e53d 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -17,6 +17,7 @@ pub struct Client { } impl Client { + /// A client that neither publishes nor subscribes until configured. pub fn new() -> Self { Default::default() } @@ -36,18 +37,6 @@ impl Client { self } - #[doc(hidden)] - #[deprecated(note = "renamed to `with_publisher`")] - pub fn with_publish(self, publish: origin::Consumer) -> Self { - self.with_publisher(publish) - } - - #[doc(hidden)] - #[deprecated(note = "renamed to `with_subscriber`")] - pub fn with_consume(self, subscribe: origin::Producer) -> Self { - self.with_subscriber(subscribe) - } - /// Attach a tier-scoped [`stats::Handle`]. Per-broadcast and per-subscription /// counters will be bumped through this handle for the lifetime of the session. /// Pass [`stats::Handle::default`] (a no-op handle) to opt out. @@ -64,6 +53,8 @@ impl Client { self.with_publisher(&origin).with_subscriber(origin) } + /// Restrict which protocol versions to offer, in preference order. + /// Defaults to every version this crate supports. pub fn with_versions(mut self, versions: Versions) -> Self { self.versions = versions; self diff --git a/rs/moq-net/src/coding/decode.rs b/rs/moq-net/src/coding/decode.rs index dfa2f47393..f3eadc2233 100644 --- a/rs/moq-net/src/coding/decode.rs +++ b/rs/moq-net/src/coding/decode.rs @@ -13,48 +13,63 @@ pub trait Decode: Sized { #[derive(Error, Debug, Clone)] #[non_exhaustive] pub enum DecodeError { + /// The buffer ran out mid-value. Retry once more bytes arrive. #[error("short buffer")] Short, + /// The value claims more bytes than the enclosing message allows. #[error("long buffer")] Long, + /// A string field was not valid UTF-8. #[error("invalid string")] InvalidString(#[from] FromUtf8Error), + /// The message type ID is unknown for the negotiated version. #[error("invalid message: {0:?}")] InvalidMessage(u64), + /// A SUBSCRIBE start/end location is malformed or out of order. #[error("invalid subscribe location")] InvalidSubscribeLocation, + /// A field held a value outside its permitted range. #[error("invalid value")] InvalidValue, + /// A repeated field exceeded the count this implementation accepts. #[error("too many")] TooMany, + /// An integer was too large for the QUIC varint range. #[error("bounds exceeded")] BoundsExceeded, + /// More data followed where the message was required to end. #[error("expected end")] ExpectedEnd, + /// The stream ended where a payload was required. #[error("expected data")] ExpectedData, + /// A parameter or field appeared more than once. #[error("duplicate")] Duplicate, + /// A required parameter or field was absent. #[error("missing")] Missing, + /// The value is well-formed but this implementation does not handle it. #[error("unsupported")] Unsupported, + /// Bytes remained after the value was fully decoded. #[error("trailing bytes")] TrailingBytes, + /// The field does not exist in the negotiated protocol version. #[error("unsupported version")] Version, } diff --git a/rs/moq-net/src/coding/encode.rs b/rs/moq-net/src/coding/encode.rs index b537a4322f..d6108ede94 100644 --- a/rs/moq-net/src/coding/encode.rs +++ b/rs/moq-net/src/coding/encode.rs @@ -8,16 +8,22 @@ use super::BoundsExceeded; #[derive(thiserror::Error, Debug, Clone)] #[non_exhaustive] pub enum EncodeError { + /// An integer was too large for the QUIC varint range. #[error("bounds exceeded")] BoundsExceeded, + /// The payload exceeds the maximum size the wire format can express. #[error("too large")] TooLarge, + /// The destination buffer had no room for the value. #[error("short buffer")] Short, + /// The message cannot be encoded from the current session state. #[error("invalid state")] InvalidState, + /// A repeated field exceeded the count the wire format permits. #[error("too many")] TooMany, + /// The field does not exist in the negotiated protocol version. #[error("unsupported version")] Version, } diff --git a/rs/moq-net/src/coding/varint.rs b/rs/moq-net/src/coding/varint.rs index a921e5bc93..a933977174 100644 --- a/rs/moq-net/src/coding/varint.rs +++ b/rs/moq-net/src/coding/varint.rs @@ -35,10 +35,12 @@ impl VarInt { Self(x as u64) } + /// Construct from a `u64`, or `None` if it exceeds [`Self::MAX`]. pub const fn from_u64(x: u64) -> Option { if x <= Self::MAX.0 { Some(Self(x)) } else { None } } + /// Construct from a `u128`, or `None` if it exceeds [`Self::MAX`]. pub const fn from_u128(x: u128) -> Option { if x <= Self::MAX.0 as u128 { Some(Self(x as u64)) @@ -278,11 +280,11 @@ impl VarInt { match ones { 0 => { - // 0xxxxxxx — 7 bits + // 0xxxxxxx: 7 bits Ok(Self(u64::from(b))) } 1 => { - // 10xxxxxx + 1 byte — 14 bits + // 10xxxxxx + 1 byte: 14 bits if !r.has_remaining() { return Err(DecodeError::Short); } @@ -291,7 +293,7 @@ impl VarInt { Ok(Self((hi << 8) | lo)) } 2 => { - // 110xxxxx + 2 bytes — 21 bits + // 110xxxxx + 2 bytes: 21 bits if r.remaining() < 2 { return Err(DecodeError::Short); } @@ -301,7 +303,7 @@ impl VarInt { Ok(Self((hi << 16) | u64::from(u16::from_be_bytes(buf)))) } 3 => { - // 1110xxxx + 3 bytes — 28 bits + // 1110xxxx + 3 bytes: 28 bits if r.remaining() < 3 { return Err(DecodeError::Short); } @@ -313,7 +315,7 @@ impl VarInt { )) } 4 => { - // 11110xxx + 4 bytes — 35 bits + // 11110xxx + 4 bytes: 35 bits if r.remaining() < 4 { return Err(DecodeError::Short); } @@ -323,7 +325,7 @@ impl VarInt { Ok(Self((hi << 32) | u64::from(u32::from_be_bytes(buf)))) } 5 => { - // 111110xx + 5 bytes — 42 bits + // 111110xx + 5 bytes: 42 bits if r.remaining() < 5 { return Err(DecodeError::Short); } @@ -351,7 +353,7 @@ impl VarInt { Ok(Self((hi << 48) | u64::from_be_bytes(buf))) } 7 => { - // 11111110 + 7 bytes — 56 bits + // 11111110 + 7 bytes: 56 bits if r.remaining() < 7 { return Err(DecodeError::Short); } @@ -361,7 +363,7 @@ impl VarInt { Ok(Self(u64::from_be_bytes(buf))) } 8 => { - // 11111111 + 8 bytes — 64 bits + // 11111111 + 8 bytes: 64 bits if r.remaining() < 8 { return Err(DecodeError::Short); } @@ -383,27 +385,27 @@ impl VarInt { let remaining = w.remaining_mut(); if x < (1 << 7) { - // 0xxxxxxx — 1 byte + // 0xxxxxxx: 1 byte if remaining < 1 { return Err(EncodeError::Short); } w.put_u8(x as u8); } else if x < (1 << 14) { - // 10xxxxxx — 2 bytes + // 10xxxxxx: 2 bytes if remaining < 2 { return Err(EncodeError::Short); } w.put_u8(0x80 | (x >> 8) as u8); w.put_u8(x as u8); } else if x < (1 << 21) { - // 110xxxxx — 3 bytes + // 110xxxxx: 3 bytes if remaining < 3 { return Err(EncodeError::Short); } w.put_u8(0xC0 | (x >> 16) as u8); w.put_u16(x as u16); } else if x < (1 << 28) { - // 1110xxxx — 4 bytes + // 1110xxxx: 4 bytes if remaining < 4 { return Err(EncodeError::Short); } @@ -411,14 +413,14 @@ impl VarInt { w.put_u8((x >> 16) as u8); w.put_u16(x as u16); } else if x < (1 << 35) { - // 11110xxx — 5 bytes + // 11110xxx: 5 bytes if remaining < 5 { return Err(EncodeError::Short); } w.put_u8(0xF0 | (x >> 32) as u8); w.put_u32(x as u32); } else if x < (1 << 42) { - // 111110xx — 6 bytes + // 111110xx: 6 bytes if remaining < 6 { return Err(EncodeError::Short); } @@ -426,7 +428,7 @@ impl VarInt { w.put_u8((x >> 32) as u8); w.put_u32(x as u32); } else if x < (1 << 56) { - // 11111110 — 8 bytes (skips 7) + // 11111110: 8 bytes (skips 7) if remaining < 8 { return Err(EncodeError::Short); } @@ -436,7 +438,7 @@ impl VarInt { w.put_u16((x >> 32) as u16); w.put_u32(x as u32); } else { - // 11111111 — 9 bytes + // 11111111: 9 bytes if remaining < 9 { return Err(EncodeError::Short); } @@ -572,7 +574,7 @@ mod tests { (&[0x25], 37), (&[0x80, 0x25], 37), (&[0xbb, 0xbd], 15_293), - // Example 4 (0xdd7f3e7d = 494,878,333) is omitted — the spec has a bug. + // Example 4 (0xdd7f3e7d = 494,878,333) is omitted. The spec has a bug. // See https://github.com/moq-wg/moq-transport/pull/1521 (&[0xfa, 0xa1, 0xa0, 0xe4, 0x03, 0xd8], 2_893_212_287_960), ( diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 87d035789b..a5cb6ffe21 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -4,9 +4,11 @@ use crate::coding; #[derive(thiserror::Error, Debug, Clone)] #[non_exhaustive] pub enum Error { + /// The underlying QUIC/WebTransport connection failed; carries the backend's message. #[error("transport: {0}")] Transport(String), + /// A message off the wire could not be parsed. #[error(transparent)] Decode(#[from] coding::DecodeError), @@ -25,6 +27,7 @@ pub enum Error { #[error("unexpected stream type")] UnexpectedStream, + /// An integer was too large for the QUIC varint range. #[error(transparent)] BoundsExceeded(#[from] coding::BoundsExceeded), @@ -33,6 +36,7 @@ pub enum Error { #[error("duplicate")] Duplicate, + /// Nobody is reading any more, so the producer stopped. Not a failure. // Cancel is returned when there are no more readers. #[error("cancelled")] Cancel, @@ -55,6 +59,7 @@ pub enum Error { #[error("app code={0}")] App(u16), + /// The requested broadcast or track does not exist at the peer. #[error("not found")] NotFound, @@ -63,27 +68,35 @@ pub enum Error { #[error("unroutable")] Unroutable, + /// A frame's payload length disagreed with its declared size. #[error("wrong frame size")] WrongSize, + /// The peer broke a protocol rule; the session is unusable. #[error("protocol violation")] ProtocolViolation, + /// The peer's token does not grant the requested path or operation. #[error("unauthorized")] Unauthorized, + /// A valid message arrived in a state where it is not allowed. #[error("unexpected message")] UnexpectedMessage, + /// The peer asked for a feature this endpoint does not implement. #[error("unsupported")] Unsupported, + /// A message could not be serialized for the negotiated version. #[error(transparent)] Encode(#[from] coding::EncodeError), + /// A message carried more parameters than this endpoint accepts. #[error("too many parameters")] TooManyParameters, + /// The peer acted against the [`Role`](crate::Role) it advertised at SETUP. #[error("invalid role")] InvalidRole, @@ -92,9 +105,11 @@ pub enum Error { #[error("unknown ALPN: {0}")] UnknownAlpn(String), + /// The producer was dropped without finishing, so the content is incomplete. #[error("dropped")] Dropped, + /// The handle was already closed by this side. #[error("closed")] Closed, @@ -179,6 +194,7 @@ impl web_transport_trait::Error for Error { } } +/// A [`Result`](std::result::Result) with this crate's [`Error`]. pub type Result = std::result::Result; #[cfg(test)] diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 9aa87bb20b..5eea03eb3d 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -746,7 +746,7 @@ impl ControlStreamAdapter { let id = decode_request_id(body, self.version)?; Ok(Route::CloseStream(id)) } - // v14/v15: namespace-keyed — decode namespace and look up request_id + // v14/v15: namespace-keyed, so decode namespace and look up request_id Version::Draft14 | Version::Draft15 => { let id = self.lookup_namespace_request_id(body)?; Ok(Route::CloseStream(id)) @@ -960,7 +960,7 @@ mod tests { streams: Mutex::new(HashMap::new()), namespaces: Mutex::new(HashMap::new()), }); - // We need a dummy inner session — but classify doesn't use it. + // We need a dummy inner session, but classify doesn't use it. // Use a struct that satisfies the trait bound. We can't easily construct one, // so we'll test via a free function wrapper instead. diff --git a/rs/moq-net/src/ietf/properties.rs b/rs/moq-net/src/ietf/properties.rs index 5d44f67ea5..5672804978 100644 --- a/rs/moq-net/src/ietf/properties.rs +++ b/rs/moq-net/src/ietf/properties.rs @@ -1,4 +1,4 @@ -/// Track Properties — relay-visible metadata attached to tracks. +/// Track Properties: relay-visible metadata attached to tracks. /// /// Draft-17 adds Track Properties to SUBSCRIBE_OK, PUBLISH, and FETCH_OK. /// They appear after the message parameters as a sequence of Key-Value-Pairs @@ -22,7 +22,7 @@ const MAX_KVP_VALUE_LEN: usize = (1 << 16) - 1; /// /// Track Properties use the same Key-Value-Pair encoding as parameters: /// delta-encoded types, even = varint value, odd = length-prefixed bytes. -/// They have no count prefix — read until the buffer is empty. +/// They have no count prefix. Read until the buffer is empty. /// /// Only call this for draft-17+; older drafts don't have Track Properties. pub fn skip(r: &mut R, version: Version) -> Result<(), DecodeError> { diff --git a/rs/moq-net/src/ietf/subscribe_namespace.rs b/rs/moq-net/src/ietf/subscribe_namespace.rs index 3f762f9db2..d7e5868b18 100644 --- a/rs/moq-net/src/ietf/subscribe_namespace.rs +++ b/rs/moq-net/src/ietf/subscribe_namespace.rs @@ -121,7 +121,7 @@ impl Message for SubscribeNamespaceLegacy<'_> { } } -/// SubscribeNamespaceOk message (0x12) — v14 only +/// SubscribeNamespaceOk message (0x12): v14 only #[derive(Clone, Debug)] pub struct SubscribeNamespaceOk { pub request_id: RequestId, @@ -141,7 +141,7 @@ impl Message for SubscribeNamespaceOk { } } -/// SubscribeNamespaceError message (0x13) — v14 only +/// SubscribeNamespaceError message (0x13): v14 only #[derive(Clone, Debug)] pub struct SubscribeNamespaceError<'a> { pub request_id: RequestId, @@ -172,7 +172,7 @@ impl Message for SubscribeNamespaceError<'_> { } } -/// UnsubscribeNamespace message (0x14) — v14/v15 only (v16 uses stream close) +/// UnsubscribeNamespace message (0x14): v14/v15 only (v16 uses stream close) #[derive(Clone, Debug)] pub struct UnsubscribeNamespace { pub request_id: RequestId, @@ -192,7 +192,7 @@ impl Message for UnsubscribeNamespace { } } -/// NAMESPACE message (0x08) — v16 only, sent on SUBSCRIBE_NAMESPACE bidi stream +/// NAMESPACE message (0x08): v16 only, sent on SUBSCRIBE_NAMESPACE bidi stream /// Indicates a namespace suffix matching the subscribed prefix is active. #[derive(Clone, Debug)] pub struct Namespace<'a> { @@ -213,7 +213,7 @@ impl Message for Namespace<'_> { } } -/// PUBLISH_BLOCKED message (0x0F) — draft-17 only +/// PUBLISH_BLOCKED message (0x0F): draft-17 only /// Indicates a track within a namespace is blocked from publishing. #[derive(Clone, Debug)] #[allow(dead_code)] // Will be used in Phase 3 bidi stream handling @@ -242,7 +242,7 @@ impl Message for PublishBlocked<'_> { } } -/// NAMESPACE_DONE message (0x0E) — v16 only, sent on SUBSCRIBE_NAMESPACE bidi stream +/// NAMESPACE_DONE message (0x0E): v16 only, sent on SUBSCRIBE_NAMESPACE bidi stream /// Indicates a namespace suffix matching the subscribed prefix is no longer active. #[derive(Clone, Debug)] pub struct NamespaceDone<'a> { diff --git a/rs/moq-net/src/lib.rs b/rs/moq-net/src/lib.rs index 550fa861ba..25cbd48991 100644 --- a/rs/moq-net/src/lib.rs +++ b/rs/moq-net/src/lib.rs @@ -69,6 +69,8 @@ //! model-layer methods (tracks, groups, frames, origins) never touch a timer and //! run on any executor. +#![warn(missing_docs)] + mod client; mod coding; mod error; diff --git a/rs/moq-net/src/lite/connecting.rs b/rs/moq-net/src/lite/connecting.rs index 0f818a2984..9e312202aa 100644 --- a/rs/moq-net/src/lite/connecting.rs +++ b/rs/moq-net/src/lite/connecting.rs @@ -7,8 +7,8 @@ //! //! Backed by `kio`: each in-flight step holds a [`ConnectingProducer`], and the //! session is connected once they've all been dropped (which closes the channel). -//! A step drops its producer when it finishes — or, on an early error, when it goes -//! out of scope — so a failed step can't hang `connect()`. Exposes both a synchronous +//! A step drops its producer when it finishes (or, on an early error, when it goes +//! out of scope), so a failed step can't hang `connect()`. Exposes both a synchronous //! poll API and an async one; prefer `kio` over `tokio` primitives for new async state //! so we keep both available. diff --git a/rs/moq-net/src/lite/priority.rs b/rs/moq-net/src/lite/priority.rs index 04411edab5..35c320d018 100644 --- a/rs/moq-net/src/lite/priority.rs +++ b/rs/moq-net/src/lite/priority.rs @@ -724,7 +724,7 @@ mod tests { // Lower top's track below every filler. Without the swap, top would land // in vec at the tail while f1 stays in overflow despite having higher - // priority — breaking the "every overflow item < every vec item" invariant. + // priority, breaking the "every overflow item < every vec item" invariant. top.set_track(0); assert!(fillers[0].current() < u8::MAX, "f1 should be promoted back into vec"); diff --git a/rs/moq-net/src/lite/setup.rs b/rs/moq-net/src/lite/setup.rs index 0e82f44a26..6b70bde522 100644 --- a/rs/moq-net/src/lite/setup.rs +++ b/rs/moq-net/src/lite/setup.rs @@ -60,6 +60,7 @@ impl ProbeLevel { /// parameter. `Option` mirrors that: `None` is the default, and it's also what /// a client that predates the parameter decodes to. #[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] pub enum Role { /// The client will publish tracks (ingest); the server must consume. Publisher, diff --git a/rs/moq-net/src/model/bandwidth.rs b/rs/moq-net/src/model/bandwidth.rs index ad1f98a50e..8634ece97f 100644 --- a/rs/moq-net/src/model/bandwidth.rs +++ b/rs/moq-net/src/model/bandwidth.rs @@ -48,7 +48,7 @@ impl Producer { } /// Close the producer with an error, notifying all consumers. - pub fn close(&self, err: Error) -> Result<()> { + pub fn abort(&self, err: Error) -> Result<()> { let mut state = self.modify()?; state.abort = Some(err); state.close(); @@ -69,7 +69,7 @@ impl Producer { pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll> { self.state.poll_unused(waiter).map(|used| match used { Some(()) => Ok(()), - None => Err(self.abort()), + None => Err(self.close_error()), }) } @@ -82,7 +82,7 @@ impl Producer { pub fn poll_used(&self, waiter: &kio::Waiter) -> Poll> { self.state.poll_used(waiter).map(|used| match used { Some(()) => Ok(()), - None => Err(self.abort()), + None => Err(self.close_error()), }) } @@ -93,7 +93,7 @@ impl Producer { } /// The close error, once the channel is closed. - fn abort(&self) -> Error { + fn close_error(&self) -> Error { self.state.read().abort.clone().unwrap_or(Error::Dropped) } } @@ -183,7 +183,7 @@ mod tests { assert_eq!(consumer.changed().await.unwrap(), None); // Gone for good. - producer.close(Error::Cancel).unwrap(); + producer.abort(Error::Cancel).unwrap(); assert!(consumer.changed().await.is_err()); // And it stays terminal rather than flapping back to a value. assert!(consumer.changed().await.is_err()); diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index c168542265..1722c0f625 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -1,3 +1,12 @@ +//! A broadcast is a named collection of tracks, split into a [Producer] and [Consumer] handle. +//! +//! A [Producer] creates tracks on demand: a [Consumer] subscribes by name, and the +//! producer either serves a track it already has or is handed a [`track::Request`] to +//! fill. Both handles are refcounted clones of one broadcast, which closes when the +//! last producer drops. +//! +//! [Info] is the static metadata; [Route] is the dynamic path the broadcast takes to +//! reach an origin, including whether it is announced to subscribers. use crate::track; use std::{ collections::{HashMap, VecDeque}, @@ -73,10 +82,25 @@ pub struct Route { impl Route { /// An unannounced direct route: no hops, best cost. + /// + /// The broadcast is reachable only by its exact path, so subscribers must already + /// know it exists. Use [`announced`](Self::announced) to advertise it instead. pub fn new() -> Self { Self::default() } + /// An announced direct route: no hops, best cost. + /// + /// The broadcast is advertised to subscribers via + /// [`crate::origin::Consumer::announced`] while this is the best route, on top of + /// staying reachable by exact path. Use [`new`](Self::new) to keep it unadvertised. + pub fn announced() -> Self { + Self { + announce: true, + ..Self::default() + } + } + /// Append a hop to the chain, oldest first. /// /// Fails with [`crate::TooManyOrigins`] once the chain is full, the same limit @@ -225,6 +249,7 @@ impl Producer { } } + /// The broadcast's static metadata, fixed when it was created. pub fn info(&self) -> &Info { &self.info } @@ -396,6 +421,7 @@ impl Drop for Producer { } #[cfg(test)] +#[allow(missing_docs)] // test-only assertion helpers impl Producer { pub fn assert_create_track( &mut self, @@ -489,6 +515,7 @@ impl Dynamic { Self { info, alive, state } } + /// The broadcast's static metadata, fixed when it was created. pub fn info(&self) -> &Info { &self.info } @@ -570,6 +597,7 @@ impl Drop for Dynamic { use futures::FutureExt; #[cfg(test)] +#[allow(missing_docs)] // test-only assertion helpers impl Dynamic { pub fn assert_request(&mut self) -> track::Request { self.requested_track() @@ -609,6 +637,7 @@ impl Clone for Consumer { } impl Consumer { + /// The broadcast's static metadata, fixed when it was created. pub fn info(&self) -> &Info { &self.info } @@ -793,6 +822,7 @@ impl super::WeakEntry for WeakConsumer { } #[cfg(test)] +#[allow(missing_docs)] // test-only assertion helpers impl Consumer { pub fn assert_not_closed(&self) { assert!(self.closed().now_or_never().is_none(), "should not be closed"); diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index efde3e367d..8bd335e2c3 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2125,6 +2125,7 @@ impl Drop for AnnounceConsumer { use futures::FutureExt; #[cfg(test)] +#[allow(missing_docs)] // test-only assertion helpers impl AnnounceConsumer { pub fn assert_next(&mut self, expected: impl AsPath, broadcast: &broadcast::Consumer) { let expected = expected.as_path(); diff --git a/rs/moq-net/src/model/time.rs b/rs/moq-net/src/model/time.rs index 87ca1d5c9f..73ebca076d 100644 --- a/rs/moq-net/src/model/time.rs +++ b/rs/moq-net/src/model/time.rs @@ -28,7 +28,8 @@ impl Timescale { Some(n) => Self(n), None => unreachable!(), }; - /// 1,000,000 units per second (`1_000_000`). Common default for media tracks. + /// 1,000,000 units per second (`1_000_000`). Widely used by container formats; + /// this crate's own default is [`Self::MILLI`]. pub const MICRO: Self = match NonZero::new(1_000_000) { Some(n) => Self(n), None => unreachable!(), diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index a297e62a25..4715db5500 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -638,6 +638,7 @@ impl Producer { } } + /// The track's name, unique within its broadcast. pub fn name(&self) -> &str { &self.name } @@ -965,6 +966,9 @@ impl Producer { snapshot_subscription(&subs, bound) } + /// Poll counterpart to [`subscription_changed`](Self::subscription_changed): the + /// aggregate subscription whenever it changes, or `None` once nobody is subscribed. + /// Errors once the track is aborted. pub fn poll_subscription_changed(&mut self, waiter: &kio::Waiter) -> Poll>> { // Surface an abort as the stream ending. `poll_closed` parks on the closed // waiters, so per-group churn on the track state never wakes this poll. @@ -1085,6 +1089,7 @@ impl Dynamic { Self { name, state, fetch } } + /// The track's name, unique within its broadcast. pub fn name(&self) -> &str { &self.name } @@ -1098,6 +1103,7 @@ impl Dynamic { kio::wait(|waiter| self.poll_requested_group(waiter)).await } + /// Poll counterpart to [`requested_group`](Self::requested_group). pub fn poll_requested_group(&self, waiter: &kio::Waiter) -> Poll> { poll_requested_group(&self.state, &self.fetch, waiter) } @@ -1507,6 +1513,8 @@ enum SubscribingKind { } impl Subscribing { + /// Poll until the peer confirms the subscription, yielding the [`Subscriber`]. + /// Errors if the track is aborted or not found. pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { match &self.inner { SubscribingKind::Plain(state) => { @@ -1573,6 +1581,7 @@ enum QueryingKind { } impl Querying { + /// Poll until the track's [`Info`] is known, without subscribing to its groups. pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { match &self.inner { QueryingKind::Plain(state) => { @@ -1895,6 +1904,7 @@ impl Subscriber { &self.info } + /// The track's name, unique within its broadcast. pub fn name(&self) -> &str { &self.name } @@ -2094,6 +2104,17 @@ impl Subscriber { } } +/// A subscriber asked for a track this broadcast doesn't have yet. +/// +/// Yielded by [`broadcast::Dynamic::requested_track`](crate::broadcast::Dynamic::requested_track), +/// or created up front with [`broadcast::Producer::reserve_track`](crate::broadcast::Producer::reserve_track). +/// Subscribers block until the request is +/// resolved: call [`accept`](Self::accept) to serve it with a [`Producer`], or +/// [`reject`](Self::reject) to fail them. Dropping it without either rejects with +/// [`Error::Dropped`]. +/// +/// Concurrent requests for one name are coalesced, so exactly one of these exists per +/// name at a time. pub struct Request { name: Arc, // The parent broadcast's info, threaded into the [`Producer`] on accept. @@ -2132,6 +2153,7 @@ impl Request { &self.name } + /// A [`Consumer`] for the eventual track, usable before the request is accepted. pub fn consume(&self) -> Consumer { Consumer::plain(self.name.clone(), self.state.consume()) } @@ -2178,6 +2200,8 @@ impl Request { } } + /// The delivery preferences aggregated across everyone waiting on this request, + /// or `None` if nobody is waiting. Useful for sizing the track before accepting. pub fn subscription(&self) -> Option { let state = self.state.read(); let (subs, bound) = (state.subscriptions.clone(), state.latency_bound()); @@ -2185,10 +2209,13 @@ impl Request { snapshot_subscription(&subs, bound) } + /// Block until the aggregate [`subscription`](Self::subscription) changes, + /// yielding `None` once nobody is waiting. pub async fn subscription_changed(&mut self) -> Option { kio::wait(|waiter| self.poll_subscription_changed(waiter)).await } + /// Poll counterpart to [`subscription_changed`](Self::subscription_changed). pub fn poll_subscription_changed(&mut self, waiter: &kio::Waiter) -> Poll> { let state = self.state.read(); let (subs, bound) = (state.subscriptions.clone(), state.latency_bound()); @@ -2224,6 +2251,7 @@ impl Request { use futures::FutureExt; #[cfg(test)] +#[allow(missing_docs)] // test-only assertion helpers impl Subscriber { pub fn assert_group(&mut self) -> group::Consumer { self.recv_group() diff --git a/rs/moq-net/src/path.rs b/rs/moq-net/src/path.rs index 1bc34c2e86..3e95ec5710 100644 --- a/rs/moq-net/src/path.rs +++ b/rs/moq-net/src/path.rs @@ -12,6 +12,7 @@ pub type PathOwned = Path<'static>; /// When providing a String/str, any leading/trailing slashes are trimmed and multiple consecutive slashes are collapsed. /// When already a Path, normalization is skipped and the underlying buffer is reused without copying. pub trait AsPath { + /// Borrow `self` as a [`Path`], normalizing slashes only when needed. fn as_path(&self) -> Path<'_>; } @@ -176,6 +177,10 @@ impl<'a> Path<'a> { s.as_bytes().get(prefix.len()) == Some(&b'/') } + /// The remainder after removing `prefix`, or `None` if it isn't a prefix. + /// + /// Only whole segments match: `a/bc` is not prefixed by `a/b`. An empty prefix + /// returns the whole path. pub fn strip_prefix(&'a self, prefix: impl AsPath) -> Option> { let prefix = prefix.as_path(); @@ -233,6 +238,7 @@ impl<'a> Path<'a> { } } + /// The normalized path as a string, with no leading or trailing slash. pub fn as_str(&self) -> &str { match &self.0 { Repr::Borrowed(s) => s, @@ -240,18 +246,22 @@ impl<'a> Path<'a> { } } + /// The empty path, which prefixes every other path. pub fn empty() -> Path<'static> { Path(Repr::Borrowed("")) } + /// Returns `true` if this is the empty path. pub fn is_empty(&self) -> bool { self.as_str().is_empty() } + /// The length in bytes, not segments. pub fn len(&self) -> usize { self.as_str().len() } + /// Clone into a `'static` path, sharing the existing buffer when there is one. pub fn to_owned(&self) -> PathOwned { match &self.0 { Repr::Borrowed("") => Path::empty(), @@ -266,6 +276,7 @@ impl<'a> Path<'a> { } } + /// Consume into a `'static` path, reusing the existing buffer when there is one. pub fn into_owned(self) -> PathOwned { match self.0 { Repr::Borrowed("") => Path::empty(), @@ -656,14 +667,17 @@ impl PathPrefixes { Self { paths: result } } + /// Returns `true` if the set contains no prefixes, so it matches nothing. pub fn is_empty(&self) -> bool { self.paths.is_empty() } + /// The number of prefixes, after redundant ones were collapsed. pub fn len(&self) -> usize { self.paths.len() } + /// Iterate the prefixes in the set. pub fn iter(&self) -> std::slice::Iter<'_, PathOwned> { self.paths.iter() } diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 3204734966..50e456699d 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -16,6 +16,7 @@ pub struct Server { } impl Server { + /// A server that neither publishes nor subscribes until configured. pub fn new() -> Self { Default::default() } @@ -36,18 +37,6 @@ impl Server { self } - #[doc(hidden)] - #[deprecated(note = "renamed to `with_publisher`")] - pub fn with_publish(self, publish: origin::Consumer) -> Self { - self.with_publisher(publish) - } - - #[doc(hidden)] - #[deprecated(note = "renamed to `with_subscriber`")] - pub fn with_consume(self, subscribe: origin::Producer) -> Self { - self.with_subscriber(subscribe) - } - /// Attach a tier-scoped [`stats::Handle`]. Per-broadcast and per-subscription /// counters will be bumped through this handle for the lifetime of the session. /// Pass [`stats::Handle::default`] (a no-op handle) to opt out. @@ -61,6 +50,8 @@ impl Server { self.with_publisher(&origin).with_subscriber(origin) } + /// Restrict which protocol versions to accept, in preference order. + /// Defaults to every version this crate supports. pub fn with_versions(mut self, versions: Versions) -> Self { self.versions = versions; self diff --git a/rs/moq-net/src/setup.rs b/rs/moq-net/src/setup.rs index 2129c76a6e..8b436dd00c 100644 --- a/rs/moq-net/src/setup.rs +++ b/rs/moq-net/src/setup.rs @@ -12,7 +12,7 @@ const SERVER_SETUP: u8 = 0x21; /// Draft-17 unified SETUP message type (varint 0x2F00) pub(crate) const SETUP_V17: u64 = 0x2F00; -/// Draft-17+ unified SETUP message — same encoding for both client and server. +/// Draft-17+ unified SETUP message, with the same encoding for both client and server. #[derive(Debug, Clone)] pub struct Setup { pub parameters: Bytes, diff --git a/rs/moq-net/src/version.rs b/rs/moq-net/src/version.rs index 5f3fc9ab3a..00d0e91341 100644 --- a/rs/moq-net/src/version.rs +++ b/rs/moq-net/src/version.rs @@ -51,7 +51,9 @@ pub(crate) const ALPN_19: &str = "moqt-19"; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] #[non_exhaustive] pub enum Version { + /// A `moq-lite` draft, the simplified protocol this project specifies. Lite(lite::Version), + /// An IETF `moq-transport` draft. Ietf(ietf::Version), } @@ -288,10 +290,12 @@ impl Versions { self.0.contains(&version).then_some(version) } + /// Returns `true` if the set includes this version. pub fn contains(&self, version: &Version) -> bool { self.0.contains(version) } + /// Iterate the set in preference order, most preferred first. pub fn iter(&self) -> impl Iterator { self.0.iter() } diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index eab70fea52..90967cb67d 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -62,7 +62,9 @@ impl Connection { let authorized = match role { Some(moq_net::Role::Publisher) => publish.is_some(), Some(moq_net::Role::Subscriber) => subscribe.is_some(), - None => publish.is_some() || subscribe.is_some(), + // Bidirectional or an unrecognized future role: require the token to grant + // something, and let the per-direction checks apply once it's used. + None | Some(_) => publish.is_some() || subscribe.is_some(), }; if !authorized { let _ = self.request.close(http::StatusCode::FORBIDDEN.as_u16()).await; @@ -102,7 +104,8 @@ impl Connection { let (publish, subscribe) = match role { Some(moq_net::Role::Publisher) => (publish, None), Some(moq_net::Role::Subscriber) => (None, subscribe), - None => (publish, subscribe), + // Bidirectional or an unrecognized future role: keep whatever the token grants. + None | Some(_) => (publish, subscribe), }; // Accept the connection. diff --git a/rs/moq-token/src/error.rs b/rs/moq-token/src/error.rs index 27d2b12cf0..a50200f768 100644 --- a/rs/moq-token/src/error.rs +++ b/rs/moq-token/src/error.rs @@ -1,5 +1,6 @@ /// Errors related to key configuration and cryptographic operations. #[derive(Debug, thiserror::Error)] +#[non_exhaustive] pub enum KeyError { #[error("invalid algorithm for key type")] InvalidAlgorithm, @@ -43,6 +44,7 @@ pub enum KeyError { /// Top-level error type for moq-token. #[derive(Debug, thiserror::Error)] +#[non_exhaustive] pub enum Error { #[error(transparent)] Key(#[from] KeyError), From 783f49c7f9dd404993c57ba5e7467414ba79c2ba Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 20 Jul 2026 13:23:11 -0700 Subject: [PATCH 2/6] refactor(libmoq)!: unify C ABI release verbs and catalog accessor prefix Give the C ABI one verb per concept before the 0.4.0 bump. `_free` now always means "release a value handed to the caller" and `_close` always means "stop a task or subscription", with no overlap: rename moq_consume_{frame,track_frame,datagram,json_value}_close to _free. Move the catalog section accessors into the moq_consume_catalog_* family that owns the handle's lifecycle: moq_catalog_section_count/_at and moq_catalog_get_section become moq_consume_catalog_section_count/_at and moq_consume_catalog_section. Regenerates moq.h and updates cpp/obs, the C smoke client, and the docs. Co-Authored-By: Claude Fable 5 --- cpp/obs/src/moq-source.cpp | 30 +++++++++--------- doc/concept/layer/hang.md | 2 +- doc/lib/c/index.md | 2 +- rs/libmoq/README.md | 4 +-- rs/libmoq/src/api.rs | 52 ++++++++++++++++---------------- rs/libmoq/src/test.rs | 45 +++++++++++++++------------ test/smoke/clients/c/subscribe.c | 2 +- 7 files changed, 71 insertions(+), 66 deletions(-) diff --git a/cpp/obs/src/moq-source.cpp b/cpp/obs/src/moq-source.cpp index 598d35cc3a..5bf653b37a 100644 --- a/cpp/obs/src/moq-source.cpp +++ b/cpp/obs/src/moq-source.cpp @@ -539,7 +539,7 @@ static void on_video_frame(void *user_data, int32_t frame_id) if (ctx->shutting_down.load() || ctx->consume < 0) { // Shutting down or disconnected: drop the frame. pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } pthread_mutex_unlock(&ctx->mutex); @@ -1020,7 +1020,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) { // Fast path: check atomic flag before taking lock if (ctx->shutting_down.load()) { - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1029,7 +1029,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) // Double-check after acquiring lock (may have changed) if (ctx->shutting_down.load()) { pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1037,7 +1037,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) // Note: sws_ctx and frame_buffer may be NULL on first frame - they're created dynamically if (!ctx->codec_ctx) { pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1046,7 +1046,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) if (moq_consume_frame(frame_id, &frame_data) < 0) { LOG_ERROR("Failed to get frame data"); pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1058,7 +1058,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) ctx->frames_waiting_for_keyframe); } pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1079,7 +1079,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) AVPacket *packet = av_packet_alloc(); if (!packet) { pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1110,7 +1110,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) } } pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1118,7 +1118,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) AVFrame *frame = av_frame_alloc(); if (!frame) { pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1143,7 +1143,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) } av_frame_free(&frame); pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1173,7 +1173,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) LOG_ERROR("Invalid decoded frame dimensions: %dx%d", frame->width, frame->height); av_frame_free(&frame); pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1182,7 +1182,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) LOG_ERROR("Invalid decoded frame pixel format: %d", decoded_pix_fmt); av_frame_free(&frame); pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1203,7 +1203,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) : "unknown"); av_frame_free(&frame); pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1216,7 +1216,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) sws_freeContext(new_sws_ctx); av_frame_free(&frame); pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); return; } @@ -1252,7 +1252,7 @@ static void moq_source_decode_frame(struct moq_source *ctx, int32_t frame_id) av_frame_free(&frame); pthread_mutex_unlock(&ctx->mutex); - moq_consume_frame_close(frame_id); + moq_consume_frame_free(frame_id); } // Registration function diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index 67c84e623c..b905cefb0a 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -98,7 +98,7 @@ The catalog is a JSON document published through the merge-patch snapshot helper A base consumer ignores them; an extension reads its own section and treats its absence as "not present". In TypeScript, build an extended schema with `z.extend(Catalog.RootSchema, { scte35: ... })`. In Rust, either flatten the catalog into your own struct with `#[serde(flatten)]` for typed access, or read sections untyped from an `Extra` catalog, which keeps unknown keys as raw JSON (`catalog.section("scte35")`). The `()` default drops sections it doesn't model. - The FFI bindings always use the untyped form, one JSON string per section keyed by name (`catalog.sections["scte35"]` in Python, `moq_catalog_get_section()` / `moq_catalog_section_at()` in C). + The FFI bindings always use the untyped form, one JSON string per section keyed by name (`catalog.sections["scte35"]` in Python, `moq_consume_catalog_section()` / `moq_consume_catalog_section_at()` in C). - **Writing**: the catalog producer holds one shared document. Each owner edits only its own keys and publishes: `producer.mutate(c => { c.scte35 = ... })` in TypeScript; the `Deref`/`DerefMut` lock guard from `producer.lock()` for a typed Rust extension, or `producer.set_section("scte35", value)` for an untyped one; `broadcast.set_catalog_section("scte35", value)` in Python; `moq_publish_catalog_section()` in C. Every edit starts from the latest value, so the base media sections and any extension sections compose instead of clobbering one another. diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 7a3791e7c4..b748827b56 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -202,7 +202,7 @@ moq_publish_json_snapshot_update(json, value, strlen(value)); // Subscribe: on_value fires with a value ID for each update; read it, then release it. int32_t task = moq_consume_json_snapshot(consume, "status", strlen("status"), &config, on_value, user_data); -// In on_value: struct moq_json_value v; moq_consume_json_value(id, &v); ... moq_consume_json_value_close(id); +// In on_value: struct moq_json_value v; moq_consume_json_value(id, &v); ... moq_consume_json_value_free(id); ``` `compression` must match on the producer and subscriber. The consumer callback follows the same lifetime contract as every other (see above): release `user_data` on the terminal `<= 0` call. diff --git a/rs/libmoq/README.md b/rs/libmoq/README.md index 64fa5b01e7..b9e469d1e4 100644 --- a/rs/libmoq/README.md +++ b/rs/libmoq/README.md @@ -74,10 +74,10 @@ int32_t moq_consume_audio_close(uint32_t track); // Consuming: Frames int32_t moq_consume_frame(uint32_t frame, moq_frame *dst); -int32_t moq_consume_frame_close(uint32_t frame); +int32_t moq_consume_frame_free(uint32_t frame); int32_t moq_consume_track(uint32_t broadcast, const char *name, uintptr_t name_len, void (*on_frame)(void *user_data, int32_t frame), void *user_data); int32_t moq_consume_track_frame(uint32_t frame, moq_frame *dst); -int32_t moq_consume_track_frame_close(uint32_t frame); +int32_t moq_consume_track_frame_free(uint32_t frame); int32_t moq_consume_track_close(uint32_t track); ``` diff --git a/rs/libmoq/src/api.rs b/rs/libmoq/src/api.rs index 03da7c0210..8f464a28d1 100644 --- a/rs/libmoq/src/api.rs +++ b/rs/libmoq/src/api.rs @@ -907,8 +907,8 @@ pub unsafe extern "C" fn moq_publish_audio_remove(broadcast: u32, name: *const c /// Set (or replace) a top-level application catalog section by name. /// -/// This is the producer counterpart to [moq_catalog_get_section] / -/// [moq_catalog_section_at]: it writes an arbitrary top-level JSON key into the +/// This is the producer counterpart to [moq_consume_catalog_section] / +/// [moq_consume_catalog_section_at]: it writes an arbitrary top-level JSON key into the /// catalog of a broadcast created with [moq_origin_publish], beyond the /// `video`/`audio` keys owned by the media pipeline. Calling it again with the /// same name replaces the section. The updated catalog is published to @@ -1354,12 +1354,12 @@ pub unsafe extern "C" fn moq_consume_audio_config(catalog: u32, index: u32, dst: /// Number of untyped application catalog sections in a catalog snapshot. /// /// These are the top-level catalog keys beyond `video`/`audio`, carried through -/// verbatim. Iterate them by index with [moq_catalog_section_at], or look one up -/// directly by name with [moq_catalog_get_section]. +/// verbatim. Iterate them by index with [moq_consume_catalog_section_at], or look one up +/// directly by name with [moq_consume_catalog_section]. /// /// Returns the count (>= 0) on success, or a negative code on failure. #[unsafe(no_mangle)] -pub extern "C" fn moq_catalog_section_count(catalog: u32) -> i32 { +pub extern "C" fn moq_consume_catalog_section_count(catalog: u32) -> i32 { ffi::enter(move || { let catalog = ffi::parse_id(catalog)?; State::lock().consume.catalog_section_count(catalog) @@ -1369,7 +1369,7 @@ pub extern "C" fn moq_catalog_section_count(catalog: u32) -> i32 { /// Query an application catalog section by index, keyed by name. /// /// Fills `dst` with the section's name and JSON value at `index`, in the range -/// `[0, moq_catalog_section_count)`. Both pointers borrow the snapshot's storage +/// `[0, moq_consume_catalog_section_count)`. Both pointers borrow the snapshot's storage /// and stay valid until it is freed with [moq_consume_catalog_free]. /// /// Returns a zero on success, or a negative code on failure (e.g. `index` out of @@ -1379,7 +1379,7 @@ pub extern "C" fn moq_catalog_section_count(catalog: u32) -> i32 { /// - The caller must ensure that `dst` is a valid pointer to a [moq_section] struct. /// - The caller must ensure that `dst` is not used after [moq_consume_catalog_free] is called. #[unsafe(no_mangle)] -pub unsafe extern "C" fn moq_catalog_section_at(catalog: u32, index: u32, dst: *mut moq_section) -> i32 { +pub unsafe extern "C" fn moq_consume_catalog_section_at(catalog: u32, index: u32, dst: *mut moq_section) -> i32 { ffi::enter(move || { let catalog = ffi::parse_id(catalog)?; let index = index as usize; @@ -1402,7 +1402,7 @@ pub unsafe extern "C" fn moq_catalog_section_at(catalog: u32, index: u32, dst: * /// - The caller must ensure that `dst` is a valid pointer to a [moq_string] struct. /// - The caller must ensure that `dst` is not used after [moq_consume_catalog_free] is called. #[unsafe(no_mangle)] -pub unsafe extern "C" fn moq_catalog_get_section( +pub unsafe extern "C" fn moq_consume_catalog_section( catalog: u32, name: *const c_char, name_len: usize, @@ -1509,7 +1509,7 @@ pub extern "C" fn moq_consume_audio_close(track: u32) -> i32 { /// Read the payload of a frame as a single contiguous slice. /// /// Frames are not chunked; the entire payload is delivered through `dst.payload` / -/// `dst.payload_size` in one call. The pointer is valid until [`moq_consume_frame_close`] +/// `dst.payload_size` in one call. The pointer is valid until [`moq_consume_frame_free`] /// is called for this frame. /// /// Returns a zero on success, or a negative code on failure. @@ -1525,11 +1525,11 @@ pub unsafe extern "C" fn moq_consume_frame(frame: u32, dst: *mut moq_frame) -> i }) } -/// Close a frame and clean up its resources. +/// Free a decoded frame delivered via a [moq_consume_video] or [moq_consume_audio] callback. /// /// Returns a zero on success, or a negative code on failure. #[unsafe(no_mangle)] -pub extern "C" fn moq_consume_frame_close(frame: u32) -> i32 { +pub extern "C" fn moq_consume_frame_free(frame: u32) -> i32 { ffi::enter(move || { let frame = ffi::parse_id(frame)?; State::lock().consume.frame_close(frame) @@ -1556,7 +1556,7 @@ pub extern "C" fn moq_consume_close(consume: u32) -> i32 { /// `on_frame` is never called again and `user_data` is never touched again, so /// release `user_data` there. The terminal callback fires even after /// [moq_consume_track_close]. Read each frame with [moq_consume_track_frame] and -/// release it with [moq_consume_track_frame_close]. Pass NULL for `subscription` +/// release it with [moq_consume_track_frame_free]. Pass NULL for `subscription` /// to use moq-net defaults. /// /// Returns a non-zero handle to the track on success, or a negative code on failure. @@ -1603,7 +1603,7 @@ pub unsafe extern "C" fn moq_consume_track_update(track: u32, subscription: *con /// Read a raw frame's payload delivered via the [moq_consume_track] callback. /// /// Fills `dst.payload` / `dst.payload_size`; the pointer is valid until the -/// frame is released with [moq_consume_frame_close]. `dst.timestamp_us` is the +/// frame is released with [moq_consume_frame_free]. `dst.timestamp_us` is the /// frame presentation timestamp in microseconds. `dst.keyframe` is reported as /// false because raw tracks do not parse codec metadata. /// @@ -1620,11 +1620,11 @@ pub unsafe extern "C" fn moq_consume_track_frame(frame: u32, dst: *mut moq_frame }) } -/// Close a raw frame and clean up its resources. +/// Free a raw frame delivered via the [moq_consume_track] callback, releasing its payload. /// /// Returns a zero on success, or a negative code on failure. #[unsafe(no_mangle)] -pub extern "C" fn moq_consume_track_frame_close(frame: u32) -> i32 { +pub extern "C" fn moq_consume_track_frame_free(frame: u32) -> i32 { ffi::enter(move || { let frame = ffi::parse_id(frame)?; State::lock().consume.raw_frame_close(frame) @@ -1637,7 +1637,7 @@ pub extern "C" fn moq_consume_track_frame_close(frame: u32) -> i32 { /// Does NOT free `user_data`; the [moq_consume_track] `on_frame` callback still /// fires once more with a terminal `0` (or a negative error), which is where /// `user_data` should be released. Frames already delivered via the callback -/// remain valid until released with [moq_consume_track_frame_close]. +/// remain valid until released with [moq_consume_track_frame_free]. #[unsafe(no_mangle)] pub extern "C" fn moq_consume_track_close(track: u32) -> i32 { ffi::enter(move || { @@ -1654,7 +1654,7 @@ pub extern "C" fn moq_consume_track_close(track: u32) -> i32 { /// terminal (`<= 0`) callback, `on_datagram` is never called again and `user_data` is never /// touched again, so release `user_data` there. The terminal callback fires even after /// [moq_consume_datagrams_close]. Read each datagram with [moq_consume_datagram] and release -/// it with [moq_consume_datagram_close]. Datagrams arrive only over datagram-capable +/// it with [moq_consume_datagram_free]. Datagrams arrive only over datagram-capable /// transports and lite-05 or newer moq-lite; there is no stream fallback. /// /// Returns a non-zero handle to the subscription on success, or a negative code on failure. @@ -1681,7 +1681,7 @@ pub unsafe extern "C" fn moq_consume_datagrams( /// Read a datagram delivered via the [moq_consume_datagrams] callback. /// /// Fills `dst.payload` / `dst.payload_size` (valid until the datagram is released with -/// [moq_consume_datagram_close]), plus `dst.timestamp_us` and `dst.sequence`. +/// [moq_consume_datagram_free]), plus `dst.timestamp_us` and `dst.sequence`. /// /// Returns a zero on success, or a negative code on failure. /// @@ -1696,11 +1696,11 @@ pub unsafe extern "C" fn moq_consume_datagram(datagram: u32, dst: *mut moq_datag }) } -/// Close a datagram and clean up its resources. +/// Free a datagram delivered via the [moq_consume_datagrams] callback, releasing its payload. /// /// Returns a zero on success, or a negative code on failure. #[unsafe(no_mangle)] -pub extern "C" fn moq_consume_datagram_close(datagram: u32) -> i32 { +pub extern "C" fn moq_consume_datagram_free(datagram: u32) -> i32 { ffi::enter(move || { let datagram = ffi::parse_id(datagram)?; State::lock().consume.datagram_close(datagram) @@ -1712,7 +1712,7 @@ pub extern "C" fn moq_consume_datagram_close(datagram: u32) -> i32 { /// Returns immediately: zero on success, or a negative code if already closed. Does NOT free /// `user_data`; the [moq_consume_datagrams] `on_datagram` callback still fires once more with a /// terminal `0` (or a negative error), which is where `user_data` should be released. Datagrams -/// already delivered via the callback remain valid until released with [moq_consume_datagram_close]. +/// already delivered via the callback remain valid until released with [moq_consume_datagram_free]. #[unsafe(no_mangle)] pub extern "C" fn moq_consume_datagrams_close(task: u32) -> i32 { ffi::enter(move || { @@ -1727,7 +1727,7 @@ pub extern "C" fn moq_consume_datagrams_close(task: u32) -> i32 { /// falls behind collapses the backlog and only sees the newest. It is called exactly once more /// with a terminal `0` (track ended / closed) or a negative error, after which `user_data` is /// never touched again, so release it there. Read each value with [moq_consume_json_value] and -/// release it with [moq_consume_json_value_close]. Pass the same compression the producer used. +/// release it with [moq_consume_json_value_free]. Pass the same compression the producer used. /// /// Returns a non-zero handle to the task on success, or a negative code on failure. /// @@ -1758,7 +1758,7 @@ pub unsafe extern "C" fn moq_consume_json_snapshot( /// /// `on_value` is called with a positive value ID for each record, in order, then once more with /// a terminal `0` or negative error where `user_data` should be released. Read each value with -/// [moq_consume_json_value] and release it with [moq_consume_json_value_close]. +/// [moq_consume_json_value] and release it with [moq_consume_json_value_free]. /// /// Returns a non-zero handle to the task on success, or a negative code on failure. /// @@ -1787,7 +1787,7 @@ pub unsafe extern "C" fn moq_consume_json_stream( /// Read a JSON value delivered via a [moq_consume_json_snapshot] or [moq_consume_json_stream] callback. /// /// Fills `dst.json` / `dst.json_len`; the pointer is valid until the value is released with -/// [moq_consume_json_value_close]. +/// [moq_consume_json_value_free]. /// /// Returns a zero on success, or a negative code on failure. /// @@ -1806,7 +1806,7 @@ pub unsafe extern "C" fn moq_consume_json_value(value: u32, dst: *mut moq_json_v /// /// Returns a zero on success, or a negative code on failure. #[unsafe(no_mangle)] -pub extern "C" fn moq_consume_json_value_close(value: u32) -> i32 { +pub extern "C" fn moq_consume_json_value_free(value: u32) -> i32 { ffi::enter(move || { let value = ffi::parse_id(value)?; State::lock().consume.json_value_close(value) @@ -1818,7 +1818,7 @@ pub extern "C" fn moq_consume_json_value_close(value: u32) -> i32 { /// Returns immediately: zero on success, or a negative code if already closed. Does NOT free /// `user_data`; the `on_value` callback still fires once more with a terminal `0` (or a negative /// error), which is where `user_data` should be released. Values already delivered remain valid -/// until released with [moq_consume_json_value_close]. +/// until released with [moq_consume_json_value_free]. #[unsafe(no_mangle)] pub extern "C" fn moq_consume_json_close(task: u32) -> i32 { ffi::enter(move || { diff --git a/rs/libmoq/src/test.rs b/rs/libmoq/src/test.rs index 7e6f6ebe4d..5cc76ef9ef 100644 --- a/rs/libmoq/src/test.rs +++ b/rs/libmoq/src/test.rs @@ -431,7 +431,7 @@ fn catalog_section_roundtrip() { let catalog_id = id(catalog_cb.recv()); // Both sections come back; iterate by index to find each by name. - let count = moq_catalog_section_count(catalog_id); + let count = moq_consume_catalog_section_count(catalog_id); assert_eq!(count, 2, "expected two sections, got {count}"); let mut found_a = false; @@ -443,7 +443,10 @@ fn catalog_section_roundtrip() { json: std::ptr::null(), json_len: 0, }; - assert_eq!(unsafe { moq_catalog_section_at(catalog_id, index, &mut section) }, 0); + assert_eq!( + unsafe { moq_consume_catalog_section_at(catalog_id, index, &mut section) }, + 0 + ); let name = unsafe { std::slice::from_raw_parts(section.name.cast::(), section.name_len) }; let json = unsafe { std::slice::from_raw_parts(section.json.cast::(), section.json_len) }; match name { @@ -466,7 +469,7 @@ fn catalog_section_roundtrip() { len: 0, }; assert_eq!( - unsafe { moq_catalog_get_section(catalog_id, name_a.as_ptr() as *const c_char, name_a.len(), &mut value) }, + unsafe { moq_consume_catalog_section(catalog_id, name_a.as_ptr() as *const c_char, name_a.len(), &mut value) }, 0 ); let got = unsafe { std::slice::from_raw_parts(value.data.cast::(), value.len) }; @@ -475,8 +478,9 @@ fn catalog_section_roundtrip() { // A missing section fails. let missing = b"nope"; assert!( - unsafe { moq_catalog_get_section(catalog_id, missing.as_ptr() as *const c_char, missing.len(), &mut value) } - < 0, + unsafe { + moq_consume_catalog_section(catalog_id, missing.as_ptr() as *const c_char, missing.len(), &mut value) + } < 0, "missing section should fail" ); @@ -487,12 +491,13 @@ fn catalog_section_roundtrip() { ); let catalog_id2 = id(catalog_cb.recv()); assert_eq!( - moq_catalog_section_count(catalog_id2), + moq_consume_catalog_section_count(catalog_id2), 1, "one section should remain after remove" ); assert!( - unsafe { moq_catalog_get_section(catalog_id2, name_a.as_ptr() as *const c_char, name_a.len(), &mut value) } < 0, + unsafe { moq_consume_catalog_section(catalog_id2, name_a.as_ptr() as *const c_char, name_a.len(), &mut value) } + < 0, "removed section should be gone" ); @@ -635,7 +640,7 @@ fn raw_track_publish_consume() { assert_eq!(received, payload); assert_eq!(frame.timestamp_us, timestamp_us); assert!(!frame.keyframe, "raw frames have no keyframe flag"); - assert_eq!(moq_consume_track_frame_close(frame_id), 0); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); // Multi-frame group via the explicit group API. let group = id(moq_publish_track_group(track)); @@ -660,7 +665,7 @@ fn raw_track_publish_consume() { let received = unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }; assert_eq!(received, expected); assert_eq!(frame.timestamp_us, timestamp_us); - assert_eq!(moq_consume_track_frame_close(frame_id), 0); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); } assert_eq!(moq_consume_track_close(consumer), 0); @@ -724,7 +729,7 @@ fn raw_track_datagram_publish_consume() { assert_eq!(received, payload); assert_eq!(datagram.timestamp_us, 120_000); assert_eq!(datagram.sequence, sequence); - assert_eq!(moq_consume_datagram_close(dg_id), 0); + assert_eq!(moq_consume_datagram_free(dg_id), 0); assert_eq!(moq_consume_datagrams_close(consumer), 0); // The task delivers one final terminal callback after close; drain it @@ -830,7 +835,7 @@ fn raw_track_subscription_options_and_update() { let received = unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }; assert_eq!(received, b"one"); assert_eq!(frame.timestamp_us, 20_000); - assert_eq!(moq_consume_track_frame_close(frame_id), 0); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); let update = moq_subscription { group_end: 2, @@ -849,7 +854,7 @@ fn raw_track_subscription_options_and_update() { let received = unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }; assert_eq!(received, b"two"); assert_eq!(frame.timestamp_us, 40_000); - assert_eq!(moq_consume_track_frame_close(frame_id), 0); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); assert_eq!(moq_consume_track_close(consumer), 0); assert_eq!(frame_cb.recv_terminal(), 0); @@ -909,7 +914,7 @@ fn json_snapshot_publish_consume() { serde_json::from_slice::(received).unwrap(), serde_json::from_str::(expected).unwrap() ); - assert_eq!(moq_consume_json_value_close(value_id), 0); + assert_eq!(moq_consume_json_value_free(value_id), 0); } assert_eq!(moq_consume_json_close(consumer), 0); @@ -972,7 +977,7 @@ fn json_stream_publish_consume() { serde_json::from_slice::(received).unwrap(), serde_json::from_str::(expected).unwrap() ); - assert_eq!(moq_consume_json_value_close(value_id), 0); + assert_eq!(moq_consume_json_value_free(value_id), 0); } assert_eq!(moq_consume_json_close(consumer), 0); @@ -991,7 +996,7 @@ fn close_invalid_or_zero_ids() { assert!(moq_session_close(9999) < 0); assert!(moq_publish_finish(9999) < 0); assert!(moq_consume_close(9999) < 0); - assert!(moq_consume_frame_close(9999) < 0); + assert!(moq_consume_frame_free(9999) < 0); assert!(moq_origin_close(0) < 0); assert!(moq_session_close(0) < 0); @@ -1091,8 +1096,8 @@ fn double_close_all_resource_types() { ); let frame_id = id(frame_cb.recv()); - assert_eq!(moq_consume_frame_close(frame_id), 0); - assert!(moq_consume_frame_close(frame_id) < 0); + assert_eq!(moq_consume_frame_free(frame_id), 0); + assert!(moq_consume_frame_free(frame_id) < 0); assert_eq!(moq_consume_audio_close(track), 0); assert_eq!(frame_cb.recv_terminal(), 0, "audio close delivers terminal 0"); @@ -1281,7 +1286,7 @@ fn local_publish_consume() { let received = unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }; assert_eq!(received, payload, "frame payload should match"); - assert_eq!(moq_consume_frame_close(frame_id), 0); + assert_eq!(moq_consume_frame_free(frame_id), 0); assert_eq!(moq_consume_audio_close(track), 0); assert_eq!(frame_cb.recv_terminal(), 0, "audio close delivers terminal 0"); assert_eq!(moq_consume_catalog_free(catalog_id), 0); @@ -1473,7 +1478,7 @@ fn video_publish_consume() { assert_eq!(frame.timestamp_us, 0); assert!(frame.payload_size > 0, "frame should have payload data"); - assert_eq!(moq_consume_frame_close(frame_id), 0); + assert_eq!(moq_consume_frame_free(frame_id), 0); assert_eq!(moq_consume_video_close(track), 0); assert_eq!(frame_cb.recv_terminal(), 0, "video close delivers terminal 0"); assert_eq!(moq_consume_catalog_free(catalog_id), 0); @@ -1637,7 +1642,7 @@ fn multiple_frames_ordering() { let expected = format!("frame-{i}"); assert_eq!(received, expected.as_bytes(), "frame {i} has wrong payload"); - assert_eq!(moq_consume_frame_close(frame_id), 0); + assert_eq!(moq_consume_frame_free(frame_id), 0); } assert_eq!(moq_consume_audio_close(track), 0); diff --git a/test/smoke/clients/c/subscribe.c b/test/smoke/clients/c/subscribe.c index a23a82d5fa..bd21c06033 100644 --- a/test/smoke/clients/c/subscribe.c +++ b/test/smoke/clients/c/subscribe.c @@ -45,7 +45,7 @@ static void on_frame(void *ud, int32_t frame) { pthread_cond_signal(&c->cv); pthread_mutex_unlock(&c->mu); } - moq_consume_frame_close((uint32_t)frame); + moq_consume_frame_free((uint32_t)frame); } // Delivers the broadcast handle once it's announced, then once more with a From f918459e3d57e1edb9c5017d383a8487f68b4504 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 20 Jul 2026 13:40:29 -0700 Subject: [PATCH 3/6] refactor: pre-bump polish for moq-stats, hang, moq-mux moq-stats: rename the producer config Config -> ProducerConfig so it reads unambiguously next to ConsumerConfig before 0.1.0 freezes the name; fix the traffic_track doc that mixed the publisher/subscriber roles. hang: accept the pre-0.20 displayRatioWidth/Height catalog keys as serde aliases for the renamed displayAspectWidth/Height fields, so an old catalog still decodes its aspect ratio instead of silently dropping it. Emitted names stay the new ones. Adds a legacy-decode regression test. moq-mux: rename the stuttering container::ts::catalog::Catalog trait to Carrier (it carries the MPEG-TS section) to clear the foo::foo::Foo path and the three-way Catalog name collision; document that Container::poll_read/read may yield an empty batch as a poll-again state and only None ends the group. Co-Authored-By: Claude Fable 5 --- rs/hang/src/catalog/video/mod.rs | 19 +++++++++++++++++ rs/moq-mux/src/container/consumer.rs | 3 ++- rs/moq-mux/src/container/mod.rs | 16 +++++++++++---- rs/moq-mux/src/container/ts/catalog.rs | 8 ++++---- rs/moq-mux/src/container/ts/export.rs | 4 ++-- rs/moq-mux/src/container/ts/export_test.rs | 2 +- rs/moq-mux/src/container/ts/import.rs | 24 +++++++++++----------- rs/moq-mux/src/import/container.rs | 12 +++++------ rs/moq-relay/src/stats.rs | 6 +++--- rs/moq-stats/src/consume.rs | 4 ++-- rs/moq-stats/src/lib.rs | 8 ++++---- rs/moq-stats/src/produce.rs | 23 +++++++++++---------- 12 files changed, 79 insertions(+), 50 deletions(-) diff --git a/rs/hang/src/catalog/video/mod.rs b/rs/hang/src/catalog/video/mod.rs index 1b4bbede41..e26a8bf79f 100644 --- a/rs/hang/src/catalog/video/mod.rs +++ b/rs/hang/src/catalog/video/mod.rs @@ -129,7 +129,12 @@ pub struct VideoConfig { /// /// This allows you to stretch/shrink pixels of the video. /// If not provided, the display aspect ratio is 1:1 + /// + /// The `displayRatio*` aliases decode catalogs from publishers predating the + /// rename to `displayAspect*`; the current name is what we emit. + #[serde(alias = "displayRatioWidth")] pub display_aspect_width: Option, + #[serde(alias = "displayRatioHeight")] pub display_aspect_height: Option, // TODO color space @@ -222,4 +227,18 @@ mod test { assert!(encoded.get("displayRatioWidth").is_none()); assert!(encoded.get("displayRatioHeight").is_none()); } + + #[test] + fn decodes_legacy_display_ratio_keys() { + // A catalog serialized by a pre-0.20 publisher used displayRatio*; the + // alias keeps the aspect ratio from being silently dropped. + let json = serde_json::json!({ + "codec": "avc1.640028", + "displayRatioWidth": 16, + "displayRatioHeight": 9, + }); + let config: VideoConfig = serde_json::from_value(json).expect("failed to decode legacy keys"); + assert_eq!(config.display_aspect_width, Some(16)); + assert_eq!(config.display_aspect_height, Some(9)); + } } diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 3bb0cf7036..395d93a5b0 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -578,7 +578,8 @@ impl GroupBuffer { if !ready!(self.buffer_once(waiter, format)?) { return Poll::Ready(Ok(false)); } - // poll_read returned Some(vec![]) — loop and try again + // poll_read returned Some(vec![]): a wire frame decoded to no media + // frames, so loop and try again. } } diff --git a/rs/moq-mux/src/container/mod.rs b/rs/moq-mux/src/container/mod.rs index f29e53df3d..843bfdea31 100644 --- a/rs/moq-mux/src/container/mod.rs +++ b/rs/moq-mux/src/container/mod.rs @@ -94,16 +94,24 @@ pub trait Container { fn write(&self, group: &mut moq_net::group::Producer, frames: &[Frame]) -> Result<(), Self::Error>; /// Poll the next moq-lite frame from `group` and decode it into media - /// frames. Returns `Ok(None)` when the group has ended. A single call - /// may produce multiple media frames (e.g. all samples in a CMAF - /// fragment). + /// frames. A single call may produce multiple media frames (e.g. all samples + /// in a CMAF fragment). + /// + /// Only `Ok(None)` signals the end of the group. `Ok(Some(batch))` may carry + /// an empty `batch`: a wire frame was consumed but decoded to no media frames + /// (e.g. a CMAF fragment with zero samples). That is not end-of-group; poll + /// again for the next batch. Callers accumulating frames must not treat an + /// empty batch as completion. fn poll_read( &self, group: &mut moq_net::group::Consumer, waiter: &kio::Waiter, ) -> Poll>, Self::Error>>; - /// Async wrapper around [`Self::poll_read`]. + /// Async wrapper around [`Self::poll_read`]. Carries the same contract: only + /// `Ok(None)` ends the group, and `Ok(Some(batch))` may hand back an empty + /// `batch` (poll again for more), so a caller loop must key completion off + /// `None`, not an empty batch. fn read( &self, group: &mut moq_net::group::Consumer, diff --git a/rs/moq-mux/src/container/ts/catalog.rs b/rs/moq-mux/src/container/ts/catalog.rs index 6e2a1014dd..b4c4809442 100644 --- a/rs/moq-mux/src/container/ts/catalog.rs +++ b/rs/moq-mux/src/container/ts/catalog.rs @@ -148,7 +148,7 @@ impl CatalogExt for Ext {} /// /// Implement this for an application extension to compose MPEG-TS carriage with /// additional sections. -pub trait Catalog: CatalogExt { +pub trait Carrier: CatalogExt { /// The section to record MPEG-TS details into, or `None` for an extension that /// doesn't carry them. /// @@ -158,7 +158,7 @@ pub trait Catalog: CatalogExt { fn mpegts_mut(&mut self) -> Option<&mut Mpegts>; } -impl Catalog for () { +impl Carrier for () { fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { None } @@ -166,13 +166,13 @@ impl Catalog for () { // The untyped passthrough carries no typed mpegts section (a TS importer driving an `Extra` // catalog records verbatim streams as raw JSON sections, not the typed `Mpegts` view). -impl Catalog for crate::catalog::hang::Extra { +impl Carrier for crate::catalog::hang::Extra { fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { None } } -impl Catalog for Ext { +impl Carrier for Ext { fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { Some(&mut self.mpegts) } diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 09a90c6f2c..f2bf99a6e8 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -53,7 +53,7 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500); /// The leading PAT/PMT rides on the first frame (so it inherits a real /// timestamp), and is re-emitted at video keyframes and periodically for /// mid-stream tune-in. Returns `None` when the broadcast ends. -pub struct Export { +pub struct Export { source: crate::Source, catalog: Option>, latency: Duration, @@ -179,7 +179,7 @@ impl Export { } } -impl Export { +impl Export { /// Shared constructor. The public entry points each live on a concrete /// `Export` impl that pins `E`, so the extension is chosen by which one you call. async fn build(source: crate::Source, catalog_format: CatalogFormat) -> Result { diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 89eefb04f5..affed5becd 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -64,7 +64,7 @@ async fn drain(consumer: moq_net::broadcast::Consumer) -> BytesMut { } /// `drain` for an exporter built with an explicit catalog extension. -async fn drain_with(mut exporter: Export) -> BytesMut { +async fn drain_with(mut exporter: Export) -> BytesMut { let mut out = BytesMut::new(); // `while let Ok` stops on the first timeout (`Pending`: no more output). while let Ok(res) = tokio::time::timeout(std::time::Duration::from_secs(1), exporter.next()).await { diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 75f1cbdfd8..7e5ceff27a 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -40,7 +40,7 @@ use moq_net::Timestamp; /// by a program-level 'CUEI' registration descriptor, and other private sections) /// are intercepted before the reader and reassembled. With a base `Catalog<()>` /// they're logged and dropped instead. -pub struct Import { +pub struct Import { broadcast: moq_net::broadcast::Producer, catalog: crate::catalog::Producer, @@ -105,7 +105,7 @@ pub struct Import { media_unwrap: PtsUnwrap, } -impl Import { +impl Import { pub fn new(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { let feed = Feed::default(); // A long-lived producer handle for catalog edits (mpegts sections, later PMTs); the passed @@ -631,7 +631,7 @@ fn to_descriptors(descriptors: &[mpeg2ts::ts::Descriptor]) -> Vec( +fn register_verbatim( broadcast: &mut moq_net::broadcast::Producer, catalog: &mut crate::catalog::Producer, pid: u16, @@ -664,7 +664,7 @@ fn register_verbatim( } /// Remove a verbatim track's entry from the `mpegts` catalog section on drop. -fn unregister_verbatim(catalog: &mut crate::catalog::Producer, name: &str) { +fn unregister_verbatim(catalog: &mut crate::catalog::Producer, name: &str) { if let Some(mpegts) = catalog.lock().mpegts_mut() { mpegts.tracks.remove(name); } @@ -677,13 +677,13 @@ fn unregister_verbatim(catalog: &mut crate::catalog::Produc /// intercepted before the mpeg2ts reader (which would PES-parse it and abort). /// The byte-level reassembly lives in [`SectionReassembler`]; this type owns the /// track and catalog entry and stamps each section with the media clock. -struct SectionStream { +struct SectionStream { track: crate::container::Producer, catalog: crate::catalog::Producer, reassembler: SectionReassembler, } -impl SectionStream { +impl SectionStream { fn new( mut broadcast: moq_net::broadcast::Producer, mut catalog: crate::catalog::Producer, @@ -746,7 +746,7 @@ impl SectionStream { } } -impl Drop for SectionStream { +impl Drop for SectionStream { fn drop(&mut self) { let name = self.track.name().to_string(); unregister_verbatim(&mut self.catalog, &name); @@ -759,7 +759,7 @@ impl Drop for SectionStream { /// /// Unlike [`SectionStream`], these ride the normal PES reassembly path, so this /// type only stamps each PES payload with its (unwrapped) PTS and writes it. -struct VerbatimStream { +struct VerbatimStream { track: crate::container::Producer, catalog: crate::catalog::Producer, unwrap: PtsUnwrap, @@ -767,7 +767,7 @@ struct VerbatimStream { stream_id_recorded: bool, } -impl VerbatimStream { +impl VerbatimStream { fn new( mut broadcast: moq_net::broadcast::Producer, mut catalog: crate::catalog::Producer, @@ -833,7 +833,7 @@ impl VerbatimStream { } } -impl Drop for VerbatimStream { +impl Drop for VerbatimStream { fn drop(&mut self) { let name = self.track.name().to_string(); unregister_verbatim(&mut self.catalog, &name); @@ -992,7 +992,7 @@ impl SectionReassembler { } /// One elementary stream's codec importer plus PTS-unwrap state. -enum Stream { +enum Stream { H264 { split: h264::Split, import: Box>, @@ -1014,7 +1014,7 @@ enum Stream { Ignored, } -impl Stream { +impl Stream { fn write(&mut self, pending: Pending, burst: Option) -> anyhow::Result<()> { match self { Stream::H264 { split, import, unwrap } => { diff --git a/rs/moq-mux/src/import/container.rs b/rs/moq-mux/src/import/container.rs index 9841e20b6e..f84a697333 100644 --- a/rs/moq-mux/src/import/container.rs +++ b/rs/moq-mux/src/import/container.rs @@ -10,7 +10,7 @@ use crate::Result; /// The concrete container importers, shared by [`Container`] and /// [`ContainerStream`]. Containers parse their own internal framing, so a whole /// chunk and a stream chunk decode identically. -enum ContainerImpl { +enum ContainerImpl { // Boxed because it's a large struct and clippy complains about the size. Fmp4(Box>), Mkv(Box>), @@ -18,7 +18,7 @@ enum ContainerImpl { Flv(Box>), } -impl ContainerImpl { +impl ContainerImpl { fn fmp4(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { ContainerImpl::Fmp4(Box::new(crate::container::fmp4::Import::new(broadcast, reserved))) } @@ -76,11 +76,11 @@ impl ContainerImpl { /// /// Use this when the caller hands over discrete buffers (the typical case for /// files and reassembled network input). May publish more than one track. -pub struct Container { +pub struct Container { inner: ContainerImpl, } -impl Container { +impl Container { /// Create a new container importer, decoding the initial chunk. pub fn new( broadcast: moq_net::broadcast::Producer, @@ -125,11 +125,11 @@ impl Container { /// /// Use this when the caller pushes arbitrary byte chunks and the container /// recovers its own framing. May publish more than one track. -pub struct ContainerStream { +pub struct ContainerStream { inner: ContainerImpl, } -impl ContainerStream { +impl ContainerStream { /// Create a new container stream importer. pub fn new( broadcast: moq_net::broadcast::Producer, diff --git a/rs/moq-relay/src/stats.rs b/rs/moq-relay/src/stats.rs index 4cef574957..9167083ca1 100644 --- a/rs/moq-relay/src/stats.rs +++ b/rs/moq-relay/src/stats.rs @@ -69,7 +69,7 @@ pub struct StatsConfig { /// single `/node/` broadcast for the whole node. Set to 1 to /// publish a per-first-segment broadcast (e.g. per tenant), so a consumer can /// announce-scope to just that group rather than slurping every node's full - /// stats. See [`moq_stats::Config::depth`]. + /// stats. See [`moq_stats::ProducerConfig::depth`]. #[arg(long = "stats-depth", env = "MOQ_STATS_DEPTH")] pub depth: Option, } @@ -84,14 +84,14 @@ impl StatsConfig { /// stops when the last clone drops). pub fn build(&self, origin: origin::Producer) -> moq_stats::Producer { if !self.enabled.unwrap_or(false) { - return moq_stats::Producer::new(moq_stats::Config::new()); + return moq_stats::Producer::new(moq_stats::ProducerConfig::new()); } let prefix = self.prefix.clone().unwrap_or_else(|| ".stats".to_string()); let interval = Duration::from_secs(self.interval.unwrap_or(1).max(1)); let node = self.node.clone().map(PathOwned::from); let depth = self.depth.unwrap_or(0); tracing::info!(prefix, interval_secs = interval.as_secs(), node = ?node, depth, "stats publishing enabled"); - let config = moq_stats::Config::new() + let config = moq_stats::ProducerConfig::new() .with_origin(origin) .with_prefix(prefix) .with_interval(interval) diff --git a/rs/moq-stats/src/consume.rs b/rs/moq-stats/src/consume.rs index 6105c5726f..9c0bba2942 100644 --- a/rs/moq-stats/src/consume.rs +++ b/rs/moq-stats/src/consume.rs @@ -105,14 +105,14 @@ mod tests { use moq_net::{Consume, Origin, PathOwned, announce, origin}; - use crate::{Config, Producer, Tier}; + use crate::{Producer, ProducerConfig, Tier}; use super::*; fn test_producer() -> (Producer, origin::Producer) { let origin = Origin::random().produce(); let producer = Producer::new( - Config::new() + ProducerConfig::new() .with_origin(origin.clone()) .with_node(PathOwned::from("sjc")), ); diff --git a/rs/moq-stats/src/lib.rs b/rs/moq-stats/src/lib.rs index fd8456fd9b..ad60fdf6ef 100644 --- a/rs/moq-stats/src/lib.rs +++ b/rs/moq-stats/src/lib.rs @@ -52,7 +52,7 @@ mod consume; mod produce; pub use consume::{Consumer, ConsumerConfig, SessionsConsumer, TrafficConsumer}; -pub use produce::{Config, Producer}; +pub use produce::{Producer, ProducerConfig}; use std::collections::BTreeMap; @@ -71,9 +71,9 @@ pub type SessionsFrame = BTreeMap; /// Suffix appended to a plain track name for its compressed sibling. pub const COMPRESSED_SUFFIX: &str = ".z"; -/// The traffic track name for a tier and role: `publisher.json` on the default -/// tier, `/subscriber.json` on a named one, plus [`COMPRESSED_SUFFIX`] -/// when `compressed`. +/// The traffic track name for a tier and role: `.json` at the prefix root +/// on the default tier (`publisher.json` / `subscriber.json`), `/.json` +/// on a named one, plus [`COMPRESSED_SUFFIX`] when `compressed`. pub fn traffic_track(tier: &Tier, role: Role, compressed: bool) -> String { let mut name = tier.track_name(&format!("{}.json", role.as_str())); if compressed { diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index 6c56d9b777..d21e705174 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -11,16 +11,17 @@ use web_async::spawn; use crate::{COMPRESSED_SUFFIX, SessionsFrame, TrafficFrame, sessions_track, traffic_track}; -/// Settings for a [`Producer`]. Construct with [`Config::new`] and chain the -/// `with_*` setters (e.g. `Config::new().with_origin(origin).with_prefix(".foo")`), -/// then hand it to [`Producer::new`]. +/// Settings for a [`Producer`]. Construct with [`ProducerConfig::new`] and chain +/// the `with_*` setters (e.g. +/// `ProducerConfig::new().with_origin(origin).with_prefix(".foo")`), then hand it +/// to [`Producer::new`]. /// /// With no origin set the resulting producer is a no-op: its registry is -/// disabled (bumps are dropped) and no task spawns. Call [`Config::with_origin`] -/// to publish. +/// disabled (bumps are dropped) and no task spawns. Call +/// [`ProducerConfig::with_origin`] to publish. #[derive(Clone)] #[non_exhaustive] -pub struct Config { +pub struct ProducerConfig { /// Origin the stats broadcasts are created on. /// When `None`, [`Producer::new`] spawns no task and publishes nothing. pub origin: Option, @@ -47,7 +48,7 @@ pub struct Config { pub depth: usize, } -impl Config { +impl ProducerConfig { /// A config with default settings: no origin (no-op), `.stats` prefix, 1s /// interval, and no node suffix. Call [`Self::with_origin`] to actually /// publish. @@ -93,7 +94,7 @@ impl Config { } } -impl Default for Config { +impl Default for ProducerConfig { fn default() -> Self { Self::new() } @@ -126,8 +127,8 @@ impl Producer { /// [`Producer`] clone is dropped. With no origin the producer is a no-op /// (its registry is disabled, nothing is published) and no task spawns, so /// it's safe to build outside an async runtime. - pub fn new(config: Config) -> Self { - let Config { + pub fn new(config: ProducerConfig) -> Self { + let ProducerConfig { origin, prefix, node, @@ -558,7 +559,7 @@ mod tests { fn test_producer(node: Option<&str>) -> (Producer, origin::Producer) { let origin = Origin::random().produce(); let producer = Producer::new( - Config::new() + ProducerConfig::new() .with_origin(origin.clone()) .with_node(node.map(|s| PathOwned::from(s.to_string()))), ); From 454aa4d52e5a4c1952ccfe4694cd3030d60a1851 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 20 Jul 2026 14:00:57 -0700 Subject: [PATCH 4/6] refactor(gateways)!: pre-bump API polish for hls, rtc, rtmp, srt moq-hls: Broadcaster::new returns crate::Result instead of leaking moq_mux::Error; make the internally-built playlist Snapshot/Segment/ render_media pub(crate) (no external users) which also clears the collision with the recorder segments::Segment; mark the renditions Event cursor enum #[non_exhaustive]. moq-rtc: re-export axum and url (leaked in router/dial signatures) with the major-bump caveat; make the empty client whep/whip submodules private; document Client::new and every Error variant. moq-rtmp, moq-srt: take path: impl AsPath uniformly across every publish/ pull/accept entry point so the mirrored surfaces read consistently; origin ownership left as each call site actually uses it. Turn on warn(missing_docs) across all four crates. Co-Authored-By: Claude Fable 5 --- rs/moq-hls/src/export/mod.rs | 4 ++-- rs/moq-hls/src/export/playlist.rs | 6 +++--- rs/moq-hls/src/export/renditions.rs | 8 +++++++- rs/moq-hls/src/lib.rs | 2 ++ rs/moq-rtc/src/client/mod.rs | 8 ++++++-- rs/moq-rtc/src/error.rs | 12 ++++++++++++ rs/moq-rtc/src/lib.rs | 15 +++++++++++++++ rs/moq-rtmp/src/dial.rs | 5 +++-- rs/moq-rtmp/src/lib.rs | 2 ++ rs/moq-rtmp/src/server.rs | 12 +++++++----- rs/moq-srt/src/dial.rs | 10 ++++++---- rs/moq-srt/src/lib.rs | 2 ++ rs/moq-srt/src/server.rs | 10 ++++++---- 13 files changed, 73 insertions(+), 23 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 2021feb4d5..ca3be24ba4 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -28,7 +28,7 @@ use std::time::Duration; use moq_mux::catalog::{self, CatalogFormat, Stream}; -pub use playlist::{Segment, Snapshot, render_media}; +pub(crate) use playlist::render_media; pub use rendition::{Kind, Rendition}; /// How long to wait before retrying the initial catalog subscription. @@ -71,7 +71,7 @@ pub struct Broadcaster { impl Broadcaster { /// Resolve `source`'s catalog broadcast and start tracking its renditions. - pub async fn new(source: moq_mux::Source, config: Config) -> Result, moq_mux::Error> { + pub async fn new(source: moq_mux::Source, config: Config) -> crate::Result> { let broadcast = source.broadcast().await?; let renditions = renditions::Producer::new(); let broadcaster = Arc::new(Self { diff --git a/rs/moq-hls/src/export/playlist.rs b/rs/moq-hls/src/export/playlist.rs index 4695638ecf..035f3a01bf 100644 --- a/rs/moq-hls/src/export/playlist.rs +++ b/rs/moq-hls/src/export/playlist.rs @@ -13,7 +13,7 @@ const VERSION: u32 = 6; /// Everything a media playlist render needs; built by the crate-internal /// `Rendition::playlist`. -pub struct Snapshot { +pub(crate) struct Snapshot { /// `EXT-X-TARGETDURATION`, in whole seconds. pub target_duration: u64, /// `EXT-X-MEDIA-SEQUENCE` of the first listed segment. @@ -28,7 +28,7 @@ pub struct Snapshot { } /// One listed segment. -pub struct Segment { +pub(crate) struct Segment { /// The starting group sequence; the URI is `seg/{group}.m4s`. pub group: u64, /// `EXTINF` duration in seconds. @@ -36,7 +36,7 @@ pub struct Segment { } /// Render a media playlist for one rendition from a [`Snapshot`]. -pub fn render_media(snapshot: &Snapshot) -> String { +pub(crate) fn render_media(snapshot: &Snapshot) -> String { let mut out = String::new(); let _ = writeln!(out, "#EXTM3U"); let _ = writeln!(out, "#EXT-X-VERSION:{VERSION}"); diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index 9d1731c052..f2876cf49f 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -21,11 +21,17 @@ use super::rendition::{Kind, Rendition}; type Key = (Kind, String); /// A change to the rendition set, yielded by [`Consumer::next`]. +#[non_exhaustive] pub enum Event { /// A rendition appeared (or was reconfigured: a `Removed` for the old one precedes it). Added(Arc), /// A rendition disappeared (removed from the catalog, or replaced by a reconfigure). - Removed { kind: Kind, name: String }, + Removed { + /// Which axis (video or audio) the removed rendition was on. + kind: Kind, + /// The removed rendition's name. + name: String, + }, } /// The producing side of a broadcast's rendition set. diff --git a/rs/moq-hls/src/lib.rs b/rs/moq-hls/src/lib.rs index 443bb8c48d..5b97dd6cda 100644 --- a/rs/moq-hls/src/lib.rs +++ b/rs/moq-hls/src/lib.rs @@ -17,6 +17,8 @@ //! crate owns the HLS manifest generation, the timeline-driven playlist //! window, and the HTTP surface. +#![warn(missing_docs)] + mod error; pub mod export; pub mod import; diff --git a/rs/moq-rtc/src/client/mod.rs b/rs/moq-rtc/src/client/mod.rs index 14279de406..52774f9a35 100644 --- a/rs/moq-rtc/src/client/mod.rs +++ b/rs/moq-rtc/src/client/mod.rs @@ -5,8 +5,8 @@ //! it to the remote URL. Once the answer arrives the same internal session //! driver takes over, so the per-codec bridges and UDP socket loop are shared. -pub mod whep; -pub mod whip; +mod whep; +mod whip; use std::net::SocketAddr; @@ -32,6 +32,10 @@ pub struct Client { } impl Client { + /// Build a dialer from the shared client [`Config`]. The underlying + /// [`reqwest::Client`] (with its connection pool and rustls config) is created + /// once here and reused across every [`subscribe`](Self::subscribe) / + /// [`publish`](Self::publish) call. pub fn new(config: Config) -> Self { Self { config, diff --git a/rs/moq-rtc/src/error.rs b/rs/moq-rtc/src/error.rs index 8374736629..4cece4f1ac 100644 --- a/rs/moq-rtc/src/error.rs +++ b/rs/moq-rtc/src/error.rs @@ -2,38 +2,50 @@ #[derive(Debug, thiserror::Error)] #[non_exhaustive] pub enum Error { + /// An SDP offer or answer failed to parse or was missing something required. #[error("invalid SDP: {0}")] InvalidSdp(String), + /// The negotiated payload used a codec this gateway can't bridge to MoQ. #[error("unsupported codec: {0}")] UnsupportedCodec(String), + /// No live session matched the resource id (e.g. a DELETE for an unknown id). #[error("session not found")] SessionNotFound, + /// The peer closed the session; the media session ended without a failure. #[error("session closed")] SessionClosed, + /// ICE connectivity was not established before the establishment deadline. #[error("ICE did not connect before the establishment deadline")] IceTimeout, + /// I/O error on the media socket (bind, send, or receive). #[error("io error: {0}")] Io(#[from] std::io::Error), + /// Error from the underlying moq-net transport. #[error("moq error: {0}")] Moq(#[from] moq_net::Error), + /// Error from the moq-mux import/export layer bridging RTP and MoQ. #[error("mux error: {0}")] Mux(#[from] moq_mux::Error), + /// Error from the str0m WebRTC engine (SDP negotiation, DTLS, media state). #[error("rtc error: {0}")] Rtc(#[from] str0m::RtcError), + /// Error feeding a received UDP datagram into the str0m WebRTC engine. #[error("rtc input error: {0}")] RtcInput(#[from] str0m::error::NetError), + /// Catch-all for gateway logic that reports via `anyhow`. #[error(transparent)] Other(#[from] anyhow::Error), } +/// Convenience alias for results from the WebRTC <-> MoQ gateway. pub type Result = std::result::Result; diff --git a/rs/moq-rtc/src/lib.rs b/rs/moq-rtc/src/lib.rs index c03d0a93c5..e4b94bfd5e 100644 --- a/rs/moq-rtc/src/lib.rs +++ b/rs/moq-rtc/src/lib.rs @@ -36,6 +36,8 @@ //! parameter sets) and that's exactly what the importers want. AV1 uses the //! shared OBU splitter/importer. Opus, VP8, and VP9 pass through. +#![warn(missing_docs)] + pub mod client; pub mod server; @@ -57,6 +59,19 @@ mod session; /// bump is therefore a breaking change for this crate. pub use str0m; +/// Re-export of the HTTP router stack, so consumers can merge the [`axum::Router`] +/// returned by [`Server::publish_router`] / [`Server::subscribe_router`] (and by +/// [`whip::router`] / [`whep::router`]) into their own app without adding their own +/// axum dependency (and risking a version mismatch). A major axum bump is therefore +/// a breaking change for this crate. +pub use axum; + +/// Re-export of the URL type, so consumers can build the [`url::Url`] that +/// [`Client::subscribe`] / [`Client::publish`] dial without adding their own url +/// dependency (and risking a version mismatch). A major url bump is therefore a +/// breaking change for this crate. +pub use url; + pub use client::Client; pub use error::*; pub use server::{Response, Server, whep, whip}; diff --git a/rs/moq-rtmp/src/dial.rs b/rs/moq-rtmp/src/dial.rs index 743ddeafea..36237d310c 100644 --- a/rs/moq-rtmp/src/dial.rs +++ b/rs/moq-rtmp/src/dial.rs @@ -239,7 +239,8 @@ impl Client { /// /// This future resolves when the remote stream ends, so callers usually run it /// on its own task. - pub async fn pull(mut self, stream_key: &str, origin: &origin::Producer, path: &str) -> Result<()> { + pub async fn pull(mut self, stream_key: &str, origin: &origin::Producer, path: impl moq_net::AsPath) -> Result<()> { + let path = path.as_path(); let request = self .session .request_playback(stream_key.to_string()) @@ -249,7 +250,7 @@ impl Client { tracing::info!(%stream_key, %path, "rtmp play accepted by remote"); - let mut publisher = Publisher::new(origin, path)?; + let mut publisher = Publisher::new(origin, path.as_str())?; let result = self.pull_media(&mut publisher).await; match &result { diff --git a/rs/moq-rtmp/src/lib.rs b/rs/moq-rtmp/src/lib.rs index b8dc58be9d..f0044cbfc8 100644 --- a/rs/moq-rtmp/src/lib.rs +++ b/rs/moq-rtmp/src/lib.rs @@ -59,6 +59,8 @@ //! the vendored `rml` module (a fork of `rml_rtmp`), with no librtmp or ffmpeg //! dependency. +#![warn(missing_docs)] + use std::time::Duration; mod dial; diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 110de76ae3..dd7f518444 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -454,11 +454,12 @@ impl Publish { /// relay's shared origin, optionally re-rooted/scoped per the authenticated /// token). This future resolves when the connection ends, so callers usually /// run it on its own task. - pub async fn accept(mut self, origin: &origin::Producer, path: &str) -> Result<()> { + pub async fn accept(mut self, origin: &origin::Producer, path: impl moq_net::AsPath) -> Result<()> { + let path = path.as_path(); // Reserve the broadcast path before telling the client the publish succeeded: // if the origin refuses `path`, reject cleanly instead of accepting and then // dropping the connection a moment later. - let mut publisher = match Publisher::new(origin, path) { + let mut publisher = match Publisher::new(origin, path.as_str()) { Ok(publisher) => publisher, Err(err) => { tracing::warn!(peer = %self.peer, %path, %err, "rejecting RTMP publish: broadcast unavailable"); @@ -591,7 +592,8 @@ impl Play { /// before the publisher), cancelling cleanly if the viewer disconnects first. /// This future resolves when playback ends, so callers usually run it on its /// own task. - pub async fn accept(mut self, origin: &origin::Consumer, path: &str) -> Result<()> { + pub async fn accept(mut self, origin: &origin::Consumer, path: impl moq_net::AsPath) -> Result<()> { + let path = path.as_path(); // Wait for the broadcast before telling the client playback started. Feed the // client's bytes through the session (not discard them) so its deserializer // stays in sync for everything `play_pump` parses next. @@ -602,7 +604,7 @@ impl Play { tracing::debug!(peer = %self.peer, %path, "viewer disconnected before play started"); return Ok(()); } - broadcast = tokio::time::timeout(PLAY_RESOLVE_TIMEOUT, origin.announced_broadcast(path)) => { + broadcast = tokio::time::timeout(PLAY_RESOLVE_TIMEOUT, origin.announced_broadcast(&path)) => { match broadcast { Ok(broadcast) => broadcast, Err(_) => { @@ -651,7 +653,7 @@ impl Play { // The export re-resolves the broadcast (and any sibling broadcast a rendition's // catalog `broadcast` field references) through the origin. - let mut export = FlvExport::new(moq_mux::Source::new(origin.consume(), path)) + let mut export = FlvExport::new(moq_mux::Source::new(origin.consume(), path.as_str())) .await .map_err(|e| anyhow::anyhow!("init FLV export: {e}"))? .with_latency(self.latency) diff --git a/rs/moq-srt/src/dial.rs b/rs/moq-srt/src/dial.rs index 2a1b9e43e3..9c051e8059 100644 --- a/rs/moq-srt/src/dial.rs +++ b/rs/moq-srt/src/dial.rs @@ -38,11 +38,12 @@ pub async fn publish( resource: &str, latency: impl Into>, origin: &origin::Consumer, - path: &str, + path: impl moq_net::AsPath, ) -> Result<()> { + let path = path.as_path(); let latency = latency.into().unwrap_or(DEFAULT_LATENCY); let socket = call(addr, resource, Mode::Publish, latency).await?; - serve_subscribe(origin, path, socket, latency).await + serve_subscribe(origin, path.as_str(), socket, latency).await } /// Dial `addr` and pull a remote stream into `origin`: connect as an SRT caller @@ -57,10 +58,11 @@ pub async fn pull( resource: &str, latency: impl Into>, origin: &origin::Producer, - path: &str, + path: impl moq_net::AsPath, ) -> Result<()> { + let path = path.as_path(); let socket = call(addr, resource, Mode::Request, latency).await?; - serve_publish(origin, path, socket).await + serve_publish(origin, path.as_str(), socket).await } /// Dial `addr` as an SRT caller for `resource`, sending the standard diff --git a/rs/moq-srt/src/lib.rs b/rs/moq-srt/src/lib.rs index 216641a411..be4be8583d 100644 --- a/rs/moq-srt/src/lib.rs +++ b/rs/moq-srt/src/lib.rs @@ -36,6 +36,8 @@ //! Pure Rust: SRT is provided by `srt-tokio`, with no libsrt or ffmpeg //! dependency. +#![warn(missing_docs)] + pub mod dial; mod error; mod listen; diff --git a/rs/moq-srt/src/server.rs b/rs/moq-srt/src/server.rs index fb449cfba6..fbef8efead 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -212,10 +212,11 @@ impl Publish { /// relay's shared origin, optionally scoped per the authenticated token). This /// future resolves when the connection ends, so callers usually run it on its /// own task. - pub async fn accept(self, origin: &origin::Producer, path: &str) -> Result<()> { + pub async fn accept(self, origin: &origin::Producer, path: impl moq_net::AsPath) -> Result<()> { + let path = path.as_path(); let socket = self.0.request.accept(None).await?; tracing::info!(peer = %self.0.peer, %path, "SRT publish accepted"); - serve_publish(origin, path, socket).await + serve_publish(origin, path.as_str(), socket).await } /// Reject the publish, sending the client a `Forbidden` rejection. @@ -262,10 +263,11 @@ impl Subscribe { /// Waits for the broadcast to be announced (so a caller may connect before the /// publisher), cancelling cleanly if the caller disconnects first. This future /// resolves when playback ends, so callers usually run it on its own task. - pub async fn accept(self, origin: &origin::Consumer, path: &str) -> Result<()> { + pub async fn accept(self, origin: &origin::Consumer, path: impl moq_net::AsPath) -> Result<()> { + let path = path.as_path(); let socket = self.0.request.accept(None).await?; tracing::info!(peer = %self.0.peer, %path, "SRT subscribe accepted"); - serve_subscribe(origin, path, socket, self.0.latency).await + serve_subscribe(origin, path.as_str(), socket, self.0.latency).await } /// Reject the subscribe, sending the client a `Forbidden` rejection. From 4642fab3dc935218f70e795068acedc50dbe416e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 20 Jul 2026 14:24:13 -0700 Subject: [PATCH 5/6] refactor(moq-mux)!: flatten the ts catalog trait to ts::Catalog Name the MPEG-TS catalog-carriage capability ts::Catalog: make the `catalog` submodule private and re-export its surface flat at the `ts` module, so the trait and its section types read as ts::Catalog, ts::Mpegts, ts::Track, ... instead of stuttering under container::ts::catalog. This clears the earlier foo::foo::Foo path without inventing a bespoke trait name, and lines up with the hang::Catalog / msf::Catalog family. Co-Authored-By: Claude Fable 5 --- rs/moq-cli/src/publish.rs | 6 ++--- rs/moq-mux/src/container/ts/catalog.rs | 8 +++---- rs/moq-mux/src/container/ts/export.rs | 4 ++-- rs/moq-mux/src/container/ts/export_test.rs | 2 +- rs/moq-mux/src/container/ts/import.rs | 26 +++++++++++----------- rs/moq-mux/src/container/ts/mod.rs | 10 +++++---- rs/moq-mux/src/import/container.rs | 12 +++++----- 7 files changed, 35 insertions(+), 33 deletions(-) diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index b178aa63d0..a51d94ba80 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -136,7 +136,7 @@ enum PublishDecoder { Fmp4(Box), // TS carries undecoded elementary streams (SCTE-35, teletext, DVB AC-3, ...) // verbatim, so it uses the `mpegts` catalog extension rather than the media-only `()`. - Ts(Box>), + Ts(Box>), Flv(Box), } @@ -225,7 +225,7 @@ impl Publish { if let PublishFormat::Ts = format { let catalog = moq_mux::catalog::Producer::with_catalog( &mut broadcast, - moq_mux::catalog::hang::Catalog::::default(), + moq_mux::catalog::hang::Catalog::::default(), )?; let ts = ts::Import::new(broadcast.clone(), catalog.reserve()); return Ok(Self { @@ -436,7 +436,7 @@ mod tests { use bytes::BytesMut; use moq_mux::catalog::CatalogFormat; use moq_mux::catalog::hang::{Catalog, Container}; - use moq_mux::container::ts::{Export, Import, catalog as tscat}; + use moq_mux::container::ts::{self as tscat, Export, Import}; use moq_mux::container::{Consumer, Frame, Producer}; use moq_net::Timestamp; diff --git a/rs/moq-mux/src/container/ts/catalog.rs b/rs/moq-mux/src/container/ts/catalog.rs index b4c4809442..6e2a1014dd 100644 --- a/rs/moq-mux/src/container/ts/catalog.rs +++ b/rs/moq-mux/src/container/ts/catalog.rs @@ -148,7 +148,7 @@ impl CatalogExt for Ext {} /// /// Implement this for an application extension to compose MPEG-TS carriage with /// additional sections. -pub trait Carrier: CatalogExt { +pub trait Catalog: CatalogExt { /// The section to record MPEG-TS details into, or `None` for an extension that /// doesn't carry them. /// @@ -158,7 +158,7 @@ pub trait Carrier: CatalogExt { fn mpegts_mut(&mut self) -> Option<&mut Mpegts>; } -impl Carrier for () { +impl Catalog for () { fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { None } @@ -166,13 +166,13 @@ impl Carrier for () { // The untyped passthrough carries no typed mpegts section (a TS importer driving an `Extra` // catalog records verbatim streams as raw JSON sections, not the typed `Mpegts` view). -impl Carrier for crate::catalog::hang::Extra { +impl Catalog for crate::catalog::hang::Extra { fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { None } } -impl Carrier for Ext { +impl Catalog for Ext { fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { Some(&mut self.mpegts) } diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index f2bf99a6e8..09a90c6f2c 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -53,7 +53,7 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500); /// The leading PAT/PMT rides on the first frame (so it inherits a real /// timestamp), and is re-emitted at video keyframes and periodically for /// mid-stream tune-in. Returns `None` when the broadcast ends. -pub struct Export { +pub struct Export { source: crate::Source, catalog: Option>, latency: Duration, @@ -179,7 +179,7 @@ impl Export { } } -impl Export { +impl Export { /// Shared constructor. The public entry points each live on a concrete /// `Export` impl that pins `E`, so the extension is chosen by which one you call. async fn build(source: crate::Source, catalog_format: CatalogFormat) -> Result { diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index affed5becd..89eefb04f5 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -64,7 +64,7 @@ async fn drain(consumer: moq_net::broadcast::Consumer) -> BytesMut { } /// `drain` for an exporter built with an explicit catalog extension. -async fn drain_with(mut exporter: Export) -> BytesMut { +async fn drain_with(mut exporter: Export) -> BytesMut { let mut out = BytesMut::new(); // `while let Ok` stops on the first timeout (`Pending`: no more output). while let Ok(res) = tokio::time::timeout(std::time::Duration::from_secs(1), exporter.next()).await { diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 7e5ceff27a..55a3af4cad 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -35,12 +35,12 @@ use moq_net::Timestamp; /// manages the track, catalog config, and keyframe-based group boundaries. /// /// Elementary streams we don't decode are carried verbatim, one MoQ track per -/// PID, when the catalog `E` carries the [`mpegts`](catalog) section: PES-framed +/// PID, when the catalog `E` carries the [`mpegts`](super::Mpegts) section: PES-framed /// streams ride the normal PES reassembly, section-framed streams (SCTE-35, marked /// by a program-level 'CUEI' registration descriptor, and other private sections) /// are intercepted before the reader and reassembled. With a base `Catalog<()>` /// they're logged and dropped instead. -pub struct Import { +pub struct Import { broadcast: moq_net::broadcast::Producer, catalog: crate::catalog::Producer, @@ -105,7 +105,7 @@ pub struct Import { media_unwrap: PtsUnwrap, } -impl Import { +impl Import { pub fn new(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { let feed = Feed::default(); // A long-lived producer handle for catalog edits (mpegts sections, later PMTs); the passed @@ -631,7 +631,7 @@ fn to_descriptors(descriptors: &[mpeg2ts::ts::Descriptor]) -> Vec( +fn register_verbatim( broadcast: &mut moq_net::broadcast::Producer, catalog: &mut crate::catalog::Producer, pid: u16, @@ -664,7 +664,7 @@ fn register_verbatim( } /// Remove a verbatim track's entry from the `mpegts` catalog section on drop. -fn unregister_verbatim(catalog: &mut crate::catalog::Producer, name: &str) { +fn unregister_verbatim(catalog: &mut crate::catalog::Producer, name: &str) { if let Some(mpegts) = catalog.lock().mpegts_mut() { mpegts.tracks.remove(name); } @@ -677,13 +677,13 @@ fn unregister_verbatim(catalog: &mut crate::catalog::Produc /// intercepted before the mpeg2ts reader (which would PES-parse it and abort). /// The byte-level reassembly lives in [`SectionReassembler`]; this type owns the /// track and catalog entry and stamps each section with the media clock. -struct SectionStream { +struct SectionStream { track: crate::container::Producer, catalog: crate::catalog::Producer, reassembler: SectionReassembler, } -impl SectionStream { +impl SectionStream { fn new( mut broadcast: moq_net::broadcast::Producer, mut catalog: crate::catalog::Producer, @@ -746,7 +746,7 @@ impl SectionStream { } } -impl Drop for SectionStream { +impl Drop for SectionStream { fn drop(&mut self) { let name = self.track.name().to_string(); unregister_verbatim(&mut self.catalog, &name); @@ -759,7 +759,7 @@ impl Drop for SectionStream { /// /// Unlike [`SectionStream`], these ride the normal PES reassembly path, so this /// type only stamps each PES payload with its (unwrapped) PTS and writes it. -struct VerbatimStream { +struct VerbatimStream { track: crate::container::Producer, catalog: crate::catalog::Producer, unwrap: PtsUnwrap, @@ -767,7 +767,7 @@ struct VerbatimStream { stream_id_recorded: bool, } -impl VerbatimStream { +impl VerbatimStream { fn new( mut broadcast: moq_net::broadcast::Producer, mut catalog: crate::catalog::Producer, @@ -833,7 +833,7 @@ impl VerbatimStream { } } -impl Drop for VerbatimStream { +impl Drop for VerbatimStream { fn drop(&mut self) { let name = self.track.name().to_string(); unregister_verbatim(&mut self.catalog, &name); @@ -992,7 +992,7 @@ impl SectionReassembler { } /// One elementary stream's codec importer plus PTS-unwrap state. -enum Stream { +enum Stream { H264 { split: h264::Split, import: Box>, @@ -1014,7 +1014,7 @@ enum Stream { Ignored, } -impl Stream { +impl Stream { fn write(&mut self, pending: Pending, burst: Option) -> anyhow::Result<()> { match self { Stream::H264 { split, import, unwrap } => { diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 35820070f6..f951c017a0 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -8,16 +8,18 @@ //! //! Elementary streams we don't decode (SCTE-35, teletext, DVB subtitles, private //! data, ...) are carried verbatim, one MoQ track per PID, described in the -//! [`catalog`] (`mpegts`) section. SCTE-35 is just one such stream (`stream_type` 0x86). +//! [`Mpegts`] catalog section. SCTE-35 is just one such stream (`stream_type` 0x86). mod adts; mod export; mod import; -/// The `mpegts` catalog section: per-track PID + descriptors plus verbatim -/// carriage of undecoded elementary streams. -pub mod catalog; +// The `mpegts` catalog section (per-track PID + descriptors plus verbatim carriage +// of undecoded elementary streams) and the `Catalog` capability, re-exported flat so +// they read as `ts::Catalog`, `ts::Mpegts`, ... instead of stuttering under `catalog`. +mod catalog; +pub use catalog::{Catalog, Descriptor, Ext, Framing, Mpegts, Track, Verbatim}; pub use export::*; pub use import::*; diff --git a/rs/moq-mux/src/import/container.rs b/rs/moq-mux/src/import/container.rs index f84a697333..2434b8ffe1 100644 --- a/rs/moq-mux/src/import/container.rs +++ b/rs/moq-mux/src/import/container.rs @@ -10,7 +10,7 @@ use crate::Result; /// The concrete container importers, shared by [`Container`] and /// [`ContainerStream`]. Containers parse their own internal framing, so a whole /// chunk and a stream chunk decode identically. -enum ContainerImpl { +enum ContainerImpl { // Boxed because it's a large struct and clippy complains about the size. Fmp4(Box>), Mkv(Box>), @@ -18,7 +18,7 @@ enum ContainerImpl { Flv(Box>), } -impl ContainerImpl { +impl ContainerImpl { fn fmp4(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { ContainerImpl::Fmp4(Box::new(crate::container::fmp4::Import::new(broadcast, reserved))) } @@ -76,11 +76,11 @@ impl ContainerImpl { /// /// Use this when the caller hands over discrete buffers (the typical case for /// files and reassembled network input). May publish more than one track. -pub struct Container { +pub struct Container { inner: ContainerImpl, } -impl Container { +impl Container { /// Create a new container importer, decoding the initial chunk. pub fn new( broadcast: moq_net::broadcast::Producer, @@ -125,11 +125,11 @@ impl Container { /// /// Use this when the caller pushes arbitrary byte chunks and the container /// recovers its own framing. May publish more than one track. -pub struct ContainerStream { +pub struct ContainerStream { inner: ContainerImpl, } -impl ContainerStream { +impl ContainerStream { /// Create a new container stream importer. pub fn new( broadcast: moq_net::broadcast::Producer, From f8606ce39834938b0b0801007c9fe5ee53a1ca79 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 20 Jul 2026 14:32:20 -0700 Subject: [PATCH 6/6] fix(moq-gst): handle the non_exhaustive reconnect Status in the sink moq-native's Status became #[non_exhaustive], so the ConnectionStatus mapping in the reconnect loop needs a wildcard. Map any non-Connected state to Disconnected (no behavior change today; future variants fall through safely). moq-gst doesn't build on the macOS dev shell (no GStreamer), so this only surfaced in CI. Co-Authored-By: Claude Fable 5 --- rs/moq-gst/src/sink/session.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-gst/src/sink/session.rs b/rs/moq-gst/src/sink/session.rs index d2eeeafa13..fc2170bb20 100644 --- a/rs/moq-gst/src/sink/session.rs +++ b/rs/moq-gst/src/sink/session.rs @@ -234,7 +234,7 @@ async fn forward( Ok(state) => { let connection = match state { moq_native::Status::Connected => ConnectionStatus::Connected, - moq_native::Status::Disconnected => ConnectionStatus::Disconnected, + _ => ConnectionStatus::Disconnected, }; status.set(connection, reconnect.version().map(|v| v.to_string())); match state {