diff --git a/cpp/moq/MIGRATE.md b/cpp/moq/MIGRATE.md index 89169b3ff78..986d21b524f 100644 --- a/cpp/moq/MIGRATE.md +++ b/cpp/moq/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`Anchor` gains a `Time` case holding `timestamp_us`, the group containing that presentation time, so exhaustive visits need a case for it. +`TrackConsumer::update` now returns a result, failing with `TimestampMismatch` on a time anchor an untimed track can't serve. + ### JSON consumers return `JsonValue` (#5677) `JsonSnapshotConsumer::next()` and `JsonStreamConsumer::next()` return `JsonValue` (`value`, `timestamp_us`) instead of a string; `timestamp_us` is empty when untimed. diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index 8154fd68df2..b280596e611 100644 --- a/dart/moq/MIGRATE.md +++ b/dart/moq/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`Anchor` gains a time variant, so exhaustive switches need a case for it. +`TrackConsumer.update` now throws on a time anchor an untimed track can't serve. + ### JSON consumers return `Value` (#5677) `SnapshotConsumer.next()` and `StreamConsumer.next()` return `Value` (`value`, `timestampUs`) instead of a string; `timestampUs` is null when untimed. diff --git a/dart/moq/lib/src/aliases.dart b/dart/moq/lib/src/aliases.dart index a72a77f225e..ee9d97750c1 100644 --- a/dart/moq/lib/src/aliases.dart +++ b/dart/moq/lib/src/aliases.dart @@ -120,7 +120,7 @@ typedef Subscription = MoqSubscription; /// A span of a track to deliver: from `start` up to `end`, or onward when `end` is null. typedef Range = MoqRange; -/// Where one end of a [Range] falls: absolute or relative to the latest group. +/// Where one end of a [Range] falls: absolute, relative to the latest group, or the group containing a time. typedef Anchor = MoqAnchor; /// Which pending group the publisher sends first. diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 8c807bd0d31..a8739d693a7 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -4268,6 +4268,12 @@ class FfiConverterMoqAnchor { lifted.value, lifted.bytesRead - subview.offsetInBytes + 4, ); + case 3: + final lifted = TimeMoqAnchor.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); default: throw UniffiInternalError( UniffiInternalError.unexpectedEnumCase, @@ -4377,6 +4383,44 @@ class RelativeMoqAnchor extends MoqAnchor { } } +class TimeMoqAnchor extends MoqAnchor { + final int timestampUs; + TimeMoqAnchor(int this.timestampUs); + TimeMoqAnchor._(int this.timestampUs); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final timestampUs_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final timestampUs = timestampUs_lifted.value; + new_offset += timestampUs_lifted.bytesRead; + return LiftRetVal(TimeMoqAnchor._(timestampUs), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterUInt64.allocationSize(timestampUs) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 3); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterUInt64.write( + timestampUs, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + enum MoqOrder { ascending, descending } class FfiConverterMoqOrder { @@ -8354,7 +8398,7 @@ class MoqTrackConsumer implements MoqTrackConsumerInterface { FfiConverterMoqSubscription.lower(subscription), status, ); - }, null); + }, moqExceptionErrorHandler); } } @@ -18558,7 +18602,7 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_recv_group() != 28712) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_update() != 24851) { + if (uniffi_moq_ffi_checksum_method_moqtrackconsumer_update() != 39083) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqgroupdemand_is_used() != 55523) { diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index edda6e031b0..1525c7a0b4b 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -229,7 +229,7 @@ Each subscription carries the knobs that decide behavior under congestion: | **Priority** (0..255) | Higher-priority tracks get bandwidth first. Audio above video, base layer above enhancement. From a native Rust publisher over QUIC or WebTransport (iroh excepted), subscriptions at the same priority split bandwidth evenly by bytes, streams and datagrams alike, however many groups each has queued. A browser publisher and other transports without send groups are not covered. | | **Order** | Which group to send first when several are pending. Newest first for live (the default), oldest first for catch-up. Needs moq-lite 07 or moq-transport; older moq-lite always sends newest first. | | **Max delay** | How long a non-latest group stays worth delivering before it is skipped. Zero means "live edge only", and raising it is also what asks for history. | -| **Ranges** | Which groups to ask for: from a group, or a number of groups back from the latest, optionally up to an end. The default asks for everything max delay allows. Max delay still caps every range. | +| **Ranges** | Which groups to ask for: from a group, a number of groups back from the latest, or the group containing a presentation time, optionally up to an end. The default asks for everything max delay allows. Max delay still caps every range. | On moq-lite 05 and later, a publisher explicitly reports groups that will not arrive. This lets a finished subscription settle without waiting for missing groups. @@ -240,7 +240,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. +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 +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. Order only decides what goes first, never what is skipped, so catching up oldest first needs a max delay that covers the backlog. When a relay serves subscribers of diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index bd15cc49975..b5bfc663b1d 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`AnchorTime` names the group containing a presentation time, and type switches over `Anchor` need a case for it. +`TrackConsumer.Update` also fails on a time anchor an untimed track can't serve. + ### JSON consumers return `*json.Timed` (#5677) `SnapshotConsumer.Next`, `StreamConsumer.Next`, and `All` return `*Timed{Value, Timestamp}` instead of the raw message; `Timestamp` is a `*time.Duration`, nil when untimed. diff --git a/go/wrapper/subscribe.go b/go/wrapper/subscribe.go index 5d6efcda25f..f9172cf133f 100644 --- a/go/wrapper/subscribe.go +++ b/go/wrapper/subscribe.go @@ -150,15 +150,14 @@ func (t *TrackConsumer) Info() (TrackInfo, error) { return trackInfoFromFFI(info), nil } -// Update changes this subscriber's delivery preferences. It fails only on a -// negative MaxDelay. +// Update changes this subscriber's delivery preferences. It fails on a +// negative MaxDelay, or a time anchor the track can't serve. func (t *TrackConsumer) Update(subscription Subscription) error { sub, err := subscription.ffi() if err != nil { return err } - t.inner.Update(sub) - return nil + return t.inner.Update(sub) } // Groups ranges over groups in sequence order. diff --git a/go/wrapper/types.go b/go/wrapper/types.go index 0ce689b03de..c55269463c2 100644 --- a/go/wrapper/types.go +++ b/go/wrapper/types.go @@ -58,12 +58,14 @@ type ( // A relative start is the first frame of its group, a relative end includes its whole // group, and an absolute end is exclusive. Range = ffi.MoqRange - // Anchor is where one end of a Range falls: AnchorAbsolute or AnchorRelative. + // Anchor is where one end of a Range falls: AnchorAbsolute, AnchorRelative, or AnchorTime. Anchor = ffi.MoqAnchor // AnchorAbsolute is an exact group and frame. AnchorAbsolute = ffi.MoqAnchorAbsolute // AnchorRelative is the group Groups before the latest one, clamped to group 0: 0 is the latest. AnchorRelative = ffi.MoqAnchorRelative + // AnchorTime is the group containing TimestampUs, a presentation time in microseconds. + AnchorTime = ffi.MoqAnchorTime // Order is which of a subscription's pending groups the publisher sends first. Order = ffi.MoqOrder // VideoDecoderOutput configures what a video.Consumer delivers: an optional resize, a max delay, and whether frames keep the decoder's surface (macOS only; refused elsewhere). diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index 51b55968ef0..9a3ad22f29e 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`Anchor` gains `AnchorTime`, so exhaustive `when` expressions need a branch for it. +`TrackConsumer.update` now throws `MoqException` on a time anchor an untimed track can't serve. + ### JSON consumers emit timestamps (#5677) `values()` emits `json.Value` (`value`, plus `timestamp` as a `Duration?`) instead of a string, and `valuesAs()` emits `Timed(value, timestamp)`. diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index 87c02cff362..6430cf04fc7 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -103,12 +103,14 @@ typealias Route = uniffi.moq.MoqRoute typealias Subscription = uniffi.moq.MoqSubscription /** A span of a track to deliver: from `start` up to `end`, or onward when `end` is null. */ typealias Range = uniffi.moq.MoqRange -/** Where one end of a [Range] falls: [AnchorAbsolute] or [AnchorRelative]. */ +/** Where one end of a [Range] falls: [AnchorAbsolute], [AnchorRelative], or [AnchorTime]. */ typealias Anchor = uniffi.moq.MoqAnchor /** An exact group and frame. */ typealias AnchorAbsolute = uniffi.moq.MoqAnchor.Absolute /** The group `groups` before the latest one, clamped to group 0: 0 is the latest. */ typealias AnchorRelative = uniffi.moq.MoqAnchor.Relative +/** The group containing `timestampUs`, a presentation time in microseconds. */ +typealias AnchorTime = uniffi.moq.MoqAnchor.Time /** Which pending group the publisher sends first: `ASCENDING` or `DESCENDING`. */ typealias Order = uniffi.moq.MoqOrder /** Options for fetching one past group by sequence. */ diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index 8dac627d188..59f06c9963e 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`Anchor` gains a `TIME` variant, the group containing a presentation time. +`TrackConsumer.update` now raises on a time anchor an untimed track can't serve. + ### JSON consumers yield `json.Timed` (#5677) `json.SnapshotConsumer` and `json.StreamConsumer` yield `Timed(value, timestamp)` instead of the bare value; `timestamp` is a `timedelta`, or `None` when untimed. diff --git a/py/moq-rs/moq/subscribe.py b/py/moq-rs/moq/subscribe.py index 5935b657de7..4648d683d6a 100644 --- a/py/moq-rs/moq/subscribe.py +++ b/py/moq-rs/moq/subscribe.py @@ -126,7 +126,10 @@ def info(self) -> TrackInfo: return TrackInfo._from_ffi(self._inner.info()) def update(self, subscription: Subscription) -> None: - """Change this subscriber's delivery preferences.""" + """Change this subscriber's delivery preferences. + + Raises on a time anchor the track can't serve. + """ self._inner.update(subscription._ffi()) def cancel(self) -> None: diff --git a/quest/m1/binding-timestamps.md b/quest/m1/binding-timestamps.md index 49014093c58..16cf4a3fee9 100644 --- a/quest/m1/binding-timestamps.md +++ b/quest/m1/binding-timestamps.md @@ -16,8 +16,10 @@ Public API: breaking in every binding. Today every binding speaks microseconds: moq-ffi `timestamp_us: Option` (about 160 sites in 9 files: `MoqFrame`, `MoqDatagram`, `MoqMediaFrame`, audio -frames, json/flate producer `update`/`append`, media `flush`), moq-c -`timestamp_us` plus `timestamp_present` pairs (about 140 sites in 6 files), and +frames, json/flate producer `update`/`append`, media `flush`, the +subscription start `MoqAnchor::Time`), moq-c `timestamp_us` plus +`timestamp_present` pairs (about 140 sites in 6 files, plus +`moq_anchor.timestamp_us` read as `MOQ_ANCHOR_TIME`), and each wrapper's native type (Python `timedelta`, Go `*time.Duration`, Swift `UInt64`, Kotlin `Duration`, Dart `int`, C++ `std::optional`, OBS microseconds). Reads floor through `as_micros()`, so any scale that does not diff --git a/quest/m1/subscribe-ranges/README.md b/quest/m1/subscribe-ranges/README.md index b67786518ed..05b6dbeb7f4 100644 --- a/quest/m1/subscribe-ranges/README.md +++ b/quest/m1/subscribe-ranges/README.md @@ -72,9 +72,17 @@ relay bounds a sparse FETCH only if the model can request ranges: has no time filter. It is part of lite-07 and must land before [lite-07 finalizes](/quest/m1/lite07-finalize.md), which this line gates. This amends the 09-29 premise that ranges are group-only because a group's timestamp isn't always - known: timed tracks know it, and untimed ones refuse. Split out of #5929 - (2026-10-19) into [time anchors](/quest/m1/subscribe-ranges/time-anchors.md), - which lands in its own PR. + known: timed tracks know it, and untimed ones refuse. The cache half landed in its own PR after #5929. + A datagram is judged on its own (maintainer, 2026-10-20): its timestamp + against the range's times, its sequence against the absolute and resolved + relative groups, never through the settle walk. Refusing time anchors on + datagram tracks was considered and can't work: nothing signals that a track + carries datagrams, and the lite draft lets a publisher send any single-frame + group as a datagram, including a copy of one it also streams. Datagrams stay + ephemeral (maintainer, 2026-10-20): a buffered datagram never indexes where + its group starts, so a time end settles across a sequence that arrived only + as a datagram, and a stream copy of that group arriving later is read even + if it starts past the end. - Ranges are frame-precise (`Position`), not whole groups. The IETF joining FETCH for a mid-group SUBSCRIBE's uncached prefix stays, because today's bridge relies on it (maintainer, 09-29). @@ -92,7 +100,6 @@ every sequence delivered or dropped. - [moq-transport ranges](/quest/m1/subscribe-ranges/ietf.md) - range FETCH upstream per missing run, and non-blocking FETCH served downstream - [JS ranges](/quest/m1/subscribe-ranges/js.md) - `@moq/net` model and lite-07 parity - [Frame bounds](/quest/m1/subscribe-ranges/frame-bounds.md) - a local subscriber reads only the frames its ranges cover -- [Time anchors](/quest/m1/subscribe-ranges/time-anchors.md) - a range start or end may name a presentation time ## Related diff --git a/quest/m1/subscribe-ranges/dynamic.md b/quest/m1/subscribe-ranges/dynamic.md index 5e7b850d66e..3a2ddfa8d82 100644 --- a/quest/m1/subscribe-ranges/dynamic.md +++ b/quest/m1/subscribe-ranges/dynamic.md @@ -12,8 +12,8 @@ kept beside it. A subscriber already reads only its own ranges, re-resolved when they are replaced. What remains is the miss: one below the cache becomes a range request -on `Dynamic` rather than a group request. Time anchors land separately; see -[time anchors](/quest/m1/subscribe-ranges/time-anchors.md). +on `Dynamic` rather than a group request. A time anchor the cache can't +resolve passes to the handler unresolved in its range request. - Replace `requested_group` and `fetch_group` rather than keep both. Update every caller (relay, moq-archive, hls, the ladder) in the same PR, diff --git a/quest/m1/subscribe-ranges/time-anchors.md b/quest/m1/subscribe-ranges/time-anchors.md deleted file mode 100644 index b66dd31b474..00000000000 --- a/quest/m1/subscribe-ranges/time-anchors.md +++ /dev/null @@ -1,56 +0,0 @@ -# [L] Time anchors - -## Goal - -A range start or end may name a presentation time: `Time(pts)` in the track's -timescale. A start resolves to the group containing pts, an end to just after that -group, and an untimed track refuses either. It lands in its own PR, before -[lite-07 finalizes](/quest/m1/lite07-finalize.md). - -## Plan - -Split out of #5929 (maintainer, 2026-10-19), which ships absolute and relative -anchors only. Start from the `quest/m1/subscribe-ranges/time-end` branch, which -holds that PR's time anchors through every binding: `Anchor::Time`, the -`Holdings` resolution in `Window`, the end settlement (`Holdings::after`, -`TrackState::accounted`), and the failover handling. Merge main into it and drop -what #5929 changed since. - -Decide first whether time starts ship alone. A start resolves once from the -cache, but keeping it without time ends needs a start-only anchor type in every -binding, which is why #5929 dropped both. - -Open findings from #5929's review, each needing a regression test: - -- **Per-read rescan behind a peer-controlled gap** (Codex 4320985614). Settling an - unresolved end walks back from the live edge on every unsettled read. A peer that - leaves an early sequence missing and keeps sending later groups makes each read - revisit a growing prefix until the gap's grace runs out. Keep an incremental - settle boundary, so each read scans only what arrived since. -- **Failover replacing an unsettled end** (Codex 4320985627). A copy capped at a - route switch reports an unsettled time end as unresolved, so the cap replaces - the `Time` anchor for good, and a group starting before the time that arrives - later on the old copy is delivered. Keep the time filter on the reader, separate - from the copy cap. -- **One speculative group past an end.** An unresolved end covers every sequence - through `u64::MAX`, so each group handed out before its first frame can leak past - it at the live edge. Cap it at the newest held group known to start at or before - pts, plus the first unstarted group above it, and recompute as groups start. - Raising the cap must wake readers: register on the speculative group's start - (`group::Producer::poll_started`) from both `Cursor::poll_recv_group` and - `poll_next_group`. A held group that never shows a start (a copy joined - mid-group) must not pin the cap. Datagrams are not in `lookup`, so on a - datagram-only timed track an end never resolves: resolve datagram starts too, - or leave datagram reads uncapped. Reproduced on #5929: range - `time(3000)..time(3500)` with three groups opened before any frame delivers - all three. - -Also pass a time anchor the cache can't resolve to the `track::Dynamic` handler -in its range request, once [dynamic range requests](/quest/m1/subscribe-ranges/dynamic.md) -land, and update `quest/m1/binding-timestamps.md` for the anchor's -`timestamp_us` fields. - -## Related - -- [Dynamic range requests](/quest/m1/subscribe-ranges/dynamic.md) - range requests a `Dynamic` handler resolves -- [Archive line](/quest/m1/archive/README.md) - DVR seek, which needs time anchors diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index dac364224f1..6ce1689b31f 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`moq_anchor` gains `timestamp_us`, read with the new `MOQ_ANCHOR_TIME` kind, which changes its ABI. +`moq_consume_track_update` fails with a negative code when the track refuses an update, such as a time anchor on an untimed track, leaving the subscription as it was. +An update made before the subscription is established applies once it is, so a refusal then ends the consumer through its callback. + ### JSON values carry their timestamp (#5677) `moq_json_value` gains `timestamp_us` and `timestamp_present`, which changes its ABI. diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 7948ed0c53e..74c55fa2952 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -601,8 +601,8 @@ pub struct moq_subscription { /// A span of a track to deliver: from `start` up to `end`, or onward without one. /// -/// A relative start is the first frame of its group, and a relative end includes its whole -/// group. An absolute end is exclusive. +/// A relative or time start is the first frame of its group, and a relative or time end +/// includes its whole group. An absolute end is exclusive. #[repr(C)] #[allow(non_camel_case_types)] #[derive(Clone, Copy)] @@ -627,6 +627,8 @@ pub struct moq_anchor { pub group: u64, /// The frame within `group`, read only with `MOQ_ANCHOR_ABSOLUTE`. pub frame: u64, + /// The presentation time in microseconds, read only with `MOQ_ANCHOR_TIME`. + pub timestamp_us: u64, } /// How a [moq_anchor] names its position. @@ -638,6 +640,8 @@ pub enum moq_anchor_kind { MOQ_ANCHOR_ABSOLUTE = 0, /// A count of groups back from the latest one. MOQ_ANCHOR_RELATIVE = 1, + /// The group containing a presentation time. An untimed track refuses it. + MOQ_ANCHOR_TIME = 2, } /// Which of a subscription's pending groups the publisher sends first. Zero is the default. @@ -660,6 +664,9 @@ fn parse_anchor(anchor: &moq_anchor) -> Result { }) } v if v == moq_anchor_kind::MOQ_ANCHOR_RELATIVE as u32 => moq_net::track::Anchor::Relative(anchor.group), + v if v == moq_anchor_kind::MOQ_ANCHOR_TIME as u32 => { + moq_net::track::Anchor::Time(moq_net::Timestamp::from_micros(anchor.timestamp_us)?) + } _ => return Err(Error::InvalidCode), }) } @@ -3734,7 +3741,10 @@ pub unsafe extern "C" fn moq_consume_track( /// /// Pass NULL for `subscription` to reset to moq-net defaults. /// -/// Returns a zero on success, or a negative code on failure. +/// Returns a zero on success, or a negative code on failure, such as a time anchor on an +/// untimed track, which leaves the subscription as it was. An update made before the +/// subscription is established applies once it is, so a refusal then ends the consumer +/// through its callback instead. /// /// # Safety /// - The caller must ensure that subscription is either NULL or a valid pointer to a [moq_subscription] struct, diff --git a/rs/moq-c/src/consume.rs b/rs/moq-c/src/consume.rs index 4b248c878c7..a2e4d7bcf51 100644 --- a/rs/moq-c/src/consume.rs +++ b/rs/moq-c/src/consume.rs @@ -5,7 +5,7 @@ use std::{ pin::Pin, task::{Poll, ready}, }; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::oneshot; use crate::ffi::OnStatus; use crate::task::Task; @@ -133,7 +133,15 @@ async fn resolve( /// A raw track task also accepts subscription updates while it is running. struct RawTaskEntry { task: Task, - update: mpsc::UnboundedSender>, + control: RawControl, +} + +/// Where a raw track's subscription updates go. +enum RawControl { + /// Waiting on SUBSCRIBE_OK, holding the latest update to apply once it arrives. + Pending(Option), + /// Subscribed: updates apply at once, so a refusal reaches the caller. + Live(moq_net::track::Control), } /// Outcome of polling a raw track source alongside its control channels. @@ -592,15 +600,28 @@ impl Consume { let name = name.to_string(); let (task, close) = Task::new(on_frame); - let (update, updates) = mpsc::unbounded_channel(); - let id = self.raw_task.insert(Some(RawTaskEntry { task, update }))?; + let control = RawControl::Pending(None); + let id = self.raw_task.insert(Some(RawTaskEntry { task, control }))?; // `subscribe` blocks on SUBSCRIBE_OK, so run it inside the task. tokio::spawn(async move { let res = crate::ffi::run(driver.watch(async move { - let mut track = broadcast.track(&name)?.subscribe(subscription.clone()).await?.ordered(); - Self::apply_raw_subscription(&mut track, subscription); - Self::run_raw(on_frame, track, close, updates).await + let track = broadcast.track(&name)?.subscribe(subscription.clone()).await?.ordered(); + let control = track.control(); + Self::update_raw(&control, Self::raw_subscription(subscription))?; + // Go live under the lock, so no later update lands before the pending one. + if let Some(entry) = State::lock() + .consume + .raw_task + .get_mut(id) + .and_then(|entry| entry.as_mut()) + { + let pending = std::mem::replace(&mut entry.control, RawControl::Live(control.clone())); + if let RawControl::Pending(Some(subscription)) = pending { + Self::update_raw(&control, subscription)?; + } + } + Self::run_raw(on_frame, track, close).await })) .await; @@ -615,24 +636,30 @@ impl Consume { Ok(id) } - /// Apply a raw reader's preferences, which join at the latest group unless a range - /// names a floor of its own. - fn apply_raw_subscription(track: &mut moq_net::track::Ordered, subscription: Option) { + /// A raw reader's preferences, which join at the latest group unless a range names + /// a floor of its own. + fn raw_subscription(subscription: Option) -> moq_net::track::Subscription { let mut subscription = subscription.unwrap_or_default(); for range in &mut subscription.ranges { if range.start == moq_net::track::Range::oldest().start { range.start = moq_net::track::Range::live().start; } } - // A closed track makes the update meaningless; the reader already sees the close. - let _ = track.update(subscription); + subscription + } + + fn update_raw(control: &moq_net::track::Control, subscription: moq_net::track::Subscription) -> Result<(), Error> { + match control.update(subscription) { + // A closed track makes the update meaningless; the reader already sees the close. + Ok(()) | Err(moq_net::Error::Closed) => Ok(()), + Err(err) => Err(err.into()), + } } async fn run_raw( callback: OnStatus, mut track: moq_net::track::Ordered, mut close: oneshot::Receiver<()>, - mut updates: mpsc::UnboundedReceiver>, ) -> Result<(), Error> { // Deliver every frame in sequence order, reading all frames within each // group rather than the one-frame-per-group convenience. This is the @@ -644,7 +671,7 @@ impl Consume { loop { let mut group = match moq_net::kio::wait(|waiter| -> Poll, Error>> { - if Self::poll_raw_control(&mut close, &mut updates, &mut track, waiter) { + if Pin::new(&mut close).poll(&mut waiter.context()).is_ready() { return Poll::Ready(Ok(RawStep::Stop)); } match ready!(track.poll_next_group(waiter))? { @@ -661,7 +688,7 @@ impl Consume { loop { let frame = match moq_net::kio::wait(|waiter| -> Poll, Error>> { - if Self::poll_raw_control(&mut close, &mut updates, &mut track, waiter) { + if Pin::new(&mut close).poll(&mut waiter.context()).is_ready() { return Poll::Ready(Ok(RawStep::Stop)); } match ready!(group.poll_read_frame(waiter))? { @@ -685,33 +712,6 @@ impl Consume { } } - /// Poll the close and update channels, applying any subscription updates inline. - /// - /// Returns `true` when the task must stop: either the consumer was closed - /// (`close` fired) or the update channel was dropped. The caller must then - /// unwind rather than poll `close` again, since a completed oneshot panics if - /// re-polled. Borrows `track` only for the duration of the call so the caller - /// can poll a track/group source afterwards. - fn poll_raw_control( - close: &mut oneshot::Receiver<()>, - updates: &mut mpsc::UnboundedReceiver>, - track: &mut moq_net::track::Ordered, - waiter: &moq_net::kio::Waiter, - ) -> bool { - let mut cx = waiter.context(); - if Pin::new(close).poll(&mut cx).is_ready() { - return true; - } - - loop { - match updates.poll_recv(&mut cx) { - Poll::Ready(Some(subscription)) => Self::apply_raw_subscription(track, subscription), - Poll::Ready(None) => return true, - Poll::Pending => return false, - } - } - } - pub fn raw_track_update( &mut self, track: Id, @@ -722,7 +722,11 @@ impl Consume { .get_mut(track) .and_then(|entry| entry.as_mut()) .ok_or(Error::TrackNotFound)?; - entry.update.send(subscription).map_err(|_| Error::TrackNotFound)?; + let subscription = Self::raw_subscription(subscription); + match &mut entry.control { + RawControl::Pending(pending) => *pending = Some(subscription), + RawControl::Live(control) => Self::update_raw(control, subscription)?, + } Ok(()) } diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 366481f9cde..b270deb4f76 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -1735,11 +1735,13 @@ fn raw_track_subscription_options_and_update() { kind: moq_anchor_kind::MOQ_ANCHOR_ABSOLUTE as u32, group: 1, frame: 0, + timestamp_us: 0, }, end: moq_anchor { kind: moq_anchor_kind::MOQ_ANCHOR_ABSOLUTE as u32, group: 2, frame: 0, + timestamp_us: 0, }, end_present: true, }]; @@ -1810,6 +1812,183 @@ fn raw_track_subscription_options_and_update() { assert_eq!(moq_origin_close(origin), 0); } +/// A time anchor starts at the group containing that presentation time. +#[test] +fn raw_track_time_anchor() { + let origin = id(moq_origin_create()); + let path = b"raw-track-time"; + let broadcast = publish_broadcast(origin, path); + + let track_name = b"data"; + let info = moq_track_info { + priority: 3, + max_age_us: 1_000_000, + max_age_present: true, + timescale: 1_000_000, + timescale_present: true, + refresh: 0, + refresh_present: false, + }; + let track = + id(unsafe { moq_publish_track(broadcast, track_name.as_ptr() as *const c_char, track_name.len(), &info) }); + + let payloads: [&[u8]; 3] = [b"zero", b"one", b"two"]; + for (i, payload) in payloads.into_iter().enumerate() { + assert_eq!( + unsafe { moq_publish_track_frame(track, payload.as_ptr(), payload.len(), i as u64 * 20_000) }, + 0 + ); + } + + let consume = request_broadcast(origin, path); + + let frame_cb = Callback::new(); + let ranges = [moq_range { + start: moq_anchor { + kind: moq_anchor_kind::MOQ_ANCHOR_TIME as u32, + group: 0, + frame: 0, + timestamp_us: 30_000, + }, + end: moq_anchor { + kind: moq_anchor_kind::MOQ_ANCHOR_ABSOLUTE as u32, + group: 0, + frame: 0, + timestamp_us: 0, + }, + end_present: false, + }]; + let subscription = moq_subscription { + priority: 5, + max_delay_us: 1_000_000, + ranges: ranges.as_ptr(), + ranges_len: ranges.len(), + order: 0, + }; + let consumer = id(unsafe { + moq_consume_track( + consume, + track_name.as_ptr() as *const c_char, + track_name.len(), + &subscription, + Some(channel_callback), + frame_cb.ptr, + ) + }); + + for expected in [&b"one"[..], b"two"] { + let frame_id = id(frame_cb.recv()); + let mut frame = moq_frame { + payload: std::ptr::null(), + payload_size: 0, + timestamp_us: 0, + timestamp_present: false, + keyframe: false, + }; + assert_eq!(unsafe { moq_consume_track_frame(frame_id, &mut frame) }, 0); + let received = unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }; + assert_eq!(received, expected); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); + } + + assert_eq!(moq_consume_track_cancel(consumer), 0); + assert_eq!(frame_cb.recv_terminal(), 0); + assert_eq!(moq_publish_track_finish(track), 0); + assert_eq!(moq_consume_close(consume), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + +/// An untimed track refuses a time anchor update at the call, keeping its ranges. The C +/// API only publishes timed tracks, so the track comes from moq-net. +#[test] +fn raw_track_update_refuses_time_on_untimed() { + let origin = id(moq_origin_create()); + let path = b"raw-track-untimed"; + let broadcast = publish_broadcast(origin, path); + + let track_name = b"data"; + let track = State::lock() + .publish + .track( + Id::try_from(broadcast).unwrap(), + "data", + Some(moq_net::track::Info::default().with_timescale(None)), + ) + .unwrap(); + let write = |payload: &'static [u8]| { + State::lock() + .publish + .track_producer(track) + .unwrap() + .write_frame(None, payload) + .unwrap() + }; + + let consume = request_broadcast(origin, path); + let frame_cb = Callback::new(); + let consumer = id(unsafe { + moq_consume_track( + consume, + track_name.as_ptr() as *const c_char, + track_name.len(), + std::ptr::null(), + Some(channel_callback), + frame_cb.ptr, + ) + }); + + // The first frame proves the subscription is established. + write(b"zero"); + let frame_id = id(frame_cb.recv()); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); + + let ranges = [moq_range { + start: moq_anchor { + kind: moq_anchor_kind::MOQ_ANCHOR_TIME as u32, + group: 0, + frame: 0, + timestamp_us: 0, + }, + end: moq_anchor { + kind: moq_anchor_kind::MOQ_ANCHOR_ABSOLUTE as u32, + group: 0, + frame: 0, + timestamp_us: 0, + }, + end_present: false, + }]; + let update = moq_subscription { + priority: 3, + max_delay_us: 1_000_000, + ranges: ranges.as_ptr(), + ranges_len: ranges.len(), + order: 0, + }; + assert!(unsafe { moq_consume_track_update(consumer, &update) } < 0); + + // The consumer keeps reading under its old ranges. + write(b"one"); + let frame_id = id(frame_cb.recv()); + let mut frame = moq_frame { + payload: std::ptr::null(), + payload_size: 0, + timestamp_us: 0, + timestamp_present: false, + keyframe: false, + }; + assert_eq!(unsafe { moq_consume_track_frame(frame_id, &mut frame) }, 0); + let received = unsafe { std::slice::from_raw_parts(frame.payload, frame.payload_size) }; + assert_eq!(received, b"one"); + assert_eq!(moq_consume_track_frame_free(frame_id), 0); + + assert_eq!(moq_consume_track_cancel(consumer), 0); + assert_eq!(frame_cb.recv_terminal(), 0); + assert_eq!(moq_consume_close(consume), 0); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn raw_track_relative_update_counts_from_the_track_edge() { let origin = id(moq_origin_create()); @@ -1835,6 +2014,7 @@ fn raw_track_relative_update_counts_from_the_track_edge() { kind: kind as u32, group, frame: 0, + timestamp_us: 0, }; let subscribe = |ranges: &[moq_range]| moq_subscription { priority: 0, diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 7c30fe86411..eca047ebf96 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -5,6 +5,11 @@ These are the UniFFI names; each language wrapper has its own `MIGRATE.md` in it ## Unreleased +### Anchors can name a time (#6074) + +`MoqAnchor` gains `Time { timestamp_us }`, the group containing that presentation time, so exhaustive matches need an arm for it. +`MoqTrackConsumer::update` now returns `Result<(), MoqError>`, failing with `TimestampMismatch` on a time anchor an untimed track can't serve. + ### JSON consumers return `MoqJsonValue` (#5677) `MoqJsonSnapshotConsumer::next` and `MoqJsonStreamConsumer::next` return `MoqJsonValue { value, timestamp_us }` instead of a string; `timestamp_us` is null when untimed. diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index 728f08e1999..40be2cadba4 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -49,8 +49,8 @@ pub struct MoqSubscription { /// A span of a track to deliver: from `start` up to `end`, or onward when `end` is null. /// -/// A relative start is the first frame of its group, and a relative end includes its whole -/// group. An absolute end is exclusive. +/// A relative or time start is the first frame of its group, and a relative or time end +/// includes its whole group. An absolute end is exclusive. #[derive(Clone, Copy, uniffi::Record)] pub struct MoqRange { /// Where delivery starts. @@ -67,6 +67,9 @@ pub enum MoqAnchor { Absolute { group: u64, frame: u64 }, /// The group this many before the latest, clamped to group 0: `0` is the latest group. Relative { groups: u64 }, + /// The group containing this presentation time, in microseconds. An untimed track + /// refuses it. + Time { timestamp_us: u64 }, } /// Which of a subscription's pending groups the publisher sends first, mirroring @@ -79,21 +82,28 @@ pub enum MoqOrder { Descending, } -impl From for moq_net::track::Anchor { - fn from(anchor: MoqAnchor) -> Self { - match anchor { +impl TryFrom for moq_net::track::Anchor { + type Error = MoqError; + + fn try_from(anchor: MoqAnchor) -> Result { + Ok(match anchor { MoqAnchor::Absolute { group, frame } => Self::Absolute(moq_net::track::Position { group, frame }), MoqAnchor::Relative { groups } => Self::Relative(groups), - } + MoqAnchor::Time { timestamp_us } => { + Self::Time(moq_net::Timestamp::from_micros(timestamp_us).map_err(|_| MoqError::TimeOverflow)?) + } + }) } } -impl From for moq_net::track::Range { - fn from(range: MoqRange) -> Self { - Self { - start: range.start.into(), - end: range.end.map(Into::into), - } +impl TryFrom for moq_net::track::Range { + type Error = MoqError; + + fn try_from(range: MoqRange) -> Result { + Ok(Self { + start: range.start.try_into()?, + end: range.end.map(TryInto::try_into).transpose()?, + }) } } @@ -127,18 +137,25 @@ impl From for moq_net::group::Fetch { } } -impl From for moq_net::track::Subscription { - fn from(s: MoqSubscription) -> Self { +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)); if let Some(ranges) = s.ranges { - subscription = subscription.with_ranges(ranges.into_iter().map(Into::into)); + subscription = subscription.with_ranges( + ranges + .into_iter() + .map(TryInto::try_into) + .collect::, _>>()?, + ); } if let Some(order) = s.order { subscription = subscription.with_order(order.into()); } - subscription + Ok(subscription) } } @@ -254,7 +271,7 @@ impl MoqBroadcastConsumer { ) -> Result, MoqError> { self.driver .watch(async { - let subscription = subscription.map(moq_net::track::Subscription::from); + let subscription = subscription.map(moq_net::track::Subscription::try_from).transpose()?; let track = self.inner.track(&name)?.subscribe(subscription).await?; Ok(Arc::new(MoqTrackConsumer::new(track, self.driver.clone()))) }) @@ -664,9 +681,13 @@ impl MoqTrackConsumer { /// Change this subscriber's delivery preferences. /// /// Silently ignored if the track already ended; the update is meaningless at - /// that point. - pub fn update(&self, subscription: MoqSubscription) { - let _ = self.control.update(subscription.into()); + /// that point. Errors on a time anchor the track can't serve, leaving the + /// preferences as they were. + pub fn update(&self, subscription: MoqSubscription) -> Result<(), MoqError> { + match self.control.update(subscription.try_into()?) { + Ok(()) | Err(moq_net::Error::Closed) => Ok(()), + Err(err) => Err(err.into()), + } } /// Cancel all current and future reads. diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 23e05fc6534..2ada4028234 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -1330,7 +1330,10 @@ impl MoqMediaTrackConsumer { // subscription if init parsing fails. let media = media_container(container)?; let resolved = broadcast.resolve_inner(reference.as_deref()).await?; - let subscription = subscription.map(moq_net::track::Subscription::from).unwrap_or_default(); + let subscription = subscription + .map(moq_net::track::Subscription::try_from) + .transpose()? + .unwrap_or_default(); let track = broadcast .driver .watch(async { Ok(resolved.track(&name)?.subscribe(subscription).await?) }) diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index e3468a11de4..5c032c16dce 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -525,7 +525,7 @@ impl MoqTrackProducer { let _guard = crate::ffi::enter(); let guard = self.inner.lock().unwrap(); let track = guard.as_ref().ok_or(MoqError::Closed)?; - let subscription = subscription.map(moq_net::track::Subscription::from); + let subscription = subscription.map(moq_net::track::Subscription::try_from).transpose()?; Ok(Arc::new(MoqTrackConsumer::new( track.subscribe(subscription), Background::default(), diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 404a497b708..ed8ccdd37e9 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -915,15 +915,17 @@ async fn raw_track_update_does_not_wait_for_pending_read() { tokio::spawn(async move { consumer.read_frame().await }) }; - consumer.update(MoqSubscription { - priority: 10, - max_delay_us: 25_000, - ranges: Some(vec![MoqRange { - start: MoqAnchor::Absolute { group: 0, frame: 0 }, - end: None, - }]), - order: None, - }); + consumer + .update(MoqSubscription { + priority: 10, + max_delay_us: 25_000, + ranges: Some(vec![MoqRange { + start: MoqAnchor::Absolute { group: 0, frame: 0 }, + end: None, + }]), + order: None, + }) + .unwrap(); let payload = b"updated subscription".to_vec(); track @@ -3936,15 +3938,17 @@ async fn raw_track_update_during_pending_group_still_reads_datagram() { spawn_parked(async move { consumer.next_group().await }).await }; - consumer.update(MoqSubscription { - priority: 10, - max_delay_us: 25_000, - ranges: Some(vec![MoqRange { - start: MoqAnchor::Absolute { group: 0, frame: 0 }, - end: None, - }]), - order: None, - }); + consumer + .update(MoqSubscription { + priority: 10, + max_delay_us: 25_000, + ranges: Some(vec![MoqRange { + start: MoqAnchor::Absolute { group: 0, frame: 0 }, + end: None, + }]), + order: None, + }) + .unwrap(); let sequence = track.append_datagram(datagram_frame(b"after-update")).unwrap(); let datagram = tokio::time::timeout(TIMEOUT, consumer.recv_datagram()) @@ -6745,7 +6749,7 @@ async fn ascending_subscription_reads_a_cached_backlog_oldest_first() { }]), order: Some(MoqOrder::Ascending), }; - let converted = moq_net::track::Subscription::from(subscription.clone()); + let converted = moq_net::track::Subscription::try_from(subscription.clone()).unwrap(); assert_eq!(converted.order, moq_net::track::Order::Ascending); assert_eq!( converted.ranges, @@ -6767,3 +6771,61 @@ async fn ascending_subscription_reads_a_cached_backlog_oldest_first() { } assert_eq!(sequences, vec![0, 1, 2]); } + +/// A time anchor starts at the group containing it, and an untimed track refuses one. +#[tokio::test] +async fn time_anchors_start_at_the_containing_group() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let track = broadcast.publish_track("events".into(), Some(timed_info())).unwrap(); + for timestamp_us in [0, 1_000, 2_000] { + let group = track.append_group().unwrap(); + group + .write_frame(MoqFrame { + payload: b"x".to_vec(), + timestamp_us: Some(timestamp_us), + }) + .unwrap(); + group.finish().unwrap(); + } + + let since = |timestamp_us| MoqSubscription { + priority: 0, + max_delay_us: 10_000_000, + ranges: Some(vec![MoqRange { + start: MoqAnchor::Time { timestamp_us }, + end: None, + }]), + order: Some(MoqOrder::Ascending), + }; + let consumer = track.consume(Some(since(1_500))).unwrap(); + let mut sequences = Vec::new(); + for _ in 0..2 { + let group = tokio::time::timeout(TIMEOUT, consumer.recv_group()) + .await + .expect("timed out waiting for a cached group") + .unwrap() + .expect("expected a group"); + sequences.push(group.sequence()); + } + assert_eq!(sequences, vec![1, 2]); + + let untimed = broadcast + .publish_track( + "status".into(), + Some(MoqTrackInfo { + timescale: None, + ..timed_info() + }), + ) + .unwrap(); + let consumer = untimed.consume(None).unwrap(); + assert!(matches!( + consumer.update(since(0)), + Err(MoqError::Protocol { + details: MoqProtocolError { + kind: MoqProtocolKind::TimestampMismatch, + .. + } + }) + )); +} diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index f86f29361e9..8009db2c319 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`track::Anchor` gains `Time(Timestamp)`, the group containing a presentation time, so exhaustive matches need an arm for it. +Subscribing or updating with one on an untimed track fails with `Error::TimestampMismatch`. + ### Each group has one writer (#6020) `group::Producer` no longer implements `Clone`: keep one writer and move it by value between owners, or put it behind a lock when several callers write it. diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 45c61b29b30..f471d193563 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -65,6 +65,11 @@ //! for the groups outside its ranges, so a widening scans only the span it newly covers, //! and a scan of the cache or the audience shows up as a slope. //! +//! `track_time_end` lands groups one at a time behind a sequence that never arrives, each +//! drained by readers whose range ends at a time no group reaches, swept over the groups and +//! the readers: an unsettled time end walks on from where it stopped, so a rescan of the +//! groups behind the gap on every read shows up as a slope over them. +//! //! Run with `cargo bench -p moq-net --bench track`. // Custom timings measure the real time the work took, not the ambient clock. @@ -79,7 +84,7 @@ use moq_net::time::Instant; use bytes::Bytes; use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; -use moq_net::{Error, Timestamp, broadcast, cache, track}; +use moq_net::{Error, Timestamp, broadcast, cache, group, track}; /// A millisecond track: the frames written here carry timestamps. fn timed() -> track::Info { @@ -1261,6 +1266,83 @@ fn bench_range_widening(c: &mut Criterion) { bench.finish(); } +/// Readers whose range ends at a time no group reaches yet, as groups land behind a missing +/// sequence; see [`bench_time_end`]. +struct TimeEnd { + _broadcast: broadcast::Producer, + track: track::Producer, + readers: Vec, + groups: u64, +} + +impl TimeEnd { + fn new(readers: usize, groups: u64) -> Self { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", timed()).unwrap(); + let mut zero = track.create_group(group::Info { sequence: 0 }).unwrap(); + zero.write_frame(Timestamp::from_millis(0).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + zero.finish().unwrap(); + // Group 1 never arrives, and the budget outlasts the run, so the end never settles. + let subscription = track::Subscription::default() + .with_max_delay(AGED_BUDGET) + .with_ranges([track::Range { + start: track::Anchor::Absolute(track::Position::group(0)), + end: Some(track::Anchor::Time(Timestamp::from_secs(u32::MAX.into()).unwrap())), + }]); + let waiter = kio::Waiter::noop(); + let readers = (0..readers) + .map(|_| { + let mut reader = track.subscribe(subscription.clone()); + while let Poll::Ready(Ok(Some(_))) = reader.poll_recv_group(&waiter) {} + reader + }) + .collect(); + Self { + _broadcast: broadcast, + track, + readers, + groups, + } + } + + fn run(&mut self) { + let waiter = kio::Waiter::noop(); + for sequence in 2..self.groups + 2 { + let mut group = self.track.create_group(group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::from_millis(sequence).unwrap(), Bytes::from_static(b"x")) + .unwrap(); + group.finish().unwrap(); + for reader in &mut self.readers { + let mut handed = 0; + while let Poll::Ready(Ok(Some(group))) = reader.poll_recv_group(&waiter) { + black_box(group); + handed += 1; + } + assert_eq!(handed, 1); + } + } + } +} + +/// Groups landing one at a time behind a missing sequence, each drained by readers whose +/// time end it cannot settle, swept over the groups and the readers. Each group must cost +/// each reader the same, so a rescan behind the gap shows up as a slope over the groups. +fn bench_time_end(c: &mut Criterion) { + let mut group = c.benchmark_group("track_time_end"); + group.sample_size(10); + for readers in [1usize, 8, 64] { + for groups in [64u64, 512, 4096] { + group.throughput(Throughput::Elements(groups * readers as u64)); + group.bench_function(BenchmarkId::new(format!("{readers}r"), groups), |b| { + b.iter_batched_ref(|| TimeEnd::new(readers, groups), TimeEnd::run, BatchSize::PerIteration); + }); + } + } + group.finish(); +} + criterion_group!( benches, bench_range_widening, @@ -1282,6 +1364,7 @@ criterion_group!( bench_sparse_read, bench_gap_flush, bench_wall, - bench_empty_run + bench_empty_run, + bench_time_end ); criterion_main!(benches); diff --git a/rs/moq-net/src/model/expiry.rs b/rs/moq-net/src/model/expiry.rs index 23e4602979b..9f66b5a0e0e 100644 --- a/rs/moq-net/src/model/expiry.rs +++ b/rs/moq-net/src/model/expiry.rs @@ -34,6 +34,10 @@ pub(super) struct Wakes { /// Relaxed: a read registers before resolving the edge, under the group locks a /// write takes before it presents, so a write the read missed sees its deadline. earliest: AtomicU64, + /// The lowest sequence a parked time end watches for a start, or `u64::MAX`, so a first + /// frame below every one skips the lock. Ordered by the starts index's lock, which a + /// read takes after registering and a first frame before checking this. + starts_from: AtomicU64, parked: kio::Lock, } @@ -53,6 +57,9 @@ struct Parked { /// Exact cache arrivals, independent of the live edge and neighbouring groups. arrivals: BTreeMap, arrival_sweep: usize, + /// Reads parked on a time end, by the lowest sequence whose start could move it. + starts: BTreeMap, + starts_sweep: usize, /// Reads by wall-clock deadline, released by the pool's driver. wall: BTreeMap, /// The earliest `wall` deadline the pool was asked to release, so arming a later one @@ -67,6 +74,7 @@ impl Default for Wakes { fn default() -> Self { Self { earliest: AtomicU64::new(u64::MAX), + starts_from: AtomicU64::new(u64::MAX), parked: kio::Lock::default(), } } @@ -179,6 +187,36 @@ impl Wakes { parked.arrivals.entry(sequence).or_default().register(waiter); } + /// Park `waiter` until a group at or above `from` starts. + pub(super) fn watch_start(&self, from: u64, waiter: &kio::Waiter) { + let mut parked = self.parked.lock(); + if parked.starts.len() >= parked.starts_sweep { + parked.starts.retain(|_, list| !list.is_empty()); + parked.starts_sweep = (parked.starts.len() * 2).max(HELD_SWEEP_MIN); + } + parked.starts.entry(from).or_default().register(waiter); + self.starts_from.fetch_min(from, Ordering::Relaxed); + } + + /// Group `sequence` started: wake the time ends it could move. + pub(super) fn started(&self, sequence: u64) { + if sequence < self.starts_from.load(Ordering::Relaxed) { + return; + } + let mut woken = Vec::new(); + { + let mut parked = self.parked.lock(); + while let Some(entry) = parked.starts.first_entry() + && *entry.key() <= sequence + { + woken.push(entry.remove()); + } + let from = parked.starts.first_key_value().map_or(u64::MAX, |(from, _)| *from); + self.starts_from.store(from, Ordering::Relaxed); + } + woken.iter_mut().for_each(kio::WaiterList::wake); + } + /// Wake only cache arrivals whose outcome this insertion or boundary changes. pub(super) fn arrived(&self, groups: impl std::ops::RangeBounds) { let groups = (groups.start_bound().cloned(), groups.end_bound().cloned()); @@ -222,13 +260,15 @@ impl Wakes { let parked = { let mut parked = self.parked.lock(); self.earliest.store(u64::MAX, Ordering::Relaxed); + self.starts_from.store(u64::MAX, Ordering::Relaxed); std::mem::take(&mut *parked) }; let lists = parked.deadlines.into_values().chain(parked.landings.into_values()); let lists = lists .chain(parked.held.into_values()) .chain(parked.wall.into_values()) - .chain(parked.arrivals.into_values()); + .chain(parked.arrivals.into_values()) + .chain(parked.starts.into_values()); for mut list in lists { list.wake(); } diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 6498967b55d..a6737f39043 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -945,9 +945,9 @@ impl Producer { self.handle.alive.checked.store(true, Ordering::Relaxed); self.advance_largest(&state); drop(state); - // The group's lock is released first: sharing reads it back through its slot. - if let (Some(track), Some(start)) = (&track, bound) { - track.share_bound(self.handle.info.sequence, start); + // The group's lock is released first: indexing reads it back through its slot. + if let Some(track) = &track { + track.first_frame(self.handle.info.sequence, bound); } drop(track); @@ -1032,9 +1032,9 @@ impl Producer { self.advance_largest(&state); } drop(state); - // The group's lock is released first: sharing reads it back through its slot. - if let (Some(track), Some(start)) = (&track, bound) { - track.share_bound(self.handle.info.sequence, start); + // The group's lock is released first: indexing reads it back through its slot. + if let Some(track) = &track { + track.first_frame(self.handle.info.sequence, bound); } drop(track); @@ -1154,9 +1154,9 @@ impl Producer { let bound = state.settle(origin); self.handle.alive.checked.store(true, Ordering::Relaxed); drop(state); - // The group's lock is released first: sharing reads it back through its slot. - if let (Some(track), Some(start)) = (&track, bound) { - track.share_bound(self.handle.info.sequence, start); + // The group's lock is released first: indexing reads it back through its slot. + if let Some(track) = &track { + track.first_frame(self.handle.info.sequence, bound); } drop(track); @@ -1542,6 +1542,13 @@ impl Handle { self.state.read().started() } + /// Whether this copy may yet present the group's frame 0: it holds it, and nothing + /// opened or ended it. + pub(super) fn may_start(&self) -> bool { + let state = self.state.read(); + state.offset == 0 && !state.started() + } + /// Where the group starts in presentation time: its first frame's timestamp, or /// `None` while no frame has been opened or when that frame is untimed. /// diff --git a/rs/moq-net/src/model/ledger.rs b/rs/moq-net/src/model/ledger.rs index 055f1d904d8..64dad02890e 100644 --- a/rs/moq-net/src/model/ledger.rs +++ b/rs/moq-net/src/model/ledger.rs @@ -114,6 +114,14 @@ impl Ledger { self.runs.first().map(|run| run.groups.start) } + /// The exclusive end of the accounted run holding `sequence`, `None` while it is not. + pub fn run_end(&self, sequence: u64) -> Option { + let run = self + .runs + .get(self.runs.partition_point(|run| run.groups.end <= sequence))?; + (run.groups.start <= sequence).then_some(run.groups.end) + } + /// Whether every group in `groups` is accounted for. pub fn covers(&self, groups: Range) -> bool { // Runs merge on insert, so one run covers the span or none does. diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index f23b4d8f831..a5c9232e0c4 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -27,7 +27,7 @@ use std::task::{Poll, ready}; use crate::{Error, Result, StreamError, group, stats, track}; use super::delivered::Delivered; -use super::subscription::{Anchor, Cap, Position, Resolved, Subscription, Window}; +use super::subscription::{Anchor, Cap, Position, Resolved, Subscription, Unheld, Window}; /// What the front says about a logical track: the copy to read now, and how it ended. #[derive(Default)] @@ -173,7 +173,7 @@ impl Consumer { /// Subscribe to the logical track with the reader's own preferences. pub(super) fn subscribe(&self, subscription: track::Preferences) -> Subscriber { let mirrored = subscription.read().clone(); - let window = Window::new(&mirrored.ranges, None); + let window = Window::new(&mirrored.ranges, None, &Unheld).expect("an unheld hop refuses nothing"); // An absolute floor holds from the start, as on a plain cursor; a relative one // counts back from an edge, so the reported largest is never below it. let floor = window.floor().unwrap_or(0); @@ -627,16 +627,20 @@ impl Reader { start: match range.start { // The farthest offset reaches back to the oldest group, which names no floor. Anchor::Relative(back) if back != u64::MAX => Anchor::Absolute(Position::group(resolved.start)), + Anchor::Time(_) => Anchor::Absolute(Position::group(resolved.start)), start => start, }, + // A time end the copy has not settled stays one: its cap is not the end. end: match (range.end, resolved.end) { - (Some(Anchor::Relative(_)), Some(end)) => Some(Anchor::Absolute(Position::group(end))), + (Some(Anchor::Relative(_) | Anchor::Time(_)), Some(end)) => { + Some(Anchor::Absolute(Position::group(end))) + } (end, _) => end, }, } }) .collect(); - let window = Window::new(&ranges, None); + let window = Window::new(&ranges, None, &Unheld).expect("an unheld hop refuses nothing"); if !window.is_pending() { self.window = window; } @@ -657,7 +661,7 @@ impl Reader { if let Some(sub) = changed { let replaced = sub.ranges != self.mirrored.ranges; if replaced { - self.window = Window::new(&sub.ranges, None); + self.window = Window::new(&sub.ranges, None, &Unheld).expect("an unheld hop refuses nothing"); } self.mirrored = sub; for copy in &mut self.copies { @@ -2524,16 +2528,23 @@ mod test { assert_eq!(recv(&mut sub).sequence, 6); } - /// A replacement copy keeps a later range disjoint: the floor raises nothing above it, - /// and capping the replaced copy never widens its end. + /// A replacement copy keeps each range where the serving copy resolved it: the floor + /// raises a later time range's start rather than lowering it to the floor, and capping + /// the replaced copy never widens a time end. #[test] #[cfg(feature = "async")] - fn a_replacement_keeps_a_later_range_disjoint() { + fn a_replacement_keeps_a_later_time_range_disjoint() { let routes = Producer::new(); let logical = logical(&routes); let a = copy(); routes.serve(a.consume()); - let ranges = [track::Range::groups(0..1), track::Range::groups(10..12)]; + let ranges = [ + track::Range::groups(0..1), + track::Range { + start: track::Anchor::Time(ts(10_000)), + end: Some(track::Anchor::Time(ts(11_000))), + }, + ]; let subscription = Subscription::default() .with_max_delay(Duration::from_secs(30)) .with_ranges(ranges); @@ -2569,11 +2580,11 @@ mod test { assert_eq!(rest, vec![10, 11]); } - /// A replacement updated to a relative range it cannot resolve yet asks for that range, + /// A replacement updated to a time range it cannot resolve yet asks for that range, /// not everything from the floor: before it answers, and once it answers with nothing /// to resolve it against. #[test] - fn a_replacement_keeps_an_updated_relative_start() { + fn a_replacement_keeps_an_updated_time_start() { let routes = Producer::new(); let a = copy(); routes.serve(a.consume()); @@ -2584,7 +2595,7 @@ mod test { let copy = &mut reader.copies[0]; copy.floor = Some(Position::group(0)); let ranges = vec![track::Range { - start: track::Anchor::Relative(3), + start: track::Anchor::Time(ts(10_000)), end: None, }]; let updated = Subscription::default().with_ranges(ranges.clone()); @@ -2653,7 +2664,7 @@ mod test { // The read subscribes the replacement, which has not answered yet. assert!(sub.recv_group().now_or_never().is_none()); let ranges = [track::Range { - start: track::Anchor::Relative(2), + start: track::Anchor::Time(ts(10_000)), end: None, }]; sub.update(sub.subscription().with_ranges(ranges)).unwrap(); @@ -2672,11 +2683,53 @@ mod test { assert!(sub.recv_group().now_or_never().is_none()); let start = c.subscription().expect("subscribed").ranges[0].start; assert!( - matches!(start, track::Anchor::Absolute(_)), - "the replacement asks from where b resolved the relative start: {start:?}" + matches!(start, track::Anchor::Absolute(position) if position.group >= 10), + "the replacement asks from where b resolved the time: {start:?}" ); } + /// 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] + #[cfg(feature = "async")] + fn capping_a_replaced_copy_keeps_its_time_end() { + let routes = Producer::new(); + let a = copy(); + routes.serve(a.consume()); + let mut zero = a.create_group(group::Info { sequence: 0 }).unwrap(); + zero.write_frame(ts(3000), b"x".as_ref()).unwrap(); + zero.finish().unwrap(); + let range = track::Range { + start: track::Anchor::Absolute(Position::group(0)), + end: Some(track::Anchor::Time(ts(3500))), + }; + let subscription = Subscription::default() + .with_max_delay(Duration::from_secs(30)) + .with_ranges([range]); + let sub = routes.consume().subscribe(kio::Producer::new(subscription).into()); + let mut reader = sub.reader.lock().unwrap(); + let (groups, mirrored) = (reader.groups, reader.mirrored.clone()); + let old = &mut reader.copies[0]; + old.replaced = true; + old.until = Some(Some(4)); + old.update(&mirrored); + assert_eq!(old.wire(&mirrored).ranges, vec![range], "the time end outlives the cap"); + + for (sequence, millis) in [(1, 3600), (2, 3700)] { + let mut group = a.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(millis), b"x".as_ref()).unwrap(); + group.finish().unwrap(); + } + let Poll::Ready(Ok(copy)) = old.poll_ready(groups, &kio::Waiter::noop()) else { + panic!("the copy answered"); + }; + let mut read = Vec::new(); + while let Some(Ok(Some(group))) = copy.recv_group().now_or_never() { + read.push(group.sequence); + } + assert_eq!(read, vec![0]); + } + #[test] fn a_replaced_copy_at_the_last_sequence_has_no_overflow_cap() { for (last, end) in [(Some(u64::MAX), None), (None, Some(Position::default()))] { diff --git a/rs/moq-net/src/model/subscription.rs b/rs/moq-net/src/model/subscription.rs index 81f53a38461..98177f43798 100644 --- a/rs/moq-net/src/model/subscription.rs +++ b/rs/moq-net/src/model/subscription.rs @@ -3,8 +3,10 @@ use std::{ time::Duration, }; +use crate::Timestamp; + mod window; -pub(super) use window::{Resolved, Window}; +pub(super) use window::{Containing, Held, Holdings, Resolved, Unheld, Window, later}; /// Subscriber-side preferences for receiving a track. /// @@ -101,28 +103,42 @@ pub enum Order { Descending, } -/// Where one end of a [`Range`] falls: a fixed position, or a count of groups back from the -/// latest one. +/// Where one end of a [`Range`] falls: a fixed position, a count of groups back from the +/// latest one, or a presentation time. /// -/// A relative anchor resolves against the latest group of the hop that serves it, when it -/// starts serving and again whenever the subscriber replaces its ranges. Before any group -/// exists it resolves at the first one, so [`Range::live`] starts there. +/// A relative or time anchor resolves at the hop that serves it, when it starts serving +/// and again whenever the subscriber replaces its ranges. Before any group exists a +/// relative anchor resolves at the first one, so [`Range::live`] starts there. /// -/// No published wire carries one yet: upstream, a relative start is no floor, the same -/// request as [`Range::oldest`], and a relative end is no end. Behind a moq-transport -/// upstream, which joins without a floor at the live group, `Relative(n)..` starts there. +/// 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. #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub enum Anchor { /// This exact position. Absolute(Position), /// The group this many before the latest, clamped to group 0: `Relative(0)` is the latest group. Relative(u64), + /// The group containing this presentation time: the newest one starting at or before it. + /// + /// Resolved from the starts of the groups the serving hop holds. When none starts at + /// or before it, a start reaches back to the oldest group, and while no group has + /// started at all it waits like `Relative(0)`. An end settles once a group starting + /// after it is known and every sequence below that one is accounted for (started, + /// dropped, or never to start). Until then reads stop at the first held group that + /// may still start, so at the live edge at most that one group past the end is read + /// before its first frame shows where it starts. A datagram skips all of that: it is + /// judged on its own, its timestamp against the range's times and its sequence against + /// the range's absolute groups and the relative ones resolved so far. An untimed track + /// refuses it with [`Error::TimestampMismatch`](crate::Error::TimestampMismatch). + Time(Timestamp), } /// A span of a track to deliver: from `start` up to `end`, or onward when `end` is `None`. /// -/// Relative anchors cover whole groups: a relative start is frame 0 of its group, and a -/// relative end includes its group, so `Absolute(0)..Relative(0)` delivers through the latest +/// Relative and time anchors cover whole groups: such a start is frame 0 of its group, and +/// such an end includes its group, so `Absolute(0)..Relative(0)` delivers through the latest /// group. An absolute end is exclusive, like the end of a [`std::ops::Range`]: see /// [`Position::after`] and [`Position::after_group`]. #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] @@ -230,6 +246,8 @@ pub(super) fn union(ranges: impl IntoIterator) -> Vec { (Anchor::Absolute(_), None | Some(Anchor::Absolute(_))) | (Anchor::Relative(_), None) => false, (Anchor::Absolute(start), _) => tail.is_none_or(|tail| start < tail), (Anchor::Relative(back), _) => farthest.is_none_or(|farthest| back > farthest), + // A time can't be placed against the others without the hop's groups. + (Anchor::Time(_), _) => true, })); union.extend(farthest.map(|back| Range { start: Anchor::Relative(back), @@ -346,10 +364,11 @@ impl Subscription { /// These preferences with every range raised to start no earlier than `floor` and /// stop no later than `end`, the exclusive end, without ever widening one. /// - /// `resolved` names where each range resolved, by index, so a relative anchor + /// `resolved` names where each range resolved, by index, so a relative or time anchor /// becomes its own group, raised to `floor` or capped at `end`. A start not resolved /// keeps its anchor, since the floor may lie before it, so the caller drops what it - /// resolves to below `floor`; an end not resolved becomes the bound. + /// resolves to below `floor`; an end not resolved becomes the bound, except a time end, + /// which stays so the caller enforces `end` apart from it. pub(super) fn clamped( &self, floor: Option, @@ -364,11 +383,15 @@ impl Subscription { .map(|(index, range)| { let resolved = resolved.get(index).copied().flatten().filter(|_| clamping); let start = match (range.start, resolved) { - (Anchor::Relative(_), Some(resolved)) => Anchor::Absolute(Position::group(resolved.start)), + (Anchor::Relative(_) | Anchor::Time(_), Some(resolved)) => { + Anchor::Absolute(Position::group(resolved.start)) + } (start, _) => start, }; let requested = match (range.end, resolved.and_then(|resolved| resolved.end)) { - (Some(Anchor::Relative(_)), Some(resolved)) => Some(Anchor::Absolute(Position::group(resolved))), + (Some(Anchor::Relative(_) | Anchor::Time(_)), Some(resolved)) => { + Some(Anchor::Absolute(Position::group(resolved))) + } (requested, _) => requested, }; Range { @@ -379,6 +402,9 @@ impl Subscription { end: match (requested, end) { (requested, None) => requested, (Some(Anchor::Absolute(requested)), Some(end)) => Some(Anchor::Absolute(requested.min(end))), + // Folding the cap in would end the copy's time filter for good, so the + // caller enforces the cap itself and the copy keeps the time. + (Some(Anchor::Time(pts)), Some(_)) => Some(Anchor::Time(pts)), (_, Some(end)) => Some(Anchor::Absolute(end)), }, } @@ -387,6 +413,13 @@ impl Subscription { Self { ranges, ..self.clone() } } + /// Whether any range ends at a presentation time, which an untimed track refuses. + pub(super) fn has_time(&self) -> bool { + self.ranges + .iter() + .any(|range| matches!(range.start, Anchor::Time(_)) || matches!(range.end, Some(Anchor::Time(_)))) + } + /// Whether a subscriber joins at the live edge: a range counting back from the latest /// group and following it. A fixed start resumes a known group instead, such as a route /// switch picking up where the reader stopped. @@ -629,7 +662,7 @@ mod tests { /// group exists only the farthest offset resolves, since it is group 0 regardless. #[test] fn relative_starts_resolve_against_the_latest_group() { - let floor = |sub: &Subscription, latest| Window::new(&sub.ranges, latest).floor(); + let floor = |sub: &Subscription, latest| Window::new(&sub.ranges, latest, &Unheld).unwrap().floor(); let live = Subscription::default().with_ranges([Range::live()]); assert_eq!(floor(&live, Some(7)), Some(7)); diff --git a/rs/moq-net/src/model/subscription/window.rs b/rs/moq-net/src/model/subscription/window.rs index 8fd46645bf8..f52c06823a9 100644 --- a/rs/moq-net/src/model/subscription/window.rs +++ b/rs/moq-net/src/model/subscription/window.rs @@ -9,6 +9,7 @@ use std::ops; use rangemap::RangeInclusiveSet; use super::{Anchor, Range}; +use crate::{Error, Result, Timestamp}; /// One range resolved to groups: from `start` up to the exclusive `end`, `None` while it /// stays open. @@ -18,16 +19,104 @@ pub(in crate::model) struct Resolved { pub end: Option, } +/// When a serving hop's groups start, for resolving time anchors. +/// +/// Each query answers from ordered indexes, so a time end that walks them forward a step +/// at a time pays for each group or run it passes once, however long it stays unsettled. +pub(in crate::model) trait Holdings { + /// Whether the track carries timestamps at all. + fn timed(&self) -> bool; + + /// Where `pts` falls among the groups with a known start. + fn containing(&self, pts: Timestamp) -> Containing; + + /// The lowest group at or above `from` known to start, with where it starts. + fn next_start(&self, from: u64) -> Option<(u64, Timestamp)>; + + /// The lowest held group at or above `from`, with what it says about its start. + fn next_held(&self, from: u64) -> Option<(u64, Held)>; + + /// The exclusive end of the run of sequences from `sequence` that can no longer arrive + /// (dropped, given up, or below where the feed starts), `None` while `sequence` may. + fn gone(&self, sequence: u64) -> Option; + + /// Whether nothing more can arrive or start here, so a time end has nothing to wait + /// on and is left uncapped. + fn complete(&self) -> bool; +} + +/// Where a presentation time falls among the groups a hop holds. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(in crate::model) enum Containing { + /// The newest group starting at or before it. + Group(u64), + /// Every group with a known start begins after it. + Before, + /// No group has a known start. + Unknown, +} + +/// What a held group says about where it starts. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(in crate::model) enum Held { + /// Its frame 0 presented at this time. + Started(Timestamp), + /// It holds frame 0 and has not presented it yet. + Pending, + /// It never will: it ended empty, was aborted, or is held from past frame 0. + Never, +} + +/// A hop that holds no groups of its own, such as a reader following route copies that +/// resolve its ranges for it: a time end there filters nothing. +pub(in crate::model) struct Unheld; + +impl Holdings for Unheld { + fn timed(&self) -> bool { + true + } + + fn containing(&self, _: Timestamp) -> Containing { + Containing::Unknown + } + + fn next_start(&self, _: u64) -> Option<(u64, Timestamp)> { + None + } + + fn next_held(&self, _: u64) -> Option<(u64, Held)> { + None + } + + fn gone(&self, _: u64) -> Option { + None + } + + fn complete(&self) -> bool { + true + } +} + +/// Whether `time` is strictly after `other`, as instants: exact across scales, unlike +/// [`Timestamp`]'s `Ord`, which breaks a tie between scales. +pub(in crate::model) fn later(time: Timestamp, other: Timestamp) -> bool { + u128::from(time.value()) * u128::from(other.scale().as_u64()) + > u128::from(other.value()) * u128::from(time.scale().as_u64()) +} + /// A subscriber's [`Range`]s resolved to the group sequences it reads. #[derive(Debug)] pub(in crate::model) struct Window { /// What these were resolved from, so only replacing the ranges resolves again. ranges: Vec, spans: Vec, - /// The groups the spans cover. + /// The groups the spans cover, an unsettled time end through its cap. groups: RangeInclusiveSet, /// A relative anchor waits on a first group, so nothing is filtered yet. pending: bool, + /// Nothing is left to resolve: see [`Self::is_settled`]. Kept so a read checks it once + /// per group, not once per range. + settled: bool, } /// One non-empty range, resolved. @@ -47,19 +136,180 @@ enum End { Before(u64), /// Past every group. Open, + /// Before the first group starting after a time, once that is known. + Time(TimeEnd), +} + +/// A time end not settled yet, and how far the walk to settle it got. +/// +/// Timestamps never rewind, so the end falls before the lowest group starting past `pts`, +/// once every sequence between it and the newest one starting at or before `pts` is +/// accounted for. Every position below only moves forward, so each group or run of +/// sequences is passed once, whatever a peer leaves missing. +#[derive(Clone, Copy, Debug)] +struct TimeEnd { + pts: Timestamp, + /// The newest group known to start at or before `pts`: it and everything below it in + /// the span is in range. + known: Option, + /// The lowest group known to start after `pts`: where the end falls once settled. + after: Option, + /// Where the walk over held groups resumes: those below it started, or never will. + held: u64, + /// Where settling resumes: every sequence from above `known` up to it is accounted for. + accounted: u64, + /// The exclusive cap on what is read meanwhile, `None` for none. + limit: Option, + /// The group the cap admits on speculation: the first above `known` that may still + /// start, which may start past `pts`. + speculative: Option, +} + +impl TimeEnd { + fn new(pts: Timestamp, start: u64) -> Self { + Self { + pts, + known: None, + after: None, + held: start, + accounted: start, + limit: Some(start), + speculative: None, + } + } + + /// The lowest sequence not known to be in range, `None` once the last sequence is. + fn unknown(&self, start: u64) -> Option { + match self.known { + Some(known) => known.checked_add(1).map(|next| next.max(start)), + None => Some(start), + } + } + + /// Walk the held groups on from where the last call stopped. Returns where the end + /// falls once it settles. + fn advance(&mut self, start: u64, holdings: &impl Holdings) -> Option { + // The starts index passes a group that may still start, so a stalled one never + // keeps a later group known to start inside the end from being read. + while let Some(from) = self.unknown(start) + && let Some((sequence, at)) = holdings.next_start(from) + && self.after.is_none_or(|after| sequence < after) + { + match later(at, self.pts) { + true => self.after = Some(sequence), + false => self.known = Some(sequence), + } + } + + // The first sequence above `known` that may still start is the one read on + // speculation: a held group yet to start, or a missing one that may arrive. Only a + // group starting inside the end is jumped to, since it puts all below it in range. + self.speculative = None; + while let Some(from) = self.unknown(start) { + let at = self.held.max(from); + if self.after.is_some_and(|after| at >= after) { + break; + } + let next = holdings + .next_held(at) + .filter(|(sequence, _)| self.after.is_none_or(|after| *sequence < after)); + match next { + Some((sequence, held)) if sequence == at => match held { + Held::Started(time) if later(time, self.pts) => self.after = Some(at), + Held::Started(_) => self.known = Some(at), + Held::Never => {} + Held::Pending => { + self.held = at; + self.speculative = Some(at); + break; + } + }, + Some((sequence, Held::Started(time))) if !later(time, self.pts) => self.known = Some(sequence), + // Past every held group: the next to arrive is the speculative one, judged + // once a settle sees it, before it is read. + None if self.after.is_none() => { + self.held = at; + break; + } + _ => match holdings.gone(at) { + Some(end) if end > at => { + self.held = end; + continue; + } + _ => { + self.held = at; + self.speculative = (!holdings.complete()).then_some(at); + break; + } + }, + } + // Nothing sits above the last sequence, so the walk ends on it. + let Some(next) = at.checked_add(1) else { break }; + self.held = next; + } + self.limit = match (self.speculative, self.after) { + _ if holdings.complete() => None, + (Some(speculative), _) => speculative.checked_add(1), + (None, Some(after)) => Some(after), + // Every held group is accounted for: the next to arrive is the speculative one. + (None, None) => None, + }; + + // Settle once every sequence up to the first group starting past `pts` can no + // longer start inside it. + // + // Datagrams stay ephemeral and never index where their group starts, so a sequence + // that arrived only as a datagram counts as gone here. A stream copy of that group + // arriving later is read even if it starts past `pts`: accepted, since a publisher + // that streams a group it also sent as a datagram is rare. + let after = self.after?; + // Everything at or below `known` is in range, and `after` lies above it. + self.accounted = self.accounted.max(self.unknown(start).unwrap_or(after)); + while self.accounted < after { + let next = holdings.next_held(self.accounted); + match next { + Some((sequence, held)) if sequence == self.accounted => match held { + Held::Pending => return None, + // It started past `pts` since the walk above stopped on it: everything + // below is accounted for, so the end falls here. + Held::Started(at) if later(at, self.pts) => { + self.after = Some(sequence); + return self.after; + } + _ => self.accounted += 1, + }, + _ => { + let below = next.map_or(after, |(sequence, _)| sequence.min(after)); + self.accounted = match holdings.complete() { + true => below, + false => holdings.gone(self.accounted)?.min(below), + }; + } + } + } + self.after + } } impl Window { /// Resolve `ranges`, counting relative anchors back from `latest`. - pub fn new(ranges: &[Range], latest: Option) -> Self { + /// + /// Refuses a time anchor on an untimed track with [`Error::TimestampMismatch`]. + pub fn new(ranges: &[Range], latest: Option, holdings: &impl Holdings) -> Result { let mut window = Self { ranges: ranges.to_vec(), spans: Vec::new(), groups: RangeInclusiveSet::new(), pending: false, + settled: false, }; - window.place(latest); - window + window.place(latest, holdings)?; + Ok(window) + } + + /// Whether nothing is left to resolve as groups arrive and start. + pub fn is_settled(&self) -> bool { + self.settled } /// Whether a relative anchor still waits on a first group. @@ -67,14 +317,53 @@ impl Window { self.pending } - /// Resolve the relative anchors waiting on a first group once `latest` exists. - /// Returns whether anything moved. - pub fn settle(&mut self, latest: Option) -> bool { - if !self.pending || latest.is_none() { - return false; + /// Resolve what the groups now held settle: relative anchors once `latest` exists, and + /// time ends as groups start. Returns whether the groups in range moved. + pub fn settle(&mut self, latest: Option, holdings: &impl Holdings) -> bool { + if self.pending { + if latest.is_none() { + return false; + } + // Checked when the ranges were first resolved, against the same track. + let _ = self.place(latest, holdings); + return !self.pending; } - self.place(latest); - !self.pending + let mut moved = false; + for span in &mut self.spans { + let End::Time(time) = &mut span.end else { continue }; + let limit = time.limit; + if let Some(end) = time.advance(span.start, holdings) { + span.end = End::Before(end); + moved = true; + } else { + moved |= time.limit != limit; + } + } + if moved { + self.rebuild(); + } + moved + } + + /// The groups an unsettled time end reads on speculation, which may start past it: a + /// read parked on the cap wakes when one of them starts. + pub fn speculative(&self) -> impl Iterator + '_ { + self.spans.iter().filter_map(|span| match span.end { + End::Time(time) => time.speculative, + _ => None, + }) + } + + /// The lowest sequence whose start could still move an unsettled time end, `None` + /// without one or once the last sequence is in range. + pub fn watch_starts(&self) -> Option { + self.spans + .iter() + .filter_map(|span| match span.end { + End::Time(time) => time.unknown(span.start), + _ => None, + }) + .min() } /// Whether group `sequence` is in range. Everything is while a relative anchor waits. @@ -82,6 +371,37 @@ impl Window { self.pending || self.groups.contains(&sequence) } + /// Whether a datagram is in range, judged on its own and never through a time end's + /// settling: its `timestamp` against each range's times, and its `sequence` against the + /// range's absolute groups and the relative ones resolved so far. + pub fn admits_datagram(&self, sequence: u64, timestamp: Option) -> bool { + let Some(timestamp) = timestamp else { + return self.contains(sequence); + }; + self.ranges + .iter() + .enumerate() + .filter(|(_, range)| !range.is_empty()) + .any(|(index, range)| { + // A relative anchor still waiting on a latest group bounds nothing yet. + let span = self.spans.iter().find(|span| span.index == index); + let started = match range.start { + Anchor::Time(pts) => !later(pts, timestamp), + Anchor::Absolute(start) => sequence >= start.group, + Anchor::Relative(_) => span.is_none_or(|span| sequence >= span.start), + }; + let ended = match range.end { + None => true, + Some(Anchor::Time(pts)) => !later(timestamp, pts), + Some(Anchor::Absolute(end)) => end.final_sequence().is_none_or(|end| sequence < end), + Some(Anchor::Relative(_)) => { + span.is_none_or(|span| !matches!(span.end, End::Before(end) if sequence >= end)) + } + }; + started && ended + }) + } + /// The lowest group in range, `None` while a relative anchor waits. Nothing in range /// floors nowhere: group 0. pub fn floor(&self) -> Option { @@ -101,7 +421,7 @@ impl Window { } /// Each range resolved, in the order given: `None` for one that is empty or still - /// waits on a first group. + /// waits on a first group. An unsettled time end is reported open, never at its cap. pub fn resolved(&self) -> Vec> { let mut resolved = vec![None; self.ranges.len()]; for span in &self.spans { @@ -109,16 +429,17 @@ impl Window { start: span.start, end: match span.end { End::Before(end) => Some(end), - End::Open => None, + End::Open | End::Time(_) => None, }, }); } resolved } - /// The exclusive end past every group in range, `None` when one stays open. + /// The exclusive end past every group in range, `None` when one stays open. An + /// unsettled time end counts as open: its cap only holds reads back meanwhile. pub fn end(&self) -> Option { - if self.pending { + if self.pending || self.spans.iter().any(|span| matches!(span.end, End::Time(_))) { return None; } match self.groups.iter().next_back() { @@ -152,14 +473,31 @@ impl Window { Some((*run.start()).max(range.start)..end) } - fn place(&mut self, latest: Option) { + fn place(&mut self, latest: Option, holdings: &impl Holdings) -> Result<()> { + let time = |anchor: Option| matches!(anchor, Some(Anchor::Time(_))); + if !holdings.timed() + && self + .ranges + .iter() + .any(|range| time(Some(range.start)) || time(range.end)) + { + return Err(Error::TimestampMismatch); + } self.spans.clear(); self.pending = false; for (index, range) in self.ranges.iter().enumerate().filter(|(_, range)| !range.is_empty()) { - let (Some((start, floored)), Some(end)) = (start(range.start, latest), end(range.end, latest)) else { + let (Some((start, floored)), Some(end)) = (start(range.start, latest, holdings), end(range.end, latest)) + else { self.pending = true; continue; }; + let end = match end { + End::Time(mut time) => { + time = TimeEnd::new(time.pts, start); + time.advance(start, holdings).map_or(End::Time(time), End::Before) + } + end => end, + }; self.spans.push(Span { index, start, @@ -167,14 +505,25 @@ impl Window { floored, }); } + self.rebuild(); + Ok(()) + } + + fn rebuild(&mut self) { + self.settled = !self.pending && !self.spans.iter().any(|span| matches!(span.end, End::Time(_))); self.groups = RangeInclusiveSet::new(); for span in &self.spans { - let last = match span.end { - End::Before(end) => match end.checked_sub(1) { + let end = match span.end { + End::Before(end) => Some(end), + End::Open => None, + End::Time(time) => time.limit, + }; + let last = match end { + Some(end) => match end.checked_sub(1) { Some(last) if last >= span.start => last, _ => continue, }, - End::Open => u64::MAX, + None => u64::MAX, }; self.groups.insert(span.start..=last); } @@ -183,16 +532,24 @@ impl Window { /// The first group a range starts at, and whether it names that group rather than /// reaching back to the oldest one. `None` while it waits on a latest group. -fn start(anchor: Anchor, latest: Option) -> Option<(u64, bool)> { +fn start(anchor: Anchor, latest: Option, holdings: &impl Holdings) -> Option<(u64, bool)> { match anchor { Anchor::Absolute(start) => Some((start.group, true)), // Nothing sits below group 0, so the farthest offset needs no latest group. Anchor::Relative(u64::MAX) => Some((0, false)), Anchor::Relative(back) => Some((latest?.saturating_sub(back), true)), + Anchor::Time(pts) => match holdings.containing(pts) { + Containing::Group(group) => Some((group, true)), + // The group holding it is older than any held, so reach back to the oldest. + Containing::Before => Some((0, false)), + // Nothing has started yet: like the live edge, the latest group. + Containing::Unknown => Some((latest?, true)), + }, } } -/// Where a range stops, or `None` while it waits on a latest group. +/// Where a range stops, or `None` while it waits on a latest group. A time end's walk +/// starts once its start is known. fn end(anchor: Option, latest: Option) -> Option { Some(match anchor { None => End::Open, @@ -201,6 +558,7 @@ fn end(anchor: Option, latest: Option) -> Option { .saturating_sub(back) .checked_add(1) .map_or(End::Open, End::Before), + Some(Anchor::Time(pts)) => End::Time(TimeEnd::new(pts, 0)), }) } @@ -209,6 +567,115 @@ mod tests { use super::super::Position; use super::*; + use std::cell::Cell; + + fn at(millis: u64) -> Timestamp { + Timestamp::from_millis(millis).unwrap() + } + + /// A hop's groups by sequence, the runs that can no longer arrive, and how many held + /// groups and starts a walk asked about. + #[derive(Default)] + struct Groups { + held: std::collections::BTreeMap, + gone: Vec>, + complete: bool, + asked: Cell, + } + + impl Groups { + fn new(held: impl IntoIterator) -> Self { + Self { + held: held.into_iter().collect(), + ..Self::default() + } + } + + fn started(sequence: u64, millis: u64) -> (u64, Held) { + (sequence, Held::Started(at(millis))) + } + } + + impl Holdings for Groups { + fn timed(&self) -> bool { + true + } + + fn containing(&self, pts: Timestamp) -> Containing { + let mut stamped = false; + for (&sequence, held) in self.held.iter().rev() { + let Held::Started(start) = held else { continue }; + stamped = true; + if !later(*start, pts) { + return Containing::Group(sequence); + } + } + match stamped { + true => Containing::Before, + false => Containing::Unknown, + } + } + + fn next_start(&self, from: u64) -> Option<(u64, Timestamp)> { + self.asked.set(self.asked.get() + 1); + self.held.range(from..).find_map(|(&sequence, held)| match held { + Held::Started(start) => Some((sequence, *start)), + _ => None, + }) + } + + fn next_held(&self, from: u64) -> Option<(u64, Held)> { + self.asked.set(self.asked.get() + 1); + self.held + .range(from..) + .next() + .map(|(&sequence, held)| (sequence, *held)) + } + + fn gone(&self, sequence: u64) -> Option { + self.gone + .iter() + .find(|gone| gone.contains(&sequence)) + .map(|gone| gone.end) + } + + fn complete(&self) -> bool { + self.complete + } + } + + struct Untimed; + + impl Holdings for Untimed { + fn timed(&self) -> bool { + false + } + + fn containing(&self, _: Timestamp) -> Containing { + Containing::Unknown + } + + fn next_start(&self, _: u64) -> Option<(u64, Timestamp)> { + None + } + + fn next_held(&self, _: u64) -> Option<(u64, Held)> { + None + } + + fn gone(&self, _: u64) -> Option { + None + } + + fn complete(&self) -> bool { + false + } + } + + fn time(millis: u64) -> Anchor { + Anchor::Time(Timestamp::from_millis(millis).unwrap()) + } + fn range(start: Anchor, end: impl Into>) -> Range { Range { start, end: end.into() } } @@ -220,7 +687,7 @@ mod tests { /// Disjoint ranges filter to exactly their groups; the gap between them is out. #[test] fn ranges_cover_only_their_groups() { - let window = Window::new(&[Range::groups(2..4), Range::groups(7..)], Some(9)); + let window = Window::new(&[Range::groups(2..4), Range::groups(7..)], Some(9), &Untimed).unwrap(); assert_eq!(covered(&window, 10), vec![2, 3, 7, 8, 9]); assert_eq!(window.floor(), Some(2)); assert_eq!(window.end(), None); @@ -243,12 +710,14 @@ mod tests { Anchor::Absolute(Position { group: 3, frame: 1 }), )], None, - ); + &Untimed, + ) + .unwrap(); assert_eq!(covered(&window, 6), vec![1, 2, 3]); assert_eq!(window.end(), Some(4)); // Nothing asked for is nothing in range, floored nowhere. - let window = Window::new(&[], None); + let window = Window::new(&[], None, &Untimed).unwrap(); assert_eq!(covered(&window, 4), Vec::::new()); assert_eq!((window.floor(), window.end()), (Some(0), Some(0))); } @@ -257,18 +726,18 @@ mod tests { /// for a first group when there is none. #[test] fn relative_anchors_count_back_from_the_latest_group() { - let window = Window::new(&[range(Anchor::Relative(3), Anchor::Relative(1))], Some(7)); + let window = Window::new(&[range(Anchor::Relative(3), Anchor::Relative(1))], Some(7), &Untimed).unwrap(); assert_eq!(covered(&window, 10), vec![4, 5, 6]); - let mut window = Window::new(&[Range::live()], None); + let mut window = Window::new(&[Range::live()], None, &Untimed).unwrap(); assert!(window.is_pending() && window.contains(3) && window.floor().is_none()); - assert!(!window.settle(None)); - assert!(window.settle(Some(5))); + assert!(!window.settle(None, &Untimed)); + assert!(window.settle(Some(5), &Untimed)); assert_eq!(covered(&window, 8), vec![5, 6, 7]); - assert!(!window.is_pending()); + assert!(window.is_settled()); // The farthest offset needs no latest group, and names no floor. - let window = Window::new(&[Range::oldest()], None); + let window = Window::new(&[Range::oldest()], None, &Untimed).unwrap(); assert_eq!((window.floor(), window.start()), (Some(0), None)); } @@ -276,7 +745,7 @@ mod tests { /// group or still waits on a first one; an explicit group 0 names one. #[test] fn the_named_start_is_the_lowest_unless_one_reaches_back() { - let start = |ranges: &[Range], latest| Window::new(ranges, latest).start(); + let start = |ranges: &[Range], latest| Window::new(ranges, latest, &Untimed).unwrap().start(); assert_eq!(start(&[Range::groups(4..), Range::live()], Some(9)), Some(4)); assert_eq!(start(&[Range::groups(4..), Range::live()], Some(2)), Some(2)); assert_eq!(start(&[Range::groups(0..)], None), Some(0)); @@ -284,4 +753,290 @@ mod tests { assert_eq!(start(&[Range::groups(4..), Range::oldest()], Some(9)), None); assert_eq!(start(&[Range::groups(4..), Range::groups(..2)], Some(9)), None); } + + /// A time start is the group containing it, and an end stops just past that group once + /// a later group is known to start after it with nothing between left to arrive. + #[test] + fn time_anchors_resolve_from_group_starts() { + let groups = Groups::new([ + Groups::started(2, 2000), + Groups::started(3, 3000), + Groups::started(4, 4000), + Groups::started(5, 5000), + (6, Held::Pending), + ]); + let window = Window::new(&[range(time(3500), time(4500))], Some(6), &groups).unwrap(); + assert_eq!(covered(&window, 8), vec![3, 4]); + assert!(window.is_settled()); + assert_eq!(window.resolved(), vec![Some(Resolved { start: 3, end: Some(5) })]); + + // Group 6 may yet start at 5000 too: read on speculation, the end left open. + let window = Window::new(&[range(time(3500), time(5000))], Some(6), &groups).unwrap(); + assert!(!window.is_settled()); + assert_eq!(covered(&window, 9), vec![3, 4, 5, 6]); + assert_eq!(window.resolved(), vec![Some(Resolved { start: 3, end: None })]); + assert_eq!(window.end(), None, "the cap is not the end"); + assert_eq!(window.speculative().collect::>(), vec![6]); + + // Before every start held: reach back to the oldest. + let window = Window::new(&[range(time(1000), time(2500))], Some(6), &groups).unwrap(); + assert_eq!(covered(&window, 8), vec![0, 1, 2]); + + // Nothing started yet: a time start waits like the live edge. + let mut window = Window::new(&[range(time(1000), None)], None, &Groups::default()).unwrap(); + assert!(window.is_pending()); + assert!(window.settle(Some(4), &Groups::new([(4, Held::Pending)]))); + assert_eq!(window.floor(), Some(4)); + } + + /// An end settles only once every sequence between the newest group starting inside it + /// and the first starting past it is accounted for. + #[test] + fn a_time_end_settles_once_nothing_below_its_end_can_arrive() { + let mut groups = Groups::new([Groups::started(3, 3000), Groups::started(6, 6000)]); + let mut window = Window::new(&[range(time(3000), time(4000))], Some(6), &groups).unwrap(); + assert!(!window.is_settled(), "4 and 5 may still arrive inside the end"); + assert_eq!( + covered(&window, 8), + vec![3, 4], + "only 4, the first that may still arrive, on speculation, and never 6" + ); + + // 4 arrives yet to start: still the one read on speculation, so 5 waits. + groups.held.insert(4, Held::Pending); + assert!(!window.settle(Some(6), &groups), "4 was already the speculative one"); + assert_eq!(covered(&window, 8), vec![3, 4]); + groups.held.insert(4, Held::Started(at(3500))); + assert!(window.settle(Some(6), &groups)); + assert!(!window.is_settled(), "5 may still arrive"); + groups.gone.push(5..6); + assert!(window.settle(Some(6), &groups)); + assert!(window.is_settled()); + assert_eq!(covered(&window, 8), vec![3, 4, 5]); + + // A group that never starts proves nothing, so it never holds the end open. + let groups = Groups::new([Groups::started(3, 3000), (4, Held::Never), Groups::started(5, 5000)]); + let window = Window::new(&[range(time(3000), time(4000))], Some(5), &groups).unwrap(); + assert!(window.is_settled()); + assert_eq!(covered(&window, 8), vec![3, 4]); + } + + /// An unsettled time end reads at most one group on speculation: the first above the + /// newest known to start inside it that may still start. A group that never starts + /// does not take that place, and a stalled one does not hold back a later group known + /// to start inside the end. + #[test] + fn an_unsettled_time_end_reads_one_group_on_speculation() { + let mut groups = Groups::new([(3, Held::Pending), (4, Held::Pending), (5, Held::Pending)]); + let mut window = Window::new( + &[range(Anchor::Absolute(Position::group(3)), time(3500))], + Some(5), + &groups, + ) + .unwrap(); + assert_eq!(covered(&window, 8), vec![3]); + + groups.held.insert(3, Held::Started(at(3000))); + assert!(window.settle(Some(5), &groups)); + assert_eq!(covered(&window, 8), vec![3, 4]); + + // 4 never starts, so 5 is the one read on speculation. + groups.held.insert(4, Held::Never); + assert!(window.settle(Some(5), &groups)); + assert_eq!(covered(&window, 8), vec![3, 4, 5]); + + // 5 stalls, 6 starts inside the end: 5 is in range then, and 7 is speculative. + groups.held.insert(6, Held::Started(at(3400))); + groups.held.insert(7, Held::Pending); + assert!(window.settle(Some(7), &groups)); + assert_eq!(covered(&window, 9), vec![3, 4, 5, 6, 7]); + + // 7 starts past the end, which settles with 5 still stalled below it. + groups.held.insert(7, Held::Started(at(3600))); + assert!(window.settle(Some(7), &groups)); + assert!(window.is_settled()); + assert_eq!(covered(&window, 9), vec![3, 4, 5, 6]); + + // Every held group accounted for: the next to arrive is the speculative one. + let groups = Groups::new([Groups::started(3, 3000)]); + let window = Window::new(&[range(time(3000), time(3500))], Some(3), &groups).unwrap(); + assert!(window.contains(4)); + + // Nothing more can arrive or start: uncapped. + let mut groups = Groups::new([Groups::started(3, 3000), (4, Held::Pending), (5, Held::Pending)]); + groups.complete = true; + let window = Window::new(&[range(time(3000), time(3500))], Some(5), &groups).unwrap(); + assert_eq!(covered(&window, 7), vec![3, 4, 5, 6]); + } + + /// A peer that leaves an early sequence missing and keeps sending later groups costs a + /// time end each group once, not a rescan of everything since the gap on every settle. + #[test] + fn a_time_end_walks_each_group_once_behind_a_missing_sequence() { + let mut groups = Groups::new([Groups::started(0, 0)]); + let mut window = Window::new(&[range(time(0), time(1_000_000))], Some(0), &groups).unwrap(); + let count = 2_000; + for sequence in 2..count { + groups.held.insert(sequence, Held::Started(at(sequence * 10))); + window.settle(Some(sequence), &groups); + } + assert!(!window.is_settled(), "group 1 never arrived"); + assert!(window.contains(count - 1)); + let asked = groups.asked.get(); + assert!(asked < 8 * count as usize, "{asked} queries for {count} groups"); + } + + /// A datagram is one frame, so its own timestamp places it against a time end. + #[test] + fn a_datagram_is_judged_by_its_own_timestamp() { + let window = Window::new( + &[range(Anchor::Absolute(Position::group(0)), time(3500))], + None, + &Groups::default(), + ) + .unwrap(); + assert!(window.admits_datagram(9, Some(at(3500)))); + assert!(!window.admits_datagram(9, Some(at(3600)))); + } + + /// Groups whose `racing` one starts past the end after the first time it is asked + /// about, as when its first frame lands between the walks of one settle. + struct Racing { + groups: Groups, + racing: u64, + asked: Cell, + } + + impl Holdings for Racing { + fn timed(&self) -> bool { + true + } + + fn containing(&self, pts: Timestamp) -> Containing { + self.groups.containing(pts) + } + + fn next_start(&self, from: u64) -> Option<(u64, Timestamp)> { + self.groups.next_start(from) + } + + fn next_held(&self, from: u64) -> Option<(u64, Held)> { + let next = self.groups.next_held(from); + if next.is_some_and(|(sequence, _)| sequence == self.racing) && self.asked.replace(true) { + return Some((self.racing, Held::Started(at(6000)))); + } + next + } + + fn gone(&self, sequence: u64) -> Option { + self.groups.gone(sequence) + } + + fn complete(&self) -> bool { + false + } + } + + #[test] + fn a_time_end_settles_on_a_group_that_starts_between_its_walks() { + let racing = Racing { + groups: Groups::new([Groups::started(0, 0), (1, Held::Pending), Groups::started(2, 7000)]), + racing: 1, + asked: Cell::new(false), + }; + let window = Window::new( + &[range(Anchor::Absolute(Position::group(0)), time(5000))], + None, + &racing, + ) + .unwrap(); + assert_eq!(window.resolved(), vec![Some(Resolved { start: 0, end: Some(1) })]); + } + + #[test] + fn a_missing_sequence_past_a_time_end_is_the_one_read_on_speculation() { + // 1 and 2 are missing, and 3 never starts: the walk may not pass the gap, or every + // sequence in it that arrives later is read past the end. + let mut groups = Groups::new([Groups::started(0, 3000), (3, Held::Never)]); + let mut window = Window::new(&[range(time(3000), time(3500))], Some(3), &groups).unwrap(); + assert_eq!(covered(&window, 5), vec![0, 1]); + assert_eq!(window.speculative().collect::>(), vec![1]); + groups.held.insert(2, Held::Pending); + groups.held.insert(1, Held::Pending); + window.settle(Some(3), &groups); + assert_eq!(covered(&window, 5), vec![0, 1], "2 waits on 1 starting"); + groups.held.insert(1, Held::Started(at(3600))); + window.settle(Some(3), &groups); + assert_eq!(window.resolved(), vec![Some(Resolved { start: 0, end: Some(1) })]); + } + + #[test] + fn a_time_end_stops_walking_at_the_last_sequence() { + let groups = Groups::new([Groups::started(u64::MAX, 3000)]); + let window = Window::new( + &[range(Anchor::Absolute(Position::group(u64::MAX)), time(5000))], + None, + &groups, + ) + .unwrap(); + assert_eq!( + window.resolved(), + vec![Some(Resolved { + start: u64::MAX, + end: None + })] + ); + assert!(window.contains(u64::MAX)); + assert_eq!(window.watch_starts(), None, "nothing above the last sequence can start"); + } + + #[test] + fn a_settled_window_still_judges_a_datagram_by_its_times() { + // 1 is a datagram the starts index never sees: the end settles before 2 by + // sequence, but 1 presents past it. + let groups = Groups { + gone: std::iter::once(1..2).collect(), + ..Groups::new([Groups::started(0, 0), Groups::started(2, 10_000)]) + }; + let window = Window::new(&[range(time(0), time(5000))], None, &groups).unwrap(); + assert_eq!(window.resolved(), vec![Some(Resolved { start: 0, end: Some(2) })]); + assert!(window.admits_datagram(1, Some(at(4000)))); + assert!(!window.admits_datagram(1, Some(at(7000)))); + } + + #[test] + fn a_waiting_window_judges_a_datagram_against_both_times() { + let window = Window::new(&[range(time(3000), time(3500))], None, &Groups::default()).unwrap(); + assert!(window.is_pending(), "no group has started"); + assert!(!window.admits_datagram(0, Some(at(2900)))); + assert!(window.admits_datagram(1, Some(at(3000)))); + assert!(window.admits_datagram(2, Some(at(3500)))); + assert!(!window.admits_datagram(3, Some(at(3600)))); + } + + #[test] + fn a_waiting_window_judges_a_datagram_against_absolute_groups_too() { + let window = Window::new( + &[range(time(3000), Some(Anchor::Absolute(Position::group(10))))], + None, + &Groups::default(), + ) + .unwrap(); + assert!(window.is_pending(), "no group has started"); + assert!(window.admits_datagram(9, Some(at(3500)))); + assert!(!window.admits_datagram(10, Some(at(3500)))); + assert!(!window.admits_datagram(9, Some(at(2900)))); + } + + #[test] + fn an_untimed_track_refuses_time_anchors() { + assert!(matches!( + Window::new(&[range(time(0), None)], None, &Untimed), + Err(Error::TimestampMismatch) + )); + assert!(matches!( + Window::new(&[range(Anchor::Relative(0), time(0))], None, &Untimed), + Err(Error::TimestampMismatch) + )); + } } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 5a7d2f66c1b..bc82a82925a 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -22,7 +22,7 @@ use super::{Datagram, Requests}; use super::Cap; pub use super::subscription::{Anchor, Order, Position, Range, Subscription}; -use super::subscription::{Resolved, Window}; +use super::subscription::{Containing, Held, Holdings, Resolved, Window, later}; use rangemap::{RangeInclusiveSet, RangeSet}; mod demand; @@ -351,6 +351,11 @@ pub(crate) struct TrackState { // the read guard a first frame holds, so it needs its own lock, never contended. bounds: Mutex>, + // Where each cached group's frame 0 presented, by sequence, received groups too: what + // a time anchor resolves against without walking groups that have not started. Filled + // through the read guard a first frame holds, like `bounds`. + starts: Mutex>, + // The sequence and start of the highest-sequence group that left the cache holding a // checked start (see [`group::Origin::Bound`]). Eviction is not a sequence prefix, so a // group above it is checked against it as well as its cached neighbors; one at or below @@ -589,6 +594,20 @@ impl Slot { } } + /// Where the group starts in presentation time, from a live copy holding frame 0. A + /// copy joined mid-group stamps a later frame, which proves nothing about the start. + fn start(&self) -> Option { + self.copies() + .filter(|group| group.live_first_frame() == Some(0)) + .find_map(group::Handle::timestamp) + } + + /// Whether the group has no start yet but may still get one: a copy holding frame 0 + /// has not presented it. + fn may_start(&self) -> bool { + self.start().is_none() && self.copies().any(group::Handle::may_start) + } + /// The checked start of whichever copy began at frame 0. fn bound(&self) -> Option { self.copies().find_map(group::Handle::bound) @@ -605,12 +624,14 @@ impl Slot { /// Heap the track keeps per cached group, excluding the group itself /// ([`group::CACHE_OVERHEAD`]). /// -/// One [`Slot`] under its sequence in `lookup`, plus a hint in each of `arrival` and -/// `evict`. Doubled because both containers run half empty in the worst case: a +/// One [`Slot`] under its sequence in `lookup`, its start in `starts` once frame 0 +/// arrives, plus a hint in each of `arrival` and `evict`. Doubled because the +/// containers run half empty in the worst case: a /// `BTreeMap` node sits between half and fully packed, and a `VecDeque` holds up to /// twice the entries in it. Half of [`cache::ENTRY_OVERHEAD`]; see it for why this is /// derived rather than measured. -pub(super) const CACHE_OVERHEAD: u64 = 2 * (size_of::() + size_of::() + 2 * size_of::<(u64, u32)>()) as u64; +pub(super) const CACHE_OVERHEAD: u64 = + 2 * (size_of::() + size_of::() + size_of::<(u64, Timestamp)>() + 2 * size_of::<(u64, u32)>()) as u64; /// Reverse state for [`Consumer::fetch_group`], beside the track state in its own /// [`kio::Shared`]: consumers enqueue (coalescing per sequence, so a relay opens one @@ -651,6 +672,71 @@ pub(crate) struct FetchOutcome { accepted: bool, } +/// Where the groups this hop holds start, for resolving time anchors: ordered lookups in +/// the starts index and the cache, so no query walks more than the run it answers. +impl Holdings for TrackState { + fn timed(&self) -> bool { + self.info.as_ref().is_none_or(|info| info.timescale.is_some()) + } + + 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()) + else { + return Containing::Unknown; + }; + if later(first_at, pts) { + return Containing::Before; + } + if !later(last_at, pts) { + return Containing::Group(last); + } + // Timestamps never rewind, so the starts ascend with the sequence: bisect the + // sequences between the newest start known at or before `pts` and one known after. + let (mut best, mut low, mut high) = (first, first, last); + while high - low > 1 { + let mid = low + (high - low) / 2; + match starts.range(low + 1..=mid).next_back() { + Some((&sequence, &at)) if later(at, pts) => high = sequence, + Some((&sequence, _)) => (best, low) = (sequence, mid), + None => low = mid, + } + } + Containing::Group(best) + } + + fn next_start(&self, from: u64) -> Option<(u64, Timestamp)> { + let starts = self.starts.lock().unwrap_or_else(PoisonError::into_inner); + starts.range(from..).next().map(|(&sequence, &at)| (sequence, at)) + } + + fn next_held(&self, from: u64) -> Option<(u64, Held)> { + let (&sequence, slot) = self.lookup.range(from..).next()?; + let held = match slot.start() { + Some(at) => Held::Started(at), + None if slot.may_start() => Held::Pending, + None => Held::Never, + }; + Some((sequence, held)) + } + + fn gone(&self, sequence: u64) -> Option { + // Without a declared start, the first arrival stands in for it, as the tail's does. + let floor = self.start_sequence.or(self.arrivals.first()).unwrap_or(0); + if sequence < floor { + return Some(floor); + } + self.drops + .get(&sequence) + .map(|dropped| dropped.end) + .or_else(|| self.arrivals.run_end(sequence)) + } + + fn complete(&self) -> bool { + self.sealed || self.abort.is_some() + } +} + impl TrackState { /// A consumer joined at the live edge: ask for a group past every sequence written when /// the newest group's bytes after its first frame exceed [`Info::refresh`] times that @@ -719,9 +805,31 @@ impl TrackState { }) } + /// Group `sequence` wrote its first frame: index where it starts, and share a + /// just-checked `bound` with its neighbors. + pub(super) fn first_frame(&self, sequence: u64, bound: Option) { + if let Some(bound) = bound { + self.share_bound(sequence, bound); + } + self.index_start(sequence); + } + + /// Index where cached group `sequence` starts, read back through its slot so only the + /// cached copy holding frame 0 counts, and wake the reads a time end parks on it. + fn index_start(&self, sequence: u64) { + let Some(start) = self.lookup.get(&sequence).and_then(Slot::start) else { + return; + }; + self.starts + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(sequence, start); + self.cache.wakes().started(sequence); + } + /// Record group `sequence`'s just-checked `start` as a bound for its neighbors and the /// name's later producers. A group stamped after it left the cache bounds nothing. - pub(super) fn share_bound(&self, sequence: u64, start: Timestamp) { + fn share_bound(&self, sequence: u64, start: Timestamp) { if self.lookup.get(&sequence).and_then(Slot::bound) != Some(start) { return; } @@ -1462,6 +1570,10 @@ impl TrackState { .get_mut() .unwrap_or_else(PoisonError::into_inner) .retain(|sequence, _| lookup.contains_key(sequence)); + self.starts + .get_mut() + .unwrap_or_else(PoisonError::into_inner) + .retain(|sequence, _| lookup.contains_key(sequence)); } /// Take `sequence` out of the cache, keeping the withheld count in step. @@ -1472,6 +1584,10 @@ impl TrackState { .get_mut() .unwrap_or_else(PoisonError::into_inner) .remove(&sequence); + self.starts + .get_mut() + .unwrap_or_else(PoisonError::into_inner) + .remove(&sequence); Some(slot) } @@ -1633,6 +1749,8 @@ impl TrackState { .unwrap_or_else(PoisonError::into_inner) .remove(&sequence); } + // A group inserted with its first frame already written started before it was here. + self.index_start(sequence); self.cache.wakes().arrived(sequence..=sequence); if visible && !pending { self.arrival.push_back((sequence, stamp)); @@ -4251,6 +4369,7 @@ impl Subscribing { // Wait until the track info is available let info = ready!(self.state.poll(waiter, |state| state.poll_info())) .map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??; + refuse_untimed(&info, &self.subscription.read())?; let resume = self.state.read().routes.clone(); let inner = match resume { @@ -5091,6 +5210,8 @@ struct Filter { resolved: u64, /// Bumped whenever the window moves, so the drop scan knows to look again. revision: u64, + /// Why the ranges can't be served here: a time anchor on an untimed track. + refused: Option, /// What the arrival walk reads: the window from the floor, below the cap. The walk /// passes a group outside it holding nothing, so widening it re-scans the cache. admitted: RangeInclusiveSet, @@ -5113,7 +5234,10 @@ impl Cursor { let preferences = subscription.read(); let ranges_revision = subscription.ranges_revision(); let state = state.read(); - let window = Window::new(&preferences.ranges, state.newest_produced); + let (window, refused) = match Window::new(&preferences.ranges, state.newest_produced, &*state) { + Ok(window) => (window, None), + Err(err) => (Window::new(&[], None, &*state).expect("nothing to refuse"), Some(err)), + }; // Arrival order need not be timestamp order. Only skip the initial stale prefix; // subsequent delivery stays in arrival order, including late datagrams. let newest = state.datagrams.iter().filter_map(|datagram| datagram.timestamp).max(); @@ -5133,6 +5257,7 @@ impl Cursor { window, resolved: ranges_revision, revision: 0, + refused, admitted: RangeInclusiveSet::new(), admitted_at: None, owed: RangeInclusiveSet::new(), @@ -5214,8 +5339,8 @@ impl Cursor { /// Bring [`Self::window`] up to date: resolved again once the subscriber replaces its /// ranges, counting relative anchors back from the latest group then, and settled as - /// the groups it waits on arrive. - fn sync(&mut self) { + /// the groups it waits on arrive. Returns whether the window moved. + fn sync(&mut self) -> bool { let replaced = { let preferences = self.subscription.read(); let revision = self.subscription.ranges_revision(); @@ -5225,18 +5350,50 @@ impl Cursor { let moved = match replaced { Some((revision, ranges)) => { self.filter.resolved = revision; - self.filter.window = Window::new(&ranges, state.newest_produced); + match Window::new(&ranges, state.newest_produced, &*state) { + Ok(window) => { + self.filter.window = window; + self.filter.refused = None; + } + Err(err) => self.filter.refused = Some(err), + } true } + None if self.filter.window.is_settled() => false, // The track had no produced group when the ranges resolved, so the newest one // then was the first to arrive, however many have arrived or left since. - None => self.filter.window.settle(state.first_produced), + None if self.filter.window.is_pending() => self.filter.window.settle(state.first_produced, &*state), + None => self.filter.window.settle(state.newest_produced, &*state), }; drop(state); if moved { self.filter.revision += 1; self.update_servable(); } + moved + } + + /// Park a read the cap of an unsettled time end holds back on what raises it: the + /// speculative group starting or ending, or any group it waits on starting. Returns + /// whether the cap already moved, so the read goes again rather than parking. + fn watch_time(&mut self, waiter: &kio::Waiter) -> bool { + if self.filter.window.is_settled() || self.filter.refused.is_some() { + return false; + } + { + let state = self.state.read(); + if let Some(from) = self.filter.window.watch_starts() { + state.cache.wakes().watch_start(from, waiter); + } + for sequence in self.filter.window.speculative() { + if let Some(slot) = state.lookup.get(&sequence) { + let _ = slot.serving().poll_started(waiter); + } + } + } + // Registered first, so a start landing since the last sync is either seen here or + // wakes the read. + self.sync() } /// The lowest group this cursor reads: its own floor, raised to its window's. @@ -5434,6 +5591,9 @@ impl Cursor { return Poll::Ready(Ok(Some(self.with_expiry(peeked.group)))); } self.sync(); + if let Some(err) = &self.filter.refused { + return Poll::Ready(Err(err.clone())); + } self.admit(); // One scan for the whole poll, so walking a backlog off stays linear in its size. @@ -5465,9 +5625,14 @@ impl Cursor { } None => { let floor = self.floor(); - let Some((producer, found_index)) = - ready!(self.poll(waiter, |state| state.poll_recv_group(self.index, floor))?) - else { + let polled = self.poll(waiter, |state| state.poll_recv_group(self.index, floor)); + // A moved window moved the servable edge `drift` was judged against. + if polled.is_pending() && self.watch_time(waiter) { + self.admit(); + drift = ready!(self.poll_drift(&self.servable, waiter))?; + continue; + } + let Some((producer, found_index)) = ready!(polled)? else { return Poll::Ready(Ok(None)); }; let consumer = producer.consume(); @@ -5478,8 +5643,8 @@ impl Cursor { } // The first group to arrive resolves a relative floor, which may land - // above it. - if self.filter.window.is_pending() { + // above it, and a later one may settle where a time end falls. + if !self.filter.window.is_settled() { self.sync(); self.admit(); // The snapshot saw the pending window, which serves everything. @@ -5550,6 +5715,11 @@ impl Cursor { /// group past the cap, nothing holds it for a later raise. fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll>> { loop { + // Refuse before reading, so a refused subscription consumes no datagram. + self.sync(); + if let Some(err) = &self.filter.refused { + return Poll::Ready(Err(err.clone())); + } let Some((datagram, found_index)) = ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?) else { @@ -5563,7 +5733,10 @@ impl Cursor { start.group == sequence && start.frame > datagram.wire.as_ref().and_then(|wire| wire.object_id).unwrap_or(0) }); - if sequence >= self.floor() && self.admits(sequence) && !mid_group { + // A datagram is a whole group, so its own timestamp places it against a time end. + let admits = super::subscription::before_end(sequence, self.end_sequence) + && self.filter.window.admits_datagram(sequence, datagram.timestamp); + if sequence >= self.floor() && admits && !mid_group { return Poll::Ready(Ok(Some(datagram))); } } @@ -5579,6 +5752,9 @@ impl Cursor { return Poll::Ready(Ok(None)); }; self.sync(); + if let Some(err) = &self.filter.refused { + return Poll::Ready(Err(err.clone())); + } let floor = next.max(self.floor()); // One scan for the whole poll, so walking a backlog off stays linear in its size. let drift = ready!(self.poll_drift(&self.servable, waiter))?; @@ -5597,8 +5773,13 @@ impl Cursor { let end = self.end_sequence; let group = loop { let window = &self.filter.window; - let Some(producer) = ready!(self.poll(waiter, |state| state.poll_next_in_range(floor, end, window))?) - else { + let polled = self.poll(waiter, |state| state.poll_next_in_range(floor, end, window)); + // A moved window moved the servable edge `drift` was judged against. + if polled.is_pending() && self.watch_time(waiter) { + drift = ready!(self.poll_drift(&self.servable, waiter))?; + continue; + } + let Some(producer) = ready!(polled)? else { // Deliberately no flush of `seek_pending` here: only a delivery commit // may count a conviction. This `None` can be an artifact of a floor // that will lower again. A conviction never committed is dropped @@ -5607,8 +5788,9 @@ impl Cursor { }; let group = producer.consume(); - // The first group to arrive resolves a relative floor, which may land above it. - if self.filter.window.is_pending() { + // The first group to arrive resolves a relative floor, which may land above it, and + // a later one may settle where a time end falls. + if !self.filter.window.is_settled() { self.sync(); // The snapshot saw the pending window, which serves everything. if !self.filter.window.is_pending() { @@ -5687,6 +5869,8 @@ pub struct Control { /// group it handed out is judged against them before the cursor next reads. `None` /// for a reader that follows a route, whose copies take them on its next read. republish: Option, + // Whether the track is timed, so a time anchor on an untimed one is refused. + timed: bool, } /// What a [`Control`] needs to republish a plain cursor's [`Published`] window. @@ -5710,7 +5894,12 @@ impl Republish { if published.read().revision >= revision { return; } - let window = Window::new(&ranges, self.state.read().newest_produced); + let state = self.state.read(); + // The update was refused before it got here if the track can't resolve it. + let Ok(window) = Window::new(&ranges, state.newest_produced, &*state) else { + return; + }; + drop(state); let runs = Arc::new(admitted(&window, 0, None)); if let Ok(mut current) = published.write() && current.revision < revision @@ -5729,8 +5918,12 @@ impl Control { /// Replace this subscriber's preferences, updating the producer's aggregate. /// /// Returns [`Error::Closed`] if the track already ended; the update is - /// meaningless at that point and can usually be ignored. + /// meaningless at that point and can usually be ignored. Refuses a time anchor on + /// an untimed track with [`Error::TimestampMismatch`], leaving the preferences as they were. pub fn update(&self, subscription: Subscription) -> Result<()> { + if !self.timed && subscription.has_time() { + return Err(Error::TimestampMismatch); + } self.subscription.update(subscription)?; if let Some(republish) = &self.republish { republish.publish(&self.subscription); @@ -5739,6 +5932,14 @@ impl Control { } } +/// Refuse `subscription` when it asks an untimed track for a presentation time. +fn refuse_untimed(info: &Info, subscription: &Subscription) -> Result<()> { + match info.timescale.is_none() && subscription.has_time() { + true => Err(Error::TimestampMismatch), + false => Ok(()), + } +} + impl Subscriber { /// The track's [`Info`], resolved when the subscription was established. /// @@ -5778,6 +5979,7 @@ impl Subscriber { /// Create a handle for updating this subscriber's delivery preferences. pub fn control(&self) -> Control { + let timed = self.info.timescale.is_some(); match &self.inner { Inner::Plain(cursor) => Control { subscription: cursor.subscription.clone(), @@ -5785,10 +5987,12 @@ impl Subscriber { state: cursor.state.weak(), published: cursor.published.weak(), }), + timed, }, Inner::Resume(resume, _) => Control { subscription: resume.subscription().clone(), republish: None, + timed, }, } } @@ -6106,7 +6310,9 @@ impl Subscriber { match &self.inner { Inner::Plain(cursor) => { let state = cursor.state.read(); - Window::new(ranges, state.newest_produced).resolved() + Window::new(ranges, state.newest_produced, &*state) + .map(|window| window.resolved()) + .unwrap_or_default() } Inner::Resume(..) => Vec::new(), } @@ -6266,8 +6472,10 @@ impl Subscriber { /// the cached groups they newly cover, even ones that arrived before the groups /// already read. Returns [`Error::Closed`] /// if the track already ended; the update is meaningless at that point and can usually - /// be ignored. + /// be ignored. Refuses a time anchor on an untimed track with + /// [`Error::TimestampMismatch`], leaving the preferences as they were. pub fn update(&mut self, subscription: Subscription) -> Result<()> { + refuse_untimed(&self.info, &subscription)?; let channel = match &self.inner { Inner::Plain(cursor) => &cursor.subscription, Inner::Resume(resume, _) => resume.subscription(), @@ -9989,6 +10197,37 @@ mod test { assert_eq!(group.sequence, 3); } + /// A subscriber read by either cursor, so one test covers both. + #[cfg(feature = "async")] + enum Reader { + Arrival(Subscriber), + Ordered(Ordered), + } + + #[cfg(feature = "async")] + impl Reader { + fn new(subscriber: Subscriber, ordered: bool) -> Self { + match ordered { + true => Self::Ordered(subscriber.ordered()), + false => Self::Arrival(subscriber), + } + } + + fn drain(&mut self) -> Vec { + match self { + Self::Arrival(subscriber) => drain(subscriber), + Self::Ordered(ordered) => drain_ordered(ordered), + } + } + + fn resolved(&mut self) -> Vec> { + match self { + Self::Arrival(subscriber) => subscriber.resolved_ranges(), + Self::Ordered(ordered) => ordered.inner.resolved_ranges(), + } + } + } + /// Every group the ordered cursor can read right now, in sequence order. #[cfg(feature = "async")] fn drain_ordered(subscriber: &mut Ordered) -> Vec { @@ -15183,4 +15422,299 @@ mod test { } assert_eq!(dropped, vec![2..4, 6..8]); } + + /// A time start is the cached group containing it and a time end stops just past that + /// group; one past every start held stays open until a group starts after it. + #[test] + #[cfg(feature = "async")] + fn time_anchors_resolve_against_the_cache() { + let mut producer = track_producer("test", Info::timed()); + for second in 0..4 { + append_at(&mut producer, second * 1000); + } + let time = |millis| Anchor::Time(Timestamp::from_millis(millis).unwrap()); + + let mut seek = producer.subscribe(replay().with_ranges([Range { + start: time(1500), + end: Some(time(2000)), + }])); + assert_eq!(drain(&mut seek), vec![1, 2]); + + // A start before every group reaches back to the oldest one. + let mut early = producer.subscribe(replay().with_ranges([Range { + start: Anchor::Time(Timestamp::from_micros(0).unwrap()), + end: Some(time(500)), + }])); + assert_eq!(drain(&mut early), vec![0]); + + // Past the newest start: open until a group is known to start after the end. + let mut live = producer.subscribe(replay().with_ranges([Range { + start: time(3000), + end: Some(time(4500)), + }])); + assert_eq!(drain(&mut live), vec![3]); + append_at(&mut producer, 4000); + append_at(&mut producer, 5000); + append_at(&mut producer, 6000); + assert_eq!(drain(&mut live), vec![4]); + } + + /// A group joined mid-group stamps a later frame than its start, so it never proves a + /// time end falls before it. + #[test] + #[cfg(feature = "async")] + fn a_partial_group_proves_no_start() { + let mut producer = track_producer("test", Info::timed()); + append_at(&mut producer, 0); + append_at(&mut producer, 1000); + let mut partial = producer.append_group().unwrap(); + partial.start_at(1).unwrap(); + partial + .write_frame(Timestamp::from_millis(2500).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + partial.finish().unwrap(); + + let time = |millis| Anchor::Time(Timestamp::from_millis(millis).unwrap()); + let mut seek = producer.subscribe(replay().with_ranges([Range { + start: time(1500), + end: Some(time(2200)), + }])); + assert_eq!(drain(&mut seek), vec![1, 2]); + } + + /// A time end stays open across a group that has yet to start: it may start before the + /// end, so a later group starting after it settles nothing. + #[test] + #[cfg(feature = "async")] + fn a_time_end_waits_on_an_unstarted_group_below_it() { + let producer = track_producer("test", Info::timed()); + let at = |millis| Timestamp::from_millis(millis).unwrap(); + let mut zero = producer.create_group(group::Info { sequence: 0 }).unwrap(); + zero.write_frame(at(0), bytes::Bytes::from_static(b"x")).unwrap(); + let mut hole = producer.create_group(group::Info { sequence: 1 }).unwrap(); + let mut two = producer.create_group(group::Info { sequence: 2 }).unwrap(); + two.write_frame(at(2000), bytes::Bytes::from_static(b"x")).unwrap(); + + let mut subscriber = producer.subscribe(replay().with_ranges([Range { + start: Anchor::Absolute(Position::group(0)), + end: Some(Anchor::Time(at(500))), + }])); + hole.write_frame(at(1000), bytes::Bytes::from_static(b"x")).unwrap(); + assert_eq!(drain(&mut subscriber), vec![0]); + } + + /// A time end stays open across a sequence yet to arrive: it may start before the end, + /// so a later group starting after it settles nothing. Once that sequence is reported + /// dropped, nothing below the later group can start before the end, so it settles. + #[test] + #[cfg(feature = "async")] + fn a_time_end_waits_on_an_absent_group_below_it() { + let at = |millis| Timestamp::from_millis(millis).unwrap(); + let ranges = [Range { + start: Anchor::Absolute(Position::group(0)), + end: Some(Anchor::Time(at(500))), + }]; + for dropped in [false, true] { + let producer = track_producer("test", Info::timed()); + let mut zero = producer.create_group(group::Info { sequence: 0 }).unwrap(); + zero.write_frame(at(0), bytes::Bytes::from_static(b"x")).unwrap(); + let mut two = producer.create_group(group::Info { sequence: 2 }).unwrap(); + two.write_frame(at(2000), bytes::Bytes::from_static(b"x")).unwrap(); + + let mut subscriber = producer.subscribe(replay().with_ranges(ranges)); + assert_eq!( + subscriber.resolved_ranges(), + vec![Some(Resolved { start: 0, end: None })], + "group 1 may still start before the end" + ); + if dropped { + producer.drop_groups(1..2); + assert_eq!(drain(&mut subscriber), vec![0]); + assert_eq!( + subscriber.resolved_ranges(), + vec![Some(Resolved { start: 0, end: Some(2) })], + "the drop settles the end" + ); + continue; + } + let mut one = producer.create_group(group::Info { sequence: 1 }).unwrap(); + one.write_frame(at(1000), bytes::Bytes::from_static(b"x")).unwrap(); + assert_eq!(drain(&mut subscriber), vec![0]); + } + } + + /// A time end at the live edge reads at most one group past it: groups opened before + /// their first frame wait on the one below them starting, rather than every one being + /// handed out while the end is unsettled. + #[test] + #[cfg(feature = "async")] + fn a_time_end_reads_one_group_past_it_at_the_live_edge() { + let at = |millis| Timestamp::from_millis(millis).unwrap(); + let ranges = [Range { + start: Anchor::Time(at(3000)), + end: Some(Anchor::Time(at(3500))), + }]; + for ordered in [false, true] { + let mut producer = track_producer("test", Info::timed()); + append_at(&mut producer, 3000); + let mut groups: Vec<_> = (1..4) + .map(|sequence| producer.create_group(group::Info { sequence }).unwrap()) + .collect(); + let subscriber = producer.subscribe(replay().with_ranges(ranges)); + let mut reader = Reader::new(subscriber, ordered); + assert_eq!(reader.drain(), vec![0, 1], "1 is read on speculation, 2 and 3 wait"); + groups[0] + .write_frame(at(3600), bytes::Bytes::from_static(b"x")) + .unwrap(); + groups[1] + .write_frame(at(4000), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert_eq!(reader.drain(), Vec::::new(), "ordered: {ordered}"); + assert_eq!( + reader.resolved(), + vec![Some(Resolved { start: 0, end: Some(1) })], + "1 starting past the end settles it" + ); + } + } + + /// A read parked on a time end's cap wakes when the speculative group starts, and when + /// a later group starts inside the end while the speculative one stalls. + #[moq_clock::test] + #[cfg(feature = "async")] + async fn a_read_parked_on_a_time_end_wakes_as_groups_start() { + let at = |millis| Timestamp::from_millis(millis).unwrap(); + let producer = track_producer("test", Info::timed()); + let mut groups: Vec<_> = (0..4) + .map(|sequence| producer.create_group(group::Info { sequence }).unwrap()) + .collect(); + let mut subscriber = producer.subscribe(replay().with_ranges([Range { + start: Anchor::Absolute(Position::group(0)), + end: Some(Anchor::Time(at(3500))), + }])); + assert_eq!(drain(&mut subscriber), vec![0]); + + let parked = moq_clock::sim::spawn(async move { + let one = subscriber.recv_group().await.unwrap().unwrap().sequence; + let two = subscriber.recv_group().await.unwrap().unwrap().sequence; + (one, two, subscriber) + }); + moq_clock::sim::yield_now().await; + assert!(!parked.is_finished(), "1 waits on 0 starting"); + groups[0] + .write_frame(at(3000), bytes::Bytes::from_static(b"x")) + .unwrap(); + moq_clock::sim::yield_now().await; + assert!(!parked.is_finished(), "1 is read, and 2 waits on 1 starting"); + + // 1 stalls, so only 2 starting inside the end lifts the cap past it. + groups[2] + .write_frame(at(3200), bytes::Bytes::from_static(b"x")) + .unwrap(); + let (one, two, mut subscriber) = parked.await.unwrap(); + assert_eq!((one, two), (1, 2)); + assert_eq!(drain(&mut subscriber), vec![3], "3 is the speculative one now"); + groups[3] + .write_frame(at(3600), bytes::Bytes::from_static(b"x")) + .unwrap(); + groups[1] + .write_frame(at(3100), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert_eq!( + subscriber.resolved_ranges(), + vec![Some(Resolved { start: 0, end: Some(3) })] + ); + } + + /// A datagram is a whole group of one frame, so its own timestamp says whether it falls + /// inside a time end, without one ever settling from the groups. + #[test] + #[cfg(feature = "async")] + fn a_datagram_is_judged_against_a_time_end_by_its_timestamp() { + let at = |millis| Timestamp::from_millis(millis).unwrap(); + let mut producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(replay().with_ranges([Range { + start: Anchor::Absolute(Position::group(0)), + end: Some(Anchor::Time(at(3500))), + }])); + for millis in [3000, 3400, 3600, 3700] { + producer.append_datagram(at(millis), &b"x"[..]).unwrap(); + } + let mut read = Vec::new(); + while let Some(Ok(Some(datagram))) = subscriber.recv_datagram().now_or_never() { + read.push(datagram.sequence); + } + assert_eq!(read, vec![0, 1]); + } + + /// A track of only datagrams starts no group, so a time start is judged against each + /// datagram's own timestamp rather than letting every one through. + #[test] + #[cfg(feature = "async")] + fn a_datagram_is_judged_against_a_time_start_by_its_timestamp() { + let at = |millis| Timestamp::from_millis(millis).unwrap(); + let mut producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(replay().with_ranges([Range { + start: Anchor::Time(at(3500)), + end: None, + }])); + for millis in [3000, 3400, 3500, 3700] { + producer.append_datagram(at(millis), &b"x"[..]).unwrap(); + } + let mut read = Vec::new(); + while let Some(Ok(Some(datagram))) = subscriber.recv_datagram().now_or_never() { + read.push(datagram.sequence); + } + assert_eq!(read, vec![2, 3]); + } + + /// An untimed track refuses a time anchor wherever the ranges reach it. + #[test] + #[cfg(feature = "async")] + fn an_untimed_track_refuses_time_anchors() { + let mut producer = untimed_producer(); + let timed = replay().with_ranges([Range { + start: Anchor::Time(Timestamp::from_millis(0).unwrap()), + end: None, + }]); + + let pending = producer.consume().subscribe(timed.clone()); + assert!(matches!( + pending.poll_ok(&kio::Waiter::noop()), + Poll::Ready(Err(Error::TimestampMismatch)) + )); + + let mut subscriber = producer.subscribe(replay()); + assert!(matches!( + subscriber.update(timed.clone()), + Err(Error::TimestampMismatch) + )); + assert!(matches!( + subscriber.control().update(timed.clone()), + Err(Error::TimestampMismatch) + )); + assert_eq!(subscriber.subscription(), replay(), "a refused update changes nothing"); + + let mut refused = producer.subscribe(timed); + assert!(matches!( + refused.recv_group().now_or_never(), + Some(Err(Error::TimestampMismatch)) + )); + producer.append_datagram(None, &b"x"[..]).unwrap(); + assert!(matches!( + refused.recv_datagram().now_or_never(), + Some(Err(Error::TimestampMismatch)) + )); + + // Ranges it can serve lift the refusal. + refused.update(replay()).unwrap(); + let mut group = producer.append_group().unwrap(); + group.finish().unwrap(); + let read = refused.recv_group().now_or_never().expect("the group is cached"); + assert_eq!( + read.unwrap().map(|group| group.sequence), + Some(1), + "the datagram took sequence 0" + ); + } } diff --git a/rs/moq-net/src/model/track/demand/ranges.rs b/rs/moq-net/src/model/track/demand/ranges.rs index 120723920d5..88bae86d296 100644 --- a/rs/moq-net/src/model/track/demand/ranges.rs +++ b/rs/moq-net/src/model/track/demand/ranges.rs @@ -4,7 +4,8 @@ //! where it ends, so a position is covered while the running sum stays above zero. A treap //! keeps each subtree's sum and lowest running sum, so a change touches two boundaries and //! asks one range-minimum question, whatever else the audience holds. Ranges anchored to -//! the latest group can't be placed on that line and are counted by value instead. +//! the latest group or to a time can't be placed on that line and are counted by value +//! instead. use std::cell::OnceCell; use std::collections::BTreeMap; @@ -139,6 +140,8 @@ impl Ranges { match range.start { Anchor::Absolute(start) => self.absolute.tail().is_some_and(|tail| tail <= start), Anchor::Relative(back) => self.farthest().is_some_and(|farthest| back <= farthest), + // Where a time falls among the groups is the serving hop's to say. + Anchor::Time(_) => false, } } } diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 1e51e61d5e4..2f6ffc91e7d 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Anchors can name a time (#6074) + +`Subscription.Anchor` gains `.time(timestampUs:)`, so exhaustive switches need a case for it. +`TrackConsumer.update(subscription:)` now `throws`, on a time anchor an untimed track can't serve; call it with `try`. + ### JSON consumers yield `Json.Timed` (#5677) `Json.SnapshotConsumer` and `Json.StreamConsumer` yield `Json.Timed` (`value`, `timestampUs`) instead of the bare value; `timestampUs` is `nil` when untimed. diff --git a/swift/Sources/Moq/Aliases.swift b/swift/Sources/Moq/Aliases.swift index 97c8cce74e2..9b9fd71b54c 100644 --- a/swift/Sources/Moq/Aliases.swift +++ b/swift/Sources/Moq/Aliases.swift @@ -115,8 +115,8 @@ public typealias ProtocolError = MoqFFI.MoqProtocolError extension Subscription { /// A span of a track to deliver: from `start` up to `end`, or onward when `end` is `nil`. public typealias Range = MoqFFI.MoqRange - /// Where one end of a `Range` falls: `.absolute(group:frame:)` or - /// `.relative(groups:)` back from the latest group. + /// Where one end of a `Range` falls: `.absolute(group:frame:)`, `.relative(groups:)` + /// back from the latest group, or `.time(timestampUs:)` for the group containing it. public typealias Anchor = MoqFFI.MoqAnchor /// Which pending group the publisher sends first: `.ascending` or `.descending`. public typealias Order = MoqFFI.MoqOrder diff --git a/swift/Sources/Moq/Track.swift b/swift/Sources/Moq/Track.swift index c4566ba3286..efd34bfe137 100644 --- a/swift/Sources/Moq/Track.swift +++ b/swift/Sources/Moq/Track.swift @@ -44,9 +44,10 @@ public final class TrackConsumer: AsyncSequence, Sendable { try ffi.info() } - /// Change this subscriber's delivery preferences. - public func update(subscription: Subscription) { - ffi.update(subscription: subscription) + /// Change this subscriber's delivery preferences. Throws on a time anchor the track + /// can't serve. + public func update(subscription: Subscription) throws { + try ffi.update(subscription: subscription) } /// Cancel all current and future reads.