diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 8be91d05c6c..fb261379f5f 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -61,6 +61,30 @@ if (session < 0) return fail(moq_error()); ``` +## Publish + +```c +const char *path = "my-stream.hang"; +uint32_t broadcast; +int publication = moq_origin_create_broadcast(origin, path, strlen(path), &broadcast); +if (publication < 0) + return fail(moq_error()); +// Fill tracks on broadcast here; announce on the publication. + +char epoch[MOQ_EPOCH_LEN]; +moq_mint_epoch(epoch, sizeof(epoch)); // a fresh epoch per run +moq_announce_route route; +memset(&route, 0, sizeof(route)); // zeroed means cost 0 and no epoch +route.epoch = epoch; +route.epoch_len = sizeof(epoch); +if (moq_publication_announce((uint32_t)publication, &route) < 0) + return fail(moq_error()); +``` + +A route carries only a cost and an epoch; relays add the hops. Mint a fresh epoch each +time the publisher starts, so viewers see a restart (`MOQ_ANNOUNCE_KIND_RESTART`) instead of +a stalled broadcast. Redundant publishers of the same bytes share one. + Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Publication track names stay separate from codec settings. For simulcast, give each rendition the next one down as the `lower` handle in its `moq_video_encode_options`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. ## Things to know diff --git a/quest/m1/c/README.md b/quest/m1/c/README.md index a8a62969b3d..af4cbf2d370 100644 --- a/quest/m1/c/README.md +++ b/quest/m1/c/README.md @@ -31,4 +31,3 @@ Public API: moq-c gains each m1 binding feature. Wire: none. - [moq-c decode start](/quest/m1/c/decode-start.md) - a moq-c decoder can read a finished track from its start - [JSON windows](/quest/m1/c/json-window.md) - moq-c publishes and reads moq-json windows, as moq-ffi does -- [Announce route](/quest/m1/c/announce-route.md) - a C publisher announces with a cost and restart epoch through its own struct, and cannot claim relay hops diff --git a/quest/m1/c/announce-route.md b/quest/m1/c/announce-route.md deleted file mode 100644 index 1d52946f378..00000000000 --- a/quest/m1/c/announce-route.md +++ /dev/null @@ -1,36 +0,0 @@ -# [S] moq-c announces through its own route struct - -## Goal - -A C publisher announces with a publish-only route that matches moq-ffi's -`MoqAnnounceRoute`: a cost and an optional epoch, so it can announce a restart -and viewers see a restart instead of a stalled broadcast. It can no longer -claim relay hops it never took. `moq_route` stays for received routes, where -hops are real. - -## Plan - -Found while landing [moq-dev/moq#7044](https://github.com/moq-dev/moq/pull/7044), -which moved announcing onto `moq_publication_announce`. Today that call takes -`moq_route` (`rs/moq-c/src/api.rs`), which has no epoch and accepts any -caller-supplied hops, while moq-ffi's `MoqAnnounceRoute` -(`rs/moq-ffi/src/origin.rs`) leaves hops to the origin. - -Decided with the maintainer on 2026-10-28: - -- **A separate announce struct**, `moq_announce_route { cost, epoch }`, - mirroring moq-ffi, rather than refusing non-empty hops on the shared - `moq_route` or validating hop values. `moq_route` keeps describing received - routes. -- **Epoch as UUIDv7 text**, as moq-ffi's `epoch`, minted the way - `moq_mint_epoch` is in moq-ffi; add a C mint call if one is missing. -- **In m1:** the hand-written moq-c ships at the cut with parity with moq-ffi - for every m1 binding feature (decided 2026-10-25), so this C ABI break lands - before it. - -Update the C header, `c/moq`'s probe if it announces, the C++ and OBS -wrappers, `doc/lib/c`, and the tests: a restart epoch reaching a subscriber as -a restart, and no hops on a locally announced route. - -Public API: breaking in the C ABI (`moq_publication_announce` takes the new -struct), with an Impact migration note. Wire: none. diff --git a/rs/moq-c/README.md b/rs/moq-c/README.md index aa3e6b93731..3eabf2ec079 100644 --- a/rs/moq-c/README.md +++ b/rs/moq-c/README.md @@ -66,7 +66,9 @@ int32_t moq_origin_announced_cancel(uint32_t announced); // config (NULL for everything): filter is relative to the literal prefix, or NULL for **. hidden also lists `.`-named paths. Updates stay relative to the origin. // Publishing: Announcement (a publication from moq_origin_create_broadcast or moq_broadcast_prepared_publish) -int32_t moq_publication_announce(uint32_t publication, const moq_route *route); +int32_t moq_publication_announce(uint32_t publication, const moq_announce_route *route); +// route (NULL for the default): a cost and an optional epoch from moq_mint_epoch. Hops are the origin's to add. +int32_t moq_mint_epoch(char *dst, uintptr_t dst_len); int32_t moq_publication_unannounce(uint32_t publication); int32_t moq_publication_free(uint32_t publication); diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index b6fe1cf13bd..e77a80b2bd1 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -748,62 +748,68 @@ pub struct moq_section { pub json_len: usize, } -/// A route advertisement: hops and a static cost. +/// The length in bytes of an epoch's UUIDv7 text, which [moq_mint_epoch] writes. +pub const MOQ_EPOCH_LEN: usize = 36; + +/// How a publication advertises its exact path: a static cost and the publisher epoch. /// -/// Pair with [moq_publication_announce] or [moq_origin_requests]. Zeroed (NULL hops, -/// hops_len 0, cost 0) is the default route. `hops` is borrowed for the duration -/// of the call that reads it. +/// Pass one to [moq_publication_announce]. The hop chain is the origin's to build, +/// so a local publisher cannot claim relay hops it never took. Zeroed (cost 0, NULL +/// epoch) is the default route. `epoch` is borrowed for the duration of the call. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct moq_announce_route { + /// Preference among routes covering the same prefix: lower wins. A publisher + /// sets its production cost here: zero for content it is already producing, + /// larger for content it would have to start producing on demand. + pub cost: u64, + /// The publisher instance the route serves, as UUIDv7 text, NOT NULL terminated, + /// or NULL (or empty) when unknown. + /// + /// A publisher that can restart sets a fresh one per run from [moq_mint_epoch], so + /// viewers see a restart instead of a stalled broadcast. Routes with the same epoch + /// serve the same bytes, so redundant publishers share one. Among routes at a + /// prefix the newest epoch wins, and changing it is a `MOQ_ANNOUNCE_KIND_RESTART`. + pub epoch: *const c_char, + /// The length in bytes of `epoch`. + pub epoch_len: usize, +} + +/// An observed route: hops, a static cost, and the publisher epoch. /// -/// `cost` is the static production and link price; lower wins. +/// Carried by [moq_announce_event]. Its pointers borrow the announcement's storage. #[repr(C)] #[derive(Clone, Copy)] pub struct moq_route { - /// Hop ids, oldest first. NULL when `hops_len` is 0. 0 is the anonymous - /// mark and is legal on a received chain. + /// Hop ids of the relay hops the route traversed, oldest first. 0 is the + /// anonymous mark, which ranks the route below every fully identified one. + /// A route announced on this origin has none. pub hops: *const u64, /// The number of entries in `hops`. pub hops_len: usize, /// Preference among routes covering the same prefix: lower wins. pub cost: u64, + /// The publisher instance the route serves, as UUIDv7 text; empty when unknown. + pub epoch: moq_string, } -impl Default for moq_route { - fn default() -> Self { - Self { - hops: std::ptr::null(), - hops_len: 0, - cost: 0, - } - } -} - -/// Parse a [moq_route], treating NULL as the default. +/// Parse a [moq_announce_route], treating NULL as the default. /// /// # Safety -/// `route` may be NULL, or must point at a readable [moq_route] whose `hops` -/// pointer is valid for `hops_len` elements. -unsafe fn parse_route(route: *const moq_route) -> Result { +/// `route` may be NULL, or must point at a readable [moq_announce_route] whose `epoch` +/// pointer is valid for `epoch_len` bytes. +unsafe fn parse_announce_route(route: *const moq_announce_route) -> Result { let Some(route) = (unsafe { route.as_ref() }) else { return Ok(moq_net::origin::Route::default()); }; - let mut route_hops = moq_net::Hops::new(); - if route.hops_len > 0 { - if route.hops.is_null() { - return Err(Error::InvalidPointer); - } - let hops = unsafe { std::slice::from_raw_parts(route.hops, route.hops_len) }; - for id in hops { - let hop = if *id == 0 { - moq_net::Hop::UNKNOWN - } else { - moq_net::Hop::new(*id).map_err(|e| Error::InvalidConfig(e.to_string()))? - }; - route_hops.push(hop).map_err(|e| Error::InvalidConfig(e.to_string()))?; - } + let mut out = moq_net::origin::Route::default().with_cost(moq_net::origin::Cost::new(route.cost)); + if let Some(epoch) = unsafe { ffi::parse_str_optional(route.epoch, route.epoch_len)? } { + let epoch = epoch + .parse() + .map_err(|_| Error::InvalidConfig(format!("epoch {epoch:?} is not canonical UUIDv7 text")))?; + out = out.with_epoch(epoch); } - Ok(moq_net::origin::Route::default() - .with_cost(moq_net::origin::Cost::new(route.cost)) - .with_hops(route_hops)) + Ok(out) } /// Which broadcasts [moq_origin_announced] lists. @@ -863,6 +869,10 @@ pub struct moq_announce_event { /// Which event this is. pub kind: moq_announce_kind, + + /// The route serving the prefix. On `MOQ_ANNOUNCE_KIND_END` it carries the last + /// advertised route. + pub route: moq_route, } /// Where a session is in its lifecycle, reported by [moq_connection_state]. @@ -1098,6 +1108,34 @@ pub unsafe extern "C" fn moq_versions(dst: *mut moq_string, count: usize) -> i32 }) } +/// Mint a fresh publisher epoch from the wall clock and secure randomness, ordered newest last. +/// +/// Writes its [MOQ_EPOCH_LEN] bytes of UUIDv7 text, NOT NULL terminated, to `dst`. Pass +/// it as [moq_announce_route]'s `epoch` or to [moq_broadcast_request_prepare]; a +/// publisher that can restart mints one per run. +/// +/// Returns [MOQ_EPOCH_LEN] on success, or a negative code when `dst` is NULL or +/// `dst_len` is shorter. +/// +/// # Safety +/// - `dst` must be NULL or point at `dst_len` writable bytes. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn moq_mint_epoch(dst: *mut c_char, dst_len: usize) -> i32 { + ffi::enter(move || { + if dst.is_null() { + return Err(Error::InvalidPointer); + } + if dst_len < MOQ_EPOCH_LEN { + return Err(Error::InvalidConfig(format!( + "an epoch needs {MOQ_EPOCH_LEN} bytes, got {dst_len}" + ))); + } + let dst = unsafe { std::slice::from_raw_parts_mut(dst.cast::(), MOQ_EPOCH_LEN) }; + dst.copy_from_slice(moq_net::Epoch::mint().as_str().as_bytes()); + Ok(MOQ_EPOCH_LEN) + }) +} + /// Whether this build can capture qlog traces. /// /// Capture is compile-time optional. [moq_client_config]'s `quic_qlog` accepts a directory @@ -2200,19 +2238,20 @@ pub extern "C" fn moq_origin_close(origin: u32) -> i32 { /// Advertise a publication's exact path as a route. /// -/// Announcing again re-prices the route in place. A NULL `route` uses the default -/// (no hops, cost 0). Until announced, the broadcast is invisible and unroutable for -/// local consumers and peers alike. Fails once the broadcast closed. +/// Announcing again re-prices the route in place, or restarts it under a new epoch. A +/// NULL `route` uses the default (cost 0, no epoch). Until announced, the broadcast is +/// invisible and unroutable for local consumers and peers alike. Fails once the +/// broadcast closed, or when `epoch` is not UUIDv7 text. /// /// Returns a zero on success, or a negative code on failure. /// /// # Safety -/// - `route` may be NULL, or must point at a readable [moq_route]. +/// - `route` may be NULL, or must point at a readable [moq_announce_route]. #[unsafe(no_mangle)] -pub unsafe extern "C" fn moq_publication_announce(publication: u32, route: *const moq_route) -> i32 { +pub unsafe extern "C" fn moq_publication_announce(publication: u32, route: *const moq_announce_route) -> i32 { ffi::enter(move || { let publication = ffi::parse_id(publication)?; - let route = unsafe { parse_route(route)? }; + let route = unsafe { parse_announce_route(route)? }; State::lock().publish.announce(publication, route) }) } diff --git a/rs/moq-c/src/origin.rs b/rs/moq-c/src/origin.rs index 5dd8745f213..77e510a9e6d 100644 --- a/rs/moq-c/src/origin.rs +++ b/rs/moq-c/src/origin.rs @@ -3,7 +3,7 @@ use tokio::sync::oneshot; use crate::ffi::OnStatus; use crate::task::Task; -use crate::{Error, Id, NonZeroSlab, State, moq_announce_event, moq_announce_kind, moq_string}; +use crate::{Error, Id, NonZeroSlab, State, moq_announce_event, moq_announce_kind, moq_route, moq_string}; /// Global state managing all active resources. /// @@ -76,6 +76,10 @@ struct AnnouncedRecord { captures: Option>, capture_views: Vec, kind: moq_announce_kind, + hops: Vec, + cost: u64, + /// Empty when the route carries no epoch. + epoch: String, } // The raw pointers only borrow immutable String allocations owned by this record. @@ -112,6 +116,9 @@ impl AnnouncedRecord { captures, capture_views, kind, + hops: update.route.hops.iter().map(|hop| hop.id()).collect(), + cost: update.route.cost.value(), + epoch: update.route.epoch.map(|epoch| epoch.to_string()).unwrap_or_default(), } } } @@ -192,6 +199,15 @@ impl Origin { captures_len: announced.capture_views.len(), has_captures: announced.captures.is_some(), kind: announced.kind, + route: moq_route { + hops: announced.hops.as_ptr(), + hops_len: announced.hops.len(), + cost: announced.cost, + epoch: moq_string { + data: announced.epoch.as_ptr().cast::(), + len: announced.epoch.len(), + }, + }, }; Ok(()) } diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 7b336585d07..6dd66d398db 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -161,11 +161,51 @@ fn announce_info(announced: u32) -> moq_announce_event { captures_len: 0, has_captures: false, kind: moq_announce_kind::MOQ_ANNOUNCE_KIND_START, + route: moq_route { + hops: std::ptr::null(), + hops_len: 0, + cost: 0, + epoch: moq_string { + data: std::ptr::null(), + len: 0, + }, + }, }; assert_eq!(unsafe { moq_origin_announced_info(announced, &mut info) }, 0); info } +/// Mint an epoch through the C ABI. +fn mint_epoch() -> String { + let mut buf = [0u8; MOQ_EPOCH_LEN]; + assert_eq!( + unsafe { moq_mint_epoch(buf.as_mut_ptr().cast(), buf.len()) }, + MOQ_EPOCH_LEN as i32 + ); + String::from_utf8(buf.to_vec()).unwrap() +} + +/// An announce route at `cost` under `epoch`, borrowing it. +fn announce_route(cost: u64, epoch: &str) -> moq_announce_route { + moq_announce_route { + cost, + epoch: epoch.as_ptr().cast(), + epoch_len: epoch.len(), + } +} + +/// The epoch a received route carries, or None when unknown. +fn route_epoch(route: &moq_route) -> Option { + (route.epoch.len > 0).then(|| unsafe { + std::str::from_utf8(std::slice::from_raw_parts( + route.epoch.data.cast::(), + route.epoch.len, + )) + .unwrap() + .to_string() + }) +} + impl Drop for Callback { fn drop(&mut self) { unsafe { drop(Box::from_raw(self.ptr as *mut mpsc::Sender)) }; @@ -2834,17 +2874,100 @@ fn create_broadcast_is_unroutable_until_announced() { } #[test] -fn announce_accepts_an_anonymous_hop() { +fn mint_epoch_writes_ordered_uuidv7_text() { + let epoch = mint_epoch(); + let parsed: moq_net::Epoch = epoch.parse().expect("a minted epoch parses"); + assert_eq!(parsed.as_str(), epoch); + assert!(mint_epoch() > epoch, "a later mint orders newer"); + + let mut short = [0u8; MOQ_EPOCH_LEN - 1]; + assert!(unsafe { moq_mint_epoch(short.as_mut_ptr().cast(), short.len()) } < 0); + assert!(unsafe { moq_mint_epoch(std::ptr::null_mut(), MOQ_EPOCH_LEN) } < 0); +} + +/// A locally announced route carries its cost and epoch, and no hops: the hop chain is +/// the origin's to build. +#[test] +fn announce_route_carries_cost_and_epoch_without_hops() { let origin = id(moq_origin_create()); - let path = b"anon"; - let (publication, broadcast) = create_broadcast(origin, path); - let hops = [0u64]; - let route = moq_route { - hops: hops.as_ptr(), - hops_len: hops.len(), - cost: 1, - }; + let cb = Callback::new(); + let announced_task = id(unsafe { moq_origin_announced(origin, std::ptr::null(), Some(channel_callback), cb.ptr) }); + let (publication, broadcast) = create_broadcast(origin, b"routed"); + + let epoch = mint_epoch(); + let route = announce_route(7, &epoch); assert_eq!(unsafe { moq_publication_announce(publication, &route) }, 0); + + let (announced, info) = cb.recv_route(); + assert_eq!(info.kind, moq_announce_kind::MOQ_ANNOUNCE_KIND_START); + assert_eq!(info.route.hops_len, 0, "a local route claims no hops"); + assert_eq!(info.route.cost, 7); + assert_eq!(route_epoch(&info.route), Some(epoch)); + assert_eq!(moq_origin_announced_free(announced), 0); + + // A zeroed route is the default: cost 0 and no epoch. + let zeroed = moq_announce_route { + cost: 0, + epoch: std::ptr::null(), + epoch_len: 0, + }; + let (publication2, broadcast2) = create_broadcast(origin, b"plain"); + assert_eq!(unsafe { moq_publication_announce(publication2, &zeroed) }, 0); + let (announced, info) = cb.recv_route(); + assert_eq!(info.route.cost, 0); + assert_eq!(route_epoch(&info.route), None); + assert_eq!(moq_origin_announced_free(announced), 0); + + // Anything but canonical UUIDv7 text is refused. + let upper = mint_epoch().to_uppercase(); + for bad in ["6ba7b810-9dad-41d1-80b4-00c04fd430c8", upper.as_str(), "not-an-epoch"] { + let route = announce_route(0, bad); + assert!( + unsafe { moq_publication_announce(publication, &route) } < 0, + "{bad:?} is not an epoch" + ); + } + + assert_eq!(moq_origin_announced_cancel(announced_task), 0); + assert_eq!(cb.recv_announce_terminal(), 0); + assert_eq!(moq_publication_free(publication), 0); + assert_eq!(moq_publication_free(publication2), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_publish_close(broadcast2), 0); + assert_eq!(moq_origin_close(origin), 0); +} + +/// Announcing under a fresh epoch is a new publisher instance: subscribers see a restart +/// rather than a stalled broadcast. +#[test] +fn announce_under_a_new_epoch_restarts() { + let origin = id(moq_origin_create()); + let cb = Callback::new(); + let announced_task = id(unsafe { moq_origin_announced(origin, std::ptr::null(), Some(channel_callback), cb.ptr) }); + let (publication, broadcast) = create_broadcast(origin, b"restarted"); + + let first = mint_epoch(); + assert_eq!( + unsafe { moq_publication_announce(publication, &announce_route(0, &first)) }, + 0 + ); + let (announced, info) = cb.recv_route(); + assert_eq!(info.kind, moq_announce_kind::MOQ_ANNOUNCE_KIND_START); + assert_eq!(route_epoch(&info.route), Some(first)); + assert_eq!(moq_origin_announced_free(announced), 0); + + let second = mint_epoch(); + assert_eq!( + unsafe { moq_publication_announce(publication, &announce_route(0, &second)) }, + 0 + ); + let (announced, info) = cb.recv_route(); + assert_eq!(info.kind, moq_announce_kind::MOQ_ANNOUNCE_KIND_RESTART); + assert_eq!(route_epoch(&info.route), Some(second)); + assert_eq!(moq_origin_announced_free(announced), 0); + + assert_eq!(moq_origin_announced_cancel(announced_task), 0); + assert_eq!(cb.recv_announce_terminal(), 0); assert_eq!(moq_publication_free(publication), 0); assert_eq!(moq_publish_close(broadcast), 0); assert_eq!(moq_origin_close(origin), 0); @@ -3913,7 +4036,7 @@ fn consume_announced_close_cancels() { assert_eq!(moq_origin_close(origin), 0); } -/// Publish `path` on `origin` through `route`, which the C ABI cannot set. +/// Publish `path` on `origin` through `route` from Rust, keeping the Rust publication. fn publish(origin: u32, path: &str, route: moq_net::origin::Route) -> moq_net::origin::Publication { let _runtime = crate::ffi::RUNTIME.enter(); let origin = crate::Id::try_from(origin).unwrap();