moq_net/model/

subscription.rs

use std::{
	ops::{Bound, RangeBounds, RangeFull, RangeTo, RangeToInclusive},
	task::Poll,
	time::Duration,
};

/// Subscriber-side preferences for receiving a track.
///
/// Each subscriber holds its own [`Subscription`]; the publisher observes an
/// aggregate across all live subscribers via [`crate::track::Producer::subscription`].
/// A subscriber can change its preferences after the fact with
/// [`crate::track::Subscriber::update`].
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct Subscription {
	/// Delivery priority. Higher values preempt lower ones when bandwidth is constrained.
	pub priority: u8,
	/// How old a group may get before this subscriber gives up on it.
	///
	/// [`Duration::ZERO`] (the default) skips immediately: group 8 arriving means group 7
	/// is abandoned. A larger budget tolerates that much reordering before giving up.
	/// This never *adds* delay, since the bound is only reached once newer data is
	/// already that far ahead.
	///
	/// This is the `Subscriber Max Age` on the wire, and it is stored here verbatim so
	/// what was asked for stays readable. Encoded as milliseconds in a QUIC varint, so
	/// a duration of `2^62` milliseconds or more cannot be put on the wire. Clamped to
	/// the publisher's [`Info::max_age`](crate::track::Info::max_age), since waiting for
	/// a group longer than it is kept around cannot produce it.
	///
	/// # Where it is enforced
	///
	/// At both ends, and neither alone is enough. The publisher skips a group that has
	/// aged out instead of putting it on the wire, which bounds a backlog before it
	/// costs bandwidth. But what reaches the publisher is the aggregate across every
	/// subscriber, resolved in favor of the most tolerant one, so that gate is only ever
	/// as tight as the most patient viewer. The subscriber applies the same budget again
	/// as it reads, where its own is the only one in play.
	///
	/// This bounds a *subscription*.
	/// [`track::Consumer::fetch_group`](crate::track::Consumer::fetch_group) is exempt:
	/// it names one old group explicitly, so there is no live edge to be late against.
	///
	/// # How age is measured
	///
	/// In presentation time only. A group is measured by its *reach*, where its immediate
	/// successor begins, against the newest frame of the latest group: it cannot present
	/// past its successor, so once everything it could still hold falls outside the budget
	/// it is provably useless. The candidate needs no timestamp of its own, so an empty or
	/// stalled group is bounded by its stamped successor the same way. Wall-clock
	/// reclamation of idle content is the cache's own policy, not this budget's.
	///
	/// Protocols whose wire can't carry a timestamp (pre-Lite05 moq-lite, moq-transport
	/// without the Timestamp property) have their frames stamped on receipt, which makes
	/// the measure burst-blind on the receiving side: thirty seconds of backlog delivered
	/// in three reads as three. The publisher's copy is stamped as it produces, so the
	/// gate there still holds; it is just the coarser of the two.
	pub max_age: Duration,
	/// The lowest [`Position`] the publisher may deliver, or `None` for no floor.
	///
	/// A floor, not a request: only [`Self::max_age`] asks for data, and the floor bounds
	/// how far back it may reach. `None` and a floor of group 0 mean the same thing, since
	/// nothing sits below group 0. Delivery starts at the oldest group at or above the
	/// floor that the budget still considers fresh, so a floor above the live edge simply
	/// waits there (a resumed subscription naming where it left off).
	///
	/// Aggregated across every live subscriber (the loosest floor wins, and any subscriber
	/// without one clears it), so it says what the publisher sends, not what any one
	/// subscriber sees. [`crate::track::Subscriber::set_groups`] is the local read cursor;
	/// setting one does not imply the other. See [Local cursor vs wire
	/// preference](crate::track::Subscriber#local-cursor-vs-wire-preference).
	pub start: Option<Position>,
	/// First [`Position`] the publisher should *not* deliver, or `None` for no end.
	///
	/// Exclusive, like the end of a [`std::ops::Range`], which is what lets one field
	/// carry both "through the end of group 5" ([`Position::after_group(5)`](Position::after_group))
	/// and "up to frame 2 of group 5" ([`Position::after(5, 2)`](Position::after)). An
	/// inclusive end cannot express the first without a sentinel frame, and the ordering
	/// falls out for free: group 6's head sorts above any frame of group 5, so a
	/// whole-group subscriber correctly absorbs a frame-capped one in the aggregate.
	///
	/// The wire agrees: `Group End` and `Frame End` are both encoded as `absolute + 1`.
	///
	/// A request, aggregated across every live subscriber (any unbounded subscriber makes
	/// the aggregate unbounded). [`crate::track::Subscriber::set_groups`] is the local read
	/// cursor; [`Position::group_end`] translates this field into its bound. Setting one
	/// does not imply the other.
	pub end: Option<Position>,
}

impl Default for Subscription {
	fn default() -> Self {
		Self {
			priority: 0,
			max_age: Duration::ZERO,
			start: None,
			end: None,
		}
	}
}

impl Subscription {
	/// Set the delivery priority, returning `self` for chaining.
	pub fn with_priority(mut self, priority: u8) -> Self {
		self.priority = priority;
		self
	}

	/// Set how old a group may get before it is skipped, returning `self` for chaining.
	pub fn with_max_age(mut self, max_age: Duration) -> Self {
		self.max_age = max_age;
		self
	}

	/// Floor delivery at `start`, or leave it unfloored when `None`. Returns `self` for
	/// chaining.
	///
	/// A floor bounds how far back [`Self::max_age`] may reach; it does not request data
	/// on its own. [`Position::group`] is the whole-group form.
	pub fn with_start(mut self, start: impl Into<Option<Position>>) -> Self {
		self.start = start.into();
		self
	}

	/// Stop delivery at `end`, or leave the subscription unbounded when `None`. Returns
	/// `self` for chaining.
	///
	/// Exclusive, matching [`Self::end`], so pass the position *after* the last one you
	/// want. [`Position::after`] and [`Position::after_group`] name that conversion so no
	/// call site has to write the `+ 1` itself.
	pub fn with_end(mut self, end: impl Into<Option<Position>>) -> Self {
		self.end = end.into();
		self
	}

	/// Request the whole groups in `groups`, replacing both [`Self::start`] and
	/// [`Self::end`]. Returns `self` for chaining.
	///
	/// Any range of group sequences works: `2..=5`, `2..6`, `..6`, `2..`, or `..` to
	/// clear both bounds. An inclusive end past the last group is unbounded, as
	/// [`Position::after_group`] spells it.
	pub fn with_groups(mut self, groups: impl RangeBounds<u64>) -> Self {
		let unbounded_start = matches!(groups.start_bound(), Bound::Unbounded);
		let (start, end) = sequence_bounds(groups);
		self.start = (!unbounded_start).then(|| Position::group(start));
		self.end = end.map(Position::group);
		self
	}

	// Fold this subscription into the running aggregate: Ready with the merged
	// result when it demands more than `combined`, Pending when it's a subset
	// (so callers can skip a redundant broadcast of the same aggregate).
	pub(super) fn poll_combined(&self, combined: &Option<Subscription>) -> Poll<Subscription> {
		let Some(combined) = combined else {
			return Poll::Ready(self.clone());
		};

		let merged = Subscription {
			priority: self.priority.max(combined.priority),
			// Sequence-first prioritization is enabled only when every subscriber wants it.
			max_age: self.max_age.max(combined.max_age),
			// Bounds fold as whole positions. Two subscribers starting in the same group
			// are separated only by their frame, so folding group and frame independently
			// would invent a bound neither asked for.
			start: min_floored(self.start, combined.start),
			end: max_unbounded(self.end, combined.end),
		};

		if &merged != combined {
			return Poll::Ready(merged);
		}

		Poll::Pending
	}
}

/// A frame-precise point in a track: a group sequence and a frame index within it.
///
/// Ordered lexicographically, so comparing positions is the same as comparing groups
/// and only falling back to frames within one. This is the model's counterpart of the
/// wire's (`Group`, `Frame`) pairs on SUBSCRIBE and FETCH.
///
/// Pairing the two is the point: a frame index counts from the start of a group, so it
/// means nothing on its own. Carrying them together makes "frame 5 of nothing"
/// unrepresentable rather than merely undefined.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
pub struct Position {
	/// The group sequence.
	pub group: u64,
	/// The frame index within the group, numbered from 0 in write order.
	pub frame: u64,
}

impl Position {
	/// The first frame of `group`.
	///
	/// As an exclusive end this means "everything before `group`"; as a start it means
	/// "`group` from the beginning".
	pub fn group(group: u64) -> Self {
		Self { group, frame: 0 }
	}

	/// The position just past `frame` of `group`: the exclusive end that includes it.
	///
	/// `None` when there is no such position, i.e. the very last frame of the very last
	/// group. That is past everything, which [`Subscription::end`] spells `None` too, so
	/// the two meanings line up and `with_end` can take this directly.
	pub fn after(group: u64, frame: u64) -> Option<Self> {
		match frame.checked_add(1) {
			Some(frame) => Some(Self { group, frame }),
			// Past the last frame of a group is the head of the next one.
			None => Self::after_group(group),
		}
	}

	/// The position just past every frame of `group`: the exclusive end that includes
	/// the group whole.
	///
	/// `None` past the last group, for the reason given on [`Self::after`].
	pub fn after_group(group: u64) -> Option<Self> {
		Some(Self::group(group.checked_add(1)?))
	}

	/// The bound this exclusive end puts on a group cursor, for
	/// [`crate::track::Subscriber::set_groups`].
	///
	/// A head-of-group end excludes its group. A mid-group end includes it, so a frame
	/// cap can apply within that group.
	pub fn group_end(self) -> Bound<u64> {
		if self.frame == 0 {
			Bound::Excluded(self.group)
		} else {
			Bound::Included(self.group)
		}
	}
}

/// Where a read cursor stops: a group sequence or frame index it will not deliver.
///
/// Built from a range so the call site says whether its bound is delivered: `..5` stops
/// before 5, `..=5` reads through it, and `..` removes the cap. A [`Bound`] converts
/// too, for callers holding one (a decoded wire field, or [`Position::group_end`]).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct Cap(Option<u64>);

impl Cap {
	/// The first index withheld, or `None` for no cap. Cursors store this form and
	/// compare with [`before_end`]. An inclusive bound at `u64::MAX` has nothing above
	/// it, so it is no cap at all.
	pub(crate) fn exclusive(self) -> Option<u64> {
		self.0
	}
}

impl From<Bound<u64>> for Cap {
	fn from(bound: Bound<u64>) -> Self {
		Self(match bound {
			Bound::Included(index) => index.checked_add(1),
			Bound::Excluded(index) => Some(index),
			Bound::Unbounded => None,
		})
	}
}

impl From<RangeTo<u64>> for Cap {
	fn from(range: RangeTo<u64>) -> Self {
		Self(Some(range.end))
	}
}

impl From<RangeToInclusive<u64>> for Cap {
	fn from(range: RangeToInclusive<u64>) -> Self {
		Bound::Included(range.end).into()
	}
}

impl From<RangeFull> for Cap {
	fn from(_: RangeFull) -> Self {
		Self(None)
	}
}

// Normalize discrete ranges once, including the empty range above the last index.
pub(super) fn sequence_bounds(range: impl RangeBounds<u64>) -> (u64, Option<u64>) {
	let start = match range.start_bound() {
		Bound::Included(&start) => start,
		Bound::Excluded(&start) => match start.checked_add(1) {
			Some(start) => start,
			None => return (u64::MAX, Some(u64::MAX)),
		},
		Bound::Unbounded => 0,
	};
	(start, Cap::from(range.end_bound().cloned()).exclusive())
}

/// Whether `sequence` is strictly below an exclusive cap. `None` is unbounded.
pub(super) fn before_end(sequence: u64, end: Option<u64>) -> bool {
	end.is_none_or(|end| sequence < end)
}

// Combining two optional bounds comes in two families, and they disagree on what `None`
// means. `_some` treats it as the neutral element (the other side wins), for intersecting
// two ranges that each restrict independently. `_floored` / `_unbounded` treat it as
// absorbing (the result is `None` too), for aggregating across subscribers, where one
// subscriber asking for everything makes the aggregate everything. Picking the wrong
// family silently narrows or widens what the publisher sends, so the suffix, not the
// `min`/`max`, is the part to read.

/// The lower of two optional bounds, `None` neutral. Pairs with [`max_some`].
pub(super) fn min_some<T: Ord>(a: Option<T>, b: Option<T>) -> Option<T> {
	match (a, b) {
		(Some(a), Some(b)) => Some(a.min(b)),
		(Some(a), None) | (None, Some(a)) => Some(a),
		(None, None) => None,
	}
}

/// The higher of two optional bounds, `None` neutral. Pairs with [`min_some`].
pub(super) fn max_some<T: Ord>(a: Option<T>, b: Option<T>) -> Option<T> {
	match (a, b) {
		(Some(a), Some(b)) => Some(a.max(b)),
		(Some(a), None) | (None, Some(a)) => Some(a),
		(None, None) => None,
	}
}

/// The lower of two optional floors, `None` absorbing (no floor). The mirror of
/// [`max_unbounded`]: both bounds only ever *restrict*, so a subscriber without one keeps
/// the aggregate unrestricted.
pub(super) fn min_floored<T: Ord>(a: Option<T>, b: Option<T>) -> Option<T> {
	match (a, b) {
		(Some(a), Some(b)) => Some(a.min(b)),
		(None, _) | (_, None) => None,
	}
}

/// The higher of two optional bounds, `None` absorbing (unbounded).
pub(super) fn max_unbounded<T: Ord>(a: Option<T>, b: Option<T>) -> Option<T> {
	match (a, b) {
		(Some(a), Some(b)) => Some(a.max(b)),
		(None, _) | (_, None) => None,
	}
}

#[cfg(test)]
mod tests {
	use super::*;

	fn combine(subscriptions: &[Subscription]) -> Option<Subscription> {
		let mut combined = None;
		for sub in subscriptions {
			if let Poll::Ready(merged) = sub.poll_combined(&combined) {
				combined = Some(merged);
			}
		}
		combined
	}

	/// The exclusive representation runs out at both extremes, and `Option` says so
	/// rather than saturating into a bound that contradicts the request.
	/// A group range spells both positions at once, in whichever form the caller has.
	#[test]
	fn group_ranges_build_whole_group_positions() {
		let sub = Subscription::default().with_groups(2..=5);
		assert_eq!(sub.start, Some(Position::group(2)));
		assert_eq!(sub.end, Some(Position::group(6)));

		let sub = Subscription::default().with_groups(2..6);
		assert_eq!(sub.end, Some(Position::group(6)));

		let sub = Subscription::default().with_groups(..6);
		assert_eq!(sub.start, None);
		assert_eq!(sub.end, Some(Position::group(6)));

		let sub = Subscription::default().with_groups(2..);
		assert_eq!(sub.start, Some(Position::group(2)));
		assert_eq!(sub.end, None);

		// Through the last group is unbounded, as `after_group` spells it.
		let sub = Subscription::default().with_groups(..=u64::MAX);
		assert_eq!(sub.end, None);

		let sub = Subscription::default().with_groups(2..=5).with_groups(..);
		assert_eq!((sub.start, sub.end), (None, None));
	}

	#[test]
	fn positions_are_total_at_the_extremes() {
		// Past the last frame of a group is the head of the next one, not a wider frame
		// in the same group.
		assert_eq!(Position::after(5, u64::MAX), Some(Position::group(6)));

		// Past everything has no position. `Subscription::end` spells that `None` too,
		// so the meanings coincide and the group is included rather than dropped.
		assert_eq!(Position::after_group(u64::MAX), None);
		assert_eq!(Position::after(u64::MAX, u64::MAX), None);
		assert_eq!(
			Subscription::default().with_end(Position::after_group(u64::MAX)).end,
			None
		);

		// A head-of-group end excludes that group; a mid-group end includes it.
		assert_eq!(Position::group(0).group_end(), Bound::Excluded(0));
		assert_eq!(Position::after_group(5).unwrap().group_end(), Bound::Excluded(6));
		assert_eq!(Position::after(5, 2).unwrap().group_end(), Bound::Included(5));

		// A cursor cap is the first index it withholds; an inclusive bound at the last
		// index withholds nothing.
		assert_eq!(Cap::from(..0).exclusive(), Some(0));
		assert_eq!(Cap::from(..=5).exclusive(), Some(6));
		assert_eq!(Cap::from(..=u64::MAX).exclusive(), None);
		assert_eq!(Cap::from(..).exclusive(), None);
		assert_eq!(Cap::from(Bound::Included(5)), Cap::from(..6));
		assert_eq!(Cap::from(Bound::Unbounded), Cap::from(..));
	}

	#[test]
	fn combined_group_start_keeps_the_loosest_floor() {
		// A floor only restricts, so the lowest one wins across floored subscribers.
		let catchup = Subscription::default().with_start(Position::group(10));
		let older_catchup = Subscription::default().with_start(Position::group(5));
		let combined = combine(&[catchup.clone(), older_catchup]).unwrap();
		assert_eq!(combined.start, Some(Position::group(5)));

		// A subscriber with no floor at all clears the aggregate: its budget may reach
		// below any floor the others set.
		let unfloored = Subscription::default();
		let combined = combine(&[catchup, unfloored]).unwrap();
		assert_eq!(combined.start, None);
	}

	#[test]
	fn combined_group_end_keeps_live_subscription_unbounded() {
		// No end at all is unbounded, which absorbs any explicit one.
		let live = Subscription::default();
		let bounded = Subscription::default().with_end(Position::after_group(10));

		let combined = combine(&[live, bounded]).unwrap();

		assert_eq!(combined.end, None);
	}

	#[test]
	fn combined_start_folds_the_whole_position() {
		let early_frame = Subscription::default().with_start(Position { group: 5, frame: 2 });
		let late_frame = Subscription::default().with_start(Position { group: 5, frame: 9 });

		// Same group: the earlier frame wins.
		let combined = combine(&[late_frame.clone(), early_frame.clone()]).unwrap();
		assert_eq!(combined.start, Some(Position { group: 5, frame: 2 }));

		// An earlier group wins outright, carrying its own frame rather than the
		// smallest frame across the two.
		let earlier_group = Subscription::default().with_start(Position { group: 4, frame: 7 });
		let combined = combine(&[early_frame, earlier_group]).unwrap();
		assert_eq!(combined.start, Some(Position { group: 4, frame: 7 }));
	}

	#[test]
	fn combined_end_folds_the_whole_position() {
		let short = Subscription::default().with_end(Position::after(5, 2));
		let long = Subscription::default().with_end(Position::after(5, 9));

		// Same group: the later frame wins. Ends are exclusive, so an inclusive frame 9
		// is stored as 10.
		let combined = combine(&[short.clone(), long.clone()]).unwrap();
		assert_eq!(combined.end, Some(Position { group: 5, frame: 10 }));

		// The whole group is the head of the next one, which outsorts every frame of
		// this one, so it absorbs any capped end without a sentinel.
		let whole = Subscription::default().with_end(Position::after_group(5));
		let combined = combine(&[long, whole]).unwrap();
		assert_eq!(combined.end, Some(Position::group(6)));

		// A later group wins outright, carrying its own frame.
		let later_group = Subscription::default().with_end(Position::after(6, 1));
		let combined = combine(&[short, later_group]).unwrap();
		assert_eq!(combined.end, Some(Position { group: 6, frame: 2 }));
	}

	#[test]
	fn combined_group_end_uses_latest_bounded_end() {
		let early = Subscription::default().with_end(Position::after_group(10));
		let late = Subscription::default().with_end(Position::after_group(20));

		let combined = combine(&[early, late]).unwrap();

		assert_eq!(combined.end, Some(Position::group(21)));
	}
}