diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 624ee44d88d..fccde265737 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -107,7 +107,7 @@ inside. The header is the reference: each function carries a doc comment. - **Discovery.** Announcement captures are always a list of wildcard matches, empty when the filter has no wildcards. - **Ending cleanly.** When the input ends, finish every track, then call `moq_publish_catalog_finish()` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `moq_publish_close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close ends with `UNROUTABLE`, which readers take as the publisher leaving, and its handle is released. - **Timeline breaks.** On a seek or pause, call `moq_publish_media_discontinuity`, then keep timestamps moving forward and resume video on a keyframe. -- **Timestamps.** A `moq_timestamp` is exact: `value` units at `scale` units per second, so a 90 kHz tick survives a read and republish. Frames, datagrams, and JSON values point `timestamp` at theirs, NULL on an untimed track and valid as long as the payload. Tracks moq-c creates are timed, so their raw writes need one. JSON and binary writes take one too, converted to their track's milliseconds; NULL stamps the broadcast clock's current time. A zero scale or an out-of-range value fails with `MOQ_ERROR_TIMESTAMP_OVERFLOW`. +- **Timestamps.** A `moq_timestamp` is exact: `value` units at `scale` units per second, so a 90 kHz tick survives a read and republish. Frames, datagrams, and JSON values point `timestamp` at theirs, NULL on an untimed track and valid as long as the payload. Tracks moq-c creates are timed, so their raw writes need one. JSON writes, including window pushes and pops, take one too, converted to their track's milliseconds; NULL stamps the broadcast clock's current time. A zero scale or an out-of-range value fails with `MOQ_ERROR_TIMESTAMP_OVERFLOW`. - **JSON values.** `moq_consume_json_value` fills `moq_json_value` with the document and its frame's timestamp. A snapshot consumer's callback fires for each state in order, but a group superseded before it is read is skipped. - **JSON windows.** `moq_publish_json_window` publishes a sliding run of records: `moq_publish_json_window_push` appends at the back, `moq_publish_json_window_pop` drops from the front, and `moq_publish_json_window_cut` closes the open group at a natural boundary. The catalog has no window mode, so a reader names the track. `moq_consume_json_window_event` fills `moq_json_window_event` with a push (the record at `start`, in `json`), a pop of `[start, end)`, or a skip of `[start, end)` that this reader joined too late to receive. Set `checkpoint_records` to bound what each new group restates for a window you never pop. - **Decoded reads.** `moq_decode_audio` and `moq_decode_video` deliver read ids in order: a frame, a gap where media was lost, or an end where the publisher stopped its encoder. Check `kind` on what `moq_decode_audio_read` or `moq_decode_video_read` writes. Free each id with `moq_decode_audio_read_free` or `moq_decode_video_read_free` promptly: a held video frame holds a decoder buffer, and the decoder stalls. A decoder reads everything the track still holds unless a live player sets `start` to `MOQ_DECODE_START_LATEST`, which joins at the newest group and skips the cached backlog. diff --git a/quest/m1/c/README.md b/quest/m1/c/README.md index c36fabbb527..f2a983a8026 100644 --- a/quest/m1/c/README.md +++ b/quest/m1/c/README.md @@ -29,5 +29,5 @@ Public API: moq-c gains each m1 binding feature. Wire: none. ## Required -- [API gaps](/quest/m1/c/api-gaps.md) - caller timestamps for JSON windows and disabling pre-encoded simulcast rungs +- [API gaps](/quest/m1/c/api-gaps.md) - disabling pre-encoded simulcast rungs after the shared importer support lands - [Every consumer cancels while pending](/quest/m1/c/consumer-cancel.md) - catalog, raw, datagram, and media consumers end cleanly on cancel before their subscription is accepted diff --git a/quest/m1/c/api-gaps.md b/quest/m1/c/api-gaps.md index d53adbd9a99..772bd9c1a27 100644 --- a/quest/m1/c/api-gaps.md +++ b/quest/m1/c/api-gaps.md @@ -1,25 +1,33 @@ -# [S] moq-c closes its small API gaps +# [S] moq-c disables pre-encoded simulcast rungs ## Goal -moq-c stamps a JSON window entry with a caller's timestamp and can disable a -simulcast rung on pre-encoded media. +moq-c can disable a simulcast rung on pre-encoded media, with its lower +rendition's priority and a rate control driven by the reservation. ## Plan -- `moq_publish_json_window_push` and `_pop` always stamp the broadcast clock. - Let them take an optional timestamp like the JSON snapshot and stream - writers do since [moq-dev/moq#6946](https://github.com/moq-dev/moq/pull/6946). - Decided 2026-10-30 (maintainer, /quest-plan of the 2026-10-30 /quest-complete follow-ups): - ✅ add it here / drop. -- Add C ABI support for disabling simulcast rungs on pre-encoded media, which - [moq-dev/moq#7029](https://github.com/moq-dev/moq/pull/7029) added to moq-ffi - for OBS. Decided 2026-10-30 (maintainer, /quest-plan of the 2026-10-30 /quest-complete follow-ups): the C epic's rule (every m1 - binding feature missing from `moq.h` lands before the cut) overrides - #7029's decision 4, which parked it until a C consumer asked: - ✅ add to m1 C gaps / exempt it. - -Update `moq.h`, `rs/moq-c/README.md` and `doc/lib/c` as needed. - -Public API: moq-c gains an optional caller timestamp on JSON window push and -pop, and a call to disable a rung of pre-encoded media. Wire: none. +- Add C ABI support for disabling simulcast rungs on pre-encoded media after + [moq-dev/moq#7029](https://github.com/moq-dev/moq/pull/7029) lands its shared + importer support. It is still open as of 2026-11-02: main lacks + `moq_mux::import::Track::set_enabled`, `import::VideoInit.priority`, and the + AV1/VP8/VP9 importer setters. That PR owns those changes and the moq-ffi + `MoqRateControl`, `MoqVideoInit.lower`, and media producer `set_enabled` + interfaces. Do not duplicate or import its unmerged source changes here. +- After it merges, expose media `set_enabled`, lower-rendition ordering, and + a rate control over the C reservation. Inspect the final shared interfaces + first. Decided 2026-10-30 (maintainer, /quest-plan of the 2026-10-30 + /quest-complete follow-ups): the C epic's rule (every m1 binding feature + missing from `moq.h` lands before the cut) overrides that PR's decision 4, + which parked it until a C consumer asked: ✅ add to m1 C gaps / exempt it. + +Update the generated `moq.h`, `rs/moq-c/README.md` and `doc/lib/c` as needed. +The independent JSON window caller timestamp gap is implemented in +[moq-dev/moq#7614](https://github.com/moq-dev/moq/pull/7614). + +Public API: moq-c gains pre-encoded rung disable, lower-rendition ordering, +and reservation rate control. Wire: none. + +## Required + +- [OBS rung disable](/quest/m1/simulcast-rung-disable.md) - land the shared importer and FFI prerequisites in [moq-dev/moq#7029](https://github.com/moq-dev/moq/pull/7029) before adding their C ABI counterparts diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 9b90aeca5d3..8a782e11d76 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -4,6 +4,13 @@ How to update your code for each breaking change, newest first. ## Unreleased +### JSON window edits take an optional timestamp + +`moq_publish_json_window_push` and `moq_publish_json_window_pop` take a final +`const moq_timestamp *timestamp`. Pass `NULL` to keep stamping the broadcast +clock, or a pointer to the edit's capture time, converted to track milliseconds. +Recompile callers against the regenerated `moq.h`. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `moq_announce_event.has_captures` is removed, changing the C ABI. Read `captures_len` directly: zero means the filter has no wildcards. Rebuild consumers against the regenerated header and matching native library. diff --git a/rs/moq-c/README.md b/rs/moq-c/README.md index bd75c5db695..16835118d7e 100644 --- a/rs/moq-c/README.md +++ b/rs/moq-c/README.md @@ -149,3 +149,12 @@ are exact `moq_timestamp` records, `value` units at `scale` units per second. track's timescale, a microsecond one unless `moq_track_info.timescale` sets another, and `moq_consume_track_frame` points `moq_frame.timestamp` at the timestamp read, valid until the frame is freed, or NULL on an untimed track. + +JSON window pushes and pops accept an optional `const moq_timestamp *timestamp`, +like snapshot updates and stream appends. It is when the edit occurred, +converted to the track's milliseconds; pass NULL to stamp the broadcast clock. + +```c +int32_t moq_publish_json_window_push(uint32_t window, const char *value, uintptr_t value_len, const moq_timestamp *timestamp); +int32_t moq_publish_json_window_pop(uint32_t window, uint64_t count, const moq_timestamp *timestamp); +``` diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 49c7782ff93..74afa774a3c 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -3567,7 +3567,8 @@ pub unsafe extern "C" fn moq_publish_json_snapshot( /// unchanged from the previous update. /// /// `timestamp` is when the value was captured, converted to the track's millisecond timescale; -/// NULL stamps the broadcast clock's current time. +/// NULL stamps the broadcast clock's current time. A timestamp too large for milliseconds fails +/// with `MOQ_ERROR_MOQ` and publishes nothing. /// /// Returns a zero on success, or a negative code on failure. /// @@ -3626,7 +3627,8 @@ pub unsafe extern "C" fn moq_publish_json_stream(broadcast: u32, name: *const c_ /// Append one record to a JSON stream track. `value` is a UTF-8 JSON document. /// /// `timestamp` is when the value was captured, converted to the track's millisecond timescale; -/// NULL stamps the broadcast clock's current time. +/// NULL stamps the broadcast clock's current time. A timestamp too large for milliseconds fails +/// with `MOQ_ERROR_MOQ` and publishes nothing. /// /// Returns a zero on success, or a negative code on failure. /// @@ -3699,28 +3701,47 @@ pub unsafe extern "C" fn moq_publish_json_window( /// Append one record to the back of a JSON window. `value` is a UTF-8 JSON document. /// +/// `timestamp` is when the edit occurred, converted to the track's millisecond timescale; +/// NULL stamps the broadcast clock's current time. A timestamp too large for milliseconds fails +/// with `MOQ_ERROR_MOQ` and publishes nothing. +/// /// Returns a zero on success, or a negative code on failure. /// /// # Safety /// - The caller must ensure `value` is a valid pointer to `value_len` bytes. +/// - `timestamp` must be NULL or a valid pointer to a [moq_timestamp]. #[unsafe(no_mangle)] -pub unsafe extern "C" fn moq_publish_json_window_push(window: u32, value: *const c_char, value_len: usize) -> i32 { +pub unsafe extern "C" fn moq_publish_json_window_push( + window: u32, + value: *const c_char, + value_len: usize, + timestamp: *const moq_timestamp, +) -> i32 { ffi::enter(move || { let window = ffi::parse_id(window)?; let value = unsafe { ffi::parse_slice(value.cast::(), value_len)? }; let value = serde_json::from_slice(value)?; - State::lock().publish.json_window_push(window, value) + let timestamp = unsafe { ffi::parse_timestamp(timestamp)? }; + State::lock().publish.json_window_push(window, value, timestamp) }) } /// Drop `count` records from the front of a JSON window, clamped to what it holds. /// +/// `timestamp` is when the edit occurred, converted to the track's millisecond timescale; +/// NULL stamps the broadcast clock's current time. A timestamp too large for milliseconds fails +/// with `MOQ_ERROR_MOQ` and publishes nothing. +/// /// Returns a zero on success, or a negative code on failure. +/// +/// # Safety +/// - `timestamp` must be NULL or a valid pointer to a [moq_timestamp]. #[unsafe(no_mangle)] -pub extern "C" fn moq_publish_json_window_pop(window: u32, count: u64) -> i32 { +pub unsafe extern "C" fn moq_publish_json_window_pop(window: u32, count: u64, timestamp: *const moq_timestamp) -> i32 { ffi::enter(move || { let window = ffi::parse_id(window)?; - State::lock().publish.json_window_pop(window, count) + let timestamp = unsafe { ffi::parse_timestamp(timestamp)? }; + State::lock().publish.json_window_pop(window, count, timestamp) }) } diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index 48789dbbf2b..8b157e5fba9 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -1369,20 +1369,28 @@ impl Publish { } /// Append one record to the back of a JSON window. - pub fn json_window_push(&mut self, window: Id, value: serde_json::Value) -> Result<(), Error> { + pub fn json_window_push( + &mut self, + window: Id, + value: serde_json::Value, + timestamp: Option, + ) -> Result<(), Error> { let window = &mut self.json_window.get_mut(window).ok_or(Error::TrackNotFound)?.data; - window - .producer - .push(moq_net::Timed::from(&value).at(window.clock.now()))?; + let at = timestamp.unwrap_or_else(|| window.clock.now()); + window.producer.push(moq_net::Timed::from(&value).at(at))?; Ok(()) } /// Drop `count` records from the front of a JSON window, clamped to what it holds. - pub fn json_window_pop(&mut self, window: Id, count: u64) -> Result<(), Error> { + pub fn json_window_pop( + &mut self, + window: Id, + count: u64, + timestamp: Option, + ) -> Result<(), Error> { let window = &mut self.json_window.get_mut(window).ok_or(Error::TrackNotFound)?.data; - window - .producer - .pop(moq_net::Timed::from(count).at(window.clock.now()))?; + let at = timestamp.unwrap_or_else(|| window.clock.now()); + window.producer.pop(moq_net::Timed::from(count).at(at))?; Ok(()) } diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index f37c82b6f66..052ec8199b8 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -2733,7 +2733,7 @@ fn recv_window(cb: &Callback) -> WindowEvent { fn window_push(handle: u32, n: u64) { let value = format!(r#"{{"n":{n}}}"#); assert_eq!( - unsafe { moq_publish_json_window_push(handle, value.as_ptr() as *const c_char, value.len()) }, + unsafe { moq_publish_json_window_push(handle, value.as_ptr() as *const c_char, value.len(), std::ptr::null()) }, 0 ); } @@ -2776,7 +2776,7 @@ fn json_window_publish_consume() { for n in 0..3 { window_push(producer, n); } - assert_eq!(moq_publish_json_window_pop(producer, 2), 0); + assert_eq!(unsafe { moq_publish_json_window_pop(producer, 2, std::ptr::null()) }, 0); let mut last = moq_net::Timestamp::ZERO; for n in 0..3 { @@ -2797,12 +2797,101 @@ fn json_window_publish_consume() { assert_eq!(moq_publish_close(broadcast), 0); assert!(event_cb.recv_terminal() < 0, "an aborted window is not a clean end"); assert!(moq_consume_json_cancel(consumer) < 0, "the task removed itself"); - assert!(moq_publish_json_window_pop(producer, 1) < 0, "the handle was released"); + assert!( + unsafe { moq_publish_json_window_pop(producer, 1, std::ptr::null()) } < 0, + "the handle was released" + ); assert!(moq_publish_json_window_finish(producer) < 0, "the handle was released"); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_origin_close(origin), 0); } +/// Caller timestamps include zero and non-millisecond scales; refusing an invalid timestamp +/// must leave both the records and the emitted events unchanged. +#[test] +fn json_window_caller_timestamps() { + use moq_json_window_kind::*; + + let origin = id(moq_origin_create()); + let path = b"json-window-timestamps"; + let broadcast = publish_broadcast(origin, path); + let name = b"log"; + let config = moq_json_window_config { checkpoint_records: 0 }; + let producer = id(unsafe { moq_publish_json_window(broadcast, name.as_ptr().cast(), name.len(), &config) }); + let consume = request_broadcast(origin, path); + let event_cb = Callback::new(); + let consumer = window_consume(consume, name, &event_cb); + let value = br#"{"n":1}"#; + let invalid = moq_timestamp { value: 1, scale: 0 }; + // Valid, but too large once converted to the track's milliseconds. + let unconvertible = moq_timestamp { + value: (1 << 61) - 1, + scale: 1, + }; + + assert_eq!( + unsafe { moq_publish_json_window_push(producer, value.as_ptr().cast(), value.len(), &invalid) }, + MOQ_ERROR_TIMESTAMP_OVERFLOW + ); + assert_eq!( + unsafe { moq_publish_json_window_push(producer, value.as_ptr().cast(), value.len(), &unconvertible) }, + MOQ_ERROR_MOQ + ); + for timestamp in [ + micros(0), + moq_timestamp { + value: 111_111, + scale: 90_000, + }, + ] { + assert_eq!( + unsafe { moq_publish_json_window_push(producer, value.as_ptr().cast(), value.len(), ×tamp) }, + 0 + ); + } + assert_eq!( + unsafe { moq_publish_json_window_pop(producer, 1, &invalid) }, + MOQ_ERROR_TIMESTAMP_OVERFLOW + ); + assert_eq!( + unsafe { moq_publish_json_window_pop(producer, 1, &unconvertible) }, + MOQ_ERROR_MOQ + ); + assert_eq!( + unsafe { moq_publish_json_window_pop(producer, 1, µs(2_500_999)) }, + 0 + ); + assert_eq!(moq_publish_json_window_finish(producer), 0); + + for (n, millis) in [0, 1234].into_iter().enumerate() { + let (kind, range, json, timestamp) = recv_window(&event_cb); + assert_eq!( + (kind, range, json, timestamp), + ( + MOQ_JSON_WINDOW_KIND_PUSH, + n as u64..n as u64 + 1, + Some(serde_json::json!({ "n": 1 })), + Some(moq_net::Timestamp::from_millis(millis).unwrap()), + ) + ); + } + let (kind, range, json, timestamp) = recv_window(&event_cb); + assert_eq!( + (kind, range, json, timestamp), + ( + MOQ_JSON_WINDOW_KIND_POP, + 0..1, + None, + Some(moq_net::Timestamp::from_millis(2500).unwrap()) + ) + ); + assert_eq!(event_cb.recv_terminal(), 0); + assert!(moq_consume_json_cancel(consumer) < 0, "the task removed itself"); + assert_eq!(moq_consume_close(consume), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + /// A join past the refresh ratio gets the window restated in a fresh group holding only the /// checkpoint's records, and a finished window still serves a reader joining afterwards, which /// learns of the older records as a skip. diff --git a/rs/moq-json/src/snapshot/mod.rs b/rs/moq-json/src/snapshot/mod.rs index 15bdb500dab..20536b583fe 100644 --- a/rs/moq-json/src/snapshot/mod.rs +++ b/rs/moq-json/src/snapshot/mod.rs @@ -127,6 +127,26 @@ mod test { } } + /// A timestamp the track's millisecond scale can't hold is refused before a replacement group + /// is published, so the last value stays readable. + #[test] + fn a_timed_track_refuses_an_unconvertible_timestamp() { + let huge = moq_net::Timestamp::from_scale((1 << 61) - 1, 1).unwrap(); + for deltas in [false, true] { + let (mut producer, track) = producer_on(cfg(deltas), timed()); + let at = moq_net::Timestamp::from_millis(100).unwrap(); + producer + .update(moq_net::Timed::from(&json!({ "n": 1 })).at(at)) + .unwrap(); + assert!(matches!( + producer.update(moq_net::Timed::from(&json!({ "n": 2 })).at(huge)), + Err(crate::Error::Net(moq_net::Error::TimestampMismatch)) + )); + producer.finish().unwrap(); + assert_eq!(drain(track), vec![json!({ "n": 1 })], "deltas={deltas}"); + } + } + #[test] fn a_bare_publish_stays_untimed() { let track = moq_net::broadcast::Info::new() @@ -971,24 +991,29 @@ mod test { assert!(matches!(producer.update(&json!({ "a": 3 })), Err(crate::Error::Net(_)))); } - /// A track whose timescale is extreme enough that converting a wall-clock timestamp into it - /// overflows, so `write_frame` rejects every frame. That stands in for any post-`append_group` - /// write failure (the reported one is a frame over moq-net's 32 MB per-group cache) without - /// allocating 32 MB to provoke it. + /// A timed track whose first group already starts at ten seconds, so a group that a frame at + /// [`rewound`] opens is published by `append_group` and then refused by `write_frame` for + /// starting before it. That stands in for any post-`append_group` write failure (the reported + /// one is a frame over moq-net's 32 MB per-group cache) without allocating 32 MB to provoke it. fn rejecting_track() -> moq_net::track::Producer { - let mut info = timed(); - info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap()); - - moq_net::broadcast::Info::new() + let track = moq_net::broadcast::Info::new() .produce() - .create_track("test", Some(info)) - .unwrap() + .create_track("test", timed()) + .unwrap(); + let mut group = track.append_group().unwrap(); + group + .write_frame( + moq_net::Timestamp::from_millis(10_000).unwrap(), + bytes::Bytes::from_static(b"{}"), + ) + .unwrap(); + group.finish().unwrap(); + track } - /// A time [`rejecting_track`] can't hold: it matches the timed track, so the publish gets as - /// far as the group before the conversion fails. - fn late() -> moq_net::Timestamp { - moq_net::Timestamp::from_millis(10_000).unwrap() + /// Before [`rejecting_track`]'s first group starts, so a group opened at it is refused. + fn rewound() -> moq_net::Timestamp { + moq_net::Timestamp::from_millis(1_000).unwrap() } /// `append_group` publishes the group before the frame is written, and a rejected frame does not @@ -1001,7 +1026,7 @@ mod test { let mut producer = Producer::::new(track, cfg(false)); assert!(matches!( - producer.update(moq_net::Timed::from(&json!({ "a": 1 })).at(late())), + producer.update(moq_net::Timed::from(&json!({ "a": 1 })).at(rewound())), Err(crate::Error::Net(_)) )); diff --git a/rs/moq-json/src/snapshot/producer.rs b/rs/moq-json/src/snapshot/producer.rs index 6602573b452..aa858fba387 100644 --- a/rs/moq-json/src/snapshot/producer.rs +++ b/rs/moq-json/src/snapshot/producer.rs @@ -362,6 +362,7 @@ impl Inner { impl Inner { fn update(&mut self, payload: Timed<&T>) -> Result> { + let timestamp = self.track.on_track(payload.at)?; // Split the borrow so `frame` can hold the encoder while `track` is written through. let Inner { track, encoder, .. } = self; @@ -373,7 +374,6 @@ impl Inner { // resynchronizes with a fresh snapshot. Most failures kill the track outright, but a rejected // frame (too large) doesn't, and a delta against a snapshot no consumer ever saw is unreadable. // The value stays the last one written, so a refresh restates that rather than the rejection. - let timestamp = payload.at; track.write(timestamp, &frame)?; let size = frame.payload.len(); frame.commit(); @@ -430,7 +430,18 @@ impl Track { Ok(()) } - /// Write one encoded frame at `timestamp`, rolling a group when it's a snapshot. + /// `timestamp` at the track's timescale, or [`moq_net::Error::TimestampMismatch`] when it is + /// missing on a timed track, present on an untimed one, or beyond what the scale can hold. + /// + /// An update checks this before encoding. Found by `write_frame` instead, the mismatch would + /// come after a snapshot published its group, leaving an empty newest group in place of the + /// last value, and the dropped frame would reset the encoder, rolling a group for nothing. + fn on_track(&self, timestamp: Option) -> Result> { + Ok(moq_net::group::on_track(timestamp, self.inner.info().timescale)?) + } + + /// Write one encoded frame at a timestamp already [on the track](Self::on_track), rolling a + /// group when it's a snapshot. fn write(&mut self, timestamp: Option, encoded: &Encoded) -> Result<()> { // Check before touching a group. `write_snapshot` closes the previous group and publishes a // new one before the frame is written, so discovering the limit inside `write_frame` would @@ -439,10 +450,6 @@ impl Track { if encoded.payload.len() as u64 > moq_net::group::MAX_CACHE_BYTES { return Err(moq_net::Error::FrameTooLarge.into()); } - // The same goes for a timestamp that doesn't match the track. - if timestamp.is_some() != self.inner.info().timescale.is_some() { - return Err(moq_net::Error::TimestampMismatch.into()); - } match encoded.keyframe { true => self.write_snapshot(timestamp, encoded.payload.clone()), diff --git a/rs/moq-json/src/stream/mod.rs b/rs/moq-json/src/stream/mod.rs index 24f36ca9576..21654d46769 100644 --- a/rs/moq-json/src/stream/mod.rs +++ b/rs/moq-json/src/stream/mod.rs @@ -73,6 +73,27 @@ mod test { assert_eq!(drain(consume(track)), vec![json!({ "n": 1 }), json!({ "n": 3 })]); } + /// A timestamp the track's millisecond scale can't hold is refused like a missing one, rather + /// than reaching the group as a hole that ends the log. + #[test] + fn a_timed_track_refuses_an_unconvertible_timestamp() { + let (mut producer, track) = producer_on(Config::default(), timed()); + let at = moq_net::Timestamp::from_millis(100).unwrap(); + let huge = moq_net::Timestamp::from_scale((1 << 61) - 1, 1).unwrap(); + producer + .append(moq_net::Timed::from(&json!({ "n": 1 })).at(at)) + .unwrap(); + assert!(matches!( + producer.append(moq_net::Timed::from(&json!({ "n": 2 })).at(huge)), + Err(crate::Error::Net(moq_net::Error::TimestampMismatch)) + )); + producer + .append(moq_net::Timed::from(&json!({ "n": 3 })).at(at)) + .unwrap(); + producer.finish().unwrap(); + assert_eq!(drain(consume(track)), vec![json!({ "n": 1 }), json!({ "n": 3 })]); + } + #[test] fn a_bare_publish_stays_untimed() { let track = moq_net::broadcast::Info::new() @@ -246,24 +267,34 @@ mod test { ); } - /// A track whose timescale is extreme enough that converting a wall-clock timestamp into it - /// overflows, so `write_frame` rejects every frame. That stands in for any post-`append_group` - /// write failure (the reported one is a frame over moq-net's 32 MB per-group cache) without - /// allocating 32 MB to provoke it. + /// A timed track whose first group already starts at ten seconds, so a group that a frame at + /// [`rewound`] opens is published by `append_group` and then refused by `write_frame` for + /// starting before it. That stands in for any post-`append_group` write failure (the reported + /// one is a frame over moq-net's 32 MB per-group cache) without allocating 32 MB to provoke it. fn rejecting_track() -> moq_net::track::Producer { - let mut info = timed(); - info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap()); - - moq_net::broadcast::Info::new() + let track = moq_net::broadcast::Info::new() .produce() - .create_track("test", Some(info)) - .unwrap() + .create_track("test", timed()) + .unwrap(); + let mut group = track.append_group().unwrap(); + group + .write_frame( + moq_net::Timestamp::from_millis(10_000).unwrap(), + bytes::Bytes::from_static(b"{}"), + ) + .unwrap(); + group.finish().unwrap(); + track + } + + /// Before [`rejecting_track`]'s first group starts, so a group opened at it is refused. + fn rewound() -> moq_net::Timestamp { + moq_net::Timestamp::from_millis(1_000).unwrap() } - /// `value` at a time [`rejecting_track`] can't hold: it matches the timed track, so the - /// append gets as far as the group before the conversion fails. + /// `value` at [`rewound`], so the append gets as far as a published group before it fails. fn late(value: &Value) -> moq_net::Timed<&Value> { - moq_net::Timed::from(value).at(moq_net::Timestamp::from_millis(10_000).unwrap()) + moq_net::Timed::from(value).at(rewound()) } /// A failed write must reach the consumer, not just the caller. A clean close drains a reader to @@ -280,9 +311,8 @@ mod test { Err(crate::Error::Net(_)) )); - let waiter = kio::Waiter::noop(); assert!( - matches!(subscriber.poll_recv_group(&waiter), Poll::Ready(Err(_))), + fails(&mut subscriber), "a truncated log must surface an error rather than read as a completed one" ); } @@ -303,13 +333,21 @@ mod test { "a second append must fail on the ended track rather than open another group" ); - // A subscriber taken after the abort still exists; it surfaces the failure on its first read, - // which is how a late reader learns the log is truncated. + // A subscriber taken after the abort still exists; it surfaces the failure once past the + // groups it can still read, which is how a late reader learns the log is truncated. + assert!(fails(&mut producer.consume())); + } + + /// Whether `subscriber` reaches a failure after reading past the groups it already has. + fn fails(subscriber: &mut moq_net::track::Subscriber) -> bool { let waiter = kio::Waiter::noop(); - assert!(matches!( - producer.consume().poll_recv_group(&waiter), - Poll::Ready(Err(_)) - )); + loop { + match subscriber.poll_recv_group(&waiter) { + Poll::Ready(Ok(Some(_))) => continue, + Poll::Ready(Err(_)) => return true, + _ => return false, + } + } } /// A completed log is still readable, so finishing must not end the track the way an abort does. diff --git a/rs/moq-json/src/stream/producer.rs b/rs/moq-json/src/stream/producer.rs index e855cc520e2..c4a6f42ce9c 100644 --- a/rs/moq-json/src/stream/producer.rs +++ b/rs/moq-json/src/stream/producer.rs @@ -69,7 +69,8 @@ impl Producer { /// /// The log accepts appends for its track's life. A record larger than a frame may be is refused /// with [`moq_net::Error::FrameTooLarge`] before anything is written, leaving the log intact. A - /// timestamp whose presence doesn't match the track's timescale is refused with + /// timestamp that doesn't match the track's timescale (missing on a timed track, present on an + /// untimed one, or beyond what its scale can hold) is refused with /// [`moq_net::Error::TimestampMismatch`] the same way, so a corrected append carries on. /// /// Any other failure ends the track. A log missing a record is not the lossless log a stream @@ -124,11 +125,9 @@ impl Inner { // Split the borrow so `record` can hold the encoder while `track` is written through. let Inner { track, encoder } = self; - // A timestamp that doesn't match the track is refused before anything changes: found by - // `write`, it would be a hole that ends the whole log. - if payload.at.is_some() != track.inner.info().timescale.is_some() { - return Err(moq_net::Error::TimestampMismatch.into()); - } + // A timestamp that doesn't match the track, or that its scale can't hold, is refused before + // anything changes: found by `write`, it would be a hole that ends the whole log. + let at = moq_net::group::on_track(payload.at, track.inner.info().timescale)?; // Encode first, so a value that can't be serialized doesn't publish an empty group that // subscribers would advance into and wait on. Opening the group afterwards is safe because @@ -149,7 +148,7 @@ impl Inner { let opened = track.open(); let published = opened.is_ok(); let result = match opened { - Ok(()) => track.write(payload.at, record.payload()), + Ok(()) => track.write(at, record.payload()), Err(err) => Err(err), }; diff --git a/rs/moq-json/src/window/mod.rs b/rs/moq-json/src/window/mod.rs index fa35439f517..6aca6243446 100644 --- a/rs/moq-json/src/window/mod.rs +++ b/rs/moq-json/src/window/mod.rs @@ -122,6 +122,43 @@ mod test { assert_eq!(pushed, vec![rec(1), rec(2)]); } + /// A timestamp the track's millisecond scale can't hold is refused before encoding: a header + /// publishes no group, and an op leaves the open group to carry on rather than rolling it. + #[test] + fn a_timed_track_refuses_an_unconvertible_timestamp() { + let (mut producer, mut track) = producer_on(ProducerConfig::default(), timed()); + let at = moq_net::Timestamp::from_millis(100).unwrap(); + let huge = moq_net::Timestamp::from_scale((1 << 61) - 1, 1).unwrap(); + let largest = |track: &mut moq_net::track::Subscriber| { + track.largest().now_or_never().flatten().map(|position| position.group) + }; + let refused = |result: crate::Result<()>| { + assert!(matches!( + result, + Err(crate::Error::Net(moq_net::Error::TimestampMismatch)) + )); + }; + + refused(producer.push(moq_net::Timed::from(&rec(1)).at(huge))); + assert_eq!(largest(&mut track), None, "a refused header published a group"); + + producer.push(moq_net::Timed::from(&rec(1)).at(at)).unwrap(); + refused(producer.push(moq_net::Timed::from(&rec(2)).at(huge))); + refused(producer.pop(moq_net::Timed::from(1u64).at(huge))); + producer.push(moq_net::Timed::from(&rec(2)).at(at)).unwrap(); + assert_eq!(largest(&mut track), Some(0), "a refused op rolled the group"); + + producer.finish().unwrap(); + let pushed: Vec<_> = drain(&mut consumer(track)) + .into_iter() + .filter_map(|event| match event { + Event::Push { value, .. } => Some(value), + _ => None, + }) + .collect(); + assert_eq!(pushed, vec![rec(1), rec(2)]); + } + #[test] fn events_carry_their_frames_timestamp() { let track = moq_net::broadcast::Info::new() @@ -250,15 +287,29 @@ mod test { Consumer::new(track, ConsumerConfig::default()) } - /// A track whose timestamp conversion rejects every frame after its group is published. + /// A timed track whose first group already starts at ten seconds, so a group that a frame at + /// [`rewound`] opens is published by `append_group` and then refused by `write_frame` for + /// starting before it. That stands in for any post-`append_group` write failure (the reported + /// one is a frame over moq-net's 32 MB per-group cache) without allocating 32 MB to provoke it. fn rejecting_track() -> moq_net::track::Producer { - let mut info = timed(); - info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap()); - - moq_net::broadcast::Info::new() + let track = moq_net::broadcast::Info::new() .produce() - .create_track("test", Some(info)) - .unwrap() + .create_track("test", timed()) + .unwrap(); + let mut group = track.append_group().unwrap(); + group + .write_frame( + moq_net::Timestamp::from_millis(10_000).unwrap(), + bytes::Bytes::from_static(b"{}"), + ) + .unwrap(); + group.finish().unwrap(); + track + } + + /// Before [`rejecting_track`]'s first group starts, so a group opened at it is refused. + fn rewound() -> moq_net::Timestamp { + moq_net::Timestamp::from_millis(1_000).unwrap() } /// Drain every event currently available without blocking. @@ -790,9 +841,8 @@ mod test { let mut subscriber = track.subscribe(None).ordered(); let mut producer = Producer::::new(track, ProducerConfig::default()); - // Too late for the timescale, so the header group is published before the write fails. - let late = moq_net::Timestamp::from_millis(10_000).unwrap(); - assert!(producer.push(moq_net::Timed::from(&rec(1)).at(late)).is_err()); + // Rewound past the first group, so the header group is published before the write fails. + assert!(producer.push(moq_net::Timed::from(&rec(1)).at(rewound())).is_err()); assert_eq!(producer.range(), 0..0); assert!(producer.window().is_empty()); diff --git a/rs/moq-json/src/window/producer.rs b/rs/moq-json/src/window/producer.rs index c79b4578976..5da7a120a93 100644 --- a/rs/moq-json/src/window/producer.rs +++ b/rs/moq-json/src/window/producer.rs @@ -157,6 +157,7 @@ struct Inner { impl Inner { fn pop(&mut self, count: Timed) -> Result<()> { self.ensure_open()?; + let at = self.track.on_track(count.at)?; let Inner { track, encoder, .. } = self; let Some(frame) = encoder.pop(count.value)? else { @@ -165,9 +166,9 @@ impl Inner { // A failed write drops the frame uncommitted. The pop is discarded, and the next edit opens a // new group, since a consumer may hold the group the frame was meant for. - track.write(count.at, &frame)?; + track.write(at, &frame)?; frame.commit(); - self.at = self.at.max(count.at); + self.at = self.at.max(at); Ok(()) } @@ -224,12 +225,13 @@ impl Inner { impl Inner { fn push(&mut self, value: Timed<&T>) -> Result<()> { self.ensure_open()?; + let at = self.track.on_track(value.at)?; let Inner { track, encoder, .. } = self; let frame = encoder.push(value.value)?; - track.write(value.at, &frame)?; + track.write(at, &frame)?; frame.commit(); - self.at = self.at.max(value.at); + self.at = self.at.max(at); Ok(()) } @@ -247,14 +249,19 @@ struct Track { } impl Track { - /// Write one encoded frame, rolling a group when it is a header. - fn write(&mut self, timestamp: Option, encoded: &Encoded) -> Result<()> { - // Check before touching a group: a header publishes a new group before its frame is - // written, so a mismatch found by `write_frame` would leave an empty newest group behind. - if timestamp.is_some() != self.inner.info().timescale.is_some() { - return Err(moq_net::Error::TimestampMismatch.into()); - } + /// `timestamp` at the track's timescale, or [`moq_net::Error::TimestampMismatch`] when it is + /// missing on a timed track, present on an untimed one, or beyond what the scale can hold. + /// + /// An edit checks this before encoding. Found by `write_frame` instead, the mismatch would + /// come after a header published its group, leaving an empty newest group behind, and the + /// dropped frame would reset the encoder, rolling a group for an edit that changed nothing. + fn on_track(&self, timestamp: Option) -> Result> { + Ok(moq_net::group::on_track(timestamp, self.inner.info().timescale)?) + } + /// Write one encoded frame at a timestamp already [on the track](Self::on_track), rolling a + /// group when it is a header. + fn write(&mut self, timestamp: Option, encoded: &Encoded) -> Result<()> { match encoded.keyframe { true => self.write_header(timestamp, encoded.payload.clone()), false => self.write_op(timestamp, encoded.payload.clone()), diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index c6c965f9724..a6cd6b5e7df 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -1368,7 +1368,10 @@ impl Drop for GroupState { /// A track is all timed or all untimed, so a frame must match it: a timed frame on an /// untimed track, or an untimed one on a timed track, is [`Error::TimestampMismatch`], /// as is a timestamp the track's scale can't hold. -pub(super) fn on_track(timestamp: Option, timescale: Option) -> Result> { +/// +/// A writer that changes state before its frame reaches [`Producer::write_frame`], such as +/// publishing a group for it, checks with this first so a refused timestamp changes nothing. +pub fn on_track(timestamp: Option, timescale: Option) -> Result> { match (timestamp, timescale) { (Some(timestamp), Some(timescale)) => Ok(Some( timestamp.convert(timescale).map_err(|_| Error::TimestampMismatch)?,