diff --git a/dart/moq/lib/src/durations.dart b/dart/moq/lib/src/durations.dart index f6ffdcaa0aa..0738ec44e14 100644 --- a/dart/moq/lib/src/durations.dart +++ b/dart/moq/lib/src/durations.dart @@ -35,8 +35,9 @@ extension BackoffDuration on MoqBackoff { /// Duration views over the subscription knobs. extension SubscriptionDuration on MoqSubscription { - /// Upper bound on buffering before a stalled group is skipped. - Duration get maxDelay => Duration(microseconds: maxDelayUs); + /// Upper bound on buffering before a stalled group is skipped, or null for the default. + Duration? get maxDelay => + maxDelayUs == null ? null : Duration(microseconds: maxDelayUs!); } /// Duration views over the publisher-side track settings. diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 48f643ad1af..3963534e6d1 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -767,12 +767,12 @@ class FfiConverterMoqRange { class MoqSubscription { final int priority; - final int maxDelayUs; + final int? maxDelayUs; final List? ranges; final MoqOrder? order; MoqSubscription({ this.priority = 0, - this.maxDelayUs = 0, + this.maxDelayUs = null, this.ranges = null, this.order = null, }); @@ -790,7 +790,7 @@ class FfiConverterMoqSubscription { ); final priority = priority_lifted.value; new_offset += priority_lifted.bytesRead; - final maxDelayUs_lifted = FfiConverterUInt64.read( + final maxDelayUs_lifted = FfiConverterOptionalUInt64.read( Uint8List.view(buf.buffer, new_offset), ); final maxDelayUs = maxDelayUs_lifted.value; @@ -819,7 +819,7 @@ class FfiConverterMoqSubscription { static RustBuffer lower(MoqSubscription value) { final total_length = FfiConverterInt8.allocationSize(value.priority) + - FfiConverterUInt64.allocationSize(value.maxDelayUs) + + FfiConverterOptionalUInt64.allocationSize(value.maxDelayUs) + FfiConverterOptionalSequenceMoqRange.allocationSize(value.ranges) + FfiConverterOptionalMoqOrder.allocationSize(value.order) + 0; @@ -834,7 +834,7 @@ class FfiConverterMoqSubscription { value.priority, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterUInt64.write( + new_offset += FfiConverterOptionalUInt64.write( value.maxDelayUs, Uint8List.view(buf.buffer, new_offset), ); @@ -851,7 +851,7 @@ class FfiConverterMoqSubscription { static int allocationSize(MoqSubscription value) { return FfiConverterInt8.allocationSize(value.priority) + - FfiConverterUInt64.allocationSize(value.maxDelayUs) + + FfiConverterOptionalUInt64.allocationSize(value.maxDelayUs) + FfiConverterOptionalSequenceMoqRange.allocationSize(value.ranges) + FfiConverterOptionalMoqOrder.allocationSize(value.order) + 0; diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index f278dcc22b7..b7d0b6205b8 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -256,9 +256,12 @@ own ranges cover, resolved again whenever it replaces them, even when another subscriber of the same track asked for more. Widening them, including lowering the start, also serves the cached groups they newly cover, never one already served. A relay merges its subscribers' ranges into one upstream request, which -today's wire versions carry as the single span covering them. A time anchor is -resolved from the groups the serving hop holds; upstream it asks for everything -max delay allows, and an untimed track refuses it. At the live edge, a time end +today's wire versions carry as the single span covering them. A relative or time +anchor resolves on the hop that publishes the track, and over moq-lite-07, which +reports the live edge and media time and asks upstream for every group the budget +allows. Over an older lite or any moq-transport version, a subscriber gets every +anchor but the live and default starts refused as unsupported, and an untimed +track refuses a time anchor. At the live edge, a time end may let one group through before its first frame shows it starts past the end: the first group that has yet to start is handed out without waiting for that frame, and the next one waits on it. A datagram is judged by its own timestamp. diff --git a/go/wrapper/moq_test.go b/go/wrapper/moq_test.go index 677e1441b2e..a0e858d367a 100644 --- a/go/wrapper/moq_test.go +++ b/go/wrapper/moq_test.go @@ -830,8 +830,9 @@ func TestAscendingSubscriptionReadsCachedBacklogOldestFirst(t *testing.T) { } order := moq.OrderAscending + maxDelay := 10 * time.Second consumer, err := track.Consume(&moq.Subscription{ - MaxDelay: 10 * time.Second, + MaxDelay: &maxDelay, Ranges: []moq.Range{{Start: moq.AnchorRelative{Groups: 2}}}, Order: &order, }) diff --git a/go/wrapper/records.go b/go/wrapper/records.go index 8eaec884773..69dbe7e1944 100644 --- a/go/wrapper/records.go +++ b/go/wrapper/records.go @@ -58,19 +58,19 @@ func datagramFromFFI(d *ffi.MoqDatagram) *Datagram { } // Subscription holds subscriber-side delivery preferences: priority, the max delay -// a non-latest group may fall behind before it is skipped (zero skips at once, and it -// caps every range), the ranges of the track to deliver (nil for everything MaxDelay +// a non-latest group may fall behind before it is skipped (nil for moq-net's default of +// zero, which skips at once; it caps every range), the ranges of the track to deliver (nil for everything MaxDelay // still allows, while an empty non-nil slice asks for nothing), and the order the // publisher sends pending groups in (nil for newest first). type Subscription struct { Priority int8 - MaxDelay time.Duration + MaxDelay *time.Duration Ranges []Range Order *Order } func (s Subscription) ffi() (ffi.MoqSubscription, error) { - maxDelay, err := bridge.Duration(s.MaxDelay, "max delay") + maxDelay, err := bridge.OptionalDuration(s.MaxDelay, "max delay") var ranges *[]ffi.MoqRange if s.Ranges != nil { ranges = &s.Ranges diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt index 436a92a6717..e58584abbe3 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt @@ -53,9 +53,9 @@ val QuicConfig.idleTimeout: Duration? val QuicConfig.keepAlive: Duration? get() = keepAliveUs?.toLong()?.microseconds -/** Upper bound on buffering before a stalled group is skipped. */ -val Subscription.maxDelay: Duration - get() = maxDelayUs.toLong().microseconds +/** Upper bound on buffering before a stalled group is skipped, or null for the default. */ +val Subscription.maxDelay: Duration? + get() = maxDelayUs?.toLong()?.microseconds /** Maximum age of a non-latest group before the publisher evicts it, or null for the default. */ val TrackInfo.maxAge: Duration? diff --git a/py/moq-rs/moq/_records.py b/py/moq-rs/moq/_records.py index f1e4be00755..baaa51c789d 100644 --- a/py/moq-rs/moq/_records.py +++ b/py/moq-rs/moq/_records.py @@ -94,21 +94,22 @@ class Subscription: """Subscriber-side delivery preferences, mirroring moq-net's ``track::Subscription``. ``priority`` is in ``-128..=127``, higher first, and defaults to 0, the midpoint. - ``max_delay`` is how far a non-latest group may fall behind before it is skipped; - zero skips at once, and it caps every range. ``ranges`` are the spans of the track to - deliver, ``None`` for everything ``max_delay`` still allows. ``order`` is which pending - group the publisher sends first, ``None`` for newest first. + ``max_delay`` is how far a non-latest group may fall behind before it is skipped, + ``None`` for moq-net's default of zero, which skips at once; it caps every range. + ``ranges`` are the spans of the track to deliver, ``None`` for everything ``max_delay`` + still allows. ``order`` is which pending group the publisher sends first, ``None`` for + newest first. """ priority: int = 0 - max_delay: timedelta = timedelta(0) + max_delay: timedelta | None = None ranges: Sequence[MoqRange] | None = None order: MoqOrder | None = None def _ffi(self) -> MoqSubscription: return MoqSubscription( priority=self.priority, - max_delay_us=_to_us(self.max_delay, "max_delay"), + max_delay_us=_opt_us(self.max_delay, "max_delay"), ranges=None if self.ranges is None else list(self.ranges), order=self.order, ) diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index ce991f34eee..3c33cf8e1d0 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -73,7 +73,6 @@ Rejected in planning: - [Container formats name inputs](/quest/m1/api-audit/container-format.md) - `ContainerFormat` names what is parsed, never how it is packaged; `ContainerStream` and TS passthrough internals are gone - [moq-net exports with a consumer](/quest/m1/api-audit/net-exports.md) - Rust and JS model exports without a consumer go private, wrong ones are deleted, and JS names mirror Rust - [One clean close](/quest/m1/api-audit/clean-close.md) - a clean session close has one spelling and failures are recorded inside moq-net -- [Ranges refuse what the wire drops](/quest/m1/api-audit/ranges-wire.md) - relative and time anchors fail loud where a wire cannot carry them, `floor()` honors the subscription, and bindings take an optional max delay - [Derived origins borrow](/quest/m1/api-audit/derived-origin-close.md) - closing a JS `scope()` or `withPreference()` handle releases only that handle - [Wrapper config names](/quest/m1/api-audit/wrapper-config.md) - Python, Go, and Dart client options use the FFI's names and nesting - [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them diff --git a/quest/m1/api-audit/ranges-wire.md b/quest/m1/api-audit/ranges-wire.md deleted file mode 100644 index 179ed5991e0..00000000000 --- a/quest/m1/api-audit/ranges-wire.md +++ /dev/null @@ -1,27 +0,0 @@ -# [S] Ranges refuse what the wire drops - -## Goal - -A subscription range with a relative or time anchor fails loud on a hop -whose upstream wire cannot carry it, instead of silently becoming live. -`floor()` honors `set_groups`, and the bindings take max delay as an -optional duration before their shape freezes. - -## Plan - -Decided 2026-10-22 (API audit). Keep the general range API (the -[subscribe ranges](/quest/m1/subscribe-ranges/README.md) line lands lite-07 -before the cut); fix only what the audit found broken: - -- `Anchor::Relative` and `Anchor::Time` are resolved at the serving hop and - otherwise degrade to live on wires that cannot carry them (published lite - versions, moq-transport without a local resolution). Refuse instead, with - a test on each wire. -- `floor()` ignores `set_groups` (`rs/moq-net/src/model/track.rs`); fix and - test. -- Bindings expose `max_delay_us` with zero meaning the live default; take - `max_delay: Option` (or the binding's optional) like Rust's - `Fetch::max_delay`. moq-hls `history.rs` works around the zero default with - a `(1 << 53) - 1` ms budget; remove that once it can say "unset". - -Public API: moq-ffi and wrappers breaking (max delay). Wire: none. diff --git a/quest/m1/subscribe-ranges/lite.md b/quest/m1/subscribe-ranges/lite.md index 6600a91b9ad..e87612bd4d1 100644 --- a/quest/m1/subscribe-ranges/lite.md +++ b/quest/m1/subscribe-ranges/lite.md @@ -35,6 +35,14 @@ that is the Largest Object's group, so a newest group with no objects yet is not counted until its first object arrives; the relay reports what it has rather than waiting for the peer. +Today a track copy received over lite-06 or earlier, or any moq-transport +draft, refuses relative and time anchors with `Unsupported`, while a lite-07 +copy resolves them locally (it learns the largest and media time and asks for +everything the budget allows; `lite::Version::resolves_anchors`). Each mapping +above that makes an older wire carry an anchor lifts that wire's refusal for +that anchor only, and extends the per-wire table in +`rs/moq-net/tests/integration/range_anchors.rs`. + Test the lite-05 relay case #5000 left out, on mock time beside `rs/moq-net/tests/integration/late_lower_group.rs`: a live subscriber already reading group 1, then one from `Absolute(0)`, and a fresh group 0 reaches only the diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index cc7a47a03ce..81ea608746c 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -591,6 +591,8 @@ pub struct moq_subscription { /// Zero skips immediately. Enforced by the publisher's cache and by any local buffering. /// It caps every range. pub max_delay_us: u64, + /// Whether `max_delay_us` is set. When false, moq-net's default applies: zero. + pub max_delay_present: bool, /// The spans of the track to deliver, or NULL for everything `max_delay_us` still /// considers fresh. A non-NULL pointer with `ranges_len` 0 asks for nothing. @@ -687,8 +689,10 @@ pub(crate) unsafe fn parse_subscription( }; let mut out = moq_net::track::Subscription::default() .with_priority(subscription.priority) - .with_max_delay(std::time::Duration::from_micros(subscription.max_delay_us)) .with_order(order); + if subscription.max_delay_present { + out = out.with_max_delay(std::time::Duration::from_micros(subscription.max_delay_us)); + } if !subscription.ranges.is_null() { let ranges = unsafe { std::slice::from_raw_parts(subscription.ranges, subscription.ranges_len) }; let ranges = ranges diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index a07199245b5..5ed88eec995 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -1486,6 +1486,7 @@ fn publish_track_invalid_broadcast() { let subscription = moq_subscription { priority: 1, max_delay_us: 0, + max_delay_present: false, ranges: std::ptr::null(), ranges_len: 0, order: 0, @@ -1595,6 +1596,7 @@ fn raw_track_publish_consume() { let subscription = moq_subscription { priority: 0, max_delay_us: 1_000_000, + max_delay_present: true, ranges: std::ptr::null(), ranges_len: 0, order: 0, @@ -1817,6 +1819,7 @@ fn raw_track_subscription_options_and_update() { let subscription = moq_subscription { priority: 5, max_delay_us: 25_000, + max_delay_present: true, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -1931,6 +1934,7 @@ fn raw_track_time_anchor() { let subscription = moq_subscription { priority: 5, max_delay_us: 1_000_000, + max_delay_present: true, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -2031,6 +2035,7 @@ fn raw_track_update_refuses_time_on_untimed() { let update = moq_subscription { priority: 3, max_delay_us: 1_000_000, + max_delay_present: true, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -2089,6 +2094,7 @@ fn raw_track_relative_update_counts_from_the_track_edge() { let subscribe = |ranges: &[moq_range]| moq_subscription { priority: 0, max_delay_us: 60_000_000, + max_delay_present: true, ranges: ranges.as_ptr(), ranges_len: ranges.len(), order: 0, @@ -2359,6 +2365,7 @@ fn publish_finish_serves_finished_tracks_until_closed() { let subscription = moq_subscription { priority: 0, max_delay_us: 0, + max_delay_present: false, ranges: std::ptr::null(), ranges_len: 0, order: 0, @@ -5875,6 +5882,7 @@ fn read_raw_frames(consume: u32, name: &[u8], count: usize) -> Vec> { let subscription = moq_subscription { priority: 0, max_delay_us: 1_000_000, + max_delay_present: true, ranges: std::ptr::null(), ranges_len: 0, order: 0, diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index 4b34bdae9d6..1c48101c65f 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -31,13 +31,14 @@ pub struct MoqSubscription { /// The default 0 is the midpoint. #[uniffi(default = 0)] pub priority: i8, - /// Maximum age of a non-latest group before it is skipped, in microseconds. - /// `0` skips immediately; a larger value tolerates that much reordering. + /// Maximum age of a non-latest group before it is skipped, in microseconds, or null + /// for moq-net's default: `0`, which skips immediately. A larger value tolerates that + /// much reordering. /// /// Enforced both by the publisher's cache (sent on the wire) and by any local /// buffering, such as [`MoqMediaTrackConsumer`]'s jitter buffer. It caps every range. - #[uniffi(default = 0)] - pub max_delay_us: u64, + #[uniffi(default = None)] + pub max_delay_us: Option, /// The spans of the track to deliver, or null for everything `max_delay_us` still /// considers fresh. An empty list asks for nothing. /// @@ -155,9 +156,10 @@ impl TryFrom for moq_net::track::Subscription { type Error = MoqError; fn try_from(s: MoqSubscription) -> Result { - let mut subscription = moq_net::track::Subscription::default() - .with_priority(s.priority) - .with_max_delay(std::time::Duration::from_micros(s.max_delay_us)); + let mut subscription = moq_net::track::Subscription::default().with_priority(s.priority); + if let Some(max_delay) = s.max_delay_us { + subscription = subscription.with_max_delay(std::time::Duration::from_micros(max_delay)); + } if let Some(ranges) = s.ranges { subscription = subscription.with_ranges( ranges diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 8a59c8bff7c..80dc97e73fa 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1098,7 +1098,7 @@ async fn raw_track_update_does_not_wait_for_pending_read() { consumer .update(MoqSubscription { priority: 10, - max_delay_us: 25_000, + max_delay_us: Some(25_000), ranges: Some(vec![MoqRange { start: MoqAnchor::Absolute { group: 0, frame: 0 }, end: None, @@ -4648,7 +4648,7 @@ async fn raw_track_update_during_pending_group_still_reads_datagram() { consumer .update(MoqSubscription { priority: 10, - max_delay_us: 25_000, + max_delay_us: Some(25_000), ranges: Some(vec![MoqRange { start: MoqAnchor::Absolute { group: 0, frame: 0 }, end: None, @@ -7543,7 +7543,7 @@ async fn ascending_subscription_reads_a_cached_backlog_oldest_first() { let subscription = MoqSubscription { priority: 0, - max_delay_us: 10_000_000, + max_delay_us: Some(10_000_000), ranges: Some(vec![MoqRange { start: MoqAnchor::Relative { groups: 2 }, end: None, @@ -7782,7 +7782,7 @@ async fn time_anchors_start_at_the_containing_group() { let since = |timestamp_us| MoqSubscription { priority: 0, - max_delay_us: 10_000_000, + max_delay_us: Some(10_000_000), ranges: Some(vec![MoqRange { start: MoqAnchor::Time { timestamp_us }, end: None, diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 9cb5a3af531..6db43720d25 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -439,7 +439,7 @@ pub enum Error { #[error("unexpected message")] UnexpectedMessage, - /// The peer asked for a feature this endpoint does not implement. + /// A feature was asked for that this endpoint, or the wire to the peer, does not implement. #[error("unsupported")] Unsupported, diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index d6561adaf95..79c135ccf19 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -4215,8 +4215,10 @@ impl Subscriber { let dynamic = request.dynamic(); // Nothing feeds this copy live: what it caches is fetched, never the live edge. Idle // from the accept, so no reader takes it as live in between. - let track = request.idle().accept(accepted_info(&ok.properties, self.version)); + let track = request.idle().stage(accepted_info(&ok.properties, self.version)); + // Marked received before any reader sees it, so none resolves an anchor this wire dropped. track.received(); + let track = track.commit(); Some(Idle { demand: track.demand(), track, diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 8b8ddd095d9..af5d60427d2 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -7178,8 +7178,13 @@ impl ServeLoop { true => request.request_start(None), false => request, }; - let serving = request.accept(info); + // Marked received before any reader sees it, so none resolves an anchor this wire dropped. + let serving = request.stage(info); serving.received(); + if serve.subscriber.version.resolves_anchors() { + serving.anchored(); + } + let serving = serving.commit(); Self { edge: serving.consume(), largest: None, diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs index 5694bbab7cf..3941f10963d 100644 --- a/rs/moq-net/src/lite/version.rs +++ b/rs/moq-net/src/lite/version.rs @@ -40,6 +40,13 @@ impl Version { ) } + /// Whether a copy received over this version resolves relative and time range anchors + /// as its publisher would: the start resolves from the budget, the answer reports the + /// largest, and frames carry media time. Added in lite-07; an earlier copy refuses them. + pub(crate) fn resolves_anchors(self) -> bool { + self.resolves_start() && self.has_largest() && self.has_media_time() + } + /// Whether the version has lite-05's dedicated TRACK stream and related stream /// layout changes. /// diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 8f058a37654..49664c13a7a 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1311,6 +1311,15 @@ impl Reader { } Poll::Ready(Ok(sub)) => sub, }; + // Each copy's wire decides what it resolves: a copy it replaced refusing the + // reader's ranges leaves the rest to the copy now serving. + if let Some(err) = sub.refused() { + if !copy.replaced { + return Some(Poll::Ready(Err(err))); + } + copy.done = Some(Err(err)); + break; + } if ordered { sub.skip_ordered(floor); } @@ -1460,6 +1469,9 @@ impl Reader { } Poll::Ready(Ok(sub)) => sub, }; + if let Some(err) = sub.refused() { + return Some(Poll::Ready(Err(err))); + } match sub.poll_recv_wire_datagram(waiter) { Poll::Pending => break, Poll::Ready(Ok(Some(datagram))) => { @@ -3965,6 +3977,51 @@ mod test { ); } + /// Each copy's wire decides which anchors it resolves: a replaced copy that refuses the + /// reader's ranges leaves them to the copy serving now, whose own refusal fails the read. + #[test] + #[cfg(feature = "async")] + fn only_the_serving_copy_refuses_an_anchor() { + let routes = Producer::new(); + let front = logical(&routes); + let a = copy(); + a.received(); + routes.serve(a.consume()); + let mut sub = subscribe(&front, Duration::from_secs(30)); + let mut zero = a.create_group(group::Info { sequence: 0 }).unwrap(); + zero.write_frame(ts(0), b"x".as_ref()).unwrap(); + assert_eq!(recv(&mut sub).sequence, 0); + + let b = copy(); + b.received(); + b.anchored(); + routes.serve(b.consume()); + let ranges = [track::Range { + start: track::Anchor::Relative(1), + end: None, + }]; + sub.update(sub.subscription().with_ranges(ranges)).unwrap(); + assert!(sub.recv_group().now_or_never().is_none()); + for sequence in 1..4 { + let mut group = b.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 1000), b"x".as_ref()).unwrap(); + group.finish().unwrap(); + } + // Asked before b held any group, the relative start resolves at its first. + let group = sub.recv_group().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(group.sequence, 1); + + // Served by a copy that can't resolve it, the same range fails. + let routes = Producer::new(); + let front = logical(&routes); + let a = copy(); + a.received(); + routes.serve(a.consume()); + let mut sub = subscribe(&front, Duration::from_secs(30)); + sub.update(sub.subscription().with_ranges(ranges)).unwrap(); + assert!(matches!(sub.recv_group().now_or_never(), Some(Err(Error::Unsupported)))); + } + /// Capping a replaced copy never replaces a time end it has not settled: the copy keeps /// filtering by time, so a group below the cap that starts past the end stays out. #[test] diff --git a/rs/moq-net/src/model/subscription.rs b/rs/moq-net/src/model/subscription.rs index f6e0f0b6dbc..dc5a7b92494 100644 --- a/rs/moq-net/src/model/subscription.rs +++ b/rs/moq-net/src/model/subscription.rs @@ -113,10 +113,11 @@ pub enum Order { /// copy waiting on a fresh upstream answer counts from the live feed's latest group once /// it answers, never from the groups an ended feed left in its cache. /// -/// No published wire carries either yet: upstream, a relative or time start is no floor, -/// the same request as [`Range::oldest`], and a relative or time end is no end. Behind a -/// moq-transport upstream, which joins without a floor at the live group, `Relative(n)..` -/// starts there. +/// Over a session, the copy resolves them only on moq-lite-07, which reports the live edge +/// and media time and asks upstream for every group the budget allows. A copy received +/// over an older lite or any moq-transport version refuses each other than [`Range::live`] +/// and [`Range::oldest`] with [`Error::Unsupported`](crate::Error::Unsupported) on read, +/// rather than resolve it against the part of the track it holds. #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub enum Anchor { /// This exact position. @@ -188,6 +189,17 @@ impl Range { } } + /// Whether a wire without ranges carries this range upstream: absolute anchors, and + /// the live and oldest starts it spells with no floor. + pub(super) fn is_wire(&self) -> bool { + let start = match self.start { + Anchor::Absolute(_) => true, + Anchor::Relative(back) => back == 0 || back == u64::MAX, + Anchor::Time(_) => false, + }; + start && matches!(self.end, None | Some(Anchor::Absolute(_))) + } + /// Whether this range certainly delivers nothing, wherever the latest group lands. pub(super) fn is_empty(&self) -> bool { match (self.start, self.end) { diff --git a/rs/moq-net/src/model/subscription/window.rs b/rs/moq-net/src/model/subscription/window.rs index 0867c7e561e..019ccdf77ac 100644 --- a/rs/moq-net/src/model/subscription/window.rs +++ b/rs/moq-net/src/model/subscription/window.rs @@ -27,6 +27,12 @@ pub(in crate::model) trait Holdings { /// Whether the track carries timestamps at all. fn timed(&self) -> bool; + /// Whether the groups are a copy received over a wire that can't resolve relative or + /// time anchors as the publisher would, so only what [`Range::is_wire`] covers reads. + fn drops_anchors(&self) -> bool { + false + } + /// Where `pts` falls among the groups with a known start. fn containing(&self, pts: Timestamp) -> Containing; @@ -311,7 +317,8 @@ impl TimeEnd { impl Window { /// Resolve `ranges`, counting relative anchors back from `latest`. /// - /// Refuses a time anchor on an untimed track with [`Error::TimestampMismatch`]. + /// Refuses a time anchor on an untimed track with [`Error::TimestampMismatch`], and an + /// anchor a received copy's wire drops with [`Error::Unsupported`]. pub fn new(ranges: &[Range], latest: Option, holdings: &impl Holdings) -> Result { let mut window = Self { ranges: ranges.to_vec(), @@ -531,6 +538,9 @@ impl Window { { return Err(Error::TimestampMismatch); } + if holdings.drops_anchors() && !self.ranges.iter().all(Range::is_wire) { + return Err(Error::Unsupported); + } self.spans.clear(); self.pending = false; for (index, range) in self.ranges.iter().enumerate().filter(|(_, range)| !range.is_empty()) { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 922acaa758a..5c310b65275 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -468,8 +468,12 @@ pub(crate) struct TrackState { start_pending: bool, // A copy received from a remote publisher, whose declared end waits for groups still - // in flight below it (see `tail`). + // in flight below it (see `tail`), and whose readers' relative and time anchors are + // refused unless `anchored` (see `Holdings::drops_anchors`). received: bool, + // Whether a received copy's wire tells it what relative and time anchors resolve + // against, so they resolve here as at the publisher. + anchored: bool, // Sequences the upstream publisher gave up, with the error each was given up with: an // incoming SUBSCRIBE_DROP or a producer cut. Kept apart from the cache so reclaiming a // slot cannot erase a disposition a relay must forward, and as ranges so a wide DROP @@ -761,6 +765,10 @@ impl Holdings for TrackState { self.info.as_ref().is_none_or(|info| info.timescale.is_some()) } + fn drops_anchors(&self) -> bool { + self.received && !self.anchored + } + fn containing(&self, pts: Timestamp) -> Containing { let starts = self.starts.lock().unwrap_or_else(PoisonError::into_inner); let (Some((&first, &first_at)), Some((&last, &last_at))) = (starts.first_key_value(), starts.last_key_value()) @@ -3139,12 +3147,24 @@ impl Producer { /// Mark this track as a copy received from a remote publisher, before writing to it. /// /// Its declared end waits for groups still in flight below it; see [`Self::finish_at`]. + /// Its readers' relative and time anchors, other than [`Range::live`](crate::track::Range::live) + /// and [`Range::oldest`](crate::track::Range::oldest), fail with [`Error::Unsupported`] + /// unless its session's wire lets it resolve them, since it holds only part of the track. pub fn received(&self) { if let Ok(mut state) = self.modify() { state.received = true; } } + /// Let this received copy's readers resolve relative and time anchors, for a wire that + /// reports the live edge and media time and asks for every group the budget allows. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn anchored(&self) { + if let Ok(mut state) = self.modify() { + state.anchored = true; + } + } + /// Record that the publisher gave up `groups` with `err`: none will arrive, so a reader /// waiting on one moves past it, a group still arriving in them ends with `err`, the /// declared end does not wait for them, and a relay serving this track forwards the drop. @@ -6918,24 +6938,28 @@ impl Subscriber { Poll::Ready(start.map(|start| start.max(cursor.floor()))) } - /// The lowest group this subscriber's ranges name, as resolved here when it last read: - /// `None` while one reaches back to the oldest group still around, or a relative - /// start still waits on the first group from the live feed. + /// The lowest group this subscriber's ranges name, as resolved here when it last read, + /// raised to the start of [`Self::set_groups`]: `None` while one reaches back to the + /// oldest group still around, or a relative start still waits on the first group from + /// the live feed. /// /// Where a reader that waits on each group in sequence starts, rather than at whatever /// arrives first, which newest-first delivery makes the latest. pub fn floor(&self) -> Option { - match &self.inner { + let start = match &self.inner { Inner::Plain(cursor) => cursor.filter.window.start(), Inner::Resume(resume, _) => resume.floor(), - } + }; + Some(start?.max(self.bounds().start)) } - /// The lowest group at or above `from` this subscriber's ranges cover, `None` when - /// none does: the groups between are never read, so a reader waiting on each group in - /// sequence need not wait for them. `Some(from)` while a relative anchor waits on the - /// first group, which can't say yet. + /// The lowest group at or above `from` this subscriber's ranges and + /// [`Self::set_groups`] floor cover, `None` when none does: the groups between are + /// never read, so a reader waiting on each group in sequence need not wait for them. + /// `Some` from there while a relative anchor waits on the first group, which can't + /// say yet. pub fn next_in_range(&self, from: u64) -> Option { + let from = from.max(self.bounds().start); match &self.inner { Inner::Plain(cursor) => cursor.filter.window.runs(from).next().map(|run| *run.start()), Inner::Resume(resume, _) => resume.next_in_range(from), @@ -6954,6 +6978,18 @@ impl Subscriber { } } + /// Why this subscriber's ranges can't be served from its track, the error its reads + /// fail with; `None` for a front's reader, whose copies answer for it. + pub(super) fn refused(&mut self) -> Option { + match &mut self.inner { + Inner::Plain(cursor) => { + cursor.sync(); + cursor.filter.refused.clone() + } + Inner::Resume(..) => None, + } + } + /// `ranges` resolved against this track's cache now, by index. pub(super) fn resolve_ranges(&self, ranges: &[Range]) -> Vec> { match &self.inner { @@ -8171,6 +8207,30 @@ mod test { } } + /// The floor a reader waiting on each group starts at is the higher of its ranges' and + /// its local one, so it never waits on a group the local floor withholds. + #[test] + fn floor_honors_set_groups() { + let producer = track_producer("test", Info::default()); + let _groups: Vec<_> = (0..6).map(|sequence| framed(&producer, sequence, 2)).collect(); + + let mut reader = producer.subscribe(replay().with_groups(2..)); + assert_eq!((reader.floor(), reader.next_in_range(0)), (Some(2), Some(2))); + reader.set_groups(4..); + assert_eq!((reader.floor(), reader.next_in_range(0)), (Some(4), Some(4))); + assert_eq!(reader.next_in_range(5), Some(5)); + + // Below the ranges' floor, the ranges still decide. + let mut reader = producer.subscribe(replay().with_groups(3..)); + reader.set_groups(1..); + assert_eq!((reader.floor(), reader.next_in_range(0)), (Some(3), Some(3))); + + // A floor reaching back to the oldest group stays unknown. + let mut reader = producer.subscribe(replay()); + reader.set_groups(4..); + assert_eq!(reader.floor(), None); + } + /// The largest reports only while it is within the bounds: a cap below it or a floor /// above it reports nothing, rather than a cached group standing in for it. #[test] diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index 33c4011a096..b03af2b9afe 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -41,6 +41,7 @@ mod model; mod multi_cdn; mod partial_group; mod quiet_catalog; +mod range_anchors; mod received_rewind; mod refresh; mod rejoin; diff --git a/rs/moq-net/tests/integration/range_anchors.rs b/rs/moq-net/tests/integration/range_anchors.rs new file mode 100644 index 00000000000..7a4dcb2fea5 --- /dev/null +++ b/rs/moq-net/tests/integration/range_anchors.rs @@ -0,0 +1,165 @@ +#![cfg(feature = "async")] + +//! A range anchored relative to the latest group or at a presentation time resolves where +//! the track is held, and on a copy received over a wire that tells it enough to resolve +//! the anchor the same way. A copy received over any other wire refuses the anchor instead +//! of resolving it against the part of the track it happens to hold. + +use crate::support; + +use std::time::Duration; + +use moq_net::track::{Anchor, Info, Order, Position, Range, Subscription}; +use moq_net::{Error, Hop, Timescale, Timestamp, Version}; +use support::harness::peer; + +const TIMEOUT: Duration = Duration::from_secs(10); + +/// A budget no group outlives, so only the range decides what is read. +const FOREVER: Duration = Duration::from_millis((1 << 53) - 1); + +const VERSIONS: &[&str] = &[ + "moq-lite-03", + "moq-lite-04", + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-16", + "moq-transport-17", + "moq-transport-20", + "moq-transport-22", +]; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +fn millis(ms: u64) -> Timestamp { + Timestamp::from_millis(ms).unwrap() +} + +/// Groups 0 to 4, group `n` presenting at `n` seconds, then what a subscriber to `range` +/// reads oldest first: through the publisher's own origin with no `version`, otherwise +/// over one session on it. +async fn read(version: Option<&str>, range: Range) -> Result, Error> { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("room").unwrap(); + let track = broadcast + .create_track( + "data", + Info::default().with_timescale(Timescale::MILLI).with_max_age(FOREVER), + ) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + for sequence in 0..5 { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + group.write_frame(millis(sequence * 1000), &b"x"[..]).unwrap(); + group.finish().unwrap(); + } + + let subscriber = produce_origin(2); + let _pair; + let remote = match version { + // Through the origin, as a reader of a local route. + None => moq_clock::timeout(TIMEOUT, publisher.consume().request_broadcast("room", None)) + .await + .expect("resolve timeout") + .expect("broadcast resolves"), + Some(version) => { + _pair = peer(version.parse::().unwrap(), &publisher, &subscriber).await; + let consumer = subscriber.consume(); + moq_clock::timeout(TIMEOUT, consumer.routed("room")) + .await + .expect("announce timeout") + .expect("routed"); + moq_clock::timeout(TIMEOUT, consumer.request_broadcast("room", None)) + .await + .expect("resolve timeout") + .expect("broadcast resolves") + } + }; + + let subscription = Subscription::default() + .with_max_delay(FOREVER) + .with_order(Order::Ascending) + .with_ranges([range]); + let mut reader = moq_clock::timeout(TIMEOUT, remote.track("data").unwrap().subscribe(subscription)) + .await + .expect("subscribe timeout")?; + + let mut read = Vec::new(); + while let Ok(group) = moq_clock::timeout(TIMEOUT, reader.recv_group()).await { + match group? { + Some(group) => read.push(group.sequence), + None => break, + } + } + track.finish().unwrap(); + Ok(read) +} + +fn range(start: Anchor, end: Option) -> Range { + Range { start, end } +} + +/// Relative and time anchors at each end, with what a reader of the held track reads. +fn anchored() -> [(Range, Vec); 6] { + let zero = Anchor::Absolute(Position::group(0)); + [ + (range(Anchor::Relative(2), None), vec![2, 3, 4]), + (range(Anchor::Time(millis(2500)), None), vec![2, 3, 4]), + (range(zero, Some(Anchor::Relative(1))), vec![0, 1, 2, 3]), + (range(zero, Some(Anchor::Time(millis(2500)))), vec![0, 1, 2]), + (range(Anchor::Relative(2), Some(Anchor::Relative(1))), vec![2, 3]), + ( + range(Anchor::Time(millis(1500)), Some(Anchor::Time(millis(3500)))), + vec![1, 2, 3], + ), + ] +} + +/// Each wire resolves the anchors it carries as the held track does and refuses the rest, +/// rather than resolve them against the part of the track its copy happens to hold. +#[moq_clock::test] +async fn each_wire_refuses_only_the_anchors_it_drops() { + for version in std::iter::once(None).chain(VERSIONS.iter().copied().map(Some)) { + // Lite-07 reports the largest and media time and asks for everything the budget + // allows, so its copy resolves every anchor where it is held. + let carried = matches!(version, None | Some("moq-lite-07-wip")); + // A wire without timestamps refuses a time anchor for that first. + let timed = !matches!( + version, + Some( + "moq-lite-03" | "moq-lite-04" | "moq-lite-05" | "moq-lite-06" | "moq-transport-14" | "moq-transport-16" + ) + ); + for (range, held) in anchored() { + let result = read(version, range).await; + let time = matches!(range.start, Anchor::Time(_)) || matches!(range.end, Some(Anchor::Time(_))); + match (&result, time && !timed, carried) { + (Err(Error::TimestampMismatch), true, _) => {} + (Err(Error::Unsupported), false, false) => {} + (Ok(read), false, true) if *read == held => {} + _ => panic!("{version:?}: {range:?} read {result:?}"), + } + } + } +} + +/// Every wire carries the live and oldest starts, which it spells with no floor. +#[moq_clock::test] +async fn every_wire_carries_the_live_and_oldest_starts() { + for version in VERSIONS { + let live = read(Some(version), Range::live()).await; + assert_eq!(live.unwrap().last(), Some(&4), "{version}: live reads the latest group"); + let oldest = read(Some(version), Range::oldest()).await; + assert_eq!( + oldest.unwrap().last(), + Some(&4), + "{version}: oldest reads through the latest group" + ); + } +} diff --git a/test/interop/clients/go/main.go b/test/interop/clients/go/main.go index 9eac1d7b12d..79dad18750b 100644 --- a/test/interop/clients/go/main.go +++ b/test/interop/clients/go/main.go @@ -44,6 +44,9 @@ const readChunk = 64 * 1024 // media.TrackConsumer max delay: how much reordering the jitter buffer tolerates. const maxDelay = time.Second +// durationPtr is d as the optional duration the wrapper's options take. +func durationPtr(d time.Duration) *time.Duration { return &d } + // Synthetic audio: a 48 kHz tone per channel at 0.5 amplitude, encoded as Opus. const ( audioTrack = "tone" @@ -333,7 +336,7 @@ func finishSubscribe(ctx context.Context, client *moq.Client, broadcast string, for id, video = range catalog.Video { break } - media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{MaxDelay: maxDelay}}) + media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{MaxDelay: durationPtr(maxDelay)}}) if err != nil { return err } @@ -449,7 +452,7 @@ func decodeAudio(ctx context.Context, consumer *moq.BroadcastConsumer, id string out += "." + codec decoder, err := moqaudio.Subscribe(ctx, consumer, id, rendition, moqaudio.DecoderOutput{ Format: moq.AudioSampleFormatF32, - MaxDelay: func() *time.Duration { d := maxDelay; return &d }(), + MaxDelay: durationPtr(maxDelay), }) if err != nil { return err @@ -540,7 +543,7 @@ func subscribe(ctx context.Context, url, broadcast string, opts dialOptions, tim break } - media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{MaxDelay: maxDelay}}) + media, err := moqmedia.NewTrackConsumer(ctx, consumer, moqmedia.TrackConfig{Rendition: moqmedia.VideoRendition(id, video), Subscription: &moq.Subscription{MaxDelay: durationPtr(maxDelay)}}) if err != nil { return err }