moq_net/model/

track.rs

//! A track is a collection of semi-reliable and semi-ordered streams, split into a [Producer] and [Subscriber] handle.
//!
//! A [Producer] creates streams with a sequence number and priority.
//! The sequence number is used to determine the order of streams, while the priority is used to determine which stream to transmit first.
//! This may seem counter-intuitive, but is designed for live streaming where the newest streams may be higher priority.
//! A cloned [Producer] can be used to create streams in parallel, but will error if a duplicate sequence number is used.
//!
//! A [Subscriber] may not receive all streams in order or at all.
//! These streams are meant to be transmitted over congested networks and the key to MoQ Transport is to not block on them.
//! Streams will be cached for a potentially limited duration added to the unreliable nature.
//! A [Consumer] is a cheap, cloneable handle; subscribing it multiple times fans the same
//! cached streams out to each independent [Subscriber].
//!
//! The track is closed with [Error] when all writers or readers are dropped.

use crate::{Error, Result, Timescale, Timestamp, coding};
use crate::{broadcast, cache, group, stats};

use super::{Datagram, Requests};

use super::Cap;
pub use super::subscription::{Position, Subscription};

use std::{
	collections::{BTreeMap, HashSet, VecDeque},
	ops::{Bound, RangeBounds},
	sync::Arc,
	sync::OnceLock,
	sync::atomic::{AtomicBool, Ordering},
	task::{Poll, ready},
	time::Duration,
};

/// Default [`Info::max_age`] when the publisher doesn't set one.
pub const DEFAULT_MAX_AGE: Duration = Duration::from_secs(5);

/// Maximum number of datagrams retained in the per-track send buffer.
///
/// Datagrams are a best-effort send buffer, not a replay cache (unlike groups): only the last
/// 64 datagrams are kept, so a stalled consumer cannot retain an unbounded backlog.
/// The payload size limit also bounds the buffer's memory use.
const MAX_DATAGRAMS: usize = 64;

/// Slack before the eviction order is rebuilt, so a track holding just a few groups
/// doesn't rebuild on every write.
const EVICT_SLACK: usize = 64;

/// How many live eviction candidates one debt payment examines (Redis-style
/// bounded sampling): enough to step over a few protected (recently accessed)
/// groups, small enough that a write never scans a long queue.
const EVICT_SCAN: usize = 4;

/// One pass over the eviction order at a fixed cache time.
#[derive(Clone, Copy)]
pub(super) struct ExpiryScan {
	start: usize,
	// Ceiling on how many entries this pass examines. `EVICT_SCAN` from the write
	// path, the whole queue from the pool's cleanup pass.
	width: usize,
	now: u64,
	max_ticks: u64,
	// Cleanup dates pending activity; write-driven scans only check dated entries.
	gc: bool,
}

/// Publisher-side properties of a track.
///
/// These are fixed by the publisher when the track is created and don't change
/// while the track is alive. A subscriber learns them via
/// [`broadcast::Consumer::track`](broadcast::Consumer::track),
/// which returns the publisher's [`Info`] once the subscription is accepted.
//
// Deliberately not `Copy`, even though it's now a plain value: adding `Copy` turns
// every existing `info.clone()` in a consumer's code into a `clippy::clone_on_copy`
// error under `-D warnings`.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct Info {
	/// Units per second for per-frame timestamps on this track.
	///
	/// Every track is timed; this defaults to [`Timescale::MILLI`]. On Lite05+ it is
	/// reported in TRACK_INFO and the publisher zigzag-delta encodes per-frame
	/// timestamps at this scale on the wire. Protocols whose wire can't carry it
	/// (pre-Lite05 moq-lite, IETF moq-transport) fall back to local monotonic milliseconds.
	pub timescale: Timescale,
	/// How far behind the live edge a group may fall, in media timestamps, before it
	/// is stale. The newest group is always retained.
	///
	/// A retention bound rather than a delivery one, the inverse of an HTTP
	/// `Cache-Control: max-age`. [`Subscription::max_age`] is clamped to this, since a
	/// group can't be waited for longer than it's kept around. Reported in TRACK_INFO so
	/// relays re-serve with the same window. Defaults to [`DEFAULT_MAX_AGE`].
	///
	/// Measured against timestamps rather than the wall clock, so a congestion stall
	/// (timestamps stop advancing) can't age content out. Wall-clock reclamation of
	/// content nobody is accessing belongs to the cache pool's own
	/// [`expiry`](crate::cache::Pool::expiry) window, not to this budget.
	///
	/// This is the `Publisher Max Age` on the wire, the publisher-side half of the
	/// budget [`Subscription::max_age`] sets for a subscriber.
	///
	/// Encoded as milliseconds in a QUIC varint, so a duration of `2^62` milliseconds
	/// or more cannot be put on the wire. Sub-millisecond precision is truncated
	/// (`Duration::as_millis`) at encode time.
	pub max_age: Duration,
	/// The publisher's priority for this track, used only to break ties between
	/// subscriptions of equal subscriber priority. Reported in TRACK_INFO (Lite05+).
	pub priority: u8,
}

impl Default for Info {
	fn default() -> Self {
		Self {
			timescale: Timescale::default(),
			max_age: DEFAULT_MAX_AGE,
			priority: 0,
		}
	}
}

impl Info {
	/// Set the per-frame timestamp scale, returning `self` for chaining.
	///
	/// Defaults to [`Timescale::MILLI`]. On Lite05+ this scale is reported in TRACK_INFO
	/// and used to encode per-frame timestamps on the wire.
	pub fn with_timescale(mut self, timescale: Timescale) -> Self {
		self.timescale = timescale;
		self
	}

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

	/// Set the publisher's tie-break priority, returning `self` for chaining.
	pub fn with_priority(mut self, priority: u8) -> Self {
		self.priority = priority;
		self
	}
}

#[derive(Default)]
pub(crate) struct TrackState {
	// The publisher's properties, once known; always Some for Subscriber/Producer.
	// Copied by value into each group it creates.
	info: Option<Info>,
	// Whether a live Producer was minted. A reverse fetch may install `info`
	// before acceptance, so the two states are deliberately separate.
	published: bool,

	// The broadcast this track belongs to. Supplies the cache pool its groups charge
	// into and the `cache_duration` ceiling clamping `Info::max_age`.
	broadcast: Arc<broadcast::Info>,

	// This track's account against the shared cache pool, shared with every group it
	// creates (see `cache::Track`). Holds the gross-write counter `charge_debt` drains,
	// and the weak link a frame write follows back here to settle its own debt.
	cache: Arc<cache::Track>,

	// Cached groups by sequence: the single source of truth for what is cached. The
	// two orderings below hold bare sequences and validate against this map, so a
	// removed or replaced group turns their entries into discarded-on-pop hints.
	// Ordered so an in-range subscriber can seek to its cursor instead of scanning
	// every cached group for each delivery.
	lookup: BTreeMap<u64, Slot>,

	// Publisher-produced groups in arrival order as (sequence, stamp), walked by
	// subscriptions; an entry only resolves while its stamp matches the slot's.
	// Fetched backfill (`insert_group_request`) is deliberately absent: it is
	// served by sequence, never replayed to arrival-order subscribers.
	arrival: VecDeque<(u64, u32)>,

	// Eviction order under memory pressure as (sequence, stamp): every cached
	// group except the protected latest. `pay_debt` scans victims from the front;
	// groups accessed more recently than the pool-wide average rotate to the back
	// instead of dying, decoupling eviction order from arrival order. Entries are
	// hints that only resolve while their stamp matches the slot's, so a re-served
	// sequence can't accumulate duplicate hints that alias its replacement.
	// Eviction is deliberately approximate: a bounded scan per write.
	evict: VecDeque<(u64, u32)>,

	// Outstanding eviction debt in bytes, accrued by writes while the shared pool
	// is over capacity (see `cache::Pool::accrue`) and paid by aborting this
	// track's own oldest groups. Per track, so eviction lands proportionally to
	// what each track writes and never touches another track's cache.
	debt: u64,

	// Datagrams in arrival order, bounded by `MAX_DATAGRAMS`. Shares the group
	// sequence namespace but is otherwise independent.
	datagrams: VecDeque<Datagram>,

	// Number of datagrams dropped off the front (over capacity), mapping a subscriber's absolute
	// cursor to an index into `datagrams` (mirrors `offset` for groups).
	datagram_offset: usize,

	// We've popped the front of `arrival` this many times, mapping a subscriber's
	// absolute cursor to an index.
	offset: usize,

	// The highest sequence number successfully appended to the track. Shared with
	// datagrams, so it can run ahead of any cached group.
	max_sequence: Option<u64>,

	// The sequence of the newest cached group: the live edge, protected from
	// eviction by never entering the eviction order. Tracked separately from
	// `max_sequence` because datagrams advance that shared counter, and the live
	// edge must still demote correctly when the next group lands past one.
	latest_group: Option<u64>,

	// Incarnation counter for `Slot::stamp`.
	next_stamp: u32,

	// The sequence number at which the track was finalized.
	final_sequence: Option<u64>,

	// The first sequence the live feed serves, once the publisher declared one
	// (the wire's SUBSCRIBE_START). Lower groups never arrive on their own; a
	// fetch can still create them.
	start_sequence: Option<u64>,

	// Where production stopped, snapshotted when the cached groups are released (an
	// abort, or the last producer dropping). Computed live from the cache otherwise;
	// see [`Self::resume_position`].
	resume: Option<Position>,

	// The error that caused the track to be aborted, if any.
	abort: Option<Error>,

	// Active subscriptions, in their own [`kio::Shared`] so a read-only `Consumer`
	// registers under that lock instead of writing back into the track state.
	// Kept here (rather than threaded through every handle) so any holder reaches it.
	subscriptions: kio::Shared<Subscriptions>,

	// The reverse fetch queue (see [`FetchState`]), same reasoning: cache-miss
	// `fetch_group` calls enqueue here and a `Dynamic` drains.
	fetch: kio::Shared<FetchState>,
}

/// A cached group plus its bookkeeping in the track's `lookup` map.
///
/// Access times and the evictable-population sample live in the group's own
/// `cache::Charge`, so they share the group's lifecycle exactly: an abort from any
/// handle releases the bytes and the sample together.
struct Slot {
	group: group::Producer,

	// Incarnation stamp, echoed by this slot's arrival entry (if any). A re-served
	// sequence (an aborted group re-created by the publisher or re-fetched as
	// backfill) gets a fresh stamp, so a historical arrival entry can't resolve to
	// the replacement and deliver it twice or at the wrong position.
	stamp: u32,

	// Whether this incarnation came from the live publisher and can replace older
	// subscription content. Fetch-only backfill stays cached but never anchors drift.
	visible: bool,
}

/// 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
/// `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(crate) const CACHE_OVERHEAD: u64 = 2 * (size_of::<u64>() + size_of::<Slot>() + 2 * size_of::<(u64, u32)>()) as u64;

/// The registered subscriptions, aggregated by the producer.
type Subscriptions = Vec<kio::Consumer<Subscription>>;

/// 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
/// upstream FETCH per group) and [`Dynamic`] handlers drain under one lock, without
/// write access to the track itself.
pub(crate) type FetchState = Requests<u64, PendingFetch>;

/// One fetch attempt for a sequence, shared by every [`Fetching`] that joined it.
pub(crate) struct PendingFetch {
	// The most demanding delivery priority across the joined fetches.
	priority: u8,

	// The lowest start across the joined fetches, so serving the attempt once satisfies
	// every one of them. Only widened while the request is still queued: once a handler
	// has taken it, its range is already on the wire.
	frame_start: u64,

	// Result channel back to the joined fetches. Written only on rejection; a
	// successful accept resolves them through the track cache instead. Dropping
	// every producer without writing (a vanished handler) closes the channel,
	// which a [`Fetching`] reads as [`Error::NotFound`].
	result: kio::Producer<FetchOutcome>,
}

/// The result of a fetch attempt. Stays empty on success (the group lands in the
/// track cache); a handler writes `rejected` to fail every joined fetch.
#[derive(Default)]
pub(crate) struct FetchOutcome {
	pub(crate) rejected: Option<Error>,
}

impl TrackState {
	fn normalize_info(broadcast: &broadcast::Info, mut info: Info) -> Info {
		info.max_age = info.max_age.min(broadcast.cache_duration);
		info
	}

	fn accept(&mut self, info: Info) {
		self.published = true;
		self.install(info);
	}

	fn poll_info(&self) -> Poll<Result<Info>> {
		if let Some(info) = &self.info {
			Poll::Ready(Ok(info.clone()))
		} else if let Some(err) = &self.abort {
			// Aborted before anyone served it, so the info can never arrive: fail the
			// waiting subscribes instead of parking them on a track nobody will fill.
			Poll::Ready(Err(err.clone()))
		} else {
			Poll::Pending
		}
	}

	/// Find the next live group at or after `index` in arrival order.
	///
	/// Returns the group and its absolute index so the consumer can advance past it.
	/// The cached producer rather than a consumer: `consume` reads the clock and can
	/// take the group's own lock, which the caller does once the track guard is gone.
	fn poll_recv_group(&self, index: usize, min_sequence: u64) -> Poll<Result<Option<(group::Producer, usize)>>> {
		let start = index.saturating_sub(self.offset);
		for (i, (sequence, stamp)) in self.arrival.iter().enumerate().skip(start) {
			if *sequence >= min_sequence
				&& let Some(slot) = self.lookup.get(sequence)
				&& slot.stamp == *stamp
				&& !slot.group.is_aborted()
			{
				return Poll::Ready(Ok(Some((slot.group.clone(), self.offset + i))));
			}
		}

		// TODO once we have drop notifications, check if index == final_sequence.
		if self.is_complete() {
			Poll::Ready(Ok(None))
		} else if let Some(err) = &self.abort {
			Poll::Ready(Err(err.clone()))
		} else {
			Poll::Pending
		}
	}

	/// Find the next datagram at or after the subscriber's absolute `index`.
	///
	/// Returns the datagram and its absolute index so the consumer can advance past it. A
	/// consumer whose `index` has fallen behind `datagram_offset` (older datagrams dropped)
	/// resumes at the oldest still-buffered datagram, skipping the lost ones.
	fn poll_recv_datagram(&self, index: usize) -> Poll<Result<Option<(Datagram, usize)>>> {
		let start = index.saturating_sub(self.datagram_offset);
		if let Some(datagram) = self.datagrams.get(start) {
			return Poll::Ready(Ok(Some((datagram.clone(), self.datagram_offset + start))));
		}

		// Nothing buffered at the cursor: the track ending terminates the datagram stream too.
		if self.is_complete() {
			Poll::Ready(Ok(None))
		} else if let Some(err) = &self.abort {
			Poll::Ready(Err(err.clone()))
		} else {
			Poll::Pending
		}
	}

	/// Push a datagram, dropping the oldest when the send buffer is full.
	fn push_datagram(&mut self, datagram: Datagram) {
		if self.datagrams.len() == MAX_DATAGRAMS {
			self.datagrams.pop_front();
			self.datagram_offset += 1;
		}
		self.datagrams.push_back(datagram);
	}

	/// Find the smallest-sequence cached group satisfying
	/// `next_sequence <= seq < end_sequence (if set)`. Used by
	/// [`Subscriber::next_group`] so the range can be widened (or unset)
	/// after the fact and previously-skipped cached groups become available
	/// without scanning past them in arrival order.
	///
	/// Returns `Poll::Pending` when no in-range group is currently cached but
	/// future groups could still arrive in range; returns `Ok(None)` only when
	/// the track is finalized and no further in-range group is possible.
	fn poll_next_in_range(
		&self,
		next_sequence: u64,
		end_sequence: Option<u64>,
	) -> Poll<Result<Option<group::Producer>>> {
		// If the exclusive end is already at or below where we'd resume, no
		// group can ever satisfy this call until the cap rises. Pending (not
		// None) so the consumer is parked rather than told the stream is over.
		// An empty range (`end == 0`) parks even at the first sequence.
		if let Some(end) = end_sequence
			&& end <= next_sequence
		{
			if let Some(err) = &self.abort {
				return Poll::Ready(Err(err.clone()));
			}
			return Poll::Pending;
		}

		let best = self
			.lookup
			.range(next_sequence..)
			.map(|(_, slot)| &slot.group)
			.take_while(|group| super::subscription::before_end(group.sequence, end_sequence))
			.find(|group| !group.is_aborted());

		if let Some(group) = best {
			// Deliberately no cache refresh here: this is a pure seek, and the spliced
			// merge consults candidates it may not deliver. The deliverer stamps the
			// winner; a loser re-seeked every poll must not be shielded from eviction.
			// The caller consumes it with the track guard released, like `poll_recv_group`.
			return Poll::Ready(Ok(Some(group.clone())));
		}

		// No in-range group is cached. Decide whether more could ever arrive.
		if let Some(err) = &self.abort {
			return Poll::Ready(Err(err.clone()));
		}
		// `final_sequence` is one past the last possible sequence. If our
		// floor is already at/past it, nothing else can land in range.
		if let Some(fin) = self.final_sequence
			&& next_sequence >= fin
		{
			return Poll::Ready(Ok(None));
		}
		Poll::Pending
	}

	/// The cached group for `sequence`, but only when it holds every frame from
	/// `frame_start` onward.
	///
	/// A group cached from a frame-bounded subscription starts partway in, and serving
	/// it to someone who asked for the whole group would silently hand back a tail. It
	/// is a miss instead, so the fetch goes upstream for the frames that are missing.
	fn covering_group(&self, sequence: u64, frame_start: u64) -> Option<&group::Producer> {
		let slot = self.lookup.get(&sequence)?;
		let first = slot.group.live_first_frame()?;
		(first as u64 <= frame_start).then_some(&slot.group)
	}

	/// The publisher's max age window, or `None` while the info is unknown (an
	/// unaccepted [`Request`]). Bounds the aggregate subscription; see [`clamp_combined`].
	fn max_age_bound(&self) -> Option<Duration> {
		self.info.as_ref().map(|info| info.max_age)
	}

	/// The live edge a subscription bounded at `cap` measures drift against: the
	/// highest-sequence group that has presented a frame, or `None` while nothing
	/// stamped is servable (an unstamped track drives no expiry at all; the cache's
	/// own wall-clock policy is what bounds it).
	///
	/// One scan answers a whole poll. The highest-sequence anchor in range is above
	/// every candidate below it and above none at or past it. Recomputing per candidate
	/// would make walking a backlog of N groups off in one poll cost O(N^2).
	///
	/// Capped because a group can only be late relative to data that would actually be
	/// served in its place: a subscription ending at a takeover boundary can't jump past
	/// it, so the groups it still wants aren't stale just because the route ran on.
	/// Fetched backfill is absent from `arrival`, so it cannot age subscription content
	/// as though it were a live replacement.
	fn live_edge(&self, cap: Option<u64>) -> Option<Edge> {
		let presentation = self
			.lookup
			.range(..)
			.rev()
			.filter(|(seq, _)| super::subscription::before_end(**seq, cap))
			.find_map(|(_, slot)| {
				if !slot.visible || slot.group.is_aborted() {
					return None;
				}
				// The map is ordered by sequence, so the first stamped group from the
				// back is the newest content that exists.
				let timestamp = slot.group.timestamp()?;
				Some(PresentationEdge {
					sequence: slot.group.sequence,
					stamp: slot.stamp,
					timestamp: slot.group.latest().unwrap_or(timestamp),
				})
			});

		presentation.map(|presentation| Edge { presentation, cap })
	}

	/// The furthest presentation time the group at `sequence` could still reach: where
	/// its immediate successor begins, or `None` when nothing proves where it stops.
	///
	/// An upper bound, deliberately. A frame's duration is not on the wire, so a group's
	/// own last timestamp says where it *starts* presenting, not where it ends; only its
	/// successor's start proves it cannot run past it. And only the *immediate* servable
	/// successor counts: timestamps need not rise with sequence (a rewind reorders them),
	/// so a later stamped group proves nothing about where an unstamped successor will
	/// begin, and shrinking the bound is the unsafe direction. An unstamped successor
	/// therefore leaves the reach unbounded until it presents its first frame.
	fn reach(&self, sequence: u64, cap: Option<u64>) -> Option<Timestamp> {
		let successor = self
			.lookup
			.range(sequence.saturating_add(1)..)
			.map(|(_, slot)| slot)
			.take_while(|slot| super::subscription::before_end(slot.group.sequence, cap))
			.find(|slot| slot.visible && !slot.group.is_aborted())?;
		successor.group.timestamp()
	}

	/// Whether the group at `sequence` has drifted further behind `edge` than `budget`
	/// tolerates, so a subscriber should skip it rather than hand it over.
	///
	/// Presentation time measures a group by how far it could still *reach*, not by how far
	/// behind it started. Being behind is survivable: priority transmits newer groups
	/// first, so a backlog bursts at whatever rate is left over and closes the gap faster
	/// than the live edge advances. What is not survivable is having nothing left worth
	/// delivering, so a group is abandoned only once everything it could still present
	/// falls outside the budget.
	///
	/// A group's reach is bounded by its immediate successor (see [`Self::reach`]): it
	/// cannot present past where the next group begins. The candidate itself needs no
	/// timestamp: an empty group is bounded by its stamped successor the same way. Only
	/// timestamps drive expiry; wall-clock reclamation of idle content is the cache's
	/// own policy, not the budget's.
	///
	/// The bound is exclusive, so the comparison is `>=` rather than `>`: the freshest frame
	/// a group could still hold sits just *below* its reach, so an age equal to the budget
	/// already puts every frame in it strictly past the budget. That also makes a zero
	/// budget fall out for free instead of needing a special case.
	///
	/// The edge must sit strictly above the candidate. The live edge is never late
	/// against itself, and backfill or the tail of a rewound timeline can carry a high
	/// timestamp on a low sequence without being an edge at all.
	fn is_stale(&self, sequence: u64, edge: Option<&Edge>, budget: Duration) -> bool {
		let Some(edge) = edge else {
			return false;
		};
		if !self.lookup.contains_key(&sequence) {
			return false;
		}

		// The anchor was resolved under an earlier lock, so confirm it still names
		// the same servable incarnation before it convicts a candidate. Failing safe
		// (delivering) is right, since the next poll resolves fresh anchors.
		let live_edge = &edge.presentation;
		let reach = self.reach(sequence, edge.cap);
		live_edge.sequence > sequence
			&& self
				.lookup
				.get(&live_edge.sequence)
				.is_some_and(|live| live.stamp == live_edge.stamp && !live.group.is_aborted())
			&& reach.is_some_and(
				|reach| matches!(live_edge.timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget),
			)
	}

	/// Resolve a one-shot fetch from the track side: the cached group, or an [`Error`]
	/// once it can never be served. A missing group is a failure ([`Error::NotFound`]), not an
	/// end-of-stream. The handler side (a rejection, or no [`Dynamic`] at all) lives
	/// in [`FetchState`]; [`Fetching`] polls both.
	fn poll_fetch_cached(&self, sequence: u64, frame_start: u64) -> Poll<Result<group::Consumer>> {
		if let Some(group) = self.covering_group(sequence, frame_start) {
			// A cache hit refreshes the group: it resets both its age (expiry keys
			// off the last access) and its standing against the pool-wide average,
			// so the eviction walk keeps it over never-read groups.
			group.cache_refresh();
			return Poll::Ready(Ok(group.consume()));
		}

		if let Some(err) = &self.abort {
			return Poll::Ready(Err(err.clone()));
		}

		// Past the final sequence: the group can never exist.
		if self.final_sequence.is_some_and(|fin| sequence >= fin) {
			return Poll::Ready(Err(Error::NotFound));
		}

		Poll::Pending
	}

	/// Expire groups idle past the pool's wall-clock LRU window, never the latest.
	///
	/// The window is the pool's [`expiry`](cache::Pool::expiry), not the track's
	/// [`max_age`](Info::max_age): retention is a media-timestamp promise, while this
	/// is the cache's own guard against unread content pinning RAM, shared by every
	/// track in the pool.
	///
	/// One bounded, rotating scan over the eviction order, which holds every cached
	/// group except the protected latest. The cursor persists in the cache account, so
	/// entries beyond one scan window can't be starved by fresh (recently read,
	/// fetched, or written) entries in front of them: every position is revisited
	/// within a few writes. Expiry throughput is therefore EVICT_SCAN groups per write; the
	/// byte budget reclaims the remainder under memory pressure, and the pool's sweep
	/// ([`Self::expiry_scan_drain`]) covers a track that stopped writing entirely.
	pub(super) fn evict_expired(&mut self) {
		let scan = self.expiry_scan();
		self.evict_expired_scan(scan);
	}

	/// Describe the next bounded expiry scan without mutating observable track state.
	pub(super) fn expiry_scan(&self) -> ExpiryScan {
		ExpiryScan {
			start: self.cache.next_expiry_scan(EVICT_SCAN),
			width: EVICT_SCAN,
			now: self.cache.pool().now(),
			max_ticks: self.cache.pool().expiry_ticks(),
			gc: false,
		}
	}

	/// Scan every cached candidate so undated accesses and old entries cannot hide
	/// behind fresh entries at the front of the eviction order.
	pub(super) fn expiry_scan_drain(&self) -> ExpiryScan {
		ExpiryScan {
			start: 0,
			width: self.evict.len(),
			now: self.cache.pool().now(),
			max_ticks: self.cache.pool().expiry_ticks(),
			gc: true,
		}
	}

	#[cfg(test)]
	pub(super) fn date_cache_accesses(&self, now: u64) {
		for slot in self.lookup.values() {
			slot.group.cache_accessed_tick(Some(now));
		}
	}

	/// Whether an expiry scan would change observable track state.
	///
	/// Mirrors [`Self::evict_expired_scan`]'s walk exactly, stop rule included: a memo
	/// that scanned further would report work the scan itself will not do.
	pub(super) fn expiry_mutation_due(&self, scan: ExpiryScan) -> bool {
		let len = self.evict.len();
		if len > 0 {
			let start = scan.start % len;
			let mut retained = 0;
			for step in 0..len.min(scan.width) {
				let (sequence, stamp) = self.evict[(start + step) % len];
				let Some(slot) = self.lookup.get(&sequence) else {
					continue;
				};
				if slot.stamp != stamp {
					continue;
				}
				if slot.group.is_aborted()
					|| (Some(sequence) != self.latest_group
						&& slot
							.group
							.cache_accessed_tick(scan.gc.then_some(scan.now))
							.is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks))
				{
					return true;
				}
				retained += 1;
				if !scan.gc && retained >= EVICT_SCAN {
					break;
				}
			}
		}

		self.arrival
			.front()
			.is_some_and(|(sequence, stamp)| !self.is_current(*sequence, *stamp))
			|| self
				.evict
				.front()
				.is_some_and(|(sequence, stamp)| !self.is_current(*sequence, *stamp))
			|| self.evict.len() > 2 * self.lookup.len() + EVICT_SLACK
	}

	/// Apply a scan previously selected by [`Self::expiry_scan`] or
	/// [`Self::expiry_scan_drain`].
	pub(super) fn evict_expired_scan(&mut self, scan: ExpiryScan) {
		let len = self.evict.len();
		if len > 0 {
			let start = scan.start % len;
			let mut retained = 0;
			for step in 0..len.min(scan.width) {
				let (sequence, stamp) = self.evict[(start + step) % len];
				let Some(slot) = self.lookup.get(&sequence) else {
					continue;
				};
				if slot.stamp != stamp {
					// A historical hint; the live entry is elsewhere in the queue.
					continue;
				}
				// Already aborted: the frames are gone, reclaim the slot so a
				// later fetch can serve the sequence again.
				if slot.group.is_aborted() {
					self.lookup.remove(&sequence);
					continue;
				}
				if Some(sequence) == self.latest_group
					|| slot
						.group
						.cache_accessed_tick(scan.gc.then_some(scan.now))
						.is_none_or(|tick| scan.now.saturating_sub(tick) <= scan.max_ticks)
				{
					// Writes keep their scan bounded. Cleanup visits the entire
					// queue to date pending accesses and find idle entries behind them.
					retained += 1;
					if !scan.gc && retained >= EVICT_SCAN {
						break;
					}
					continue;
				}
				// Take the group out of the cache and abort it, so any consumer
				// still reading surfaces `Error::Old` instead of blocking forever
				// on a frame that will never arrive.
				let slot = self.lookup.remove(&sequence).unwrap();
				let _ = slot.group.abort(Error::Old);
			}
		}

		// Trim dead leading arrival entries to advance the subscriber offset. An
		// entry is dead once its slot is gone or re-stamped by a newer incarnation.
		while let Some((sequence, stamp)) = self.arrival.front() {
			if self.lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp) {
				break;
			}
			self.arrival.pop_front();
			self.offset += 1;
		}

		// Drop dead leading eviction entries so scans stay over live candidates.
		while let Some((sequence, stamp)) = self.evict.front() {
			if self.lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp) {
				break;
			}
			self.evict.pop_front();
		}

		// Dead entries behind a live front can linger; rebuild once they clearly
		// outnumber the live slots.
		if self.evict.len() > 2 * self.lookup.len() + EVICT_SLACK {
			let lookup = &self.lookup;
			self.evict
				.retain(|(sequence, stamp)| lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp));
		}
	}

	/// Whether `(sequence, stamp)` names the currently cached incarnation.
	fn is_current(&self, sequence: u64, stamp: u32) -> bool {
		self.lookup.get(&sequence).is_some_and(|slot| slot.stamp == stamp)
	}

	/// Drop every cached group and reset the eviction bookkeeping. Each group's
	/// access sample lives in its own charge, released when the group itself dies.
	fn clear_cache(&mut self) {
		self.lookup.clear();
		self.arrival.clear();
		self.evict.clear();
		self.latest_group = None;
		self.debt = 0;
	}

	/// Attach `info` to this track, clamping the publisher's window down to the
	/// origin's [`cache_duration`](crate::origin::Config::cache_duration) ceiling so a
	/// group is never retained longer than the origin allows. Every path that binds an
	/// info to a track funnels through here, covering local publishers and relayed
	/// (lite / IETF) tracks alike.
	fn install(&mut self, info: Info) {
		let info = Self::normalize_info(&self.broadcast, info);
		self.info = Some(info);
	}

	/// Create the shared state for a track under `broadcast`, along with the cache
	/// account it and its groups charge into.
	///
	/// The account holds a [`kio::Weak`] back to this state: a group must be able to
	/// settle the track's eviction debt as it writes, but the track owns its cached
	/// groups, so anything stronger would make the pair immortal.
	fn spawn(broadcast: Arc<broadcast::Info>) -> kio::Producer<Self> {
		let state = kio::Producer::new(Self {
			broadcast: broadcast.clone(),
			..Default::default()
		});
		let cache = cache::Track::new(broadcast.pool.clone(), state.downgrade());
		state.write().ok().expect("a new track is open").cache = cache;
		state
	}

	/// Reject a sequence that is still cached; a dead (aborted or evicted)
	/// incarnation is removed so a fresh group can serve the sequence again.
	///
	/// Best effort: nothing remembers a sequence whose slot is already gone, so a
	/// publisher re-sending a long-evicted sequence is accepted as new.
	/// A group that starts above `frame_start` is also replaceable: it cannot answer a
	/// request from there, so the wider producer takes the slot. Readers already
	/// draining the old one keep their own handle.
	fn claim_sequence(&mut self, sequence: u64, frame_start: u64) -> Result<()> {
		if let Some(slot) = self.lookup.get(&sequence) {
			// The same question `covering_group` asks: can this slot still answer from
			// `frame_start`? If it can, the sequence is taken; if it can't, it is dead
			// and the caller gets to replace it.
			if slot
				.group
				.live_first_frame()
				.is_some_and(|first| first as u64 <= frame_start)
			{
				return Err(Error::Duplicate);
			}
			self.lookup.remove(&sequence);
		}
		Ok(())
	}

	/// Insert a freshly-created group into the cache.
	///
	/// Updates the live edge, demoting the previous latest into the eviction order;
	/// the current latest is never enqueued, which is what protects it from
	/// eviction. `visible` controls arrival-order delivery: publisher-produced
	/// groups reach subscribers, fetched backfill is served by sequence only.
	fn insert_group(&mut self, group: &group::Producer, visible: bool) {
		let sequence = group.sequence;
		self.next_stamp = self.next_stamp.wrapping_add(1);
		let stamp = self.next_stamp;

		// The live edge is tracked separately from `max_sequence`, which datagrams
		// share and can push past any cached group: demotion must still fire when
		// the next group lands beyond a datagram-advanced counter.
		if self.latest_group.is_none_or(|latest| sequence >= latest) {
			// Demote the previous latest: it joins the eviction order (and the
			// pool's access average) like any other cached group.
			if let Some(latest) = self.latest_group
				&& sequence > latest
				&& let Some(prev) = self.lookup.get(&latest)
			{
				prev.group.cache_demote();
				self.evict.push_back((latest, prev.stamp));
			}
			self.latest_group = Some(sequence);
		} else {
			group.cache_demote();
			self.evict.push_back((sequence, stamp));
		}

		self.max_sequence = Some(self.max_sequence.map_or(sequence, |max| max.max(sequence)));
		self.lookup.insert(
			sequence,
			Slot {
				group: group.clone(),
				stamp,
				visible,
			},
		);
		if visible {
			self.arrival.push_back((sequence, stamp));
		}
	}

	/// Admit a freshly-created group: settle eviction debt first (so the newcomer
	/// can never be a victim of the very write that created it), insert it, then
	/// expire idle groups.
	fn commit_group(&mut self, group: &group::Producer, visible: bool) {
		self.charge_debt();
		self.insert_group(group, visible);
		self.evict_expired();
	}

	/// Accrue and pay eviction debt for everything written since the last charge:
	/// this track's account, which the groups' charges feed on every frame (so
	/// growth on already-demoted groups and backfill is billed too).
	///
	/// Runs BEFORE the new group is inserted, so a brand-new entry is never a
	/// victim of the very write that created it. A track whose oldest content is
	/// staler than the pool-wide average access time accrues at double rate, so
	/// stale-heavy tracks drain first.
	///
	/// Also runs from the frame-write path via [`cache::Track::settle`], which is
	/// why it's reachable from the account, so a track that only appends frames to
	/// open groups still pays.
	pub(super) fn charge_debt(&mut self) {
		let written = self.cache.take_written();
		let pool = self.cache.pool().clone();
		match pool.accrue(written) {
			Some(mut accrued) => {
				if self.oldest_is_stale(&pool) {
					accrued = accrued.saturating_mul(2);
				}
				// `used` bounds what eviction could ever free, keeping a track that
				// can't pay (everything protected) from hoarding a stale schedule.
				self.debt = self.debt.saturating_add(accrued).min(pool.used());
				// Cap each payment at twice what was written so one write never dumps
				// a deep backlog at once; the remainder carries to the next write.
				self.pay_debt(&pool, written.saturating_mul(2));
			}
			// Under capacity there is nothing to work off, and stale debt would
			// cause a spurious eviction burst at the next pressure spike.
			None => self.debt = 0,
		}
	}

	/// Whether this track's oldest evictable group was accessed at or before the
	/// pool-wide average, doubling the debt it accrues. A dead entry at the front
	/// just reads as not-stale until the next payment or expiry cleans it up.
	fn oldest_is_stale(&self, pool: &cache::Pool) -> bool {
		let Some(average) = pool.average() else {
			return false;
		};
		let Some((sequence, stamp)) = self.evict.front() else {
			return false;
		};
		let Some(slot) = self.lookup.get(sequence) else {
			return false;
		};
		slot.stamp == *stamp && !slot.group.is_aborted() && slot.group.cache_accessed() <= average
	}

	/// Abort this track's stalest groups until the outstanding debt is paid, or
	/// `cap` bytes have been freed by this call.
	///
	/// Deliberately approximate, Redis-style: at most a handful of live candidates
	/// are examined per call, from the front of the eviction order. A group
	/// accessed more recently than the pool-wide average is protected and rotates
	/// to the back, so fresh content in this track never dies while staler content
	/// survives elsewhere; the unfreed bytes keep the pool over budget, shifting
	/// the debt onto the tracks holding that staler content. When the next victim
	/// is larger than the remaining debt it is left in place and the debt carries
	/// over, so a small write never evicts a huge group (once the debt does cover
	/// it, that one victim may overshoot `cap`).
	fn pay_debt(&mut self, pool: &cache::Pool, cap: u64) {
		let average = pool.average().unwrap_or(0);
		let mut paid = 0u64;
		let mut scanned = 0usize;
		for _ in 0..self.evict.len() {
			if self.debt == 0 || paid >= cap || scanned >= EVICT_SCAN {
				return;
			}
			let Some((sequence, stamp)) = self.evict.pop_front() else {
				return;
			};
			let Some(slot) = self.lookup.get(&sequence) else {
				// Evicted or expired; discard the dead entry.
				continue;
			};
			if slot.stamp != stamp {
				// A historical hint; the live entry is elsewhere in the queue.
				continue;
			}
			if slot.group.is_aborted() {
				// Aborted upstream: the frames are already gone, reclaim the slot.
				self.lookup.remove(&sequence);
				continue;
			}
			if Some(sequence) == self.latest_group {
				// The live edge is never enqueued, but tolerate finding it anyway.
				self.evict.push_back((sequence, stamp));
				continue;
			}

			scanned += 1;
			// Protected: accessed more recently than the average (a fresh insert,
			// an active reader, or a FETCH hit, which also covers a backfill still
			// being filled). Rotate to the back.
			if slot.group.cache_accessed() > average {
				self.evict.push_back((sequence, stamp));
				continue;
			}
			// The full footprint including overhead, so even empty groups repay
			// their share of the budget when evicted.
			let size = slot.group.cache_size();
			if size > self.debt {
				self.evict.push_front((sequence, stamp));
				return;
			}

			self.debt -= size;
			paid = paid.saturating_add(size);
			let slot = self.lookup.remove(&sequence).unwrap();
			let _ = slot.group.abort(Error::Evicted);
		}
	}

	/// Record the declared first sequence of the live feed, replacing any earlier
	/// declaration: the signal is scoped to the current subscription's demand,
	/// which may legitimately move in either direction. `None` clears it (the
	/// demand dropped to the live edge, whose floor is unknown until declared).
	fn set_start(&mut self, start_sequence: Option<u64>) {
		self.start_sequence = start_sequence;
	}

	/// Record the exclusive final sequence, rejecting a re-finish or a boundary that
	/// would orphan already-produced groups.
	fn set_final(&mut self, final_sequence: u64) -> Result<()> {
		if self.final_sequence.is_some() {
			return Err(Error::Closed);
		}
		if let Some(max) = self.max_sequence
			&& final_sequence <= max
		{
			return Err(Error::ProtocolViolation);
		}
		self.final_sequence = Some(final_sequence);
		Ok(())
	}

	/// Whether the track has reached its end: the final boundary is set and the live
	/// edge has caught up to it, so no further group can arrive. A future boundary
	/// (declared via [`Producer::finish_at`] ahead of the live edge) stays incomplete
	/// until the remaining groups are produced. Drives the end-of-stream signal from
	/// the read methods (`recv_group` / `next_group` / `read_frame` return `None`).
	fn is_complete(&self) -> bool {
		self.final_sequence
			.is_some_and(|fin| self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin)
	}

	/// Where a replacement route should pick this track up: one past the last frame
	/// produced, rolling to the start of the next group once the latest group is
	/// complete (nothing more can be appended to it).
	///
	/// `None` while the track has produced nothing, which is an unbounded takeover.
	fn resume_position(&self) -> Option<Position> {
		// A snapshot taken when the cache was released wins; the groups it was derived
		// from are gone.
		if self.resume.is_some() {
			return self.resume;
		}

		let max = self.latest_group?;
		match self.lookup.get(&max).and_then(|slot| slot.group.resume_frame()) {
			// Still open *and* carrying frames, so the replacement continues it
			// frame-by-frame.
			Some(frame) => Some(Position {
				group: max,
				frame: frame as u64,
			}),
			// A copy that wrote nothing has no frames to splice onto, and the reader
			// already holds its (empty) handle. Pointing a replacement at frame 0 would
			// hand the same sequence out twice, so roll to the next group exactly as a
			// finished one does.
			None => Some(Position::group(max.saturating_add(1))),
		}
	}

	fn poll_finished(&self) -> Poll<Result<u64>> {
		if let Some(fin) = self.final_sequence {
			Poll::Ready(Ok(fin))
		} else if let Some(err) = &self.abort {
			Poll::Ready(Err(err.clone()))
		} else {
			Poll::Pending
		}
	}

	fn modify(producer: &kio::Producer<Self>) -> Result<kio::Mut<'_, Self>> {
		producer.write().map_err(|r| r.abort.clone().unwrap_or(Error::Dropped))
	}

	/// Insert a group fetched for a [`group::Request`], setting the track's [`Info`]
	/// if it isn't accepted yet. The group's timescale comes from that info, so a
	/// fetch can serve an as-yet-unaccepted track (e.g. a relay with no live
	/// subscription). The group lands in the cache so a waiting
	/// [`Fetching`] resolves via [`Self::poll_fetch`].
	pub(crate) fn insert_group_request(
		&mut self,
		sequence: u64,
		frame_start: u64,
		info: Option<Info>,
	) -> Result<group::Producer> {
		if let Some(err) = &self.abort {
			return Err(err.clone());
		}
		if let Some(fin) = self.final_sequence
			&& sequence >= fin
		{
			return Err(Error::Closed);
		}

		// Adopt the supplied info only if the track hasn't been accepted yet. Groups
		// created here charge the same account as any other, so backfill written
		// before the track is accepted settles its debt like the rest.
		if self.info.is_none() {
			self.install(info.unwrap_or_default());
		}
		let info = self.info.clone().unwrap();

		// An evicted sequence can be re-fetched; a live one is a duplicate.
		self.claim_sequence(sequence, frame_start)?;

		let group = group::Producer::new(group::Info { sequence }, info, self.cache.clone());
		// A backfill exists because someone is fetching it right now: stamp that
		// access so the eviction walk can't kill it before the fetch resolves.
		// It is also invisible to arrival-order subscribers: fetched on demand,
		// not produced live by the publisher.
		group.cache_refresh();
		self.commit_group(&group, false);
		Ok(group)
	}
}

/// Record `err` and close the track: the shared tail of [`Producer::abort`] and
/// [`Producer::abort_unused`].
fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) {
	// Snapshot the frame boundary before the cache it's derived from goes away: an
	// abort is exactly when a replacement route asks where to resume.
	state.resume = state.resume_position();
	state.abort = Some(err);
	state.clear_cache();
	state.datagrams.clear();
	state.close();
}

/// A producer for a track, used to create new groups.
#[derive(Clone)]
pub struct Producer {
	name: Arc<str>,
	info: Info,
	// The parent broadcast's info, inherited from [`broadcast::Producer::create_track`].
	// Top link of the ownership chain; carried for identity and future inheritance.
	broadcast: Arc<broadcast::Info>,
	state: kio::Producer<TrackState>,
	prev_subscription: Option<Subscription>,
	// Shared with every clone and every `Dynamic`: its `Drop` is the teardown.
	alive: Arc<Alive>,
	// Ingress stats scope, inherited from a tagged [`broadcast::Producer`]. Bumped as
	// one subscription on tag and closed when the last producer clone drops. Empty
	// (no-op) for an untagged broadcast.
	stats: stats::Scope,
}

impl Producer {
	/// Build a producer for the given track metadata.
	///
	/// Crate-private: tracks are born from their broadcast via
	/// [`broadcast::Producer::create_track`] (or served on demand through a
	/// [`Request`]), which threads the broadcast's `Arc<broadcast::Info>` down. The
	/// track opens its cache account against that broadcast's origin pool, and every
	/// group it creates charges into it.
	pub(crate) fn new(
		broadcast: Arc<broadcast::Info>,
		name: impl Into<Arc<str>>,
		info: impl Into<Option<Info>>,
	) -> Self {
		let name = name.into();
		let info = TrackState::normalize_info(&broadcast, info.into().unwrap_or_default());
		let state = TrackState::spawn(broadcast.clone());
		state.write().ok().expect("a new track is open").accept(info.clone());
		let alive = Alive::new(name.clone(), state.clone());
		alive.publish(None);
		Self {
			name,
			info,
			state,
			broadcast,
			prev_subscription: None,
			alive,
			stats: stats::Scope::default(),
		}
	}

	/// Attach the parent broadcast's ingress stats scope, counting this track as one
	/// ingress subscription (closed when the last producer clone drops). Called by a
	/// tagged [`broadcast::Producer`] when it creates the track.
	pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self {
		self.alive.publish(Some(&scope));
		self.stats = scope;
		self
	}

	/// The track's name, unique within its broadcast.
	pub fn name(&self) -> &str {
		&self.name
	}

	/// The parent broadcast this track belongs to.
	pub fn broadcast(&self) -> &broadcast::Info {
		&self.broadcast
	}

	/// Cache an already-produced group without rewriting its frames.
	///
	/// Used by an origin front to keep a warm copy of groups it already delivered
	/// after dropping the source track that produced them.
	pub(crate) fn adopt_group(&mut self, group: group::Producer, visible: bool) -> Result<()> {
		let mut state = self.modify()?;
		if let Some(fin) = state.final_sequence
			&& group.sequence >= fin
		{
			return Err(Error::Closed);
		}
		if state.lookup.contains_key(&group.sequence) {
			return Err(Error::Duplicate);
		}
		state.insert_group(&group, visible);
		Ok(())
	}

	/// Create a new group with the given sequence number.
	pub fn create_group(&self, group: group::Info) -> Result<group::Producer> {
		let mut state = self.modify()?;
		if let Some(fin) = state.final_sequence
			&& group.sequence >= fin
		{
			return Err(Error::Closed);
		}
		let track = state.info.clone().unwrap();

		// An evicted sequence can be re-created; a live one is a duplicate.
		state.claim_sequence(group.sequence, 0)?;

		let group = group::Producer::new(group, track, state.cache.clone()).with_meter(self.stats.meter());
		state.commit_group(&group, true);

		Ok(group)
	}

	/// Create a new group with the next sequence number.
	pub fn append_group(&self) -> Result<group::Producer> {
		let mut state = self.modify()?;
		let sequence = match state.max_sequence {
			Some(s) => s.checked_add(1).ok_or(coding::BoundsExceeded)?,
			None => 0,
		};
		if let Some(fin) = state.final_sequence
			&& sequence >= fin
		{
			return Err(Error::Closed);
		}

		let track = state.info.clone().unwrap();

		let group =
			group::Producer::new(group::Info { sequence }, track, state.cache.clone()).with_meter(self.stats.meter());
		state.commit_group(&group, true);

		Ok(group)
	}

	/// Append a datagram with the next sequence number, returning the assigned sequence.
	///
	/// A datagram is delivered best-effort over a single QUIC datagram, parallel to the
	/// track's groups but drawing from the same sequence namespace (so interleaving with
	/// [`Self::append_group`] never reuses a number). There is no group fallback: each
	/// session drops (with a debug log) any datagram whose encoded body exceeds the
	/// transport's datagram size, and sessions that can't carry datagrams at all (IETF
	/// moq-transport, moq-lite before 05, or stream-only transports like WebSocket) never
	/// deliver them. Keep payloads well under the 1200-byte minimum path MTU. An origin
	/// publisher uses this; a relay preserving upstream numbering uses
	/// [`Self::insert_datagram`].
	pub fn append_datagram<B: crate::IntoBytes>(&mut self, timestamp: Timestamp, payload: B) -> Result<u64> {
		let payload = payload.into_bytes();
		if payload.len() > super::datagram::MAX_DATAGRAM_PAYLOAD {
			return Err(Error::FrameTooLarge);
		}
		// Resolved before the state guard borrows `self`.
		let meter = self.stats.meter();
		let mut state = self.modify()?;
		// Normalize into the track's timescale, like frames (see `group::Producer::create_frame`).
		let timescale = state.info.as_ref().unwrap().timescale;
		let timestamp = timestamp.convert(timescale).map_err(|_| Error::TimestampMismatch)?;
		let sequence = match state.max_sequence {
			Some(s) => s.checked_add(1).ok_or(coding::BoundsExceeded)?,
			None => 0,
		};
		if let Some(fin) = state.final_sequence
			&& sequence >= fin
		{
			return Err(Error::Closed);
		}
		state.max_sequence = Some(sequence);
		meter.datagram(payload.len() as u64);
		state.push_datagram(Datagram {
			sequence,
			timestamp,
			payload,
		});
		let cache = state.cache.clone();
		drop(state);
		cache.settle(None);
		Ok(sequence)
	}

	/// Insert a datagram with an explicit sequence number.
	///
	/// Preserves the supplied sequence (bumping the shared `max_sequence` if needed), so a
	/// relay can forward a datagram without renumbering it. Most origin publishers want
	/// [`Self::append_datagram`] instead.
	pub fn insert_datagram<B: crate::IntoBytes>(
		&mut self,
		sequence: u64,
		timestamp: Timestamp,
		payload: B,
	) -> Result<()> {
		let payload = payload.into_bytes();
		if payload.len() > super::datagram::MAX_DATAGRAM_PAYLOAD {
			return Err(Error::FrameTooLarge);
		}
		// Resolved before the state guard borrows `self`.
		let meter = self.stats.meter();
		let mut state = self.modify()?;
		// Normalize into the track's timescale, like frames (see `group::Producer::create_frame`).
		let timescale = state.info.as_ref().unwrap().timescale;
		let timestamp = timestamp.convert(timescale).map_err(|_| Error::TimestampMismatch)?;
		if let Some(fin) = state.final_sequence
			&& sequence >= fin
		{
			return Err(Error::Closed);
		}
		state.max_sequence = Some(state.max_sequence.unwrap_or(0).max(sequence));
		meter.datagram(payload.len() as u64);
		state.push_datagram(Datagram {
			sequence,
			timestamp,
			payload,
		});
		let cache = state.cache.clone();
		drop(state);
		cache.settle(None);
		Ok(())
	}

	/// Create a group with a single frame, at the given presentation timestamp.
	///
	/// The timestamp is converted into the track's timescale. For data without
	/// a presentation time, pass [`Timestamp::now`] explicitly.
	pub fn write_frame<B: crate::IntoBytes>(&mut self, timestamp: Timestamp, frame: B) -> Result<()> {
		let frame = crate::IntoBytes::into_bytes(frame);
		if frame.len() as u64 > group::MAX_CACHE_BYTES {
			return Err(Error::FrameTooLarge);
		}
		let mut group = self.append_group()?;
		group.write_frame(timestamp, frame)?;
		group.finish()?;
		Ok(())
	}

	/// Mark the track as finished after the last appended group.
	///
	/// Sets the final sequence to one past the current max_sequence.
	/// No new groups at or above this sequence can be appended.
	/// NOTE: Old groups with lower sequence numbers can still arrive.
	pub fn finish(&self) -> Result<()> {
		let mut state = self.modify()?;
		let final_sequence = match state.max_sequence {
			Some(max) => max.checked_add(1).ok_or(coding::BoundsExceeded)?,
			None => 0,
		};
		state.set_final(final_sequence)
	}

	/// Declare the track's exclusive final sequence, possibly ahead of the live edge.
	///
	/// `final_sequence` is the first sequence that will never be produced, so a track
	/// whose last group is 89 finishes at `90`. Passing a boundary beyond the current
	/// max_sequence records a known ending before the remaining groups arrive (e.g.
	/// learning a track ends at group 89 while only 87 has been received). The boundary
	/// must be strictly greater than the highest produced group, otherwise it would
	/// orphan groups that already exist ([`Error::ProtocolViolation`]).
	///
	/// Groups below `final_sequence` may still be created afterwards; groups at or above
	/// it are rejected. Consumers only see end-of-stream once the live edge reaches the
	/// boundary. Use [`Self::finish`] to finish exactly at the live edge.
	pub fn finish_at(&mut self, final_sequence: u64) -> Result<()> {
		self.modify()?.set_final(final_sequence)
	}

	/// Declare the first group the live feed serves (the wire's SUBSCRIBE_START,
	/// or the start the subscription itself requested): groups below `sequence`
	/// will never arrive on their own, so a reader waiting for one fails over
	/// instead of stalling. A fetch can still retrieve them.
	///
	/// Scoped to the current subscription's demand, so a later declaration
	/// replaces this one in either direction: a re-subscription may start
	/// earlier, and a narrowed subscription skips groups an earlier declaration
	/// still promised. Pass `None` to clear it, for demand at the live edge:
	/// its floor is unknown until the feed declares one.
	pub fn start_at(&mut self, sequence: impl Into<Option<u64>>) -> Result<()> {
		self.modify()?.set_start(sequence.into());
		Ok(())
	}

	/// The declared first sequence of the live feed, once [`Self::start_at`]
	/// declared one. `None` while nothing was declared.
	#[cfg(test)]
	pub(crate) fn start_sequence(&self) -> Option<u64> {
		self.state.read().start_sequence
	}

	/// The exclusive final sequence, once [`Self::finish`] or [`Self::finish_at`] declared one.
	///
	/// `None` while the track is still open ended. Both methods reject a second boundary, so
	/// callers that may have already declared one check here first.
	pub fn final_sequence(&self) -> Option<u64> {
		self.state.read().final_sequence
	}

	/// Abort the track with the given error.
	///
	/// Consumes the handle, since nothing can be written to an aborted track. Drops the
	/// cached groups so a stale [`Consumer`] can't pin them (and their frame buffers) in
	/// memory forever. Consumers that haven't drained yet surface the abort error instead
	/// of the leftover cache. Child groups are independent: a consumer that already pulled
	/// a [`group::Consumer`] keeps its own handle and can finish reading it.
	///
	/// [`finish`](Self::finish) is deliberately not terminal: it declares the final
	/// sequence, and lower-numbered groups may still be written afterwards.
	pub fn abort(self, err: Error) -> Result<()> {
		commit_abort(self.modify()?, err);
		Ok(())
	}

	/// Abort an unused track, returning the producer unchanged if consumers remain.
	///
	/// Consumer creation and the unused check share a lock, so demand returning
	/// after [`poll_unused`](Self::poll_unused) prevents the abort. `Ok(())` means
	/// the track is closed, including when it was already closed; existing handles
	/// may still observe its final state. `Err(producer)` leaves the track unchanged
	/// so the caller can continue serving and wait for the next unused wake.
	#[expect(
		clippy::result_large_err,
		reason = "return the owned producer without allocating on an idle check"
	)]
	pub fn abort_unused(self, err: Error) -> std::result::Result<(), Self> {
		match self.state.write_unused() {
			kio::Unused::Idle(guard) => {
				commit_abort(guard, err);
				return Ok(());
			}
			kio::Unused::Closed => return Ok(()),
			kio::Unused::Used => {}
		}
		Err(self)
	}

	/// Whether the track is open and anyone is consuming it right now.
	///
	/// A point-in-time snapshot for gating work on demand (on-demand capture,
	/// dropping cached state nobody is watching). Acting on it to *end* the track
	/// is the race [`abort_unused`](Self::abort_unused) exists for; use
	/// [`unused`](Self::unused) to wait for the edge.
	pub fn is_used(&self) -> bool {
		!self.is_closed() && self.state.is_used()
	}

	/// Block until there are no active consumers.
	pub async fn unused(&self) -> Result<()> {
		self.state.unused().await.map_err(|_| self.abort_reason())
	}

	/// Block until there is at least one active consumer.
	pub async fn used(&self) -> Result<()> {
		self.state.used().await.map_err(|_| self.abort_reason())
	}

	/// Block until the track is closed or aborted, returning the cause.
	pub async fn closed(&self) -> Error {
		kio::wait(|waiter| self.poll_closed(waiter)).await
	}

	/// Poll until the track is closed or aborted; ready with the cause.
	pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<Error> {
		self.state.poll_closed(waiter).map(|()| self.abort_reason())
	}

	/// The recorded abort reason, or [`Error::Dropped`] if the track closed without one.
	fn abort_reason(&self) -> Error {
		self.state.read().abort.clone().unwrap_or(Error::Dropped)
	}

	/// Return true if the track has been closed.
	pub fn is_closed(&self) -> bool {
		self.state.read().is_closed()
	}

	/// Return the latest sequence number successfully appended to the track.
	pub fn latest(&self) -> Option<u64> {
		self.state.read().max_sequence
	}

	/// Return true if this is the same track.
	pub fn is_clone(&self, other: &Self) -> bool {
		self.state.same_channel(&other.state)
	}

	/// Create a weak reference that doesn't prevent auto-close.
	pub(crate) fn weak(&self) -> TrackWeak {
		TrackWeak {
			name: self.name.clone(),
			state: self.state.weak(),
		}
	}

	/// Create a [`Demand`]: a cloneable, watch-only handle to this track's
	/// subscriber demand.
	///
	/// Lets a publisher gate work (e.g. on-demand capture) on whether anyone is
	/// subscribed, without the ability to publish frames or close the track. The
	/// handle is weak, so holding one neither keeps the track alive nor pins its
	/// cached groups.
	pub fn demand(&self) -> Demand {
		Demand {
			name: self.name.clone(),
			state: self.state.weak(),
		}
	}

	/// Get a consumer handle for this in-process track.
	///
	/// Unlike a wire subscription, the info is already known, so a subscription
	/// opened from this handle resolves immediately.
	pub fn consume(&self) -> Consumer {
		Consumer::plain(self.name.clone(), self.state.consume())
	}

	/// Subscribing to this in-process track, resolving synchronously.
	///
	/// The info is fixed at creation, so there's nothing to wait for (no
	/// SUBSCRIBE_OK round trip). Pass `None` for [`Subscription::default`].
	///
	/// The read cursor starts at the group the subscription named (its floor), or 0.
	/// [`Subscription::max_age`] is what asks for data: delivery skips everything above
	/// the floor that the budget convicts, so the default budget of zero delivers only
	/// the latest group and a larger one reaches back over what it can still use.
	pub fn subscribe(&self, subscription: impl Into<Option<Subscription>>) -> Subscriber {
		let preferences = subscription.into().unwrap_or_default();

		// Info is fixed at creation and survives a close/abort, so read it without
		// requiring a live producer state. If the track already ended, the returned
		// subscriber surfaces the close/abort on its first read; the preferences are
		// simply never registered (nothing aggregates them anymore).
		let info = self.info.clone();
		let min_sequence = floor_of(&preferences);
		let subscription = kio::Producer::new(preferences);
		register_subscription(self.state.read(), &subscription);
		let drift_cap = kio::Producer::new(None);

		// Hoisted: an inline `read()` guard would live to the end of the struct literal,
		// deadlocking against the `consume()` below.
		let broadcast = self.state.read().broadcast.clone();
		Subscriber {
			name: self.name.clone(),
			broadcast,
			info,
			inner: SubscriberKind::Plain(PlainSubscriber {
				state: self.state.consume(),
				subscription,
				min_sequence,
				index: 0,
				datagram_index: 0,
				next_sequence: 0,
				end_sequence: None,
				parked: BTreeMap::new(),
				stale_cap: None,
				drift_cap,
				stale: stats::Content::default(),
				seek_pending: BTreeMap::new(),
			}),
			// A producer-side (in-process) subscribe is not egress: stay untagged.
			stats: stats::Scope::default(),
			_stats_sub: stats::Subscription::default(),
		}
	}

	/// Block until the aggregate subscription changes, then return the new value.
	///
	/// Yields the most demanding request across all live subscribers, or `None`
	/// once the last one drops. Used by relays to forward downstream demand
	/// upstream (e.g. SUBSCRIBE_UPDATE).
	pub async fn subscription_changed(&mut self) -> Result<Option<Subscription>> {
		kio::wait(|waiter| self.poll_subscription_changed(waiter)).await
	}

	/// A non-blocking snapshot of the current aggregate subscription, or `None`
	/// when there are no live subscribers. Unlike [`Self::subscription`], this
	/// doesn't wait for a change or advance the change cursor.
	///
	/// The aggregate's [`Subscription::max_age`] is clamped to this track's
	/// [`Info::max_age`]: no subscriber can wait for a late group longer than the
	/// publisher keeps it.
	pub fn subscription(&self) -> Option<Subscription> {
		let state = self.state.read();
		let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
		drop(state);
		snapshot_subscription(&subs, bound)
	}

	/// Poll counterpart to [`subscription_changed`](Self::subscription_changed): the
	/// aggregate subscription whenever it changes, or `None` once nobody is subscribed.
	/// Errors once the track is aborted.
	pub fn poll_subscription_changed(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Subscription>>> {
		// Surface an abort as the stream ending. `poll_closed` parks on the closed
		// waiters, so per-group churn on the track state never wakes this poll.
		if self.state.poll_closed(waiter).is_ready() {
			let abort = self.state.read().abort.clone();
			return Poll::Ready(Err(abort.unwrap_or(Error::Dropped)));
		}

		// Read the bound before locking `subs`, so the aggregation never nests the two locks.
		let state = self.state.read();
		let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
		drop(state);

		let prev = &self.prev_subscription;
		let mut combined = None;
		let mut guard = ready!(subs.poll(waiter, |subs| {
			let next = combined_subscription(subs, bound, waiter);
			if &next == prev {
				Poll::Pending
			} else {
				combined = next;
				Poll::Ready(())
			}
		}));
		// The aggregate changed: prune any closed subscribers now that we hold the lock.
		guard.retain(|sub| !sub.is_closed());
		drop(guard);
		self.prev_subscription = combined.clone();
		Poll::Ready(Ok(combined))
	}

	/// Poll for the producer becoming unused (every consumer dropped).
	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
		self.state.poll_unused(waiter).map(|used| match used {
			Some(()) => Ok(()),
			None => Err(self.abort_reason()),
		})
	}

	/// Create a [`Dynamic`] handle that serves on-demand fetches of uncached
	/// (old) groups. Most producers never need this; a relay creates one to fetch
	/// past groups from upstream.
	pub fn dynamic(&self) -> Dynamic {
		Dynamic::new(self.name.clone(), self.state.clone(), self.alive.clone())
	}

	fn modify(&self) -> Result<kio::Mut<'_, TrackState>> {
		TrackState::modify(&self.state)
	}
}

/// Pop the next queued group fetch off the fetch queue and wrap it in a
/// [`group::Request`] bound to a fresh producer handle. Shared by every
/// [`Dynamic`] handle on the track.
fn poll_requested_group(
	state: &kio::Producer<TrackState>,
	fetch: &kio::Shared<FetchState>,
	waiter: &kio::Waiter,
) -> Poll<Result<group::Request>> {
	// Prefer serving a queued fetch, even if the track has since aborted.
	if let Poll::Ready(mut guard) = fetch.poll(waiter, |fetch| {
		if fetch.has_queued() {
			Poll::Ready(())
		} else {
			Poll::Pending
		}
	}) {
		let sequence = guard.pop().expect("predicate guaranteed a request");
		// The popped attempt stays pending, so a fetch in the window between hand-off
		// and accept joins it instead of queueing a duplicate.
		// `group::Request::{accept, reject, drop}` removes the entry.
		let pending = guard.get(&sequence).expect("popped key must be pending");
		let priority = pending.priority;
		let frame_start = pending.frame_start;
		let result = pending.result.clone();
		drop(guard);
		return Poll::Ready(Ok(group::Request {
			state: state.clone(),
			fetch: fetch.clone(),
			sequence,
			priority,
			frame_start,
			result,
			done: false,
		}));
	}

	// No fetch queued: surface a track abort so the handler loop can exit.
	match state.poll_ref(waiter, |state| match &state.abort {
		Some(err) => Poll::Ready(err.clone()),
		None => Poll::Pending,
	}) {
		Poll::Ready(Ok(err)) => Poll::Ready(Err(err)),
		Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))),
		Poll::Pending => Poll::Pending,
	}
}

/// Serves on-demand fetches of uncached (old) groups for a track, the group-level
/// analogue of [`broadcast::Dynamic`].
///
/// Most tracks never serve old content, so this capability lives on a dedicated
/// handle rather than [`Producer`]: a relay creates one (via
/// [`Producer::dynamic`] or [`Request::dynamic`]) to pull past groups
/// from upstream. While at least one is alive the track will block a cache-miss
/// [`Consumer::fetch_group`] waiting to be served; with none, an accepted track's
/// miss fails fast with [`Error::NotFound`].
pub struct Dynamic {
	name: Arc<str>,
	// Kept to insert served groups into the cache and observe track abort.
	state: kio::Producer<TrackState>,
	// The fetch queue this handle drains; its `dynamic` count gates `fetch_group`.
	fetch: kio::Shared<FetchState>,
	// Shared with the track's producers: a handler still serving fetches keeps the
	// track alive, like a producer clone does.
	alive: Arc<Alive>,
}

impl Dynamic {
	fn new(name: Arc<str>, state: kio::Producer<TrackState>, alive: Arc<Alive>) -> Self {
		let fetch = state.read().fetch.clone();
		fetch.lock().add_handler();
		Self {
			name,
			state,
			fetch,
			alive,
		}
	}

	/// The track's name, unique within its broadcast.
	pub fn name(&self) -> &str {
		&self.name
	}

	/// Block until a consumer fetches a group that isn't cached, returning a
	/// [`group::Request`] to serve via [`group::Request::accept`].
	///
	/// A relay issues a wire FETCH first; an origin already has the group cached, so
	/// the fetch resolves without ever reaching here. Errors once the track is aborted.
	pub async fn requested_group(&self) -> Result<group::Request> {
		kio::wait(|waiter| self.poll_requested_group(waiter)).await
	}

	/// Poll counterpart to [`requested_group`](Self::requested_group).
	pub fn poll_requested_group(&self, waiter: &kio::Waiter) -> Poll<Result<group::Request>> {
		poll_requested_group(&self.state, &self.fetch, waiter)
	}

	/// Poll for the track becoming unused (every consumer dropped).
	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
		self.state.poll_unused(waiter).map(|_| ())
	}
}

impl Clone for Dynamic {
	fn clone(&self) -> Self {
		// Count each live handle (mirrors `broadcast::Dynamic`).
		self.fetch.lock().add_handler();
		Self {
			name: self.name.clone(),
			state: self.state.clone(),
			fetch: self.fetch.clone(),
			alive: self.alive.clone(),
		}
	}
}

impl Drop for Dynamic {
	fn drop(&mut self) {
		// Unlike `broadcast::Dynamic`, dropping the last handle doesn't abort the track:
		// a live `Producer` may still be serving the subscription. It just stops fetch
		// serving. Queued attempts no handler will ever pop are dropped, closing their
		// result channels so every joined `Fetching` resolves NotFound; an attempt
		// already handed to a handler stays, resolved by its `group::Request` instead.
		let mut fetch = self.fetch.lock();
		if fetch.remove_handler() {
			fetch.drain_queued();
		}
	}
}

/// Ends the track when the last [`Producer`] or [`Dynamic`] drops.
///
/// A refcount rather than a "am I the last one?" check inside `Drop`: that answer is a
/// snapshot, and acting on it is exactly what invalidates it. The track state's own
/// producer count can't answer it either, since a group settling its eviction debt
/// upgrades the account's weak handle and counts there for the duration (see
/// [`cache::Track::settle`]). Holding a producer of its own also keeps the state
/// writable until the teardown has run, whatever order the last owner's fields drop in.
struct Alive {
	name: Arc<str>,
	state: kio::Producer<TrackState>,

	// Set when a `Producer` is first minted, so a `Request` nobody accepted (its
	// `Dynamic` holds this guard too) isn't reported as an abandoned publisher.
	published: AtomicBool,

	// Ingress subscription for this track, opened by the tagged producer that claimed
	// it and closed when this guard drops.
	stats: OnceLock<stats::Subscription>,
}

impl Alive {
	fn new(name: Arc<str>, state: kio::Producer<TrackState>) -> Arc<Self> {
		Arc::new(Self {
			name,
			state,
			published: Default::default(),
			stats: Default::default(),
		})
	}

	/// Note that a [`Producer`] was minted from this track, optionally under a tagged
	/// broadcast's ingress scope (counted as one subscription for as long as the track
	/// has a publisher).
	fn publish(&self, stats: Option<&stats::Scope>) {
		self.published.store(true, Ordering::Relaxed);
		if let Some(scope) = stats {
			// At most one scope ever arrives: a track is minted either through
			// `Producer::new` (+ `with_stats`) or through `Request::accept`, never both.
			let _ = self.stats.set(scope.subscribe());
		}
	}
}

impl Drop for Alive {
	fn drop(&mut self) {
		// A request nobody accepted was never publishing; there's nothing to tear down.
		if !self.published.load(Ordering::Relaxed) {
			return;
		}
		// The last producer going away without finishing is an abrupt teardown:
		// release the cached groups so a stale consumer can't pin them (and their
		// frame buffers) forever, the same as an explicit abort. A cleanly
		// finished track keeps its cache so consumers can still drain it.
		// `abort()` closes the channel, so `write()` returns `Err(Ref)`. `finish()`
		// leaves it open with `final_sequence` set, so inspect both outcomes.
		match self.state.write() {
			Ok(mut state) => {
				if state.final_sequence.is_some() || state.abort.is_some() {
					return;
				}
				tracing::warn!(
					track = %self.name,
					"track::Producer dropped without finish() or abort()"
				);
				// See `abort`: keep the frame boundary once its groups go away.
				state.resume = state.resume_position();
				state.clear_cache();
				state.datagrams.clear();
			}
			Err(state) => {
				if state.final_sequence.is_some() || state.abort.is_some() {
					return;
				}
				tracing::warn!(
					track = %self.name,
					"track::Producer dropped without finish() or abort()"
				);
			}
		}
	}
}

/// Aggregate every live subscriber's preferences into the most demanding request.
///
/// Read-only: iterates the subscriptions immutably and registers `waiter` on each, so a
/// preference update (or a subscriber dropping) wakes the caller's poll. Callers decide
/// readiness from the returned value, then prune closed subscribers through the `Mut`.
fn combined_subscription(subs: &Subscriptions, bound: Option<Duration>, waiter: &kio::Waiter) -> Option<Subscription> {
	let mut combined = None;
	for sub in subs.iter() {
		// A closed consumer means the subscriber dropped: it holds no live demand.
		// `Consumer::poll` evaluates the closure before the closed flag, so it would
		// still replay the final value into the aggregate; skip it explicitly so a
		// departed subscriber can't keep the aggregate pinned to its last request.
		if sub.is_closed() {
			continue;
		}
		// Arm both waiters explicitly. `poll` registers on the value channel only
		// when it returns Pending, and a subscriber that contributes demand (always
		// the case for the first one) folds as Ready, so nothing would watch for its
		// departure or its next update: the last one leaving would never wake this
		// poll, and a reader lifting the cap it had set would leave the publisher's
		// upstream parked at that cap for good.
		let _ = sub.poll_closed(waiter);
		let _ = sub.poll(waiter, |_| Poll::<()>::Pending);
		if let Poll::Ready(merged) = sub.read().poll_combined(&combined) {
			combined = Some(merged);
		}
	}
	clamp_combined(combined, bound)
}

/// A non-blocking aggregate of the current subscriptions, without arming any waiter.
fn snapshot_subscription(subs: &kio::Shared<Subscriptions>, bound: Option<Duration>) -> Option<Subscription> {
	let mut combined: Option<Subscription> = None;
	for sub in subs.read().iter() {
		// Skip dropped subscribers, matching `combined_subscription`.
		if sub.is_closed() {
			continue;
		}
		if let Poll::Ready(merged) = sub.read().poll_combined(&combined) {
			combined = Some(merged);
		}
	}
	clamp_combined(combined, bound)
}

/// The highest sequence this subscriber could actually be handed: its read cursor's cap
/// ([`Subscriber::set_groups`]) and any cap imposed from outside, whichever is lower.
///
/// Deliberately not [`Subscription::end`]. That is a *request to the publisher*, folded
/// in with every other subscriber's, and it does not filter this handle (see
/// [`Subscriber`]): another unbounded subscriber widens the aggregate and the groups
/// arrive here anyway. Capping the drift anchor with it would pin the live edge at the
/// requested end while delivery ran past it, and everything above would then have nothing
/// newer to be late against.
///
/// `outer` is what a reader wrapping this cursor imposes. A spliced segment is
/// deliberately not given an inner cursor cap (it would park boundary-crossing groups out
/// of sight), so its reader passes down its own cap and the segment boundary that way.
/// Without it, a segment would anchor drift on groups it can never hand over.
fn servable_cap(cursor: Option<u64>, outer: Option<u64>) -> Option<u64> {
	super::subscription::min_some(cursor, outer)
}

/// The read cursor's floor: the group the subscription named, or 0 (no floor).
///
/// A floor is the only thing a start contributes; [`Subscription::max_age`] is what asks
/// for data. Delivery walks everything at or above the floor and skips what the budget
/// convicts, so a zero budget (the default) delivers only the live edge, a larger one
/// reaches back over what it can still use, and a floor above the live edge simply waits
/// there (a resumed subscription naming where it left off). One bound decides both what
/// is sent and what is expired, so the two cannot disagree.
fn floor_of(subscription: &Subscription) -> u64 {
	subscription.start.map(|start| start.group).unwrap_or(0)
}

/// Clamp a drift budget to the publisher's retention window: nobody can wait for a late
/// group longer than the publisher keeps it around.
///
/// The single clamp point. Subscribers hold their preferences verbatim, so what they asked
/// for stays readable, and it is applied here on both sides of the aggregate: to the
/// combined request the publisher sees ([`clamp_combined`]) and to one subscriber's own
/// budget when it decides a group is stale ([`TrackState::is_stale`]). Those agree because
/// `min` distributes over the `max` that combines them. `bound` is `None` on a track whose
/// info isn't known yet (an unaccepted [`Request`]), which imposes no window.
fn clamp_max_age(mut max_age: Duration, bound: Option<Duration>) -> Duration {
	if let Some(bound) = bound {
		max_age = max_age.min(bound);
	}
	max_age
}

/// Clamp the aggregate's max age budget to the publisher's window; see [`clamp_max_age`].
fn clamp_combined(combined: Option<Subscription>, bound: Option<Duration>) -> Option<Subscription> {
	let mut combined = combined?;
	combined.max_age = clamp_max_age(combined.max_age, bound);
	Some(combined)
}

/// Register a subscription if the track is live: clone the shared list out of the
/// state, release the track lock, then push under the list's own lock. A closed
/// track skips the push; nothing aggregates the preferences anymore.
fn register_subscription(state: kio::Ref<'_, TrackState>, subscription: &kio::Producer<Subscription>) {
	if state.is_closed() {
		return;
	}
	let subs = state.subscriptions.clone();
	drop(state);
	subs.lock().push(subscription.consume());
}

/// A weak reference to a track that doesn't prevent auto-close.
#[derive(Clone)]
pub(crate) struct TrackWeak {
	name: Arc<str>,
	state: kio::ProducerWeak<TrackState>,
}

impl TrackWeak {
	/// A [`Consumer`] for the cached track, or `None` once it has closed.
	///
	/// The count moves under the same lock the close takes, which is the other half
	/// of [`Producer::abort_unused`]: a lookup either gets a consumer in time to
	/// decline the teardown, or gets nothing and re-requests the track. It is never
	/// handed a track that is already on its way out.
	pub fn try_consume(&self) -> Option<Consumer> {
		Some(Consumer::plain(self.name.clone(), self.state.try_consume()?))
	}

	/// The shared name handle, for use as a broadcast lookup key (clone is a
	/// refcount bump, and the same `Arc` is shared with the track's handles).
	pub(crate) fn name(&self) -> &Arc<str> {
		&self.name
	}

	/// Reject a track nothing ever served, resolving its pending subscribes with `err`.
	///
	/// A track whose [`Producer`] was minted is left alone and this returns false;
	/// so is one that already carries an abort reason. Fetched backfill can install
	/// [`Info`] before acceptance, so metadata alone does not prove a publisher exists.
	///
	/// Closes the state like [`Producer::abort`], so a [`Request`] still held by the
	/// publisher can't `accept` its way back to life afterwards.
	pub(crate) fn reject(&self, err: Error) -> bool {
		let Some(producer) = self.state.produce() else {
			return false;
		};
		let Ok(mut state) = producer.write() else {
			return false;
		};
		if state.published || state.abort.is_some() {
			return false;
		}
		state.abort = Some(err);
		state.close();
		true
	}

	/// Whether anyone is consuming the track right now. A closed track doesn't
	/// count even if consumers linger to drain its cache: no new work is owed.
	pub(crate) fn is_used(&self) -> bool {
		!self.state.is_closed() && self.state.is_used()
	}

	/// Park `waiter` for the next consumer appearing; a no-op once one exists.
	/// Feeds [`crate::broadcast::Demand`], which recomputes on wake.
	pub(crate) fn poll_used(&self, waiter: &kio::Waiter) {
		let _ = self.state.poll_used(waiter);
	}

	/// Park `waiter` for the last consumer (or the track) going away; a no-op
	/// once none remain. Feeds [`crate::broadcast::Demand`].
	pub(crate) fn poll_unused(&self, waiter: &kio::Waiter) {
		let _ = self.state.poll_unused(waiter);
	}
}

impl super::WeakEntry for TrackWeak {
	fn is_closed(&self) -> bool {
		self.state.is_closed()
	}

	fn same_channel(&self, other: &Self) -> bool {
		self.state.same_channel(&other.state)
	}
}

/// A cloneable, watch-only handle to a track's subscriber demand.
///
/// Obtained from [`Producer::demand`]. A publisher uses it to react to
/// whether anyone is subscribed (on-demand capture / encoding) without being able
/// to publish frames or close the track. It's a weak handle, so it neither keeps
/// the track alive nor pins its cached groups; once the owning [`Producer`]
/// goes away, [`used`](Self::used) / [`unused`](Self::unused) report the track's
/// closure.
#[derive(Clone)]
pub struct Demand {
	name: Arc<str>,
	state: kio::ProducerWeak<TrackState>,
}

impl Demand {
	/// The track name this handle is bound to.
	pub fn name(&self) -> &str {
		&self.name
	}

	/// Block until there is at least one active consumer.
	pub async fn used(&self) -> Result<()> {
		self.state.used().await.map_err(|_| self.abort_reason())
	}

	/// Block until there are no active consumers.
	pub async fn unused(&self) -> Result<()> {
		self.state.unused().await.map_err(|_| self.abort_reason())
	}

	/// Block until the track is closed or aborted, returning the cause.
	pub async fn closed(&self) -> Error {
		self.state.closed().await;
		self.abort_reason()
	}

	/// The publisher's tie-break priority, as set in [`Info::priority`].
	pub(crate) fn priority(&self) -> u8 {
		// Always Some once the track exists; a closed one reads its last value.
		self.state.read().info.as_ref().map_or(0, |info| info.priority)
	}

	/// Whether anyone is subscribed right now, without waiting.
	pub fn is_used(&self) -> bool {
		self.state.is_used()
	}

	/// Poll-based variant of [`Self::used`].
	pub fn poll_used(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
		self.state.poll_used(waiter).map(|used| match used {
			Some(()) => Ok(()),
			None => Err(self.abort_reason()),
		})
	}

	/// Poll-based variant of [`Self::unused`].
	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
		self.state.poll_unused(waiter).map(|used| match used {
			Some(()) => Ok(()),
			None => Err(self.abort_reason()),
		})
	}

	/// Whether the track is gone, without waiting.
	pub(crate) fn is_closed(&self) -> bool {
		self.state.is_closed()
	}

	/// Read the current demand and arm `waiter` for the next transition.
	///
	/// Level-triggered, unlike [`used`](Self::used) / [`unused`](Self::unused): it
	/// answers what the demand is *now* and wakes on whichever way it can move
	/// next, so a poll-driven caller watching several tracks at once doesn't have
	/// to track which edge it's waiting for.
	pub(crate) fn poll_state(&self, waiter: &kio::Waiter) -> DemandState {
		loop {
			match self.state.poll_used(waiter) {
				Poll::Ready(None) => return DemandState::Closed,
				// Not used, and armed for it becoming used.
				Poll::Pending => return DemandState::Idle,
				// Used, so arm for the reverse.
				Poll::Ready(Some(())) => match self.state.poll_unused(waiter) {
					Poll::Ready(None) => return DemandState::Closed,
					Poll::Pending => return DemandState::Active,
					// Went idle between the two reads, so neither poll armed anything.
					// Start over rather than returning a state with no waker behind it.
					Poll::Ready(Some(())) => continue,
				},
			}
		}
	}

	/// The recorded abort reason, or [`Error::Dropped`] if the track closed without one.
	fn abort_reason(&self) -> Error {
		self.state.read().abort.clone().unwrap_or(Error::Dropped)
	}
}

/// What [`Demand::poll_state`] found.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub(crate) enum DemandState {
	/// At least one subscriber.
	Active,
	/// No subscribers, but the track is still open.
	Idle,
	/// The track is gone; it will never be demanded again.
	Closed,
}

/// A handle to a single track within a broadcast.
///
/// Obtained from [`broadcast::Consumer::track`]. Holding it sends nothing
/// to the publisher; it just names a track you can [`subscribe`](Self::subscribe)
/// to (a live, ongoing stream of groups) later. The same handle can be subscribed
/// to multiple times, and clones are cheap.
///
/// A track reached through a route-fed broadcast is *spliced*: it is backed by one
/// or more per-session tracks joined at group boundaries, and this handle reads
/// across them transparently.
#[derive(Clone)]
pub struct Consumer {
	name: Arc<str>,
	// The broadcast this track belongs to, so a catalog track can name the path its
	// relative references resolve against. Rebound by `broadcast::Consumer::track` to
	// that handle's view, which may name the broadcast differently than its producer did.
	broadcast: Arc<broadcast::Info>,
	inner: ConsumerKind,
	// Egress stats scope, set by a tagged [`broadcast::Consumer`] via
	// [`Self::with_stats`]. Empty (no-op) for an untagged track.
	stats: stats::Scope,
}

#[derive(Clone)]
enum ConsumerKind {
	Plain(kio::Consumer<TrackState>),
	Spliced(super::resume::Consumer),
}

impl Consumer {
	fn plain(name: Arc<str>, state: kio::Consumer<TrackState>) -> Self {
		let broadcast = state.read().broadcast.clone();
		Self {
			name,
			broadcast,
			inner: ConsumerKind::Plain(state),
			stats: stats::Scope::default(),
		}
	}

	/// A consumer over a spliced logical track (a route-fed broadcast's track).
	pub(crate) fn spliced(name: Arc<str>, broadcast: Arc<broadcast::Info>, resume: super::resume::Consumer) -> Self {
		Self {
			name,
			broadcast,
			inner: ConsumerKind::Spliced(resume),
			stats: stats::Scope::default(),
		}
	}

	/// Attach an egress stats scope, inherited by the subscriptions, fetches, and
	/// groups derived from this handle. Called by a tagged [`broadcast::Consumer`].
	pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self {
		self.stats = scope;
		self
	}

	/// Rebind the track to the broadcast handle it was reached through, so
	/// [`broadcast`](Self::broadcast) reports the path that handle was handed out at (see
	/// [`broadcast::Info::path`]). Called by [`broadcast::Consumer::track`].
	pub(crate) fn with_broadcast(mut self, broadcast: Arc<broadcast::Info>) -> Self {
		self.broadcast = broadcast;
		self
	}

	/// Groups this copy still holds, so an origin can keep them after dropping the source.
	///
	/// Publisher-produced groups come back in arrival order, matching what
	/// `recv_group` would deliver; fetched backfill absent from arrival follows
	/// in sequence order for sequence fetches.
	pub(crate) fn cached_groups(&self) -> Vec<(group::Producer, bool)> {
		match &self.inner {
			ConsumerKind::Plain(state) => {
				let state = state.read();
				let mut out = Vec::with_capacity(state.lookup.len());
				for (sequence, stamp) in state.arrival.iter() {
					if let Some(slot) = state.lookup.get(sequence)
						&& slot.stamp == *stamp
						&& !slot.group.is_aborted()
					{
						out.push((slot.group.clone(), slot.visible));
					}
				}
				// Fetched backfill never enters arrival; keep it for sequence fetches.
				let mut copied: HashSet<u64> = out.iter().map(|(group, _)| group.sequence).collect();
				for (sequence, slot) in state.lookup.iter() {
					if !slot.group.is_aborted() && copied.insert(*sequence) {
						out.push((slot.group.clone(), slot.visible));
					}
				}
				out
			}
			ConsumerKind::Spliced(resume) => resume.cached_groups(),
		}
	}

	/// A cached group that contains every frame from `frame_start` onward.
	///
	/// Unlike [`Self::peek_group`], this is suitable for satisfying a FETCH: a
	/// partial cached copy is a miss so the caller can ask upstream for its head.
	pub(crate) fn cached_group(&self, sequence: u64, frame_start: u64) -> Option<group::Consumer> {
		match &self.inner {
			ConsumerKind::Plain(state) => {
				let state = state.read();
				let group = state.covering_group(sequence, frame_start)?;
				group.cache_refresh();
				Some(group.consume())
			}
			ConsumerKind::Spliced(resume) => resume.cached_group(sequence, frame_start),
		}
	}

	/// Publisher properties already resolved on this copy, if any.
	pub(crate) fn cached_info(&self) -> Option<Info> {
		match &self.inner {
			ConsumerKind::Plain(state) => state.read().info.clone(),
			ConsumerKind::Spliced(resume) => resume.cached_info(),
		}
	}

	/// The track name this handle is bound to.
	pub fn name(&self) -> &str {
		&self.name
	}

	/// The broadcast this track belongs to, as reached through the handle it came from.
	/// Its [`path`](broadcast::Info::path) is what a catalog's relative `broadcast`
	/// references resolve against.
	pub fn broadcast(&self) -> &broadcast::Info {
		&self.broadcast
	}

	/// Open a live subscription.
	///
	/// Registers the subscription on the track and returns a [`kio::Pending`] that resolves to the
	/// [`Subscriber`] once the track info is available, or the track's abort error (or
	/// [`Error::Dropped`]) if it is already closed.
	///
	/// The read cursor starts at the group the subscription named (its floor), or 0.
	/// [`Subscription::max_age`] is what asks for data: delivery skips everything above
	/// the floor that the budget convicts, so the default budget of zero delivers only
	/// the latest group and a larger one reaches back over what it can still use.
	pub fn subscribe(&self, subscription: impl Into<Option<Subscription>>) -> kio::Pending<Subscribing> {
		let subscription = kio::Producer::new(subscription.into().unwrap_or_default());

		let inner = match &self.inner {
			ConsumerKind::Plain(state) => {
				// Register the subscription if the track is live. If it is already closed, the
				// returned future resolves to the abort error via `Subscribing::poll_ok`.
				register_subscription(state.read(), &subscription);
				SubscribingKind::Plain(state.clone())
			}
			// A spliced subscription registers per segment once the subscriber polls.
			ConsumerKind::Spliced(resume) => SubscribingKind::Spliced(resume.clone()),
		};

		kio::Pending::new(Subscribing {
			name: self.name.clone(),
			broadcast: self.broadcast.clone(),
			inner,
			subscription,
			stats: self.stats.clone(),
		})
	}

	/// The newest group, when it is already cached: resolved synchronously, without
	/// counting as a fetch or a delivery. The IETF publisher snapshots its frame count to
	/// resolve Largest Object; a group that is not immediately available reads as no edge.
	pub(crate) fn peek_latest(&self) -> Option<group::Consumer> {
		match &self.inner {
			ConsumerKind::Plain(state) => {
				let sequence = state.read().max_sequence?;
				self.peek_group(sequence)
			}
			ConsumerKind::Spliced(resume) => resume.peek_latest(),
		}
	}

	/// The nearest cached group below `sequence`, under the same terms as
	/// [`Self::peek_group`]. Walks the cache's own order, so gaps in the group numbering
	/// are crossed and aborted (evicted) entries are skipped.
	pub(crate) fn peek_before(&self, sequence: u64) -> Option<group::Consumer> {
		match &self.inner {
			ConsumerKind::Plain(state) => {
				let state = state.read();
				state
					.lookup
					.range(..sequence)
					.rev()
					.map(|(_, slot)| &slot.group)
					.find(|group| !group.is_aborted())
					.map(|group| group.consume())
			}
			ConsumerKind::Spliced(resume) => resume.peek_before(sequence),
		}
	}

	/// A cached group by sequence, under the same terms as [`Self::peek_latest`]. Unlike a
	/// fetch, a peek does not refresh the group's cache standing, so it never keeps a
	/// group alive over one a subscriber actually read; an aborted (evicted) group is a
	/// miss.
	pub(crate) fn peek_group(&self, sequence: u64) -> Option<group::Consumer> {
		match &self.inner {
			ConsumerKind::Plain(state) => {
				let state = state.read();
				let slot = state.lookup.get(&sequence)?;
				if slot.group.is_aborted() {
					return None;
				}
				Some(slot.group.consume())
			}
			ConsumerKind::Spliced(resume) => resume.peek_group(sequence),
		}
	}

	/// Attach a subscription's drift policy to a cached group resolved outside its cursor.
	pub(crate) fn guard_group(
		&self,
		group: group::Consumer,
		subscription: kio::Consumer<Subscription>,
		cap: kio::Consumer<Option<u64>>,
		bound: Option<u64>,
	) -> group::Consumer {
		let ConsumerKind::Plain(state) = &self.inner else {
			return group;
		};
		let sequence = group.sequence;
		group.with_expiry(Arc::new(GroupExpiry {
			state: state.weak(),
			subscription,
			cap,
			bound,
			sequence,
		}))
	}

	/// Poll for a cached group by sequence, parking `waiter` until it lands.
	///
	/// `Ready(None)` once it can never arrive: the track ended below the sequence, or
	/// closed. Unlike [`Self::fetch_group`] this never asks anyone to produce the group,
	/// so it only resolves for one a live subscription is already pulling. That is what
	/// the spliced reader in [`super::resume`] wants: it waits for the route serving a
	/// group to deliver it, rather than opening a redundant fetch alongside.
	pub(crate) fn poll_peek_group(&self, sequence: u64, waiter: &kio::Waiter) -> Poll<Option<group::Consumer>> {
		let ConsumerKind::Plain(state) = &self.inner else {
			// A segment's track is never itself spliced.
			return Poll::Pending;
		};

		let res = state.poll(waiter, |state| {
			match state.lookup.get(&sequence) {
				Some(slot) if !slot.group.is_aborted() => Poll::Ready(Some(slot.group.consume())),
				// An aborted slot is a hole this route can no longer fill.
				Some(_) => Poll::Ready(None),
				// Past the declared end, so it can never exist.
				None if state.final_sequence.is_some_and(|fin| sequence >= fin) => Poll::Ready(None),
				// Below the declared start, so the live feed skipped it for good.
				None if state.start_sequence.is_some_and(|start| sequence < start) => Poll::Ready(None),
				None => Poll::Pending,
			}
		});

		match res {
			Poll::Ready(Ok(res)) => Poll::Ready(res),
			// The track died; whatever it cached went with it.
			Poll::Ready(Err(_)) => Poll::Ready(None),
			Poll::Pending => Poll::Pending,
		}
	}

	/// Poll for a live cached copy of `sequence` that can serve frame `index`,
	/// parking until one exists.
	///
	/// Unlike [`Self::poll_peek_group`] this never gives a verdict: a missing
	/// group parks (registered for its arrival) even below the declared start,
	/// since demand may move backward and revive it. The spliced reader uses it
	/// to reconsider a route it gave up on, so it must be exact about what
	/// "available" means: a copy that cannot start at `index` (its head is gone)
	/// leaves the route buried rather than reviving it into a peek that would
	/// bury it again. That exactness is load-bearing, since the reader consults
	/// this ahead of its terminal checks: relaxing it to "the group exists" makes
	/// revive and re-bury alternate forever inside one poll
	/// (`resume::test::misaligned_copy_is_lost_without_spinning`, where the
	/// regression surfaces as a hang).
	pub(crate) fn poll_serving_group(&self, sequence: u64, index: u64, waiter: &kio::Waiter) -> Poll<()> {
		let ConsumerKind::Plain(state) = &self.inner else {
			// A segment's track is never itself spliced.
			return Poll::Pending;
		};
		let res = state.poll(waiter, |state| match state.lookup.get(&sequence) {
			Some(slot) if !slot.group.is_aborted() => {
				// `start_at` clamps up, so landing higher means the copy no longer
				// holds this position.
				let mut group = slot.group.consume();
				group.start_at(index);
				match group.index() == index {
					true => Poll::Ready(()),
					false => Poll::Pending,
				}
			}
			_ => Poll::Pending,
		});
		match res {
			Poll::Ready(Ok(())) => Poll::Ready(()),
			// The track died; whatever would arrive never will, and the caller's
			// terminal checks settle the wait.
			Poll::Ready(Err(_)) | Poll::Pending => Poll::Pending,
		}
	}

	/// Fetching a single past group, without holding a live subscription.
	///
	/// Returns a [`kio::Pending`] that resolves to the [`group::Consumer`]:
	/// immediately if the group is cached, otherwise once a [`Dynamic`] serves
	/// the request (a wire FETCH for a relay). `options` accepts `None`, a [`group::Fetch`],
	/// or `group::Fetch::default()`.
	///
	/// The returned future resolves to [`Error::NotFound`] when the group can never be served
	/// (past the final sequence, or no [`Dynamic`] on the track), or the track's abort error
	/// if it's already closed. Concurrent fetches for the same sequence coalesce onto one
	/// handler request.
	pub fn fetch_group(&self, sequence: u64, options: impl Into<Option<group::Fetch>>) -> kio::Pending<Fetching> {
		let options = options.into().unwrap_or_default();

		// One fetch per calling context, counted here (coalesced upstream work is
		// still one request served). Independent of `subscriptions` and the viewer
		// refcount.
		self.stats.fetch();

		let state = match &self.inner {
			ConsumerKind::Plain(state) => state,
			// Spliced: routed to the newest segment's (plain) track, waiting for a
			// segment to exist if no route has served the track yet.
			ConsumerKind::Spliced(resume) => {
				return kio::Pending::new(Fetching {
					inner: FetchingKind::Spliced(resume.fetch_group(sequence, options)),
					stats: self.stats.clone(),
				});
			}
		};

		let mut result = None;

		// Queue a request only when the group isn't already resolvable from the track
		// (cached, aborted, or past-final all resolve through `Fetching::poll` without
		// a queue entry).
		let (fetch, unresolved) = {
			let state = state.read();
			(
				state.fetch.clone(),
				state.poll_fetch_cached(sequence, options.frame_start).is_pending(),
			)
		};

		if unresolved {
			let mut fetch = fetch.lock();
			if let Some(pending) = fetch.join(&sequence) {
				// Join the in-flight attempt for this sequence (queued or already being
				// served): share its result channel, raising its priority if ours is higher
				// and widening its range if ours starts earlier.
				//
				// Widening only reaches the handler while the attempt is still queued;
				// once popped, the `group::Request` holds an immutable copy and its range is
				// already on the wire. What protects the late caller either way is the
				// coverage check above: it refuses a group starting above what it asked
				// for, so it fails cleanly and its retry queues a fresh attempt.
				pending.priority = pending.priority.max(options.priority);
				pending.frame_start = pending.frame_start.min(options.frame_start);
				result = Some(pending.result.consume());
			} else {
				// Queue a new attempt. The handler gate is atomic with a handler
				// dropping (no fetch stranded on a queue nobody drains); with no
				// handler, `Fetching::poll` fails fast instead.
				let producer = kio::Producer::<FetchOutcome>::default();
				let consumer = producer.consume();
				let attempt = PendingFetch {
					priority: options.priority,
					frame_start: options.frame_start,
					result: producer,
				};
				if fetch.insert(sequence, attempt).is_ok() {
					result = Some(consumer);
				}
			}
		}

		kio::Pending::new(Fetching {
			inner: FetchingKind::Plain {
				state: state.clone(),
				fetch,
				sequence,
				frame_start: options.frame_start,
				result,
			},
			stats: self.stats.clone(),
		})
	}

	/// Resolve the track's [`Info`] without subscribing.
	///
	/// A [`Consumer`] is a lazy handle, so the info may not be known yet: this waits
	/// for the producer to [`Request::accept`] the track (a wire TRACK_INFO round-trip
	/// for a relay), and errors with the track's abort error if it closes first.
	/// [`Subscriber::info`] is the already-resolved counterpart.
	pub fn query(&self) -> kio::Pending<Querying> {
		kio::Pending::new(Querying {
			inner: match &self.inner {
				ConsumerKind::Plain(state) => QueryingKind::Plain(state.clone()),
				ConsumerKind::Spliced(resume) => QueryingKind::Spliced(resume.clone()),
			},
		})
	}

	/// Return the latest group sequence in the track, or `None` before any group.
	pub fn latest(&self) -> Option<u64> {
		match &self.inner {
			ConsumerKind::Plain(state) => state.read().max_sequence,
			ConsumerKind::Spliced(resume) => resume.latest(),
		}
	}

	/// The frame-precise point a replacement route should resume from: one past the
	/// last frame this copy produced. `None` if it produced nothing.
	///
	/// Survives the track aborting, which is when a route change asks.
	pub(crate) fn resume_position(&self) -> Option<Position> {
		match &self.inner {
			ConsumerKind::Plain(state) => state.read().resume_position(),
			ConsumerKind::Spliced(resume) => resume.resume_position(),
		}
	}

	/// Poll for the track reaching a terminal state: `Ok(())` once it is complete
	/// (the final group was produced), `Err` once it closed or aborted before
	/// completing. The origin's dispatcher uses this to tell a track that truly
	/// ended from one whose serving route died mid-stream.
	pub(crate) fn poll_complete(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
		let ConsumerKind::Plain(state) = &self.inner else {
			// Spliced tracks are compositions; the dispatcher never monitors one.
			return Poll::Pending;
		};
		match ready!(state.poll(waiter, |state| {
			if state.is_complete() {
				Poll::Ready(())
			} else {
				Poll::Pending
			}
		})) {
			Ok(_) => Poll::Ready(Ok(())),
			// Closed before completing. Read through the returned guard: it holds
			// the lock, so re-locking the channel here would deadlock.
			Err(closed) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))),
		}
	}
}

/// The pollable state of a [`Consumer::subscribe`]; awaited via the
/// [`kio::Pending`] wrapper, whose `DerefMut` exposes [`Self::update`].
pub struct Subscribing {
	name: Arc<str>,
	broadcast: Arc<broadcast::Info>,
	inner: SubscribingKind,
	subscription: kio::Producer<Subscription>,
	stats: stats::Scope,
}

enum SubscribingKind {
	Plain(kio::Consumer<TrackState>),
	Spliced(super::resume::Consumer),
}

impl Subscribing {
	/// Poll until the peer confirms the subscription, yielding the [`Subscriber`].
	/// Errors if the track is aborted or not found.
	pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<Subscriber>> {
		match &self.inner {
			SubscribingKind::Plain(state) => {
				// Wait until the track info is available
				let info = ready!(state.poll(waiter, |state| state.poll_info()))
					.map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??;

				let drift_cap = kio::Producer::new(None);
				let min_sequence = floor_of(&self.subscription.read());
				Poll::Ready(Ok(Subscriber {
					name: self.name.clone(),
					broadcast: self.broadcast.clone(),
					info,
					inner: SubscriberKind::Plain(PlainSubscriber {
						state: state.clone(),
						subscription: self.subscription.clone(),
						min_sequence,
						index: 0,
						datagram_index: 0,
						next_sequence: 0,
						end_sequence: None,
						parked: BTreeMap::new(),
						stale_cap: None,
						drift_cap,
						stale: stats::Content::default(),
						seek_pending: BTreeMap::new(),
					}),
					stats: self.stats.clone(),
					_stats_sub: self.stats.subscribe(),
				}))
			}
			SubscribingKind::Spliced(resume) => {
				// Resolved from the first segment's track. The publisher's max age
				// window is applied to each per-session aggregate, not here.
				let info = ready!(resume.poll_info(waiter))?;

				Poll::Ready(Ok(Subscriber {
					name: self.name.clone(),
					broadcast: self.broadcast.clone(),
					info,
					inner: SubscriberKind::Spliced(Box::new(resume.subscribe_shared(self.subscription.clone()))),
					stats: self.stats.clone(),
					_stats_sub: self.stats.subscribe(),
				}))
			}
		}
	}

	/// Change the subscription preferences before (or after) it resolves.
	///
	/// Returns [`Error::Closed`] if the track already ended; the update is
	/// meaningless at that point and can usually be ignored.
	pub fn update(&mut self, subscription: Subscription) -> Result<()> {
		let mut state = self.subscription.write().map_err(|_| Error::Closed)?;
		*state = subscription;
		Ok(())
	}
}

impl kio::Pollable for Subscribing {
	type Output = Result<Subscriber>;

	fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
		self.poll_ok(waiter)
	}
}

/// The pollable state of a [`Consumer::query`]; awaited via the
/// [`kio::Pending`] wrapper.
pub struct Querying {
	inner: QueryingKind,
}

enum QueryingKind {
	Plain(kio::Consumer<TrackState>),
	Spliced(super::resume::Consumer),
}

impl Querying {
	/// Poll until the track's [`Info`] is known, without subscribing to its groups.
	pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<Info>> {
		match &self.inner {
			QueryingKind::Plain(state) => {
				// Wait until the track info is available
				let info = ready!(state.poll(waiter, |state| state.poll_info()))
					.map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??;
				Poll::Ready(Ok(info))
			}
			QueryingKind::Spliced(resume) => resume.poll_info(waiter),
		}
	}
}

impl kio::Pollable for Querying {
	type Output = Result<Info>;

	fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
		self.poll_ok(waiter)
	}
}

impl group::Request {
	/// The group sequence the consumer wants.
	pub fn sequence(&self) -> u64 {
		self.sequence
	}

	/// The delivery priority the consumer requested for this group.
	pub fn priority(&self) -> u8 {
		self.priority
	}

	/// The first frame of the group the consumer wants; 0 is the whole group.
	///
	/// A handler serving this must [`start_at`](group::Producer::start_at) it, so the frames
	/// it writes carry the indices they have in the group rather than restarting at 0.
	///
	/// There is no end: the handler fetches through the end of the group so the result
	/// is cacheable for anyone (see [`group::Fetch::frame_start`]).
	pub fn frame_start(&self) -> u64 {
		self.frame_start
	}

	/// Insert the fetched group into the track cache, resolving the waiting
	/// [`Consumer::fetch_group`], and return a [`group::Producer`] to fill.
	///
	/// The group's timescale comes from the track's [`Info`]. `info` sets that
	/// info if the track hasn't been accepted yet (a fetch with no live subscription),
	/// and is ignored once accepted. Returns [`Error::Duplicate`] if the group is
	/// already present, or the track's abort error if it closed while pending.
	pub fn accept(mut self, info: impl Into<Option<Info>>) -> Result<group::Producer> {
		self.done = true;
		// Cache the group before removing the attempt: the joined fetches resolve
		// through the cache, and removal closes their result channel (which alone
		// would read as NotFound).
		let res = TrackState::modify(&self.state)
			.and_then(|mut state| state.insert_group_request(self.sequence, self.frame_start, info.into()));
		self.remove();
		res
	}

	/// Reject the fetch, resolving every joined [`Consumer::fetch_group`] with `err`.
	pub fn reject(mut self, err: Error) {
		self.done = true;
		// Remove before writing, so a fetch arriving now starts a fresh attempt
		// instead of joining a rejected one.
		self.remove();
		if let Ok(mut outcome) = self.result.write() {
			outcome.rejected = Some(err);
		}
	}

	/// Remove this attempt from the fetch state, unless a newer attempt for the same
	/// sequence has already replaced it.
	fn remove(&self) {
		self.fetch
			.lock()
			.remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result));
	}
}

impl Drop for group::Request {
	fn drop(&mut self) {
		if self.done {
			return;
		}
		self.remove();
		if let Ok(mut outcome) = self.result.write() {
			outcome.rejected = Some(Error::Dropped);
		}
	}
}

/// The pollable state of a [`Consumer::fetch_group`].
///
/// Awaited via the [`kio::Pending`] wrapper; resolves to the
/// [`group::Consumer`] once the group lands in the track's cache (already present,
/// or produced after a wire FETCH), or [`Error::NotFound`] if it can never exist.
pub struct Fetching {
	inner: FetchingKind,
	// Egress stats scope, so the resolved group carries a payload meter (and counts
	// as one delivered group). Empty (no-op) for an untagged track.
	stats: stats::Scope,
}

enum FetchingKind {
	Plain {
		state: kio::Consumer<TrackState>,
		fetch: kio::Shared<FetchState>,
		sequence: u64,
		// This caller's own start, so a cached group that begins above it is a miss
		// rather than a short answer.
		frame_start: u64,
		// The joined attempt's result channel; `None` when no handler existed to queue on.
		result: Option<kio::Consumer<FetchOutcome>>,
	},
	/// A spliced track's fetch: waits for a segment, then fetches from it.
	Spliced(kio::Pending<super::resume::Fetching>),
}

impl kio::Pollable for Fetching {
	type Output = Result<group::Consumer>;

	fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
		let (state, fetch, sequence, frame_start, result) = match &self.inner {
			FetchingKind::Plain {
				state,
				fetch,
				sequence,
				frame_start,
				result,
			} => (state, fetch, *sequence, *frame_start, result.as_ref()),
			FetchingKind::Spliced(spliced) => {
				// A fetched group is metered here (once), at the tagged handle: the
				// spliced source track it comes from is the origin's own, untagged.
				return kio::Pollable::poll(&**spliced, waiter)
					.map(|res| res.map(|group| group.with_meter(self.stats.meter())));
			}
		};

		// Track side: the cached group, the abort error, or past-final. The outer
		// error is the channel closing without any of those.
		match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frame_start)) {
			Poll::Ready(Ok(res)) => {
				return Poll::Ready(res.map(|mut group| {
					// Hand back a consumer sitting where the caller asked rather than at
					// the group's own start. Coverage was checked above, so this can only
					// skip frames they excluded.
					group.start_at(frame_start);
					group.with_meter(self.stats.meter())
				}));
			}
			Poll::Ready(Err(closed)) => {
				return Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped)));
			}
			Poll::Pending => {}
		}

		// Handler side.
		let Some(result) = result else {
			// Never queued: no handler existed when the fetch was made. Fail fast while
			// that's still true; a handler that appeared since may yet fill the cache.
			return match fetch.poll(waiter, |fetch| match fetch.has_handlers() {
				false => Poll::Ready(()),
				true => Poll::Pending,
			}) {
				Poll::Ready(_guard) => Poll::Ready(Err(Error::NotFound)),
				Poll::Pending => Poll::Pending,
			};
		};

		// A written rejection fails every joined fetch. The channel closing without
		// one means the attempt was dropped unserved (its handlers went away).
		match result.poll(waiter, |outcome| match &outcome.rejected {
			Some(err) => Poll::Ready(err.clone()),
			None => Poll::Pending,
		}) {
			Poll::Ready(Ok(err)) => Poll::Ready(Err(err)),
			Poll::Ready(Err(_closed)) => Poll::Ready(Err(Error::NotFound)),
			Poll::Pending => Poll::Pending,
		}
	}
}

/// A live subscription to a track, used to read its groups.
///
/// Created via [`Consumer::subscribe`](Consumer::subscribe), or
/// directly from a [`Producer`] for an in-process track. Carries this
/// subscriber's [`Subscription`] preferences, which feed the producer's aggregate.
///
/// # Local cursor vs wire preference
///
/// Group bounds exist at two levels, and setting one does not imply the other:
///
/// - [`Self::set_groups`] limits **this subscriber's reads**, filtering exactly what
///   this handle returns without changing the publisher's demand.
/// - [`Subscription::start`] / [`Subscription::end`], set via [`Self::update`],
///   are a **request to the publisher**. They're aggregated across every live subscriber
///   (earliest start, widest end), so they say what the publisher should send, not what
///   this subscriber sees.
///
/// They stay separate because their scopes differ: a subscriber can't filter by the
/// aggregate, since another subscriber can widen it, and the publisher can't honor a
/// cursor it's never told about. So setting only the cursor still transfers the skipped
/// groups, and setting only the preference still returns groups another subscriber asked
/// for. Set both to skip them *and* avoid the transfer.
///
/// The one place they meet is where the cursor comes from. A new subscriber's cursor is
/// floored at the group its own subscription named (or 0), and its
/// [`Subscription::max_age`] decides what above the floor is worth delivering. Every later
/// move is the caller's.
pub struct Subscriber {
	name: Arc<str>,
	// The broadcast this track belongs to; see [`Self::broadcast`].
	broadcast: Arc<broadcast::Info>,
	info: Info,
	inner: SubscriberKind,
	// Egress stats scope, used to meter the groups this subscriber reads. Empty
	// (no-op) for an untagged track.
	stats: stats::Scope,
	// The subscription guard: bumps `subscriptions` (and the egress viewer refcount)
	// while held, closing them on drop. Empty (no-op) for an untagged track.
	_stats_sub: stats::Subscription,
}

enum SubscriberKind {
	Plain(PlainSubscriber),
	// Boxed: the spliced cursor set dwarfs the plain cursor.
	Spliced(Box<super::resume::Subscriber>),
}

/// One poll's view of how far this subscription may drift: the clamped budget and the
/// live edge to measure a candidate group against. Resolved once, then applied to every
/// group that poll considers.
#[derive(Clone)]
struct Drift {
	budget: Duration,
	edge: Option<Edge>,
}

/// Keeps one handed-out group tied to the subscription whose cursor selected it.
struct GroupExpiry {
	/// Weak so a handed-out group never joins the track's consumer count, which is what
	/// [`crate::broadcast::Demand`] reads to decide the track still has readers. A group
	/// outlives the cursor that produced it (a relay serves one for the life of its
	/// stream), and pinning demand on that would hold an upstream subscription open past
	/// the last real subscriber.
	state: kio::ConsumerWeak<TrackState>,
	subscription: kio::Consumer<Subscription>,
	cap: kio::Consumer<Option<u64>>,
	bound: Option<u64>,
	sequence: u64,
}

impl group::Expiry for GroupExpiry {
	fn is_expired(&self, waiter: &kio::Waiter) -> bool {
		let mut max_age = Duration::default();
		let _ = self.subscription.poll(waiter, |subscription| {
			max_age = subscription.max_age;
			Poll::<()>::Pending
		});

		let mut cap = None;
		let _ = self.cap.poll(waiter, |current| {
			cap = **current;
			Poll::<()>::Pending
		});
		let cap = super::subscription::min_some(cap, self.bound);

		let mut expired = false;
		let _ = self.state.poll(waiter, |state| {
			let budget = clamp_max_age(max_age, state.max_age_bound());
			loop {
				let edge = state.live_edge(cap);
				expired = state.is_stale(self.sequence, edge.as_ref(), budget);
				if expired {
					break;
				}

				// A first timestamp can change the verdict without mutating the track:
				// on a group past the edge (a new edge), or on the candidate's
				// unstamped immediate successor (a reach where there was none).
				// Register on the candidate and every unstamped servable group above
				// it. If one raced this scan, resolve the edge again before Pending.
				let mut timestamp_raced = false;
				if let Some(slot) = state.lookup.get(&self.sequence) {
					let group = &slot.group;
					if group.timestamp().is_none()
						&& group.poll_timestamp(waiter).is_ready()
						&& group.timestamp().is_some()
					{
						timestamp_raced = true;
					}
				}
				for (_, slot) in state
					.lookup
					.range((std::ops::Bound::Excluded(self.sequence), std::ops::Bound::Unbounded))
				{
					let group = &slot.group;
					if !super::subscription::before_end(group.sequence, cap) {
						break;
					}
					if slot.visible
						&& !group.is_aborted()
						&& group.timestamp().is_none()
						&& group.poll_timestamp(waiter).is_ready()
						&& group.timestamp().is_some()
					{
						timestamp_raced = true;
						break;
					}
				}
				if !timestamp_raced {
					break;
				}
			}

			// Register on track changes even though the current answer is known: a
			// newer group can move either live edge while this group read is pending.
			Poll::<()>::Pending
		});

		expired
	}
}

/// The group a poll's drift is measured against, identified well enough to tell it apart
/// from whatever may occupy its sequence by the time a candidate is judged.
#[derive(Clone)]
struct Edge {
	presentation: PresentationEdge,
	/// The cap the edge was resolved under, so per-candidate reach lookups measure
	/// against the same servable window.
	cap: Option<u64>,
}

/// The newest servable group that has presented at least one frame.
#[derive(Clone, Copy)]
struct PresentationEdge {
	sequence: u64,
	/// The slot incarnation this timestamp was read from, so an eviction or a re-served
	/// sequence between resolving the anchor and using it is detectable.
	stamp: u32,
	/// The newest frame this group has presented, not its first. The candidate side of the
	/// comparison is an upper bound on what a group could still reach, so the edge side has
	/// to be the newest content that actually exists, or the two meet and nothing is ever
	/// convicted. Only ever grows as the group fills.
	timestamp: Timestamp,
}

/// The cursor state for a subscription over a single (per-session) track.
struct PlainSubscriber {
	state: kio::Consumer<TrackState>,

	subscription: kio::Producer<Subscription>,
	/// Arrival-order cursor used by `recv_group`.
	index: usize,
	/// Arrival-order cursor used by `recv_datagram`, independent of groups.
	datagram_index: usize,
	/// Minimum sequence to return from any `recv` method. Set by `start_at`.
	min_sequence: u64,
	/// One past the highest sequence returned by `next_group`.
	/// Used only by that method to skip late arrivals; does not affect `recv_group`.
	next_sequence: u64,
	/// Exclusive upper sequence bound for `next_group` and `recv_group`, in the form
	/// [`Cap::exclusive`] produces. `None` means no cap. Set by `end_at`;
	/// can be raised, lowered, or unset at any time. Groups at or past the cap stay in
	/// the producer's cache and become eligible again when the cap rises (or is removed).
	/// `Some(0)` is the empty range.
	end_sequence: Option<u64>,
	/// Groups received beyond the [`Self::end_sequence`] cap, held for `recv_group`
	/// until the cap rises (arrival-order reads consume the shared cursor, so they
	/// are parked here instead of dropped). Keyed by sequence so the lowest is
	/// re-offered first.
	parked: BTreeMap<u64, group::Consumer>,
	/// A cap imposed by a reader wrapping this cursor (a [`super::resume::Subscriber`]
	/// segment), folded into the drift anchor only. Delivery is still bounded by
	/// `end_sequence`, which stays unset on a segment so its completion is visible.
	stale_cap: Option<u64>,
	/// Shared effective cap used by groups after this cursor hands them out.
	drift_cap: kio::Producer<Option<u64>>,
	/// Groups the drift budget skipped since the count was last drained. Accumulated
	/// here rather than metered in place because the handle that owns the stats scope
	/// is the outer [`Subscriber`], which may be reading this cursor through a
	/// [`super::resume::Subscriber`] segment (untagged, so the outer wrapper is the
	/// only place attribution happens once).
	stale: stats::Content,
	/// Groups the seek path has convicted but whose sequences no caller has committed
	/// past yet, keyed by sequence; see [`Self::commit_seek_stale`].
	seek_pending: BTreeMap<u64, stats::Content>,
}

impl PlainSubscriber {
	fn update_drift_cap(&mut self) {
		if let Ok(mut cap) = self.drift_cap.write() {
			*cap = servable_cap(self.end_sequence, self.stale_cap);
		}
	}

	// A helper to automatically apply Dropped if the state is closed without an error.
	fn poll<F, R>(&self, waiter: &kio::Waiter, f: F) -> Poll<Result<R>>
	where
		F: Fn(&kio::Ref<'_, TrackState>) -> Poll<Result<R>>,
	{
		Poll::Ready(match ready!(self.state.poll(waiter, f)) {
			Ok(res) => res,
			// We try to clone abort just in case the function forgot to check for terminal state.
			Err(state) => Err(state.abort.clone().unwrap_or(Error::Dropped)),
		})
	}

	/// Take the groups skipped since the last call, for the owner to meter.
	fn take_stale(&mut self) -> stats::Content {
		std::mem::take(&mut self.stale)
	}

	/// Note a group an outer reader skipped on this cursor's behalf, so it lands in the
	/// same counter as the ones skipped here.
	fn note_stale(&mut self, group: &group::Consumer) {
		self.stale.add(group.content());
	}

	/// This subscriber's clamped drift budget and the live edge to measure against,
	/// resolved once per poll.
	///
	/// `cap` bounds the anchor: the caller passes the same window it reads from, so a
	/// group is only ever judged against content that could actually be served in its
	/// place. Read fresh each poll, so a mid-stream [`Control::update`] applies
	/// to the very next group, and shared across every candidate that poll walks off, so
	/// discarding a backlog of N groups costs one scan rather than N. Only ever
	/// [`Poll::Ready`]; the track ending surfaces as the error the caller was going to
	/// get anyway.
	fn poll_drift(&self, cap: Option<u64>, waiter: &kio::Waiter) -> Poll<Result<Drift>> {
		let mut max_age = Duration::default();
		let _ = self.subscription.poll(waiter, |subscription| {
			max_age = subscription.max_age;
			Poll::<()>::Pending
		});
		self.poll(waiter, move |state| {
			Poll::Ready(Ok(Drift {
				budget: clamp_max_age(max_age, state.max_age_bound()),
				edge: state.live_edge(cap),
			}))
		})
	}

	/// Whether the drift budget says to skip `group`, against a [`Drift`] already resolved
	/// for this poll.
	fn poll_stale(&self, group: &group::Consumer, drift: &Drift, waiter: &kio::Waiter) -> Poll<Result<bool>> {
		self.poll(waiter, move |state| {
			Poll::Ready(Ok(state.is_stale(group.sequence, drift.edge.as_ref(), drift.budget)))
		})
	}

	fn with_expiry(&self, group: group::Consumer) -> group::Consumer {
		let sequence = group.sequence;
		group.with_expiry(Arc::new(GroupExpiry {
			state: self.state.weak(),
			subscription: self.subscription.consume(),
			cap: self.drift_cap.consume(),
			bound: None,
			sequence,
		}))
	}

	fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
		// An eviction aborts a parked group without touching any cursor this
		// subscriber polls, so each entry needs a waiter or this poll would never
		// rerun. `poll_closed` observes-or-registers under one lock: `Pending`
		// parks the waiter while the group is open (an open group cannot be
		// aborted), and `Ready` means closed, where only an abort invalidates the
		// entry. Checking `is_aborted` separately from the registration would leave
		// a window where an abort lands between the two and wakes nobody.
		let watch = |group: &group::Consumer| match group.poll_closed(waiter) {
			Poll::Pending => true,
			Poll::Ready(()) => !group.is_aborted(),
		};

		// A raised `start_at` drops parked groups it overtook, and eviction/expiry
		// (which aborts a cached group) drops its parked entry. The latter is what
		// bounds parking: a subscription capped indefinitely holds only what the
		// track's cache policy still retains, not every group it ever observed.
		let min_sequence = self.min_sequence;
		self.parked
			.retain(|sequence, group| *sequence >= min_sequence && watch(group));

		// One scan for the whole poll, so walking a backlog off stays linear in its size.
		let drift = ready!(self.poll_drift(servable_cap(self.end_sequence, self.stale_cap), waiter))?;

		loop {
			// Re-offer the lowest parked group back inside the cap once it rises,
			// ahead of the arrival cursor: it is the oldest thing still owed.
			let consumer = match self.parked.keys().next().copied() {
				Some(sequence) if super::subscription::before_end(sequence, self.end_sequence) => {
					let group = self.parked.remove(&sequence).expect("just looked it up");
					// A re-offer is a delivery: stamp it like a fresh hand-out.
					group.cache_refresh();
					group
				}
				_ => {
					let Some((producer, found_index)) =
						ready!(self.poll(waiter, |state| state.poll_recv_group(self.index, self.min_sequence))?)
					else {
						// Parked groups survive a finished track: they become deliverable
						// again if the cap rises, so the stream isn't over while any are held.
						if self.parked.is_empty() {
							return Poll::Ready(Ok(None));
						}
						return Poll::Pending;
					};
					let consumer = producer.consume();
					// Stamp with the track guard released, so delivery never nests the
					// group's state lock under the track's.
					consumer.cache_refresh();
					self.index = found_index + 1;

					// Park a group beyond the cap instead of dropping it, and keep scanning
					// so an in-range group that arrived behind it still flows.
					if !super::subscription::before_end(consumer.sequence, self.end_sequence) {
						// Watch it from the moment it parks: the retain pass above already
						// ran, so an entry admitted here would otherwise sit unwatched for
						// the rest of this poll, and an abort could wake nobody.
						if watch(&consumer) {
							self.parked.insert(consumer.sequence, consumer);
						}
						continue;
					}
					consumer
				}
			};

			// Drop a group the drift budget has given up on and keep scanning, so one
			// poll walks a whole backlog off rather than handing it out group by group.
			if ready!(self.poll_stale(&consumer, &drift, waiter))? {
				self.stale.add(consumer.content());
				continue;
			}
			return Poll::Ready(Ok(Some(self.with_expiry(consumer))));
		}
	}

	fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
		let Some((datagram, found_index)) =
			ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?)
		else {
			return Poll::Ready(Ok(None));
		};

		self.datagram_index = found_index + 1;
		Poll::Ready(Ok(Some(datagram)))
	}

	fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
		let floor = self.next_sequence.max(self.min_sequence);
		let Some(group) = ready!(self.poll_seek_group(floor, self.end_sequence, waiter))? else {
			return Poll::Ready(Ok(None));
		};
		self.next_sequence = group.sequence.saturating_add(1);
		// The delivery commits everything the seek stepped over to reach this group.
		self.commit_seek_stale(self.next_sequence);
		// Delivery is a cache access, same as the arrival-order path.
		group.cache_refresh();
		Poll::Ready(Ok(Some(group)))
	}

	/// Seek the lowest servable group in `floor..end` without advancing any cursor,
	/// walking off everything the drift budget has convicted on the way.
	///
	/// The drift anchor is built from the same window the seek reads (`end` folded into
	/// the cap), so a candidate is only judged against content that could be served in
	/// its place. A nested splice depends on this: its outer boundary arrives only as
	/// this call's `end`, and an anchor past it would convict groups the caller still
	/// owes its reader.
	///
	/// Repeated polls return the same group until the caller's own floor passes it,
	/// which is the point: the spliced sequence path picks the lowest candidate across
	/// segments and must not commit a segment past a group it has not delivered.
	fn poll_seek_group(
		&mut self,
		floor: u64,
		end: Option<u64>,
		waiter: &kio::Waiter,
	) -> Poll<Result<Option<group::Consumer>>> {
		let mut floor = floor.max(self.min_sequence);
		let end = super::subscription::min_some(end, self.end_sequence);
		// One scan for the whole poll, so walking a backlog off stays linear in its size.
		let drift = ready!(self.poll_drift(servable_cap(end, self.stale_cap), waiter))?;

		loop {
			let Some(producer) = ready!(self.poll(waiter, |state| state.poll_next_in_range(floor, end))?) 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, and a mid-group boundary can serve a
				// convicted sequence through another segment even after this cursor
				// retires. A conviction never committed is dropped uncounted with the
				// cursor, the same tail bound a retired segment already has.
				return Poll::Ready(Ok(None));
			};
			let group = producer.consume();

			// Skip a group the budget has given up on and keep scanning, so one poll
			// walks a whole backlog off rather than handing it out group by group.
			if ready!(self.poll_stale(&group, &drift, waiter))? {
				// Not counted yet: a seek advances no cursor, and a conviction is not
				// permanent (the budget can widen, the edge can be evicted), so the
				// group may still be delivered. Re-snapshot on every re-examination so
				// the eventual count reflects the group's latest observed content.
				self.seek_pending.insert(group.sequence, group.content());
				floor = group.sequence.saturating_add(1);
				continue;
			}

			// A conviction the budget walked back: the group is handed over after all,
			// so it must never reach the stale count.
			self.seek_pending.remove(&group.sequence);
			return Poll::Ready(Ok(Some(self.with_expiry(group))));
		}
	}

	/// Count the seek path's convictions below `committed` into [`Self::stale`],
	/// exactly once each.
	///
	/// A seek is speculative: the spliced path seeks every segment and delivers from
	/// one, so a conviction only becomes a real skip once a delivery commits past it.
	/// Until then the entry waits in [`Self::seek_pending`], where a delivered group
	/// removes itself (see [`Self::poll_seek_group`]). `committed` must be the
	/// deliverer's sequence watermark (one past its last delivery), which only rises.
	/// The seek's floor is NOT that: it includes `start_at`, which can be lowered
	/// again, and a conviction it flushed could then be delivered after all.
	fn commit_seek_stale(&mut self, committed: u64) {
		let keep = self.seek_pending.split_off(&committed);
		for (_, content) in std::mem::replace(&mut self.seek_pending, keep) {
			self.stale.add(content);
		}
	}

	/// Drop a conviction for `sequence` without counting it: the sequence was
	/// delivered by another cursor over the same content (a splice boundary inside
	/// the group leaves a copy in two segments), so its content is not stale.
	fn discard_seek_conviction(&mut self, sequence: u64) {
		self.seek_pending.remove(&sequence);
	}
}

/// A cloneable handle to a subscriber's delivery preferences.
///
/// This updates the same subscription as the owning [`Subscriber`] without
/// borrowing its read cursor, so callers can change delivery priority, the max age
/// budget, or group bounds while another task is waiting for groups.
#[derive(Clone)]
pub struct Control {
	subscription: kio::Producer<Subscription>,
}

impl Control {
	/// This subscriber's current preferences.
	pub fn subscription(&self) -> Subscription {
		self.subscription.read().clone()
	}

	/// 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.
	pub fn update(&self, subscription: Subscription) -> Result<()> {
		let mut state = self.subscription.write().map_err(|_| Error::Closed)?;
		*state = subscription;
		Ok(())
	}
}

impl Subscriber {
	/// The track's [`Info`], resolved when the subscription was established.
	///
	/// Free, unlike [`Consumer::query`]: subscribing already waited for the info
	/// (SUBSCRIBE_OK on the wire), so a subscriber always has it.
	pub fn info(&self) -> &Info {
		&self.info
	}

	/// The track's name, unique within its broadcast.
	pub fn name(&self) -> &str {
		&self.name
	}

	/// The broadcast this track belongs to, as reached through the handle it came from.
	/// Its [`path`](broadcast::Info::path) is what a catalog's relative `broadcast`
	/// references resolve against.
	pub fn broadcast(&self) -> &broadcast::Info {
		&self.broadcast
	}

	/// Attribute the groups the drift budget skipped since the last read.
	///
	/// Drained rather than metered where the skip happens, for the same reason a
	/// delivered group is metered here: a spliced subscriber reads through untagged
	/// per-segment cursors, so this wrapper is the one place that counts exactly once.
	fn count_stale(&mut self, meter: &stats::Meter) {
		// An untagged subscriber leaves the count where it is. A spliced reader polls
		// its segments through this same method, and those segments are untagged, so
		// draining here would throw the count away before the tagged handle wrapping
		// them ever sees it.
		if meter.is_tracked() {
			meter.stale(self.take_stale());
		}
	}

	/// Take the groups the drift budget skipped since the last call, so a nesting
	/// handle (a spliced subscriber reading this one as a segment) can attribute them.
	pub(crate) fn take_stale(&mut self) -> stats::Content {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.take_stale(),
			SubscriberKind::Spliced(spliced) => spliced.take_stale(),
		}
	}

	/// Bound the drift anchor from outside, for a reader that caps this subscriber
	/// without capping its cursor.
	///
	/// A [`super::resume::Subscriber`] segment is deliberately left uncapped
	/// ([`Self::set_groups`] would park boundary-crossing groups where its completion can't
	/// be seen), so its own cap has to reach the anchor this way or the segment measures
	/// drift against groups its reader will never be served. A spliced segment folds the
	/// cap into what it pushes onto its own segments, so the bound reaches the plain
	/// cursors at the leaves however deep the splices nest.
	pub(crate) fn set_stale_cap(&mut self, cap: Option<u64>) {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => {
				plain.stale_cap = cap;
				plain.update_drift_cap();
			}
			SubscriberKind::Spliced(spliced) => spliced.set_stale_cap(cap),
		}
	}

	/// Count the seek path's convictions below the deliverer's `committed` watermark
	/// as stale, exactly once each; see [`PlainSubscriber::commit_seek_stale`].
	///
	/// The spliced seek calls this on every segment before it delivers, so a losing
	/// segment's convictions are counted once the winning delivery moves the cursor
	/// past them, and never before.
	pub(crate) fn commit_seek_stale(&mut self, committed: u64) {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.commit_seek_stale(committed),
			SubscriberKind::Spliced(spliced) => spliced.commit_seek_stale(committed),
		}
	}

	/// Drop any conviction for `sequence` without counting it; see
	/// [`PlainSubscriber::discard_seek_conviction`]. The spliced deliverer calls this
	/// for the sequence it just handed out, since a boundary inside that group leaves
	/// a convictable copy in the next segment that the delivery splices into.
	pub(crate) fn discard_seek_conviction(&mut self, sequence: u64) {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.discard_seek_conviction(sequence),
			SubscriberKind::Spliced(spliced) => spliced.discard_seek_conviction(sequence),
		}
	}

	/// Whether the drift budget says to skip `group`, for a reader holding a group this
	/// subscriber handed it earlier (a parked one, re-offered once a cap rose). Counts
	/// the skip here so it reaches the stats with the rest. A spliced subscriber asks
	/// the segment whose window covers the group, so a nested park is re-checked
	/// against the same anchor a fresh read would use.
	pub(crate) fn poll_stale(&mut self, group: &group::Consumer, waiter: &kio::Waiter) -> Poll<Result<bool>> {
		let plain = match &mut self.inner {
			SubscriberKind::Plain(plain) => plain,
			SubscriberKind::Spliced(spliced) => return spliced.poll_stale(group, waiter),
		};
		let drift = ready!(plain.poll_drift(servable_cap(plain.end_sequence, plain.stale_cap), waiter))?;
		let stale = ready!(plain.poll_stale(group, &drift, waiter))?;
		if stale {
			plain.note_stale(group);
		}
		Poll::Ready(Ok(stale))
	}

	/// Create a handle for updating this subscriber's delivery preferences.
	pub fn control(&self) -> Control {
		Control {
			subscription: match &self.inner {
				SubscriberKind::Plain(plain) => plain.subscription.clone(),
				SubscriberKind::Spliced(spliced) => spliced.prefs(),
			},
		}
	}

	/// Poll for the next group in arrival order, without blocking.
	///
	/// Returns each group it delivers exactly once, in the order it landed on the wire,
	/// which may be out of sequence due to network reordering or loss. Use
	/// [`Self::ordered`] if you only want groups whose sequence number is higher than any
	/// previously returned.
	///
	/// Groups are semi-reliable, and the [`Subscription::max_age`] budget is the other
	/// thing (alongside eviction and a moving start) that decides which of them arrive:
	/// one that has drifted further behind the live edge than the budget tolerates is
	/// skipped rather than handed over, so a single poll walks off a whole backlog. The
	/// default is [`Duration::ZERO`], which takes the live
	/// edge and writes the rest off; raise it to read history. [`Self::set_groups`] and
	/// [`Subscription::start`] are filters, not exemptions: backfill needs a budget that
	/// covers it. [`Consumer::fetch_group`] is the way to ask for one old group outright.
	/// The budget remains attached to a returned group: if it stalls while newer data
	/// advances, its pending frame read ends with [`Error::Old`].
	///
	/// Honors the group range set by [`Self::set_groups`]:
	/// a group beyond the cap is parked (not dropped) and re-offered once the cap rises
	/// (lowest sequence first), without blocking in-range groups that arrive behind it.
	/// A parked group that the producer evicts or expires in the meantime is dropped,
	/// so parking never outlives the track's cache policy.
	///
	/// Returns `Poll::Ready(Ok(Some(group)))` when a group is available,
	/// `Poll::Ready(Ok(None))` when the track is finished,
	/// `Poll::Ready(Err(e))` when the track has been aborted, or
	/// `Poll::Pending` when no group is available yet.
	pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
		let meter = self.stats.meter();
		let res = match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.poll_recv_group(waiter),
			SubscriberKind::Spliced(spliced) => spliced.poll_recv_group(waiter),
		};
		self.count_stale(&meter);
		res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
	}

	/// Receive the next group in arrival order.
	///
	/// Every group is returned exactly once, in the order it landed on the wire, which may
	/// be out of sequence due to network reordering or loss. Use [`Self::ordered`] if you
	/// only want groups whose sequence number is higher than any previously returned.
	/// See [`Self::poll_recv_group`] for how [`Self::set_groups`] applies.
	pub async fn recv_group(&mut self) -> Result<Option<group::Consumer>> {
		kio::wait(|waiter| self.poll_recv_group(waiter)).await
	}

	/// Poll for the next datagram in arrival order, without blocking.
	///
	/// Datagrams are a separate best-effort channel from groups (see
	/// [`Producer::append_datagram`]); they share only the sequence namespace, and
	/// neither cursor moves the other. A consumer that falls too far behind silently
	/// loses the oldest datagrams.
	///
	/// Returns `Poll::Ready(Ok(Some(datagram)))` when one is available,
	/// `Poll::Ready(Ok(None))` when the track is finished, `Poll::Ready(Err(e))` when the track
	/// is aborted, or `Poll::Pending` when none is buffered yet.
	pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
		let meter = self.stats.meter();
		let res = match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.poll_recv_datagram(waiter),
			SubscriberKind::Spliced(spliced) => spliced.poll_recv_datagram(waiter),
		};
		// Unlike a group (metered lazily as its frames are read), a datagram is
		// delivered whole here, so count it as the single-frame group it stands in for.
		if let Poll::Ready(Ok(Some(datagram))) = &res {
			meter.datagram(datagram.payload.len() as u64);
		}
		res
	}

	/// Receive the next datagram in arrival order.
	///
	/// A best-effort channel parallel to [`Self::recv_group`]; the two share only the sequence
	/// namespace. To receive both concurrently from one subscriber, poll
	/// [`Self::poll_recv_group`] and [`Self::poll_recv_datagram`] together in a single `poll`
	/// closure (sequential `&mut` borrows), rather than awaiting the two `recv` futures at once.
	pub async fn recv_datagram(&mut self) -> Result<Option<Datagram>> {
		kio::wait(|waiter| self.poll_recv_datagram(waiter)).await
	}

	/// The sequence cursor behind [`Ordered`], which owns the only public door to it.
	fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
		let meter = self.stats.meter();
		let res = match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.poll_next_group(waiter),
			SubscriberKind::Spliced(spliced) => spliced.poll_next_group(waiter),
		};
		self.count_stale(&meter);
		res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
	}

	/// Seek the lowest cached group in `floor..end` without advancing any cursor.
	/// Crate-visible for [`super::resume::Subscriber`], whose sequence path picks the
	/// lowest candidate across its segments and advances only its own floor; see
	/// [`PlainSubscriber::poll_seek_group`].
	pub(crate) fn poll_seek_group(
		&mut self,
		floor: u64,
		end: Option<u64>,
		waiter: &kio::Waiter,
	) -> Poll<Result<Option<group::Consumer>>> {
		let meter = self.stats.meter();
		let res = match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.poll_seek_group(floor, end, waiter),
			SubscriberKind::Spliced(spliced) => spliced.poll_seek_group(floor, end, waiter),
		};
		self.count_stale(&meter);
		res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
	}

	/// Read this track's groups in sequence order instead of arrival order.
	///
	/// Consumes the subscriber, so one handle carries exactly one cursor: an
	/// arrival-order [`Subscriber`] or a sequence-order [`Ordered`], never both at once.
	/// The two advance independently, and interleaving them produces a group stream that
	/// is neither, which is why the choice is a handle rather than a method.
	///
	/// Datagrams come along: they are a separate cursor either way, so the choice of group
	/// order says nothing about them.
	pub fn ordered(self) -> Ordered {
		Ordered { inner: self }
	}

	/// Whether `other` was cloned from this subscriber (shares the same underlying state).
	pub fn is_clone(&self, other: &Self) -> bool {
		match (&self.inner, &other.inner) {
			(SubscriberKind::Plain(a), SubscriberKind::Plain(b)) => a.state.same_channel(&b.state),
			(SubscriberKind::Spliced(a), SubscriberKind::Spliced(b)) => a.is_clone(b),
			_ => false,
		}
	}

	/// Poll for the track's declared final sequence, without blocking.
	pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll<Result<u64>> {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.poll(waiter, |state| state.poll_finished()),
			SubscriberKind::Spliced(spliced) => spliced.poll_finished(waiter),
		}
	}

	/// Block until the track declares its end, returning the exclusive final sequence
	/// (also the total group count), or the cause on an abort.
	///
	/// Resolves as soon as the boundary is known, which may be ahead of the live edge
	/// when the producer finished via [`Producer::finish_at`]. This reports the declared
	/// end, not that every group has arrived: drive [`Self::recv_group`] (or
	/// [`Ordered::next_group`]) until it yields `None` to observe the track fully drained.
	pub async fn finished(&mut self) -> Result<u64> {
		kio::wait(|waiter| self.poll_finished(waiter)).await
	}

	/// Limit subsequent reads to these group sequences without rewinding read progress.
	///
	/// `2..=5` includes groups 2 through 5; `2..5` excludes group 5. An omitted
	/// start preserves the current floor, and an omitted end removes the cap.
	/// Groups above the end remain buffered and can be read after raising the cap.
	/// This changes only local delivery; use [`Subscription::with_groups`] and
	/// [`Self::update`] to change the requested groups.
	pub fn set_groups(&mut self, groups: impl RangeBounds<u64>) {
		let (start, end) = super::subscription::sequence_bounds(groups);
		self.raise_start_to(start);
		self.end_at(end.map_or(Bound::Unbounded, Bound::Excluded));
	}

	/// Start this subscriber's read cursor at the given sequence.
	///
	/// A local filter, not a request: it doesn't tell the publisher anything, so the
	/// skipped groups are still delivered and simply not returned. To ask the publisher
	/// to start there instead, set [`Subscription::start`] via [`Self::update`].
	/// See [Local cursor vs wire preference](Self#local-cursor-vs-wire-preference).
	pub(crate) fn start_at(&mut self, sequence: u64) {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.min_sequence = sequence,
			SubscriberKind::Spliced(spliced) => spliced.start_at(sequence),
		}
	}

	/// Raise the read cursor's floor to `sequence`, keeping any higher floor already set.
	///
	/// The spliced layer positions a segment's inner cursor with this instead of
	/// [`Self::set_groups`]: the inner subscription already resolved a start from its own
	/// budget and floor, and an assignment would discard it.
	pub(crate) fn raise_start_to(&mut self, sequence: u64) {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => plain.min_sequence = plain.min_sequence.max(sequence),
			SubscriberKind::Spliced(spliced) => spliced.raise_start_to(sequence),
		}
	}

	/// Cap this subscriber's read cursor at `end`, or remove the cap with `..`.
	///
	/// The range says whether its sequence is delivered: `..=5` serves through group 5,
	/// `..5` stops before it. `..0` is the empty range: no group is delivered.
	/// [`Position::group_end`] translates a [`Subscription::end`].
	///
	/// A local filter, not a request; [`Subscription::end`] is the wire-level
	/// counterpart. See [Local cursor vs wire preference](Self#local-cursor-vs-wire-preference).
	///
	/// Groups beyond the cap are held rather than skipped past, so a later call to
	/// [`Self::set_groups`] with a higher bound (or unbounded) makes them available again.
	/// Lowering the cap below the consumer's current cursor parks the consumer until the
	/// cap is raised.
	pub(crate) fn end_at(&mut self, end: impl Into<Cap>) {
		let end = end.into();
		match &mut self.inner {
			SubscriberKind::Plain(plain) => {
				plain.end_sequence = end.exclusive();
				plain.update_drift_cap();
			}
			SubscriberKind::Spliced(spliced) => spliced.end_at(end),
		}
	}

	/// This subscriber's current preferences.
	pub fn subscription(&self) -> Subscription {
		self.control().subscription()
	}

	/// Replace this subscriber's delivery preferences.
	///
	/// Stored verbatim; the publisher's max age window is applied to the aggregate, not
	/// here (see [`Producer::subscription`]). Returns [`Error::Closed`] if the track
	/// already ended; the update is meaningless at that point and can usually be ignored.
	pub fn update(&mut self, subscription: Subscription) -> Result<()> {
		match &mut self.inner {
			SubscriberKind::Plain(plain) => {
				let mut state = plain.subscription.write().map_err(|_| Error::Closed)?;
				*state = subscription;
			}
			SubscriberKind::Spliced(spliced) => spliced.update(subscription),
		}
		Ok(())
	}

	/// Return the latest sequence number in the track.
	pub fn latest(&self) -> Option<u64> {
		match &self.inner {
			SubscriberKind::Plain(plain) => plain.state.read().max_sequence,
			SubscriberKind::Spliced(spliced) => spliced.latest(),
		}
	}
}

/// A [`Subscriber`] that reads groups in sequence order.
///
/// Created by [`Subscriber::ordered`], which consumes the subscriber, so a track is read
/// one way or the other and never both. Every group it returns has a higher sequence
/// than the last, so a late arrival (network reordering, or a gap the cache filled after
/// the fact) is skipped rather than delivered out of turn.
///
/// # Age and skipping
///
/// [`Subscription::max_age`] applies as this cursor reads, exactly as it does on the
/// arrival cursor: a group is skipped once its *reach*, where its immediate successor
/// begins, is that far behind the newest frame on the track. Nothing weaker convicts it,
/// because the reach is the only proof that *every* frame it could still hold is past the
/// budget: a group's own timestamps say where it starts presenting, not where it stops,
/// and an unstamped successor leaves it unbounded and therefore kept. A backlog inside
/// the budget is still delivered whole, as a burst in order, which is what a decoder
/// reading a gap-free sequence needs.
///
/// The same budget follows a group already handed out: once newer content pulls that far
/// ahead, a read still waiting on it ends with [`Error::Old`] and the cursor moves on. A
/// gap costs nothing either way, since the cursor seeks to the lowest cached group in
/// range rather than waiting for the sequence it would have read next.
///
/// The budget is updatable mid-stream through [`Self::control`]. [`Duration::ZERO`] (the
/// default) keeps only what nothing newer has superseded, so a consumer that stalls and
/// resumes rejoins the live edge instead of replaying at 1x.
pub struct Ordered {
	inner: Subscriber,
}

impl Ordered {
	/// The track's [`Info`], resolved when the subscription was established.
	pub fn info(&self) -> &Info {
		self.inner.info()
	}

	/// The track's name, unique within its broadcast.
	pub fn name(&self) -> &str {
		self.inner.name()
	}

	/// The broadcast this track belongs to, as reached through the handle it came from.
	pub fn broadcast(&self) -> &broadcast::Info {
		self.inner.broadcast()
	}

	/// Poll for the next group with a higher sequence number than any previously
	/// returned, without blocking.
	///
	/// Honors the group range set by [`Self::set_groups`]:
	/// a group past the cap stays in the producer's cache and becomes eligible again if
	/// the cap rises or is removed.
	///
	/// Returns `Poll::Ready(Ok(Some(group)))` when a group is available,
	/// `Poll::Ready(Ok(None))` when the track is finished,
	/// `Poll::Ready(Err(e))` when the track has been aborted, or
	/// `Poll::Pending` when no group is available yet.
	pub fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
		self.inner.poll_next_group(waiter)
	}

	/// Return the next group with a higher sequence number than any previously returned.
	pub async fn next_group(&mut self) -> Result<Option<group::Consumer>> {
		kio::wait(|waiter| self.poll_next_group(waiter)).await
	}

	/// Poll for the next datagram in arrival order, without blocking.
	///
	/// Datagrams are a separate best-effort channel from groups (see
	/// [`Producer::append_datagram`]); they share only the sequence namespace, and neither
	/// cursor moves the other. Unordered by construction, so this behaves identically on
	/// either handle; it is here so a track carrying both channels needs one subscription
	/// rather than two.
	pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
		self.inner.poll_recv_datagram(waiter)
	}

	/// Receive the next datagram in arrival order.
	///
	/// To read groups and datagrams concurrently from one handle, poll
	/// [`Self::poll_next_group`] and [`Self::poll_recv_datagram`] together in a single
	/// `poll` closure (sequential `&mut` borrows), rather than awaiting the two `recv`
	/// futures at once.
	pub async fn recv_datagram(&mut self) -> Result<Option<Datagram>> {
		kio::wait(|waiter| self.poll_recv_datagram(waiter)).await
	}

	/// Limit subsequent reads to these group sequences without rewinding read progress.
	///
	/// See [`Subscriber::set_groups`] for inclusive, exclusive, and omitted bounds.
	pub fn set_groups(&mut self, groups: impl RangeBounds<u64>) {
		self.inner.set_groups(groups);
	}

	/// Create a handle for updating this subscriber's delivery preferences without
	/// borrowing the read cursor.
	pub fn control(&self) -> Control {
		self.inner.control()
	}

	/// This subscriber's current preferences.
	pub fn subscription(&self) -> Subscription {
		self.inner.subscription()
	}

	/// Replace this subscriber's delivery preferences.
	///
	/// Returns [`Error::Closed`] if the track already ended.
	pub fn update(&mut self, subscription: Subscription) -> Result<()> {
		self.inner.update(subscription)
	}

	/// Poll for the track's declared final sequence, without blocking.
	pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll<Result<u64>> {
		self.inner.poll_finished(waiter)
	}

	/// Block until the track declares its end, returning the exclusive final sequence.
	///
	/// See [`Subscriber::finished`]: this reports the declared end, not that every group
	/// has arrived.
	pub async fn finished(&mut self) -> Result<u64> {
		kio::wait(|waiter| self.poll_finished(waiter)).await
	}

	/// The latest sequence number in the track.
	pub fn latest(&self) -> Option<u64> {
		self.inner.latest()
	}

	/// Whether `other` reads the same underlying track state.
	pub fn is_clone(&self, other: &Self) -> bool {
		self.inner.is_clone(&other.inner)
	}
}

/// A subscriber asked for a track this broadcast doesn't have yet.
///
/// Yielded by [`broadcast::Dynamic::requested_track`](crate::broadcast::Dynamic::requested_track),
/// or created up front with [`broadcast::Producer::reserve_track`](crate::broadcast::Producer::reserve_track).
/// Subscribers block until the request is
/// resolved: call [`accept`](Self::accept) to serve it with a [`Producer`], or
/// [`reject`](Self::reject) to fail them. Dropping it without either rejects with
/// [`Error::Dropped`].
///
/// Concurrent requests for one name are coalesced, so exactly one of these exists per
/// name at a time.
pub struct Request {
	name: Arc<str>,
	// The parent broadcast's info, threaded into the [`Producer`] on accept.
	broadcast: Arc<broadcast::Info>,
	state: kio::Producer<TrackState>,

	// The previous subscription that was combined, used to detect changes.
	prev_subscription: Option<Subscription>,

	// Shared with the accepted [`Producer`] and every [`Dynamic`]: its `Drop` is the
	// teardown, and it stays inert until a producer is minted.
	alive: Arc<Alive>,

	// A requested track is served on demand, so it counts as fetch-capable from
	// birth: a consumer's cache-miss `fetch_group` waits to be served instead of
	// racing the producer (e.g. a relay) into creating its own handler. Released
	// when the request is accepted or dropped; by then the relay holds its own.
	_dynamic: Dynamic,

	// Ingress stats scope, threaded into the accepted [`Producer`]. Empty (no-op)
	// unless this request was reserved on a tagged broadcast.
	stats: stats::Scope,
}

impl Request {
	pub(crate) fn new(broadcast: Arc<broadcast::Info>, name: impl Into<Arc<str>>) -> Self {
		let name = name.into();
		let state = TrackState::spawn(broadcast.clone());
		let alive = Alive::new(name.clone(), state.clone());
		let dynamic = Dynamic::new(name.clone(), state.clone(), alive.clone());
		Self {
			name,
			broadcast,
			state,
			prev_subscription: None,
			alive,
			_dynamic: dynamic,
			stats: stats::Scope::default(),
		}
	}

	/// Attach an ingress stats scope, applied to the [`Producer`] on accept. Set by
	/// a tagged [`broadcast::Producer::reserve_track`].
	pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self {
		self.stats = scope;
		self
	}

	/// The requested track name.
	pub fn name(&self) -> &str {
		&self.name
	}

	/// A [`Consumer`] for the eventual track, usable before the request is accepted.
	pub fn consume(&self) -> Consumer {
		Consumer::plain(self.name.clone(), self.state.consume())
	}

	/// Create a [`Dynamic`] handle that serves on-demand fetches of uncached
	/// groups, before [`Self::accept`] is even called. A relay creates one to fetch
	/// past groups from upstream while (or instead of) serving a live subscription.
	pub fn dynamic(&self) -> Dynamic {
		Dynamic::new(self.name.clone(), self.state.clone(), self.alive.clone())
	}

	/// Poll for the request becoming unused (every consumer dropped), so a relay can
	/// stop serving and drop the request.
	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
		self.state.poll_unused(waiter).map(|_| ())
	}

	/// Serve the request with the given track, resolving every waiting subscriber.
	///
	/// The name is taken from [`Self::name`]; `info` supplies the remaining knobs
	/// (`None` for the defaults). If the track was already aborted, the returned
	/// [`Producer`] is inert: writes fail with the abort error, as if it had been
	/// aborted immediately after accepting.
	pub fn accept(self, info: impl Into<Option<Info>>) -> Producer {
		let info = TrackState::normalize_info(&self.broadcast, info.into().unwrap_or_default());
		// A closed state means the track was aborted under us. Mirror `reject` and
		// tolerate it: the Producer we hand back simply can't write.
		if let Ok(mut state) = self.state.write() {
			state.accept(info.clone());
		}
		// Accepting the request creates the track producer: count it as one ingress
		// subscription (closed when the last handle drops). No-op when untagged.
		self.alive.publish(Some(&self.stats));
		Producer {
			name: self.name,
			info,
			broadcast: self.broadcast,
			state: self.state,
			prev_subscription: None,
			alive: self.alive,
			stats: self.stats,
		}
	}

	/// Reject the request, waking all waiting subscribers with `err`.
	pub fn reject(self, err: Error) {
		if let Ok(mut state) = self.state.write() {
			state.abort = Some(err);
		}
	}

	/// The delivery preferences aggregated across everyone waiting on this request,
	/// or `None` if nobody is waiting. Useful for sizing the track before accepting.
	pub fn subscription(&self) -> Option<Subscription> {
		let state = self.state.read();
		let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
		drop(state);
		snapshot_subscription(&subs, bound)
	}

	/// Block until the aggregate [`subscription`](Self::subscription) changes,
	/// yielding `None` once nobody is waiting.
	pub async fn subscription_changed(&mut self) -> Option<Subscription> {
		kio::wait(|waiter| self.poll_subscription_changed(waiter)).await
	}

	/// Poll counterpart to [`subscription_changed`](Self::subscription_changed).
	pub fn poll_subscription_changed(&mut self, waiter: &kio::Waiter) -> Poll<Option<Subscription>> {
		let state = self.state.read();
		let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
		drop(state);

		let prev = &self.prev_subscription;
		let mut combined = None;
		let mut guard = ready!(subs.poll(waiter, |subs| {
			let next = combined_subscription(subs, bound, waiter);
			if &next == prev {
				Poll::Pending
			} else {
				combined = next;
				Poll::Ready(())
			}
		}));
		// The aggregate changed: prune any closed subscribers now that we hold the lock.
		guard.retain(|sub| !sub.is_closed());
		drop(guard);
		self.prev_subscription = combined.clone();
		Poll::Ready(combined)
	}

	pub(super) fn weak(&self) -> TrackWeak {
		TrackWeak {
			name: self.name.clone(),
			state: self.state.weak(),
		}
	}
}

#[cfg(test)]
use futures::FutureExt;

#[cfg(test)]
#[allow(missing_docs)] // test-only assertion helpers
impl Subscriber {
	pub fn assert_group(&mut self) -> group::Consumer {
		self.recv_group()
			.now_or_never()
			.expect("group would have blocked")
			.expect("would have errored")
			.expect("track was closed")
	}

	pub fn assert_no_group(&mut self) {
		assert!(
			self.recv_group().now_or_never().is_none(),
			"recv_group would not have blocked"
		);
	}

	pub fn assert_not_closed(&mut self) {
		assert!(self.finished().now_or_never().is_none(), "should not be closed");
	}

	pub fn assert_closed(&mut self) {
		assert!(self.finished().now_or_never().is_some(), "should be closed");
	}

	// TODO assert specific errors after implementing PartialEq
	pub fn assert_error(&mut self) {
		assert!(
			self.finished().now_or_never().expect("should not block").is_err(),
			"should be error"
		);
	}

	pub fn assert_is_clone(&self, other: &Self) {
		assert!(self.is_clone(other), "should be clone");
	}

	pub fn assert_not_clone(&self, other: &Self) {
		assert!(!self.is_clone(other), "should not be clone");
	}
}

#[cfg(test)]
mod test {
	use super::*;
	use crate::frame;
	use crate::model::test_tracing::count_drop_warnings;
	use std::time::Duration;

	/// Mint a track for tests with a default parent broadcast, since tracks are
	/// normally born from a [`broadcast::Producer`].
	fn track_producer(name: impl Into<Arc<str>>, info: impl Into<Option<Info>>) -> Producer {
		Producer::new(Arc::new(broadcast::Info::default()), name, info)
	}

	/// A bounded replay window for tests whose subject requires every buffered group.
	fn replay() -> Subscription {
		Subscription::default().with_max_age(Duration::from_secs(30))
	}

	/// Helper: count live cached groups in state.
	fn live_groups(state: &TrackState) -> usize {
		state.lookup.len()
	}

	/// Helper: get the sequence number of the first live group in arrival order.
	fn first_live_sequence(state: &TrackState) -> u64 {
		state
			.arrival
			.iter()
			.find(|(sequence, stamp)| state.lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp))
			.map(|(sequence, _)| *sequence)
			.unwrap()
	}

	/// Helper: non-blocking datagram receive that must be ready with a datagram.
	fn recv_datagram(dg: &mut Subscriber) -> Datagram {
		dg.recv_datagram()
			.now_or_never()
			.expect("datagram would have blocked")
			.expect("would have errored")
			.expect("track was closed")
	}

	/// A declared start (SUBSCRIBE_START) resolves a peek below it as a permanent
	/// miss, while a group that is already cached below the start stays readable
	/// (a fetch can create one; the declaration only covers the live feed).
	#[tokio::test]
	async fn peek_resolves_below_the_declared_start() {
		let mut producer = track_producer("test", None);
		let consumer = producer.consume();

		let mut cached = producer.create_group(group::Info { sequence: 1 }).unwrap();
		cached.write_frame(Timestamp::ZERO, b"backfill".to_vec()).unwrap();
		cached.finish().unwrap();

		// Nothing declared yet: a missing group parks.
		let waiter = kio::Waiter::noop();
		assert!(consumer.poll_peek_group(0, &waiter).is_pending());

		// The declaration turns the missing group into a permanent miss, but the
		// cached one below it still resolves.
		producer.start_at(3).unwrap();
		assert!(matches!(consumer.poll_peek_group(0, &waiter), Poll::Ready(None)));
		assert!(matches!(consumer.poll_peek_group(1, &waiter), Poll::Ready(Some(_))));
		assert!(consumer.poll_peek_group(3, &waiter).is_pending());

		// The declaration follows the demand in either direction: forward retires
		// the skipped range, backward reopens it.
		producer.start_at(4).unwrap();
		assert!(matches!(consumer.poll_peek_group(3, &waiter), Poll::Ready(None)));
		producer.start_at(0).unwrap();
		assert!(consumer.poll_peek_group(0, &waiter).is_pending());
	}

	#[tokio::test]
	async fn append_datagram_shares_group_sequence() {
		let mut producer = track_producer("test", None);
		let ts = Timestamp::from_millis(10).unwrap();

		// Interleave groups and datagrams: they draw from one monotonic counter.
		assert_eq!(producer.append_group().unwrap().sequence, 0);
		assert_eq!(producer.append_datagram(ts, &b"a"[..]).unwrap(), 1);
		assert_eq!(producer.append_group().unwrap().sequence, 2);
		assert_eq!(producer.append_datagram(ts, &b"b"[..]).unwrap(), 3);
		assert_eq!(producer.latest(), Some(3));
	}

	#[tokio::test]
	async fn append_datagram_roundtrip() {
		let mut producer = track_producer("test", None);
		let mut dg = producer.subscribe(None);

		let ts = Timestamp::from_millis(42).unwrap();
		let seq = producer.append_datagram(ts, &b"hello"[..]).unwrap();

		let got = recv_datagram(&mut dg);
		assert_eq!(got.sequence, seq);
		assert_eq!(got.timestamp, ts);
		assert_eq!(&got.payload[..], b"hello");
	}

	#[tokio::test]
	async fn insert_datagram_preserves_sequence() {
		let mut producer = track_producer("test", None);
		let mut dg = producer.subscribe(None);

		let ts = Timestamp::from_millis(5).unwrap();
		// A relay forwarding an upstream datagram keeps its sequence number.
		producer
			.insert_datagram(100, ts, bytes::Bytes::from_static(b"x"))
			.unwrap();

		assert_eq!(recv_datagram(&mut dg).sequence, 100);
		// max_sequence advanced, so the next appended group/datagram continues past it.
		assert_eq!(producer.append_group().unwrap().sequence, 101);
	}

	#[tokio::test]
	async fn insert_datagram_leaves_a_gap() {
		let mut producer = track_producer("test", None);
		let mut dg = producer.subscribe(None);
		let ts = Timestamp::from_millis(0).unwrap();

		producer
			.insert_datagram(10, ts, bytes::Bytes::from_static(b"gap"))
			.unwrap();
		assert_eq!(recv_datagram(&mut dg).sequence, 10);
		assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 11);
		assert_eq!(producer.append_group().unwrap().sequence, 12);
	}

	#[tokio::test]
	async fn insert_datagram_out_of_order_does_not_rewind() {
		let mut producer = track_producer("test", None);
		let mut dg = producer.subscribe(None);
		let ts = Timestamp::from_millis(0).unwrap();

		producer
			.insert_datagram(10, ts, bytes::Bytes::from_static(b"high"))
			.unwrap();
		producer
			.insert_datagram(5, ts, bytes::Bytes::from_static(b"low"))
			.unwrap();

		assert_eq!(recv_datagram(&mut dg).sequence, 10);
		assert_eq!(recv_datagram(&mut dg).sequence, 5);
		assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 11);
	}

	#[tokio::test]
	async fn insert_datagram_duplicate_is_best_effort() {
		let mut producer = track_producer("test", None);
		let mut dg = producer.subscribe(None);
		let ts = Timestamp::from_millis(0).unwrap();

		producer
			.insert_datagram(3, ts, bytes::Bytes::from_static(b"first"))
			.unwrap();
		producer
			.insert_datagram(3, ts, bytes::Bytes::from_static(b"again"))
			.unwrap();

		assert_eq!(&recv_datagram(&mut dg).payload[..], b"first");
		assert_eq!(&recv_datagram(&mut dg).payload[..], b"again");
		assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 4);
	}

	#[tokio::test]
	async fn insert_datagram_stale_does_not_rewind_after_append() {
		let mut producer = track_producer("test", None);
		let mut dg = producer.subscribe(None);
		let ts = Timestamp::from_millis(0).unwrap();

		assert_eq!(producer.append_datagram(ts, &b"0"[..]).unwrap(), 0);
		assert_eq!(producer.append_datagram(ts, &b"1"[..]).unwrap(), 1);
		producer
			.insert_datagram(0, ts, bytes::Bytes::from_static(b"stale"))
			.unwrap();

		assert_eq!(recv_datagram(&mut dg).sequence, 0);
		assert_eq!(recv_datagram(&mut dg).sequence, 1);
		assert_eq!(recv_datagram(&mut dg).sequence, 0);
		assert_eq!(producer.append_datagram(ts, &b"2"[..]).unwrap(), 2);
		assert_eq!(producer.append_group().unwrap().sequence, 3);
	}

	#[tokio::test]
	async fn insert_datagram_cloned_producers_share_counter() {
		let mut producer = track_producer("test", None);
		let mut other = producer.clone();
		let mut dg = producer.subscribe(None);
		let ts = Timestamp::from_millis(0).unwrap();

		producer
			.insert_datagram(4, ts, bytes::Bytes::from_static(b"a"))
			.unwrap();
		assert_eq!(other.append_datagram(ts, &b"b"[..]).unwrap(), 5);
		other.insert_datagram(8, ts, bytes::Bytes::from_static(b"c")).unwrap();
		assert_eq!(producer.append_group().unwrap().sequence, 9);

		assert_eq!(recv_datagram(&mut dg).sequence, 4);
		assert_eq!(recv_datagram(&mut dg).sequence, 5);
		assert_eq!(recv_datagram(&mut dg).sequence, 8);
	}

	#[test]
	fn insert_datagram_after_finish_is_closed() {
		let mut producer = track_producer("test", None);
		let ts = Timestamp::from_millis(0).unwrap();
		producer.finish().unwrap();
		assert!(matches!(
			producer.insert_datagram(0, ts, bytes::Bytes::from_static(b"x")),
			Err(Error::Closed)
		));
		assert!(matches!(producer.append_datagram(ts, &b"x"[..]), Err(Error::Closed)));
	}

	#[test]
	fn insert_datagram_after_abort_fails() {
		let producer = track_producer("test", None);
		let mut other = producer.clone();
		let ts = Timestamp::from_millis(0).unwrap();
		producer.abort(Error::Cancel).unwrap();
		assert!(other.insert_datagram(0, ts, bytes::Bytes::from_static(b"x")).is_err());
	}

	#[test]
	fn insert_datagram_respects_finish_at() {
		let mut producer = track_producer("test", None);
		let ts = Timestamp::from_millis(0).unwrap();
		producer.finish_at(10).unwrap();
		producer
			.insert_datagram(5, ts, bytes::Bytes::from_static(b"ok"))
			.unwrap();
		assert!(matches!(
			producer.insert_datagram(10, ts, bytes::Bytes::from_static(b"late")),
			Err(Error::Closed)
		));
		assert_eq!(producer.append_group().unwrap().sequence, 6);
	}

	/// Datagram sequence advances do not move a route takeover past an open group.
	#[test]
	fn resume_position_uses_the_latest_group() {
		let mut datagram_only = track_producer("datagram-only", None);
		let datagram_only_consumer = datagram_only.consume();
		datagram_only
			.insert_datagram(8, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
			.unwrap();
		assert_eq!(
			datagram_only_consumer.resume_position(),
			None,
			"a datagram creates no group position to resume"
		);

		let mut producer = track_producer("mixed", None);
		let consumer = producer.consume();
		let mut group = producer.create_group(group::Info { sequence: 3 }).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
			.unwrap();
		producer
			.insert_datagram(8, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
			.unwrap();

		assert_eq!(
			consumer.resume_position(),
			Some(Position { group: 3, frame: 1 }),
			"the replacement must continue the open group"
		);
		group.abort(Error::Cancel).unwrap();
	}

	/// Datagrams and groups are separate channels that share only a sequence namespace,
	/// so consuming one must not move the other's cursor.
	#[tokio::test]
	async fn recv_datagram_leaves_the_ordered_cursor_alone() {
		let mut producer = track_producer("test", None);
		let mut datagrams = producer.subscribe(None);
		let mut subscriber = producer.subscribe(None).ordered();
		let ts = Timestamp::from_millis(5).unwrap();

		producer
			.insert_datagram(5, ts, bytes::Bytes::from_static(b"x"))
			.unwrap();
		assert_eq!(recv_datagram(&mut datagrams).sequence, 5);

		producer.create_group(group::Info { sequence: 3 }).unwrap();
		producer.create_group(group::Info { sequence: 6 }).unwrap();

		let mut next = || {
			subscriber
				.next_group()
				.now_or_never()
				.expect("group would have blocked")
				.expect("would have errored")
				.expect("track was closed")
				.sequence
		};
		assert_eq!(next(), 3, "the datagram at sequence 5 did not consume group 3");
		assert_eq!(next(), 6);
	}

	#[tokio::test]
	async fn datagram_normalized_to_track_timescale() {
		let info = Info::default().with_timescale(Timescale::MICRO);
		let mut producer = track_producer("test", info);
		let mut dg = producer.subscribe(None);

		// Supplied at millis; stored/emitted at the track's micro timescale.
		producer
			.append_datagram(Timestamp::from_millis(2).unwrap(), &b"z"[..])
			.unwrap();
		let got = recv_datagram(&mut dg);
		assert_eq!(got.timestamp.scale(), Timescale::MICRO);
		assert_eq!(got.timestamp.value(), 2_000);
	}

	#[tokio::test]
	async fn datagram_rejects_oversized() {
		let mut producer = track_producer("test", None);
		let big = bytes::Bytes::from(vec![0u8; crate::model::datagram::MAX_DATAGRAM_PAYLOAD + 1]);
		let ts = Timestamp::from_millis(0).unwrap();
		assert!(matches!(
			producer.append_datagram(ts, big.clone()),
			Err(Error::FrameTooLarge)
		));
		assert!(matches!(
			producer.insert_datagram(0, ts, big),
			Err(Error::FrameTooLarge)
		));
	}

	#[tokio::test]
	async fn datagram_fanout_to_subscribers() {
		let mut producer = track_producer("test", None);
		// Two independent subscribers, each with its own datagram cursor.
		let mut a = producer.subscribe(None);
		let mut b = producer.subscribe(None);
		let ts = Timestamp::from_millis(1).unwrap();

		producer.append_datagram(ts, &b"first"[..]).unwrap();
		producer.append_datagram(ts, &b"second"[..]).unwrap();

		// Both receive every datagram in order, independently.
		assert_eq!(&recv_datagram(&mut a).payload[..], b"first");
		assert_eq!(&recv_datagram(&mut a).payload[..], b"second");
		assert_eq!(&recv_datagram(&mut b).payload[..], b"first");
		assert_eq!(&recv_datagram(&mut b).payload[..], b"second");
	}

	#[test]
	fn datagram_buffer_drops_oldest_at_capacity() {
		let mut producer = track_producer("test", None);
		let mut slow = producer.subscribe(None);
		let mut fast = producer.subscribe(None);
		let count = MAX_DATAGRAMS * 3;
		for sequence in 0..count {
			producer.append_datagram(Timestamp::ZERO, b"x".as_slice()).unwrap();
			assert_eq!(recv_datagram(&mut fast).sequence, sequence as u64);
		}
		assert_eq!(producer.state.read().datagrams.len(), MAX_DATAGRAMS);
		for sequence in count - MAX_DATAGRAMS..count {
			assert_eq!(recv_datagram(&mut slow).sequence, sequence as u64);
		}
		assert!(slow.poll_recv_datagram(&kio::Waiter::noop()).is_pending());
	}

	#[tokio::test]
	async fn datagram_recv_pends_until_written() {
		let mut producer = track_producer("test", None);
		let mut dg = producer.subscribe(None);

		assert!(
			dg.recv_datagram().now_or_never().is_none(),
			"should block with no datagrams"
		);

		producer
			.append_datagram(Timestamp::from_millis(0).unwrap(), &b"go"[..])
			.unwrap();
		assert_eq!(&recv_datagram(&mut dg).payload[..], b"go");
	}

	/// Exercises the full producer -> publisher-encode -> subscriber-decode -> producer seam
	/// (everything but the QUIC datagram send/recv), catching any field-order mismatch between
	/// the wire codec and the model.
	#[tokio::test]
	async fn datagram_wire_roundtrip_between_tracks() {
		use crate::coding::{Decode, Encode};
		use crate::lite;

		let version = lite::Version::Lite05;

		// Origin publishes a datagram; the publisher reads it and encodes the wire body.
		let mut origin = track_producer("test", None);
		let mut origin_dg = origin.subscribe(None);
		let ts = Timestamp::from_millis(7).unwrap();
		let seq = origin.append_datagram(ts, &b"payload"[..]).unwrap();

		let d = recv_datagram(&mut origin_dg);
		let body = lite::Datagram {
			subscribe: 5,
			sequence: d.sequence,
			timestamp: d.timestamp.value(),
			payload: d.payload.clone(),
		}
		.encode_bytes(version)
		.unwrap();

		// Subscriber decodes the body and writes it downstream, preserving the sequence.
		let mut slice = &body[..];
		let wire = lite::Datagram::decode(&mut slice, version).unwrap();
		let mut downstream = track_producer("test", None);
		let mut downstream_dg = downstream.subscribe(None);
		downstream
			.insert_datagram(
				wire.sequence,
				Timestamp::new(wire.timestamp, Timescale::MILLI).unwrap(),
				wire.payload,
			)
			.unwrap();

		let got = recv_datagram(&mut downstream_dg);
		assert_eq!(got.sequence, seq);
		assert_eq!(got.timestamp, ts);
		assert_eq!(&got.payload[..], b"payload");
	}

	#[tokio::test]
	async fn evict_expired_groups() {
		let producer = track_producer("test", None);

		// Create 3 groups at time 0.
		producer.append_group().unwrap(); // seq 0
		producer.append_group().unwrap(); // seq 1
		producer.append_group().unwrap(); // seq 2

		{
			let state = producer.state.read();
			assert_eq!(live_groups(&state), 3);
			assert_eq!(state.offset, 0);
		}

		// Advance time past the pool's LRU window.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));

		// Append a new group to trigger eviction.
		producer.append_group().unwrap(); // seq 3

		// Groups 0, 1, 2 are expired but seq 3 (the live edge) is kept. Their arrival
		// entries no longer resolve, so the leading ones are trimmed and the offset
		// advances past them.
		{
			let state = producer.state.read();
			assert_eq!(live_groups(&state), 1);
			assert_eq!(first_live_sequence(&state), 3);
			assert_eq!(state.offset, 3);
			assert!(!state.lookup.contains_key(&0));
			assert!(!state.lookup.contains_key(&1));
			assert!(!state.lookup.contains_key(&2));
			assert!(state.lookup.contains_key(&3));
		}
	}

	/// A group whose frames outlive `max_age` is aged out when the next group starts, but
	/// a subscriber that already drained it must still see the clean end of group. Otherwise a
	/// track with long groups (a per-minute rollup, say) fails its readers at every boundary.
	#[tokio::test]
	async fn aging_out_a_finished_group_keeps_the_clean_end() {
		let producer = track_producer("test", None);
		let mut group = producer.create_group(group::Info { sequence: 0 }).unwrap();
		let mut consumer = group.consume();

		group
			.write_frame(Timestamp::from_millis(0).unwrap(), b"hello".as_slice())
			.unwrap();
		assert_eq!(consumer.next_frame().await.unwrap().unwrap().size, 5);

		// The group stays open well past the LRU window, then the next period starts.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY * 2);
		group.finish().unwrap();
		let _next = producer.create_group(group::Info { sequence: 1 }).unwrap();

		assert!(consumer.next_frame().await.unwrap().is_none());
	}

	/// An actively-read group is not expired out from under its reader: every frame
	/// read restarts the retention clock. A group nobody reads still ages out on
	/// schedule, so reclamation stays intact.
	#[tokio::test]
	async fn active_reader_survives_expiry() {
		let producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);

		// A finished group with one frame per step of the read loop below.
		let mut group = producer.create_group(0u64.into()).unwrap();
		for _ in 0..10 {
			group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
		}
		group.finish().unwrap();
		let mut reading = subscriber.assert_group();

		// A sibling written at the same time that nobody ever reads.
		producer.create_group(1u64.into()).unwrap().finish().unwrap();

		// Each step stays well inside the retention window, but the whole read
		// spans several windows. New groups keep the expiry scan running.
		for seq in 2..12u64 {
			crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
			let frame = reading.next_frame().await;
			assert!(
				matches!(frame, Ok(Some(_))),
				"an actively-read group must not expire mid-read (step {seq})"
			);
			producer.create_group(seq.into()).unwrap().finish().unwrap();
		}

		let state = producer.state.read();
		assert!(state.lookup.contains_key(&0), "the read group survived");
		assert!(!state.lookup.contains_key(&1), "the unread group still expired");
	}

	/// Whole-frame reads served from the prefetch batch must also keep the group
	/// alive: the batch is filled (and stamped) once per `Prefetch::CAP` frames,
	/// which bounds frames, not elapsed time, so a slow `read_frame` reader has to
	/// re-stamp on a time bound between refills.
	#[tokio::test]
	async fn slow_prefetch_reader_survives_expiry() {
		let producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);

		let mut group = producer.create_group(0u64.into()).unwrap();
		for _ in 0..20 {
			group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
		}
		group.finish().unwrap();
		let mut reading = subscriber.assert_group();

		// One whole-frame read per half-window: most are served straight from the
		// prefetch without locking. New groups keep the expiry scan running.
		for seq in 1..20u64 {
			crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
			let frame = reading.read_frame().await;
			assert!(
				matches!(frame, Ok(Some(_))),
				"a slow prefetch reader must not expire mid-read (step {seq})"
			);
			producer.create_group(seq.into()).unwrap().finish().unwrap();
		}
	}

	/// Receiving a group is itself a cache access: a subscriber that takes
	/// delivery just before the group would age out still gets to read it a full
	/// window later.
	#[tokio::test]
	async fn delivery_restarts_the_expiry_clock() {
		let producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(replay());

		let mut group = producer.create_group(0u64.into()).unwrap();
		group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
		group.finish().unwrap();
		// A second group so seq 0 leaves the protected live edge.
		producer.create_group(1u64.into()).unwrap().finish().unwrap();

		// Deliver just inside the window: the delivery stamps the group.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
		let mut reading = subscriber.assert_group();

		// Almost another full window passes: far beyond the write, inside the
		// delivery stamp. The new group runs the expiry scan.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
		producer.create_group(2u64.into()).unwrap().finish().unwrap();

		let frame = reading.read_frame().await.unwrap();
		assert!(frame.is_some(), "a just-delivered group must not expire unread");
	}

	/// Streaming chunks into an in-flight frame is a write access: a straggler
	/// group (behind the live edge) trickling a large frame across several
	/// retention windows must not be expired mid-write.
	#[tokio::test]
	async fn streaming_frame_writes_keep_the_group_alive() {
		let producer = track_producer("test", None);
		let mut straggler = producer.create_group(0u64.into()).unwrap();
		// The live edge moves on, so the straggler is demoted and expirable.
		producer.create_group(1u64.into()).unwrap().finish().unwrap();

		let mut frame = straggler
			.create_frame(frame::Info {
				size: 10,
				timestamp: Timestamp::ZERO,
			})
			.unwrap();
		// One chunk per half-window; the whole frame spans several windows. New
		// groups keep the expiry scan running.
		for seq in 2..12u64 {
			crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
			frame.write(bytes::Bytes::from_static(b"x")).unwrap();
			producer.create_group(seq.into()).unwrap().finish().unwrap();
		}
		frame.finish().unwrap();
		straggler.finish().unwrap();

		let state = producer.state.read();
		assert!(
			state.lookup.contains_key(&0),
			"a group streaming a frame survives expiry"
		);
	}

	/// The wire ingest coalesces its chunk wakes to the poll boundary, so a payload
	/// whose tail arrives all at once completes without a single `frame_notify`.
	/// Committing is itself a write access: a group that just finished a frame must
	/// not be expired by the next track write on its stale frame-open stamp.
	#[tokio::test]
	async fn coalesced_frame_completion_keeps_the_group_alive() {
		let producer = track_producer("test", None);
		let mut straggler = producer.create_group(0u64.into()).unwrap();
		// The live edge moves on, so the straggler is demoted and expirable.
		producer.create_group(1u64.into()).unwrap().finish().unwrap();

		let mut frame = straggler
			.create_frame_owned(frame::Info {
				size: 3,
				timestamp: Timestamp::ZERO,
			})
			.unwrap();

		// The sender stalls past the retention window, then the whole payload lands in
		// one poll turn: the loop never returns `Pending`, so `notify` is never reached
		// and `finish` is the only write the charge sees.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
		frame.write(bytes::Bytes::from_static(b"abc")).unwrap();
		frame.finish().unwrap();
		straggler.finish().unwrap();

		// A new group runs the expiry scan.
		producer.create_group(2u64.into()).unwrap().finish().unwrap();

		let state = producer.state.read();
		assert!(
			state.lookup.contains_key(&0),
			"a group whose frame just completed must not expire"
		);
	}

	/// Re-offering a parked group (once the cap rises) is a delivery: it restarts
	/// the expiry clock so the subscriber gets to read what it was just handed.
	#[tokio::test]
	async fn parked_reoffer_restarts_the_expiry_clock() {
		let producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);
		subscriber.set_groups(..1);

		for seq in 0..2u64 {
			let mut group = producer.create_group(seq.into()).unwrap();
			group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
			group.finish().unwrap();
		}

		// Group 0 is in range; group 1 is beyond the cap and parks.
		assert_eq!(subscriber.assert_group().sequence, 0);
		subscriber.assert_no_group();

		// Just inside the window, the cap rises and the re-offer stamps group 1.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
		subscriber.set_groups(..2);
		let mut reading = subscriber.assert_group();
		assert_eq!(reading.sequence, 1);

		// Almost another full window passes: far beyond the write, inside the
		// re-offer stamp. The new group runs the expiry scan.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
		producer.create_group(2u64.into()).unwrap().finish().unwrap();

		let frame = reading.read_frame().await.unwrap();
		assert!(frame.is_some(), "a just-re-offered group must not expire unread");
	}

	#[tokio::test]
	async fn evict_keeps_max_sequence() {
		let producer = track_producer("test", None);
		producer.append_group().unwrap(); // seq 0

		// Advance time past the LRU window.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));

		// Append another group; seq 0 is expired and evicted.
		producer.append_group().unwrap(); // seq 1

		{
			let state = producer.state.read();
			assert_eq!(live_groups(&state), 1);
			assert_eq!(first_live_sequence(&state), 1);
			assert_eq!(state.offset, 1);
		}
	}

	#[tokio::test]
	async fn no_eviction_when_fresh() {
		let producer = track_producer("test", None);
		producer.append_group().unwrap(); // seq 0
		producer.append_group().unwrap(); // seq 1
		producer.append_group().unwrap(); // seq 2

		{
			let state = producer.state.read();
			assert_eq!(live_groups(&state), 3);
			assert_eq!(state.offset, 0);
		}
	}

	#[tokio::test]
	async fn consumer_skips_evicted_groups() {
		let producer = track_producer("test", None);
		producer.append_group().unwrap(); // seq 0

		let mut consumer = producer.subscribe(None);

		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
		producer.append_group().unwrap(); // seq 1

		// Group 0 was evicted. Consumer should get group 1.
		let group = consumer.assert_group();
		assert_eq!(group.sequence, 1);
	}

	/// Mint a track under an origin whose pool has the given wall-clock LRU window.
	fn track_producer_expiring(name: impl Into<Arc<str>>, expiry: impl Into<Option<Duration>>) -> Producer {
		track_producer_pooled(name, cache::Pool::new(cache::Config::default().with_expiry(expiry)))
	}

	/// Mint a track under an origin caching into `pool`.
	fn track_producer_pooled(name: impl Into<Arc<str>>, pool: cache::Pool) -> Producer {
		Producer::new(
			Arc::new(broadcast::Info {
				pool,
				..Default::default()
			}),
			name,
			None,
		)
	}

	/// The write path is not the only thing that runs expiry: a pool sweep reclaims a
	/// track's idle groups even when the track never writes again, which is the only
	/// bound on a publisher that stalls with a group still open.
	#[tokio::test]
	async fn pool_sweep_expires_without_a_write() {
		let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
		let producer = track_producer_pooled("test", pool.clone());
		let mut stalled = producer.append_group().unwrap(); // seq 0, left open
		stalled.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
		producer.append_group().unwrap(); // seq 1, the live edge

		crate::model::clock::advance(Duration::from_secs(2));
		pool.sweep();

		assert!(
			!producer.state.read().lookup.contains_key(&0),
			"the sweep reclaimed an idle open group with no write behind it"
		);
	}

	#[test]
	fn cache_gc_dates_activity_before_expiring_it() {
		let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
		let now = crate::model::clock::now();
		assert_eq!(pool.gc(now), Some(now + Duration::from_millis(500)));
		let producer = track_producer_pooled("test", pool.clone());
		let mut group = producer.append_group().unwrap();
		group.write_frame(Timestamp::ZERO, b"first".as_slice()).unwrap();
		producer.append_group().unwrap();
		// Activity written while cleanup was idle gets the new supplied time.
		let later = now + Duration::from_secs(60);
		pool.gc(later);
		assert!(producer.state.read().lookup.contains_key(&0));
		pool.gc(later + Duration::from_secs(2));
		assert!(!producer.state.read().lookup.contains_key(&0));
	}

	#[test]
	fn cache_gc_reaches_old_entries_behind_a_fresh_front() {
		let expiry = Duration::from_secs(1);
		let pool = cache::Pool::new(cache::Config::default().with_expiry(expiry));
		let producer = track_producer_pooled("test", pool.clone());
		let now = crate::model::clock::now();
		let groups: Vec<_> = (0..EVICT_SCAN * 3).map(|_| producer.append_group().unwrap()).collect();
		producer.append_group().unwrap();
		pool.gc(now);
		// Keep more than a write scan's worth of leading entries fresh.
		for group in &groups[..EVICT_SCAN * 2] {
			group.cache_refresh();
		}
		pool.gc(now + expiry * 2);
		let state = producer.state.read();
		for sequence in 0..EVICT_SCAN * 2 {
			assert!(state.lookup.contains_key(&(sequence as u64)), "fresh front survives");
		}
		for sequence in EVICT_SCAN * 2..EVICT_SCAN * 3 {
			assert!(!state.lookup.contains_key(&(sequence as u64)), "old tail is reclaimed");
		}
	}

	/// One sweep drains a whole idle backlog, not a rotating window of it: a quiet
	/// track has no writes left to revisit the rest of the queue with, so a bounded
	/// pass would leave the oldest groups parked for a backlog's length in windows.
	#[tokio::test]
	async fn pool_sweep_drains_a_deep_backlog() {
		let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
		let producer = track_producer_pooled("test", pool.clone());

		// Comfortably more than one write-driven scan window (EVICT_SCAN).
		let backlog = 4 * EVICT_SCAN;
		for _ in 0..backlog {
			let mut group = producer.append_group().unwrap();
			group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
		}
		producer.append_group().unwrap(); // the live edge, always protected

		crate::model::clock::advance(Duration::from_secs(2));
		pool.sweep();

		let state = producer.state.read();
		let stale = (0..backlog as u64).filter(|seq| state.lookup.contains_key(seq)).count();
		assert_eq!(stale, 0, "one sweep reclaimed the whole idle backlog");
	}

	#[tokio::test]
	async fn pool_expiry_controls_eviction() {
		// A shorter LRU window on the pool evicts sooner than the default.
		let producer = track_producer_expiring("test", Duration::from_secs(1));
		producer.append_group().unwrap(); // seq 0

		// Past the pool's window but well within cache::DEFAULT_EXPIRY.
		crate::model::clock::advance(Duration::from_secs(2));
		producer.append_group().unwrap(); // seq 1

		// Seq 0 is gone because the pool only keeps idle groups for 1s.
		let state = producer.state.read();
		assert_eq!(live_groups(&state), 1);
		assert_eq!(first_live_sequence(&state), 1);
	}

	#[tokio::test]
	async fn small_frame_write_expires_idle_siblings() {
		let producer = track_producer_expiring("test", Duration::from_secs(1));
		producer.append_group().unwrap().finish().unwrap(); // seq 0
		let mut live = producer.append_group().unwrap(); // seq 1

		crate::model::clock::advance(Duration::from_secs(2));
		live.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();

		let expired = !producer.state.read().lookup.contains_key(&0);
		assert!(expired, "a small frame write runs expiry");
	}

	#[tokio::test]
	async fn fresh_expiry_scan_does_not_wake_track_consumers() {
		use std::sync::atomic::{AtomicBool, Ordering};

		let producer = track_producer_expiring("test", cache::DEFAULT_EXPIRY);
		producer.append_group().unwrap().finish().unwrap();
		let mut live = producer.append_group().unwrap();
		let mut consumer = producer.subscribe(None);
		assert_eq!(consumer.assert_group().sequence, 0);
		assert_eq!(consumer.assert_group().sequence, 1);

		let woken = Arc::new(AtomicBool::new(false));
		let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));
		assert!(consumer.poll_recv_group(&waiter).is_pending());

		live.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
		assert!(
			!woken.load(Ordering::SeqCst),
			"a no-op expiry scan must not wake track consumers"
		);
	}

	#[tokio::test]
	async fn streaming_frame_write_expires_idle_siblings() {
		let producer = track_producer_expiring("test", Duration::from_secs(1));
		producer.append_group().unwrap().finish().unwrap(); // seq 0
		let mut live = producer.append_group().unwrap(); // seq 1
		let mut frame = live
			.create_frame(frame::Info {
				size: 1,
				timestamp: Timestamp::ZERO,
			})
			.unwrap();

		crate::model::clock::advance(Duration::from_secs(2));
		frame.write(b"x".as_slice()).unwrap();

		let expired = !producer.state.read().lookup.contains_key(&0);
		assert!(expired, "a streamed chunk runs expiry");
	}

	#[tokio::test]
	async fn appended_datagram_expires_idle_groups() {
		let mut producer = track_producer_expiring("test", Duration::from_secs(1));
		producer.append_group().unwrap().finish().unwrap(); // seq 0
		producer.append_group().unwrap().finish().unwrap(); // seq 1

		crate::model::clock::advance(Duration::from_secs(2));
		producer.append_datagram(Timestamp::ZERO, b"x".as_slice()).unwrap();

		let expired = !producer.state.read().lookup.contains_key(&0);
		assert!(expired, "an appended datagram runs expiry");
	}

	#[tokio::test]
	async fn forwarded_datagram_expires_idle_groups() {
		let mut producer = track_producer_expiring("test", Duration::from_secs(1));
		producer.append_group().unwrap().finish().unwrap(); // seq 0
		producer.append_group().unwrap().finish().unwrap(); // seq 1

		crate::model::clock::advance(Duration::from_secs(2));
		producer
			.insert_datagram(2, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
			.unwrap();

		let expired = !producer.state.read().lookup.contains_key(&0);
		assert!(expired, "a forwarded datagram runs expiry");
	}

	/// A track's `max_age` is a media-timestamp budget: it does not drive wall-clock
	/// eviction, so a stall (no accesses, no timestamp progress) shorter than the
	/// pool's LRU window can't age content out no matter how small the window is.
	#[tokio::test]
	async fn max_age_does_not_drive_wall_eviction() {
		let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(1)));
		producer.append_group().unwrap(); // seq 0

		// Far past max_age in wall time, but inside the pool's LRU window.
		crate::model::clock::advance(Duration::from_secs(10));
		producer.append_group().unwrap(); // seq 1

		let state = producer.state.read();
		assert_eq!(live_groups(&state), 2, "max_age is media time, not a wall clock");
	}

	/// Disabling the pool's expiry keeps idle groups until byte pressure reclaims them.
	#[tokio::test]
	async fn disabled_pool_expiry_never_reclaims() {
		let producer = track_producer_expiring("test", None);
		producer.append_group().unwrap(); // seq 0

		crate::model::clock::advance(Duration::from_secs(3600));
		producer.append_group().unwrap(); // seq 1

		let state = producer.state.read();
		assert_eq!(live_groups(&state), 2);
	}

	#[test]
	fn max_age_clamped_to_cache() {
		let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));

		// A max age budget beyond the cache is capped in the aggregate; a group can't be
		// waited for longer than the publisher keeps it. The subscriber's own preference
		// is stored verbatim, so what it asked for stays readable.
		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
		assert_eq!(subscriber.subscription().max_age, Duration::from_secs(10));
		assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));

		// A budget within the cache is left alone, and ZERO (skip immediately) stays ZERO.
		subscriber
			.update(Subscription::default().with_max_age(Duration::from_millis(500)))
			.unwrap();
		assert_eq!(producer.subscription().unwrap().max_age, Duration::from_millis(500));

		subscriber
			.update(Subscription::default().with_max_age(Duration::ZERO))
			.unwrap();
		assert_eq!(producer.subscription().unwrap().max_age, Duration::ZERO);
	}

	/// Mint a track under an origin whose retention ceiling is `cap`, so the
	/// track's own window is clamped down to it on bind.
	fn track_producer_capped(name: impl Into<Arc<str>>, info: Info, cap: Duration) -> Producer {
		Producer::new(
			Arc::new(broadcast::Info {
				cache_duration: cap,
				..Default::default()
			}),
			name,
			info,
		)
	}

	#[test]
	fn origin_cache_duration_clamps_max_age() {
		// A publisher asking to keep groups for a minute is capped to the origin's 1s
		// ceiling; a publisher already below the ceiling is left alone (it's a min).
		let capped = track_producer_capped(
			"test",
			Info::default().with_max_age(Duration::from_secs(60)),
			Duration::from_secs(1),
		);
		assert_eq!(capped.state.read().max_age_bound(), Some(Duration::from_secs(1)));
		assert_eq!(capped.subscribe(None).info().max_age, Duration::from_secs(1));

		let under = track_producer_capped(
			"test",
			Info::default().with_max_age(Duration::from_millis(500)),
			Duration::from_secs(1),
		);
		assert_eq!(under.state.read().max_age_bound(), Some(Duration::from_millis(500)));
	}

	/// The origin ceiling clamps the media-timestamp budget only; wall-clock
	/// reclamation belongs to the pool's LRU window, not the ceiling.
	#[tokio::test]
	async fn origin_cache_duration_does_not_wall_evict() {
		let producer = track_producer_capped(
			"test",
			Info::default().with_max_age(Duration::from_secs(60)),
			Duration::from_secs(1),
		);
		producer.append_group().unwrap(); // seq 0

		// Far past the ceiling in wall time, but inside the pool's LRU window.
		crate::model::clock::advance(Duration::from_secs(2));
		producer.append_group().unwrap(); // seq 1

		let state = producer.state.read();
		assert_eq!(live_groups(&state), 2);
	}

	#[test]
	fn max_age_clamped_via_every_update_path() {
		let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));
		let over = Subscription::default().with_max_age(Duration::from_secs(10));

		// The clamp lives in the aggregation, so it applies no matter which entry point
		// wrote the raw preference. Previously only `Subscriber::update` clamped.
		let mut subscriber = producer.subscribe(over.clone());
		assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));

		subscriber.control().update(over.clone()).unwrap();
		assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));

		subscriber.update(over).unwrap();
		assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
	}

	#[test]
	fn max_age_aggregate_clamps_across_subscribers() {
		let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));

		// The aggregate takes the max, then clamps once. Equivalent to clamping each
		// subscriber first, since `min` distributes over `max`.
		let _a = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
		let _b = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));

		assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
	}

	/// Append a finished group presenting at `millis`, so the track carries a media
	/// timeline for the drift budget to measure against.
	fn append_at(producer: &mut Producer, millis: u64) -> u64 {
		let mut group = producer.append_group().unwrap();
		group
			.write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x"))
			.unwrap();
		group.finish().unwrap();
		group.sequence
	}

	/// Every group the subscriber can read right now, in delivery order.
	fn drain(subscriber: &mut Subscriber) -> Vec<u64> {
		let mut sequences = Vec::new();
		while let Some(Ok(Some(group))) = subscriber.recv_group().now_or_never() {
			sequences.push(group.sequence);
		}
		sequences
	}

	#[test]
	fn real_time_skips_a_backlog_to_the_live_edge() {
		let mut producer = track_producer("test", None);
		for second in 0..5 {
			append_at(&mut producer, second * 1000);
		}

		// The default budget is REAL_TIME: a subscriber joining a track that already
		// holds five seconds of history takes the live edge, not the history. This is
		// the ceiling a takeover backlog runs into.
		let mut subscriber = producer.subscribe(None);
		assert_eq!(drain(&mut subscriber), vec![4]);

		// And it stays caught up: the next group is live when it lands.
		append_at(&mut producer, 5000);
		assert_eq!(drain(&mut subscriber), vec![5]);
	}

	#[test]
	fn real_time_skips_a_backlog_after_catching_up() {
		let mut producer = track_producer("test", None);
		append_at(&mut producer, 0);
		let mut subscriber = producer.subscribe(None);
		assert_eq!(drain(&mut subscriber), vec![0]);

		// Catch-up is a subscriber state, not a startup-only choice. A reader can pause
		// between groups and still needs to shed the backlog that accumulated meanwhile.
		for second in 1..6 {
			append_at(&mut producer, second * 1000);
		}
		assert_eq!(drain(&mut subscriber), vec![5]);
	}

	#[test]
	fn a_newer_edge_changes_an_active_catch_up() {
		let mut producer = track_producer("test", None);
		for second in 0..5 {
			append_at(&mut producer, second * 1000);
		}
		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));
		assert_eq!(
			subscriber
				.recv_group()
				.now_or_never()
				.unwrap()
				.unwrap()
				.unwrap()
				.sequence,
			3
		);

		// Group 4 was the edge at the first delivery. Advancing twice puts it outside
		// the budget before the subscriber asks for another group.
		append_at(&mut producer, 5000);
		append_at(&mut producer, 10000);
		assert_eq!(drain(&mut subscriber), vec![5, 6]);
	}

	#[test]
	fn a_growing_edge_changes_an_active_catch_up() {
		let mut producer = track_producer("test", None);
		append_at(&mut producer, 0);
		append_at(&mut producer, 1000);
		let mut edge = producer.append_group().unwrap();
		edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"a"))
			.unwrap();

		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(2)));
		assert_eq!(
			subscriber
				.recv_group()
				.now_or_never()
				.unwrap()
				.unwrap()
				.unwrap()
				.sequence,
			0
		);

		// The edge is still the same group, but its newest presentation moved far
		// enough that group 1 has fallen out of range.
		edge.write_frame(Timestamp::from_millis(5000).unwrap(), bytes::Bytes::from_static(b"b"))
			.unwrap();
		assert_eq!(drain(&mut subscriber), vec![2]);
	}

	#[test]
	fn a_budget_admits_groups_within_it() {
		let mut producer = track_producer("test", None);
		for second in 0..5 {
			append_at(&mut producer, second * 1000);
		}

		// Two seconds of tolerance keeps the groups presenting within 2s of the live
		// edge (2s, 3s, 4s) and drops the two below it. Group 1 reaches exactly 2s behind
		// the edge, and the reach bound is exclusive, so every frame it could hold is
		// already past the budget.
		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(2)));
		assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
	}

	#[test]
	fn a_budget_reaches_back_over_the_cache() {
		let mut producer = track_producer("test", None);
		for second in 0..5 {
			append_at(&mut producer, second * 1000);
		}

		// The default zero budget calls every non-latest group stale, so a subscription
		// that says nothing joins at the live edge.
		let mut live = producer.subscribe(None);
		assert_eq!(drain(&mut live), vec![4]);

		// Two seconds of tolerance covers the groups presenting within 2s of the live
		// edge, so the same join is handed the head of what it can still use. One bound
		// decides both what is sent and what is expired, so a subscriber is never sent
		// history it would discard on arrival.
		let budget = Subscription::default().with_max_age(Duration::from_secs(2));
		let mut subscriber = producer.subscribe(budget);
		assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
	}

	#[test]
	fn a_named_start_is_a_floor_not_a_request() {
		let mut producer = track_producer("test", None);
		for second in 0..5 {
			append_at(&mut producer, second * 1000);
		}

		// The budget is the only thing that asks for data; a named start only bounds how
		// far back it may reach. Naming group 1 at real time still delivers the live edge
		// alone, since the zero budget calls everything older stale.
		let named = Subscription::default().with_start(Position::group(1));
		let mut subscriber = producer.subscribe(named);
		assert_eq!(drain(&mut subscriber), vec![4]);

		// A budget reaching further back than the floor is cut off at it.
		let floored = Subscription::default()
			.with_start(Position::group(3))
			.with_max_age(Duration::from_secs(10));
		let mut subscriber = producer.subscribe(floored);
		assert_eq!(drain(&mut subscriber), vec![3, 4]);

		// A floor below what the budget admits changes nothing.
		let slack = Subscription::default()
			.with_start(Position::group(1))
			.with_max_age(Duration::from_secs(2));
		let mut subscriber = producer.subscribe(slack);
		assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
	}

	#[test]
	fn a_floor_above_the_live_edge_waits_there() {
		let mut producer = track_producer("test", None);
		for second in 0..3 {
			append_at(&mut producer, second * 1000);
		}

		// A resumed subscription names where it left off, which may not exist yet. The
		// cursor sits at the floor rather than sliding back to what is cached.
		let resumed = Subscription::default()
			.with_start(Position::group(7))
			.with_max_age(Duration::from_secs(10));
		let mut subscriber = producer.subscribe(resumed);
		assert_eq!(drain(&mut subscriber), Vec::<u64>::new());
		append_at(&mut producer, 3000); // sequence 3: still below the floor
		assert_eq!(drain(&mut subscriber), Vec::<u64>::new());
		for second in 4..8 {
			append_at(&mut producer, second * 1000);
		}
		assert_eq!(drain(&mut subscriber), vec![7]);
	}

	#[test]
	fn a_late_lower_group_within_the_budget_is_delivered() {
		let producer = track_producer("test", None);
		for (sequence, millis) in [(5, 0), (6, 1000), (7, 2000)] {
			let mut group = producer.create_group(group::Info { sequence }).unwrap();
			group
				.write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x"))
				.unwrap();
			group.finish().unwrap();
		}

		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(5)));
		assert_eq!(drain(&mut subscriber), vec![5, 6, 7]);

		// Arriving below everything already delivered is not what makes content stale:
		// the budget is the only gate, and this straggler's timestamp is within it. A
		// consumer that needs sequence order reorders (or drops) it itself.
		let mut late = producer.create_group(group::Info { sequence: 4 }).unwrap();
		late.write_frame(Timestamp::from_millis(500).unwrap(), bytes::Bytes::from_static(b"late"))
			.unwrap();
		late.finish().unwrap();
		assert_eq!(drain(&mut subscriber), vec![4]);
	}

	#[test]
	fn drift_is_measured_in_presentation_time_not_arrival_time() {
		let mut producer = track_producer("test", None);
		// A relay ingesting a backlog creates every group at once, so arrival time says
		// they are all equally fresh. Their timestamps say otherwise, which is the whole
		// point of measuring in presentation time.
		for second in 0..4 {
			append_at(&mut producer, second * 1000);
		}

		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(1500)));
		assert_eq!(drain(&mut subscriber), vec![1, 2, 3]);
	}

	#[tokio::test]
	async fn a_stamped_successor_expires_an_unstamped_group() {
		let mut producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);
		producer.append_group().unwrap(); // seq 0 stalls before its first frame

		append_at(&mut producer, 1000); // seq 1 proves the live feed moved on

		// The candidate needs no timestamp of its own: its reach is where its stamped
		// successor begins, which the zero budget already puts out of range.
		assert_eq!(drain(&mut subscriber), vec![1]);
	}

	#[tokio::test]
	async fn a_handed_out_group_expires_while_its_first_frame_is_stalled() {
		let mut producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);
		producer.append_group().unwrap();

		let mut stalled = subscriber.recv_group().await.unwrap().expect("stalled group");
		let pending = tokio::spawn(async move { stalled.read_frame().await });
		tokio::task::yield_now().await;
		assert!(
			!pending.is_finished(),
			"the empty live edge still waits for its first frame"
		);

		crate::model::clock::advance(Duration::from_secs(1));
		append_at(&mut producer, 1000);

		// It ends rather than fails: the reader took every frame the group ever had
		// (none), so nothing was truncated. What it was waiting for was the producer,
		// and a group abandoned where its reader stands looks exactly like one that
		// ended there.
		let result = pending.await.unwrap();
		assert!(matches!(result, Ok(None)), "the held group ends: {result:?}");
	}

	/// A first timestamp on a *newer* group can convict a held one, so the held reader has
	/// to be woken by it. The conviction needs a group beyond the held one's successor:
	/// a group is bounded by where its successor begins, so the successor itself never
	/// proves it stale.
	#[tokio::test]
	async fn a_handed_out_group_wakes_when_a_newer_group_gets_its_first_timestamp() {
		let mut producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
		let mut old = producer.append_group().unwrap();
		old.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"old"))
			.unwrap();

		// Group 1 bounds group 0's reach at 1s.
		append_at(&mut producer, 1000);

		let mut held = subscriber.recv_group().await.unwrap().expect("old group");
		assert!(held.read_frame().await.unwrap().is_some());

		let mut live = producer.append_group().unwrap();
		let pending = tokio::spawn(async move { held.read_frame().await });
		tokio::task::yield_now().await;
		assert!(!pending.is_finished(), "the newer group has no timestamp yet");

		// 2s edge against group 0's 1s reach: a full second past the budget.
		live.write_frame(
			Timestamp::from_millis(2000).unwrap(),
			bytes::Bytes::from_static(b"live"),
		)
		.unwrap();
		tokio::task::yield_now().await;

		assert!(pending.is_finished(), "the new presentation edge wakes the held reader");
		// Drained, so the budget ends the group rather than truncating it.
		let result = pending.await.unwrap();
		assert!(matches!(result, Ok(None)), "the held group ends: {result:?}");
	}

	/// The ordinary live case, at the default real-time budget: 2s GOPs produced one at
	/// a time and read as they arrive. The budget must take the live edge without
	/// shortening the group the reader is already on, so every frame of every group is
	/// delivered and each group *ends* rather than fails at its boundary.
	///
	/// Two things would break this. Measuring drift from a group's first frame rather
	/// than its reader's position convicts every group the moment its successor opens,
	/// since a live reader is always a little behind the edge. And the reader is parked
	/// at its group's end when that happens, because the FIN and the next group's first
	/// frame are separate events and the wire does not order them.
	#[tokio::test]
	async fn real_time_reads_a_live_stream_without_truncating_it() {
		let producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);

		let gop = |n: u64| {
			[
				Timestamp::from_millis(n * 2000).unwrap(),
				Timestamp::from_millis(n * 2000 + 1900).unwrap(),
			]
		};
		let write = |group: &mut group::Producer, timestamp| {
			group.write_frame(timestamp, bytes::Bytes::from_static(b"x")).unwrap();
		};

		let mut open = producer.append_group().unwrap();
		write(&mut open, gop(0)[0]);
		write(&mut open, gop(0)[1]);
		let mut reading = subscriber.recv_group().await.unwrap().expect("the live group");

		let mut read = Vec::new();
		for n in 1..5u64 {
			let sequence = reading.sequence;
			let mut frames = 0;
			while let Some(res) = reading.read_frame().now_or_never() {
				match res.expect("no truncation while draining") {
					Some(_) => frames += 1,
					None => panic!("group {sequence} ended early"),
				}
			}
			read.push((sequence, frames));

			let next = {
				// Parked at the end of the current group: every frame is read and no FIN
				// has landed.
				let mut end = std::pin::pin!(reading.read_frame());
				assert!(futures::poll!(end.as_mut()).is_pending(), "parked on the FIN");

				// The next keyframe opens its group. The verdict is taken here, in the
				// window before the previous group's FIN arrives.
				let mut opened = producer.append_group().unwrap();
				write(&mut opened, gop(n)[0]);
				let verdict = futures::poll!(end.as_mut());

				open.finish().unwrap();
				let res = match verdict {
					Poll::Ready(res) => res,
					Poll::Pending => end.await,
				};
				assert!(
					matches!(res, Ok(None)),
					"group {sequence} ends at the boundary rather than failing: {res:?}"
				);
				opened
			};
			let mut next = next;
			write(&mut next, gop(n)[1]);

			reading = subscriber.recv_group().await.unwrap().expect("the next live group");
			open = next;
		}

		assert_eq!(read, vec![(0, 2), (1, 2), (2, 2), (3, 2)], "every frame of every group");
	}

	/// A budget is spent from where the reader stands, not from where its group opened.
	///
	/// A 2s GOP with a 1s budget: the reader has drained to 1900ms when the next group
	/// opens at 2000ms, so it is 100ms behind the live edge and well inside what it
	/// asked for. A straggling frame of the old group arriving after the new one opened
	/// (which is ordinary, the two are separate streams) must still reach it.
	///
	/// Measuring from the group's first frame instead makes the drift 2000ms, so a
	/// budget shorter than one GOP would drop the tail of every GOP.
	#[tokio::test]
	async fn a_budget_is_measured_from_the_readers_position() {
		let producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));

		let mut open = producer.append_group().unwrap();
		open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"key"))
			.unwrap();
		open.write_frame(
			Timestamp::from_millis(1900).unwrap(),
			bytes::Bytes::from_static(b"tail"),
		)
		.unwrap();

		let mut reading = subscriber.recv_group().await.unwrap().expect("the live group");
		assert!(reading.read_frame().await.unwrap().is_some());
		assert!(reading.read_frame().await.unwrap().is_some());

		let mut end = std::pin::pin!(reading.read_frame());
		assert!(futures::poll!(end.as_mut()).is_pending(), "parked at 1900ms");

		// The next GOP opens. The reader is 100ms behind it, inside its 1s budget.
		let mut next = producer.append_group().unwrap();
		next.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"key"))
			.unwrap();
		assert!(
			futures::poll!(end.as_mut()).is_pending(),
			"a reader inside its budget is not expired by the next group opening"
		);

		// A straggler from the old group, still within the budget.
		open.write_frame(
			Timestamp::from_millis(1950).unwrap(),
			bytes::Bytes::from_static(b"late"),
		)
		.unwrap();
		let late = end.await.expect("the straggler is not truncated");
		assert_eq!(
			late.map(|frame| frame.timestamp),
			Some(Timestamp::from_millis(1950).unwrap())
		);
	}

	/// A group the budget ends rather than fails stays ended. `expired` alone would
	/// turn the clean answer into [`Error::Old`] on the next poll, and a caller is
	/// allowed to probe again past the end of a group.
	#[tokio::test]
	async fn an_ended_group_stays_ended_when_probed_again() {
		let mut producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);
		let mut open = producer.append_group().unwrap();
		open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a"))
			.unwrap();

		let mut group = subscriber.recv_group().await.unwrap().expect("group");
		assert!(group.read_frame().await.unwrap().is_some());

		let probes = tokio::spawn(async move {
			let first = group.read_frame().await;
			let second = group.read_frame().await;
			let finished = group.finished().await;
			(first, second, finished)
		});
		tokio::task::yield_now().await;

		append_at(&mut producer, 1000);

		let (first, second, finished) = probes.await.unwrap();
		assert!(matches!(first, Ok(None)), "the group ends: {first:?}");
		assert!(matches!(second, Ok(None)), "and stays ended: {second:?}");
		assert!(matches!(finished, Ok(1)), "reporting what it delivered: {finished:?}");
	}

	#[tokio::test]
	async fn a_drained_group_finishes_cleanly_after_the_live_edge_advances() {
		let mut producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);
		append_at(&mut producer, 0);

		let mut group = subscriber.recv_group().await.unwrap().expect("first group");
		assert!(group.read_frame().await.unwrap().is_some());

		append_at(&mut producer, 1000);

		assert!(group.read_frame().await.unwrap().is_none());
		assert!(!group.latency_expired());
	}

	#[tokio::test]
	async fn a_handed_out_partial_frame_expires_while_its_payload_is_stalled() {
		let mut producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);
		let mut source = producer.append_group().unwrap();
		let mut writing = source
			.create_frame(frame::Info {
				size: 6,
				timestamp: Timestamp::ZERO,
			})
			.unwrap();
		writing.write(bytes::Bytes::from_static(b"old")).unwrap();

		let mut group = subscriber.recv_group().await.unwrap().expect("partial group");
		let mut frame = group.next_frame().await.unwrap().expect("partial frame");
		assert_eq!(
			frame.read_chunk().await.unwrap(),
			Some(bytes::Bytes::from_static(b"old"))
		);
		let pending = tokio::spawn(async move { frame.read_chunk().await });
		tokio::task::yield_now().await;
		assert!(!pending.is_finished(), "the partial payload is still stalled");

		crate::model::clock::advance(Duration::from_secs(1));
		append_at(&mut producer, 1000);

		let result = pending.await.unwrap();
		assert!(
			matches!(result, Err(Error::Old)),
			"the in-flight frame expires: {result:?}"
		);
		writing.abort(Error::Cancel).unwrap();
	}

	#[test]
	fn max_age_bounds_the_budget() {
		// The publisher only keeps a group around for 500ms, so a subscriber asking to
		// wait ten seconds for one still gives up at 500ms: the same clamp the aggregate
		// applies, on the subscriber's own side of it.
		let mut producer = track_producer("test", Info::default().with_max_age(Duration::from_millis(500)));
		append_at(&mut producer, 0);
		append_at(&mut producer, 1000);
		append_at(&mut producer, 2000);

		// Group 0 reaches 1s, a full second behind the 2s edge, so the clamped 500ms
		// budget drops it. An unclamped ten seconds would have kept it.
		let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
		assert_eq!(drain(&mut subscriber), vec![1, 2]);
	}

	#[test]
	fn an_explicit_start_gets_no_exemption_from_the_budget() {
		let mut producer = track_producer("test", None);
		for second in 0..4 {
			append_at(&mut producer, second * 1000);
		}

		// Asking to start at the beginning is a filter, not a request for reliability.
		// Backfill needs a budget that covers it; without one the live edge still wins.
		let mut subscriber = producer.subscribe(Subscription::default().with_start(Position::group(0)));
		subscriber.start_at(0);
		assert_eq!(drain(&mut subscriber), vec![3]);

		let mut patient = producer.subscribe(
			Subscription::default()
				.with_start(Position::group(0))
				.with_max_age(Duration::from_secs(10)),
		);
		patient.start_at(0);
		assert_eq!(drain(&mut patient), vec![0, 1, 2, 3]);
	}

	/// A group's age is where its content *ends*, not where it began. A long group whose
	/// tail is level with the live edge still owes the reader every frame in it, so
	/// judging it by its first timestamp would discard exactly the group being filled.
	#[test]
	fn a_long_group_is_not_stale_while_its_tail_reaches_the_edge() {
		let mut producer = track_producer("test", None);

		// Group 0 spans 0..2000ms; group 1 starts at 2000ms, where group 0 ends.
		let mut long = producer.append_group().unwrap();
		for ms in [0u64, 500, 1000, 1500, 2000] {
			long.write_frame(Timestamp::from_millis(ms).unwrap(), bytes::Bytes::from_static(b"x"))
				.unwrap();
		}
		long.finish().unwrap();
		append_at(&mut producer, 2000);

		// A budget far shorter than the group's own span still keeps it.
		let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
		assert_eq!(drain(&mut sub), vec![0, 1]);
	}

	/// The same group, once its tail has fallen behind, is stale like any other.
	/// The same long group, once its *successor* has itself fallen behind. A frame's
	/// duration is not on the wire, so group 0's own last timestamp proves nothing about
	/// where it ends; only group 1's start bounds it. Group 0 is convicted once that bound
	/// is further behind the edge than the budget allows.
	#[test]
	fn a_long_group_is_stale_once_its_successor_falls_behind() {
		let mut producer = track_producer("test", None);

		let mut long = producer.append_group().unwrap();
		for ms in [0u64, 500, 1000] {
			long.write_frame(Timestamp::from_millis(ms).unwrap(), bytes::Bytes::from_static(b"x"))
				.unwrap();
		}
		long.finish().unwrap();
		// Group 0 reaches at most 3s (where group 1 starts), a full second behind the 4s
		// edge and so past the budget. Group 1 reaches the edge itself and is kept.
		append_at(&mut producer, 3000);
		append_at(&mut producer, 4000);

		let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
		assert_eq!(drain(&mut sub), vec![1, 2]);
	}

	/// An unstamped immediate successor leaves a group's reach unbounded: a later
	/// stamped group proves nothing about where the successor will begin, and
	/// shrinking the bound is the unsafe direction.
	#[tokio::test]
	async fn an_unstamped_immediate_successor_leaves_reach_unbounded() {
		let mut producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);

		append_at(&mut producer, 0); // seq 0
		producer.append_group().unwrap(); // seq 1 stalls before its first frame
		append_at(&mut producer, 10_000); // seq 2

		// Group 1's reach is group 2's start, a full edge behind: stale at zero budget.
		// Group 0's reach is unknown until group 1 presents its first frame, so it is
		// kept rather than convicted on a bound that could shrink the wrong way.
		assert_eq!(drain(&mut subscriber), vec![0, 2]);
	}

	/// Reach is the *immediate* successor's start, never a minimum across later groups.
	/// Timestamps need not rise with sequence: a rewind can put a much earlier timestamp on
	/// a much later group, and that group proves nothing about where the candidate's own
	/// successor begins. Taking the minimum would shrink the bound and discard content that
	/// is still well inside the budget.
	#[test]
	fn reach_follows_the_immediate_successor_not_a_later_rewind() {
		let mut producer = track_producer("test", None);

		// Group 0 is bounded by group 1 at 10s. Groups 2 and 3 rewind to 1s and 2s.
		append_at(&mut producer, 0);
		append_at(&mut producer, 10_000);
		append_at(&mut producer, 1_000);
		append_at(&mut producer, 2_000);

		// Group 0 reaches 10s, so nothing here proves it is past a 500ms budget: it could
		// hold frames through nearly 10s. A minimum over later groups would put its reach
		// at 1s and drop it.
		let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
		assert!(
			drain(&mut sub).contains(&0),
			"group 0 is bounded by its successor at 10s, not by a later rewind"
		);
	}

	/// Datagrams are unordered by construction, so the sequence cursor carries them too:
	/// a track using both channels needs one subscription, not two.
	#[tokio::test]
	async fn ordered_carries_datagrams() {
		let mut producer = track_producer("test", None);
		let mut sub = producer.subscribe(None).ordered();

		producer
			.insert_datagram(5, Timestamp::from_millis(5).unwrap(), bytes::Bytes::from_static(b"x"))
			.unwrap();
		producer.create_group(group::Info { sequence: 3 }).unwrap();

		let datagram = sub
			.recv_datagram()
			.now_or_never()
			.expect("datagram would have blocked")
			.expect("would have errored")
			.expect("track was closed");
		assert_eq!(datagram.sequence, 5);

		// The datagram did not consume the group cursor.
		let group = sub
			.next_group()
			.now_or_never()
			.expect("group would have blocked")
			.expect("would have errored")
			.expect("track was closed");
		assert_eq!(group.sequence, 3);
	}

	/// Every group the ordered cursor can read right now, in sequence order.
	fn drain_ordered(subscriber: &mut Ordered) -> Vec<u64> {
		let mut sequences = Vec::new();
		while let Some(Ok(Some(group))) = subscriber.next_group().now_or_never() {
			sequences.push(group.sequence);
		}
		sequences
	}

	/// A cached backlog is not free to deliver: a consumer that stalls and resumes would
	/// otherwise replay it at 1x and stay behind forever, since nothing it reads ever
	/// blocks. The budget applies as the cursor reads, exactly as on the arrival cursor.
	#[test]
	fn next_group_sheds_a_stale_backlog() {
		let mut producer = track_producer("test", None);
		for second in 0..4 {
			append_at(&mut producer, second * 1000);
		}

		let mut subscriber = producer.subscribe(None).ordered();
		assert_eq!(drain_ordered(&mut subscriber), vec![3]);

		// The arrival cursor, which the relay forwarders use, sheds the same backlog.
		let mut arrival = producer.subscribe(None);
		assert_eq!(drain(&mut arrival), vec![3]);
	}

	/// Only what is provably too old goes. A group is convicted by its *reach* (where its
	/// successor begins), so a budget spanning part of the backlog keeps every group that
	/// could still present something inside it, and the burst is delivered gap-free.
	#[test]
	fn next_group_keeps_a_backlog_inside_the_budget() {
		let mut producer = track_producer("test", None);
		for second in 0..4 {
			append_at(&mut producer, second * 1000);
		}

		// Group 0 reaches 1s, a full 2s behind the 3s edge, so it is gone. Group 1
		// reaches 2s and could still present up to it: inside a 1.5s budget.
		let mut subscriber = producer
			.subscribe(Subscription::default().with_max_age(Duration::from_millis(1500)))
			.ordered();
		assert_eq!(drain_ordered(&mut subscriber), vec![1, 2, 3]);

		// A budget covering the whole history still bursts it in full.
		let mut replay = producer.subscribe(replay()).ordered();
		assert_eq!(drain_ordered(&mut replay), vec![0, 1, 2, 3]);
	}

	/// A group whose immediate successor has not presented a frame has no proven reach,
	/// so nothing says its frames are too old and the ordered cursor keeps it.
	#[test]
	fn next_group_keeps_a_group_with_no_proven_reach() {
		let mut producer = track_producer("test", None);
		append_at(&mut producer, 0); // seq 0
		producer.append_group().unwrap(); // seq 1 stalls before its first frame
		append_at(&mut producer, 10_000); // seq 2

		// Group 1 reaches 10s, a full edge behind: convicted. Group 0 is bounded only by
		// group 1, which has yet to say where it begins.
		let mut subscriber = producer.subscribe(None).ordered();
		assert_eq!(drain_ordered(&mut subscriber), vec![0, 2]);
	}

	#[tokio::test]
	async fn real_time_skips_older_sequences_with_equal_ages() {
		let mut producer = track_producer("test", None);
		append_at(&mut producer, 0);
		append_at(&mut producer, 0);

		let mut subscriber = producer.subscribe(None);
		assert_eq!(drain(&mut subscriber), vec![1]);
	}

	#[test]
	fn fetch_ignores_the_budget() {
		let mut producer = track_producer("test", None);
		for second in 0..4 {
			append_at(&mut producer, second * 1000);
		}

		// A fetch names one old group explicitly, so there is no live edge to drift
		// from: the budget bounds a subscription, not a request for a specific group.
		let consumer = producer.consume();
		let group = consumer.fetch_group(0, None).now_or_never().unwrap().unwrap();
		assert_eq!(group.sequence, 0);
	}

	/// A one-shot fetch populates the shared cache but never the live arrival cursor,
	/// so it cannot make content available to a subscription look stale.
	#[tokio::test]
	async fn fetched_group_is_not_a_live_drift_edge() {
		let mut producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();
		append_at(&mut producer, 0);

		let pending = consumer.fetch_group(100, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("fetch request is ready")
			.unwrap();
		let mut fetched = req.accept(None).unwrap();
		fetched
			.write_frame(
				Timestamp::from_millis(100_000).unwrap(),
				bytes::Bytes::from_static(b"fetched"),
			)
			.unwrap();
		fetched.finish().unwrap();
		pending.await.unwrap();

		let mut groups = producer.subscribe(None);
		assert_eq!(groups.assert_group().sequence, 0);
		groups.assert_no_group();
	}

	/// The live edge is resolved once per poll and applied to every group that poll
	/// walks off, so it has to be revalidated before it convicts anything: an eviction
	/// in between would otherwise discard a group on the strength of content that is no
	/// longer there to jump to.
	#[test]
	fn an_evicted_live_edge_convicts_nothing() {
		let mut producer = track_producer("test", None);
		append_at(&mut producer, 0);
		let edge = append_at(&mut producer, 30_000);

		let state = producer.state.read();
		let drift = Drift {
			budget: Duration::ZERO,
			edge: state.live_edge(None),
		};
		assert!(
			state.is_stale(0, drift.edge.as_ref(), drift.budget),
			"stale against a live edge"
		);
		drop(state);

		// The edge dies between resolving it and judging the candidate.
		let slot = producer.modify().unwrap().lookup.remove(&edge).unwrap();
		let _ = slot.group.abort(Error::Evicted);

		let state = producer.state.read();
		assert!(
			!state.is_stale(0, drift.edge.as_ref(), drift.budget),
			"a vanished edge is no reason to drop what is left"
		);
	}

	#[tokio::test]
	async fn a_lower_sequence_is_never_the_live_edge() {
		let producer = track_producer("test", None);
		// A high timestamp on a lower sequence is not a live edge: backfill served on
		// demand sits there, and so does the tail of a timeline the publisher rewound.
		// Only groups above the candidate anchor the measure, so neither can convict
		// the group that follows it.
		let mut straggler = producer.create_group(0u64.into()).unwrap();
		straggler
			.write_frame(Timestamp::from_millis(60_000).unwrap(), bytes::Bytes::from_static(b"x"))
			.unwrap();
		straggler.finish().unwrap();

		let mut rewound = producer.create_group(1u64.into()).unwrap();
		rewound
			.write_frame(Timestamp::from_millis(0).unwrap(), bytes::Bytes::from_static(b"x"))
			.unwrap();
		rewound.finish().unwrap();

		let mut subscriber = producer.subscribe(replay());
		assert_eq!(drain(&mut subscriber), vec![0, 1]);
	}

	#[test]
	fn a_requested_end_does_not_cap_the_live_edge() {
		let mut producer = track_producer("test", None);
		for second in 0..4 {
			append_at(&mut producer, second * 1000);
		}

		// `Subscription::end` is a request to the publisher, folded in with every other
		// subscriber's, and it does not filter this handle: the groups above it arrive
		// anyway. Capping the live edge with it would pin the edge below them, leaving
		// everything past it with nothing newer to be late against.
		let mut subscriber = producer.subscribe(Subscription::default().with_end(Position::after_group(1)));
		assert_eq!(drain(&mut subscriber), vec![3]);
	}

	#[test]
	fn a_capped_subscriber_measures_drift_against_its_cap() {
		let mut producer = track_producer("test", None);
		append_at(&mut producer, 0);
		append_at(&mut producer, 1000);

		// Capped at group 0: the route running on past the cap is data this subscriber
		// can never be served, so it isn't a live edge to be late against. This is what
		// keeps a spliced segment from dropping the groups either side of a takeover
		// boundary.
		let mut subscriber = producer.subscribe(Subscription::default().with_end(Position::after_group(0)));
		subscriber.set_groups(..1);
		assert_eq!(drain(&mut subscriber), vec![0]);

		// Raising the cap re-offers the parked group, now measured against the wider
		// edge it just admitted.
		subscriber.set_groups(..);
		assert_eq!(drain(&mut subscriber), vec![1]);
	}

	#[test]
	fn subscriber_control_updates_while_read_future_is_pending() {
		let producer = track_producer("test", None);
		let mut subscriber = producer.subscribe(None);
		let control = subscriber.control();

		let mut recv = Box::pin(subscriber.recv_group());
		assert!(recv.as_mut().now_or_never().is_none());

		control.update(Subscription::default().with_priority(7)).unwrap();

		let aggregate = producer.subscription().expect("expected an active subscription");
		assert_eq!(aggregate.priority, 7);
	}

	#[test]
	fn dropped_subscriber_leaves_no_ghost_in_aggregate() {
		// Regression (#2351): a departed subscriber must not keep contributing its
		// last subscription to the aggregate. When it did, a relay's linger loop
		// never observed the track going idle, and an identical viewer reconnecting
		// within the linger window was reset when the stale timer fired.
		let mut producer = track_producer("test", None);
		let a = producer.subscribe(Subscription::default().with_priority(5));

		// Prime the change cursor: the aggregate currently has one subscriber.
		let waiter = kio::Waiter::noop();
		assert!(
			matches!(producer.poll_subscription_changed(&waiter), Poll::Ready(Ok(Some(_)))),
			"one live subscriber should aggregate to Some",
		);

		// The only subscriber leaves.
		drop(a);

		// The aggregate must report the drop to None, not the ghost's last value.
		assert!(
			matches!(producer.poll_subscription_changed(&waiter), Poll::Ready(Ok(None))),
			"a dropped subscriber must not linger in the aggregate",
		);

		// And the snapshot used by the linger loop must agree.
		assert!(
			producer.subscription().is_none(),
			"snapshot must exclude a dropped subscriber",
		);
	}

	#[test]
	fn dropped_subscriber_wakes_the_aggregate() {
		// The value being right isn't enough: nothing re-polls the aggregate on its
		// own, so the drop has to wake the waiter. A subscriber contributing demand
		// takes `kio::Consumer::poll`'s Ready path, which registers no waiter, so
		// the departure needs the closed waiter armed explicitly. Without it a relay
		// never learns the last viewer left and holds the upstream subscription (and
		// the upstream's viewer count) open forever.
		use std::sync::atomic::{AtomicBool, Ordering};

		let mut producer = track_producer("test", None);
		let a = producer.subscribe(Subscription::default().with_priority(5));

		let woken = Arc::new(AtomicBool::new(false));
		let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));

		// Prime the cursor, then confirm the next poll parks.
		assert!(matches!(
			producer.poll_subscription_changed(&waiter),
			Poll::Ready(Ok(Some(_)))
		));
		assert!(
			producer.poll_subscription_changed(&waiter).is_pending(),
			"the aggregate is unchanged, so this poll must park",
		);
		assert!(!woken.load(Ordering::SeqCst), "nothing happened yet");

		drop(a);
		assert!(
			woken.load(Ordering::SeqCst),
			"the last subscriber leaving must wake the aggregate watcher",
		);
	}

	#[test]
	fn widest_subscriber_update_wakes_the_aggregate() {
		// The value counterpart of the drop above. A subscriber that widens the
		// fold takes the Ready path too, so its next update registered no waiter.
		// A relay whose upstream cap came from a downstream reader then never
		// learned that the reader lifted it, and the groups parked upstream never
		// resumed: every session stayed up and nothing flowed.
		use std::sync::atomic::{AtomicBool, Ordering};

		let mut producer = track_producer("test", None);
		let _narrow = producer.subscribe(Subscription::default().with_end(Position::after_group(3)));
		let mut wide = producer.subscribe(Subscription::default());

		let woken = Arc::new(AtomicBool::new(false));
		let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));

		assert!(matches!(
			producer.poll_subscription_changed(&waiter),
			Poll::Ready(Ok(Some(_)))
		));
		assert!(producer.poll_subscription_changed(&waiter).is_pending());
		assert!(!woken.load(Ordering::SeqCst), "nothing happened yet");

		wide.update(Subscription::default().with_end(Position::after_group(5)))
			.unwrap();
		assert!(
			woken.load(Ordering::SeqCst),
			"the widest subscriber changing must wake the aggregate watcher",
		);
		match producer.poll_subscription_changed(&waiter) {
			Poll::Ready(Ok(Some(sub))) => assert_eq!(sub.end, Position::after_group(5)),
			other => panic!("expected the narrowed aggregate, got {other:?}"),
		}
	}

	/// An [`ArcWake`] that just records that it was woken.
	struct FlagWake(Arc<std::sync::atomic::AtomicBool>);

	impl futures::task::ArcWake for FlagWake {
		fn wake_by_ref(arc_self: &Arc<Self>) {
			arc_self.0.store(true, std::sync::atomic::Ordering::SeqCst);
		}
	}

	#[tokio::test]
	async fn out_of_order_max_sequence_at_front() {
		let producer = track_producer("test", None);

		// Arrive out of order: seq 5 first, then 3, then 4.
		producer.create_group(group::Info { sequence: 5 }).unwrap();
		producer.create_group(group::Info { sequence: 3 }).unwrap();
		producer.create_group(group::Info { sequence: 4 }).unwrap();

		// max_sequence = 5, which is at the front of the VecDeque.
		{
			let state = producer.state.read();
			assert_eq!(state.max_sequence, Some(5));
		}

		// Expire all three groups.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));

		// Append seq 6 (becomes new max_sequence).
		producer.append_group().unwrap(); // seq 6

		// Seq 3, 4, 5 are all expired. Seq 5 was the old max_sequence but now 6 is.
		// All old groups are evicted.
		{
			let state = producer.state.read();
			assert_eq!(live_groups(&state), 1);
			assert_eq!(first_live_sequence(&state), 6);
			assert!(!state.lookup.contains_key(&3));
			assert!(!state.lookup.contains_key(&4));
			assert!(!state.lookup.contains_key(&5));
			assert!(state.lookup.contains_key(&6));
		}
	}

	#[tokio::test]
	async fn max_sequence_at_front_blocks_trim() {
		let producer = track_producer("test", None);

		// Arrive: seq 5, then seq 3.
		producer.create_group(group::Info { sequence: 5 }).unwrap();

		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));

		// Seq 3 arrives late; max_sequence is still 5 (at front).
		producer.create_group(group::Info { sequence: 3 }).unwrap();

		// Seq 5 is max_sequence (protected). Seq 3 is not expired (just created).
		// Nothing should be evicted.
		{
			let state = producer.state.read();
			assert_eq!(live_groups(&state), 2);
			assert_eq!(state.offset, 0);
		}

		// Expire seq 3 as well.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));

		// Seq 2 arrives late, triggering eviction.
		producer.create_group(group::Info { sequence: 2 }).unwrap();

		// Seq 5 is the live edge (protected) and still resolves at the front of
		// `arrival`, so nothing is trimmed and the offset stays. Seq 3 expired out of
		// `lookup`, leaving a hole its arrival entry no longer resolves; seq 2 is
		// fresh and kept.
		{
			let state = producer.state.read();
			assert_eq!(live_groups(&state), 2);
			assert_eq!(state.offset, 0);
			assert!(state.lookup.contains_key(&5));
			assert!(!state.lookup.contains_key(&3));
			assert!(state.lookup.contains_key(&2));
		}

		// Consumer should still be able to read through the hole.
		let mut consumer = producer.subscribe(None);
		let group = consumer.assert_group();
		// consume() starts at index 0; the first arrival entry that still resolves is seq 5.
		assert_eq!(group.sequence, 5);
	}

	#[tokio::test]
	async fn abort_clears_cached_groups() {
		let producer = track_producer("test", None);
		producer.append_group().unwrap();
		producer.append_group().unwrap();

		// A stale consumer that never drains must not pin the cached groups.
		let mut consumer = producer.subscribe(None);
		assert_eq!(live_groups(&producer.state.read()), 2);

		producer.clone().abort(Error::Cancel).unwrap();

		{
			let state = producer.state.read();
			assert!(state.lookup.is_empty(), "cached groups should be dropped on abort");
			assert!(state.arrival.is_empty());
			assert!(state.evict.is_empty());
		}

		// The consumer now surfaces the abort error rather than the leftover cache.
		let result = consumer.recv_group().now_or_never().expect("should not block");
		assert!(matches!(result, Err(Error::Cancel)));
	}

	#[tokio::test]
	async fn drop_unfinished_clears_cached_groups() {
		let producer = track_producer("test", None);
		let writer = producer.clone();
		writer.append_group().unwrap();

		// A stale consumer keeps the channel (and thus the cache) alive.
		let mut consumer = producer.subscribe(None);
		assert_eq!(live_groups(&producer.state.read()), 1);

		// Drop every producer without finishing: the cache is released.
		drop(writer);
		drop(producer);

		let result = consumer.recv_group().now_or_never().expect("should not block");
		assert!(matches!(result, Err(Error::Dropped)));
	}

	#[tokio::test]
	async fn drop_after_abort_does_not_warn() {
		// abort() closes the channel after recording `abort`. Drop must treat the
		// read-only guard returned by write() as clean or it emits a false WARN.
		let warns = count_drop_warnings("track::Producer dropped without finish", || {
			let producer = track_producer("test", None);
			let keep = producer.clone();
			let writer = producer.clone();
			let group = writer.append_group().unwrap();
			group.finish().unwrap();
			let _consumer = producer.subscribe(None);
			writer.abort(Error::Cancel).unwrap();
			drop(keep);
		});
		assert_eq!(warns, 0, "abort-then-drop must not emit unfinished-producer WARN");
	}

	#[tokio::test]
	async fn drop_unfinished_warns() {
		let warns = count_drop_warnings("track::Producer dropped without finish", || {
			let producer = track_producer("test", None);
			let writer = producer.clone();
			writer.append_group().unwrap();
			let _consumer = producer.subscribe(None);
			drop(writer);
			drop(producer);
		});
		assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN");
	}

	#[tokio::test]
	async fn drop_finished_keeps_cached_groups() {
		let producer = track_producer("test", None);
		producer.append_group().unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.subscribe(None);
		drop(producer);

		// A cleanly finished track keeps its cache so the consumer can still drain.
		assert_eq!(consumer.assert_group().sequence, 0);
		let done = consumer.recv_group().now_or_never().expect("should not block").unwrap();
		assert!(done.is_none(), "consumer should drain then see clean finish");
	}

	#[tokio::test]
	async fn cached_groups_preserve_arrival_order() {
		let producer = track_producer("test", None);
		producer.create_group(group::Info { sequence: 5 }).unwrap();
		producer.create_group(group::Info { sequence: 3 }).unwrap();

		let groups = producer.consume().cached_groups();
		let sequences: Vec<u64> = groups.iter().map(|(group, _)| group.sequence).collect();
		assert_eq!(
			sequences,
			vec![5, 3],
			"warm snapshot must follow arrival, not sequence order"
		);
	}

	#[test]
	fn append_finish_cannot_be_rewritten() {
		let producer = track_producer("test", None);

		// Finishing an empty track is valid (fin = 0, total groups = 0).
		assert!(producer.finish().is_ok());
		assert!(producer.finish().is_err());
		assert!(producer.append_group().is_err());
	}

	#[test]
	fn finish_after_groups() {
		let producer = track_producer("test", None);

		producer.append_group().unwrap();
		assert!(producer.finish().is_ok());
		assert!(producer.finish().is_err());
		assert!(producer.append_group().is_err());
	}

	#[test]
	fn finish_at_rejects_a_boundary_at_or_below_the_live_edge() {
		let mut producer = track_producer("test", None);
		producer.create_group(group::Info { sequence: 5 }).unwrap();

		// The boundary is exclusive, so it must be strictly above the highest produced
		// group. 5 or below would orphan groups that already exist.
		assert!(producer.finish_at(4).is_err());
		assert!(producer.finish_at(5).is_err());
		assert!(producer.finish_at(6).is_ok());

		{
			let state = producer.state.read();
			assert_eq!(state.final_sequence, Some(6));
		}

		// Re-finishing is rejected, and no group at or above the boundary can be created.
		assert!(producer.finish_at(6).is_err());
		assert!(producer.create_group(group::Info { sequence: 4 }).is_ok());
		assert!(producer.create_group(group::Info { sequence: 6 }).is_err());
	}

	#[test]
	fn final_sequence_reports_the_declared_boundary() {
		let mut producer = track_producer("test", None);
		assert_eq!(producer.final_sequence(), None);

		producer.create_group(group::Info { sequence: 5 }).unwrap();
		assert_eq!(producer.final_sequence(), None, "a group does not declare a boundary");

		producer.finish_at(9).unwrap();
		assert_eq!(producer.final_sequence(), Some(9));

		// finish() would try to declare a second boundary, so callers check first.
		assert!(producer.finish().is_err());
	}

	#[test]
	fn final_sequence_reports_the_live_edge_after_finish() {
		let producer = track_producer("test", None);
		producer.create_group(group::Info { sequence: 5 }).unwrap();
		producer.finish().unwrap();
		assert_eq!(producer.final_sequence(), Some(6));
	}

	#[tokio::test]
	async fn finish_at_declares_a_future_boundary() {
		let mut producer = track_producer("test", None);
		producer.create_group(group::Info { sequence: 5 }).unwrap();

		// Learn the track ends at group 6 (exclusive 7) while the live edge is still 5.
		producer.finish_at(7).unwrap();

		let mut consumer = producer.subscribe(None);
		assert_eq!(consumer.assert_group().sequence, 5);

		// The boundary is known immediately, but the track isn't done: group 6 is still
		// outstanding, so the consumer parks rather than seeing end-of-stream.
		let boundary = consumer
			.finished()
			.now_or_never()
			.expect("boundary is known immediately")
			.expect("would have errored");
		assert_eq!(boundary, 7);
		assert!(
			consumer.recv_group().now_or_never().is_none(),
			"should wait for the outstanding group"
		);

		// The trailing group arrives (below the boundary), then the track completes.
		producer.create_group(group::Info { sequence: 6 }).unwrap();
		assert_eq!(consumer.assert_group().sequence, 6);
		let done = consumer
			.recv_group()
			.now_or_never()
			.expect("should not block")
			.expect("would have errored");
		assert!(done.is_none(), "track completes once the boundary is reached");
	}

	#[tokio::test]
	async fn recv_group_finishes_without_waiting_for_gaps() {
		let producer = track_producer("test", None);
		producer.create_group(group::Info { sequence: 1 }).unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.subscribe(None);
		assert_eq!(consumer.assert_group().sequence, 1);

		let done = consumer
			.recv_group()
			.now_or_never()
			.expect("should not block")
			.expect("would have errored");
		assert!(done.is_none(), "track should finish without waiting for gaps");
	}

	#[tokio::test]
	async fn next_group_skips_late_arrivals() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(None).ordered();

		// Seq 5 arrives first.
		producer.create_group(group::Info { sequence: 5 }).unwrap();
		let group = consumer
			.next_group()
			.now_or_never()
			.expect("should not block")
			.expect("would have errored")
			.expect("track should not be closed");
		assert_eq!(group.sequence, 5);

		// Seq 3 arrives late, skipped because 3 <= 5.
		producer.create_group(group::Info { sequence: 3 }).unwrap();
		// Seq 4 arrives late and is also skipped.
		producer.create_group(group::Info { sequence: 4 }).unwrap();
		// Seq 7 arrives and is returned.
		producer.create_group(group::Info { sequence: 7 }).unwrap();

		let group = consumer
			.next_group()
			.now_or_never()
			.expect("should not block")
			.expect("would have errored")
			.expect("track should not be closed");
		assert_eq!(group.sequence, 7);

		// No more groups. This would block.
		assert!(
			consumer.next_group().now_or_never().is_none(),
			"should block waiting for a higher sequence"
		);
	}

	#[tokio::test]
	async fn next_group_returns_arrivals_in_order() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay()).ordered();

		// Seq 3 arrives first, then seq 5. Both should be returned in arrival order.
		producer.create_group(group::Info { sequence: 3 }).unwrap();
		producer.create_group(group::Info { sequence: 5 }).unwrap();

		let group = consumer
			.next_group()
			.now_or_never()
			.expect("should not block")
			.expect("would have errored")
			.expect("track should not be closed");
		assert_eq!(group.sequence, 3);

		let group = consumer
			.next_group()
			.now_or_never()
			.expect("should not block")
			.expect("would have errored")
			.expect("track should not be closed");
		assert_eq!(group.sequence, 5);
	}

	#[tokio::test]
	async fn ordered_and_arrival_cursors_are_independent() {
		let producer = track_producer("test", None);
		let mut ordered = producer.subscribe(replay()).ordered();
		let mut arrival = producer.subscribe(replay());

		// Out-of-order arrivals: seq 5 first, then seq 3.
		producer.create_group(group::Info { sequence: 5 }).unwrap();
		producer.create_group(group::Info { sequence: 3 }).unwrap();

		// The ordered handle returns the smallest sequence first, regardless of
		// arrival order.
		let group = ordered
			.next_group()
			.now_or_never()
			.expect("should not block")
			.expect("would have errored")
			.expect("track should not be closed");
		assert_eq!(group.sequence, 3);

		// The plain handle walks arrivals, so it still starts at seq 5.
		assert_eq!(arrival.assert_group().sequence, 5);
	}

	#[tokio::test]
	async fn end_at_caps_next_group() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay()).ordered();

		for s in 0..6 {
			producer.create_group(group::Info { sequence: s }).unwrap();
		}

		consumer.set_groups(..3);

		// Groups 0, 1, 2 are within the cap.
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			1
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			2
		);

		// Group 3 is beyond the cap: next_group parks even though cached groups exist.
		assert!(
			consumer.next_group().now_or_never().is_none(),
			"capped consumer must block instead of returning out-of-range groups"
		);
	}

	#[tokio::test]
	async fn end_at_release_drains_cached_groups() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay()).ordered();

		for s in 0..6 {
			producer.create_group(group::Info { sequence: s }).unwrap();
		}

		consumer.set_groups(..2);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			1
		);
		assert!(consumer.next_group().now_or_never().is_none(), "capped at 2");

		// Raise the cap; previously-blocked cached groups become available again.
		consumer.set_groups(..5);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			2
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			3
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			4
		);
		assert!(consumer.next_group().now_or_never().is_none(), "capped at 5");

		// Remove the cap; everything remaining flows.
		consumer.set_groups(..);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			5
		);
		assert!(consumer.next_group().now_or_never().is_none(), "no more groups");
	}

	#[tokio::test]
	async fn end_at_lower_than_cursor_parks_consumer() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay()).ordered();

		for s in 0..3 {
			producer.create_group(group::Info { sequence: s }).unwrap();
		}

		// Drain everything with no cap.
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			1
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			2
		);

		// Lower the cap below the cursor. New groups beyond the cap are blocked.
		consumer.set_groups(..2);
		producer.create_group(group::Info { sequence: 3 }).unwrap();
		producer.create_group(group::Info { sequence: 4 }).unwrap();
		assert!(
			consumer.next_group().now_or_never().is_none(),
			"cap is below cursor; nothing returnable until cap rises"
		);

		// Restoring the cap to no-limit (or any value >= cursor) releases them.
		consumer.set_groups(..);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			3
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			4
		);
	}

	#[tokio::test]
	async fn end_at_toggling_around_late_arrivals() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay()).ordered();

		consumer.set_groups(..6);

		// Out-of-order arrivals all within the cap.
		producer.create_group(group::Info { sequence: 2 }).unwrap();
		producer.create_group(group::Info { sequence: 5 }).unwrap();
		producer.create_group(group::Info { sequence: 3 }).unwrap();
		// One beyond the cap; should be held even though it arrived in the middle.
		producer.create_group(group::Info { sequence: 8 }).unwrap();
		producer.create_group(group::Info { sequence: 4 }).unwrap();

		// next_group walks in sequence order through everything <= cap.
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			2
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			3
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			4
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			5
		);
		// Now blocked: 8 is still beyond the cap.
		assert!(consumer.next_group().now_or_never().is_none());

		// Raise the cap; cached seq 8 is finally served.
		consumer.set_groups(..11);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			8
		);
	}

	/// `recv_group` (arrival order) honors the `end_at` cap by parking, like
	/// `next_group`: beyond-cap groups are held, not dropped, and a raised cap
	/// re-offers them, even after the track finishes.
	#[tokio::test]
	async fn end_at_parks_recv_group() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay());

		for s in 0..3 {
			producer.create_group(group::Info { sequence: s }).unwrap();
		}

		consumer.set_groups(..2);
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			1
		);
		assert!(consumer.recv_group().now_or_never().is_none(), "capped at 2");

		// A finished track keeps the parked group claimable: the cap may rise.
		producer.finish().unwrap();
		assert!(
			consumer.recv_group().now_or_never().is_none(),
			"still parked after finish"
		);

		consumer.set_groups(..);
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			2
		);
		assert!(
			matches!(consumer.recv_group().now_or_never(), Some(Ok(None))),
			"finished once the parked group drains"
		);
	}

	/// A group beyond the cap must not block in-range groups that arrive behind
	/// it: a relay can ingest a burst micro-reordered (newest first).
	#[tokio::test]
	async fn recv_group_serves_arrivals_behind_the_cap() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay());

		consumer.set_groups(..2);

		// Reordered burst: the beyond-cap group arrives first.
		producer.create_group(group::Info { sequence: 2 }).unwrap();
		producer.create_group(group::Info { sequence: 0 }).unwrap();
		producer.create_group(group::Info { sequence: 1 }).unwrap();

		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			1
		);
		assert!(consumer.recv_group().now_or_never().is_none(), "capped at 2");

		consumer.set_groups(..3);
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			2
		);
	}

	#[tokio::test]
	async fn group_ranges_preserve_the_floor_when_the_cap_changes() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(None);
		consumer.set_groups(2..=2);
		producer.create_group(group::Info { sequence: 2 }).unwrap();
		assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 2);
		consumer.set_groups(..4);
		producer.create_group(group::Info { sequence: 1 }).unwrap();
		producer.create_group(group::Info { sequence: 3 }).unwrap();
		assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 3);
		consumer.set_groups(0..=4);
		producer.create_group(group::Info { sequence: 0 }).unwrap();
		producer.create_group(group::Info { sequence: 4 }).unwrap();
		assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 4);
	}

	/// A raised `start_at` drops parked groups it overtook instead of re-offering
	/// them once the cap rises.
	#[tokio::test]
	async fn start_at_drops_parked_recv_groups() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(None);

		consumer.set_groups(..1);
		producer.create_group(group::Info { sequence: 1 }).unwrap();
		assert!(
			consumer.recv_group().now_or_never().is_none(),
			"group 1 parked at the cap"
		);

		consumer.start_at(2);
		consumer.set_groups(..);
		producer.create_group(group::Info { sequence: 2 }).unwrap();
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			2,
			"the overtaken parked group is dropped, not re-offered"
		);
	}

	/// A parked group the producer aborts (eviction/expiry) is dropped: it is
	/// neither delivered once the cap rises nor allowed to hold the stream open
	/// after the track finishes. This is what bounds parking by the cache policy.
	#[tokio::test]
	async fn evicted_parked_recv_groups_are_dropped() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(None);

		producer.create_group(group::Info { sequence: 0 }).unwrap();
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);

		consumer.set_groups(..1);
		let straggler = producer.create_group(group::Info { sequence: 1 }).unwrap();
		assert!(
			consumer.recv_group().now_or_never().is_none(),
			"group 1 parked at the cap"
		);

		// The cache evicts the parked group (abort-as-tombstone), then the track ends.
		straggler.abort(Error::Old).unwrap();
		producer.finish().unwrap();

		consumer.set_groups(..);
		assert!(
			matches!(consumer.recv_group().now_or_never(), Some(Ok(None))),
			"a dead parked group must not be delivered or hold the stream open"
		);
	}

	/// Eviction aborts a parked group behind a sleeping subscriber's back. Nothing
	/// else will poll it (the track already finished), so the entry has to carry a
	/// waiter or the subscription sleeps forever holding its stream open.
	#[tokio::test]
	async fn evicted_parked_group_wakes_the_clean_end() {
		use std::sync::atomic::{AtomicUsize, Ordering};
		use std::task::{Context, Wake};

		/// A waker that counts its wakes, for asserting a pending poll left a live
		/// registration behind.
		struct CountWaker(AtomicUsize);
		impl Wake for CountWaker {
			fn wake(self: std::sync::Arc<Self>) {
				self.0.fetch_add(1, Ordering::SeqCst);
			}
		}

		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(None);

		producer.create_group(group::Info { sequence: 0 }).unwrap();
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);

		consumer.set_groups(..1);
		let straggler = producer.create_group(group::Info { sequence: 1 }).unwrap();
		assert!(consumer.recv_group().now_or_never().is_none(), "parked at the cap");
		producer.finish().unwrap();

		let counter = std::sync::Arc::new(CountWaker(AtomicUsize::new(0)));
		let waker = std::task::Waker::from(counter.clone());
		let mut cx = Context::from_waker(&waker);
		let mut fut = std::pin::pin!(consumer.recv_group());
		assert!(
			fut.as_mut().poll(&mut cx).is_pending(),
			"the parked group holds it open"
		);

		straggler.abort(Error::Old).unwrap();
		assert!(counter.0.load(Ordering::SeqCst) > 0, "the eviction wakeup was lost");
		assert!(matches!(fut.as_mut().poll(&mut cx), Poll::Ready(Ok(None))));
	}

	/// An exclusive cap at 0 is the empty range: no group is delivered, even group 0.
	#[tokio::test]
	async fn end_at_zero_is_the_empty_range() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay());
		producer.create_group(group::Info { sequence: 0 }).unwrap();
		producer.create_group(group::Info { sequence: 1 }).unwrap();

		consumer.set_groups(..0);
		assert!(
			consumer.recv_group().now_or_never().is_none(),
			"empty cap delivers nothing"
		);

		consumer.set_groups(..1);
		assert_eq!(
			consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		assert!(consumer.recv_group().now_or_never().is_none(), "group 1 stays parked");
	}

	/// A requested empty range plus a local empty cap still parks while another
	/// subscriber keeps aggregate demand unbounded.
	#[tokio::test]
	async fn empty_local_cap_holds_while_another_subscriber_requests_everything() {
		let producer = track_producer("test", None);
		let mut everything = producer.subscribe(replay());
		let mut empty = producer.subscribe(Subscription::default().with_end(Position::group(0)));
		empty.set_groups((Bound::Unbounded, Position::group(0).group_end()));

		for s in 0..3 {
			producer.create_group(group::Info { sequence: s }).unwrap();
		}

		assert_eq!(
			everything
				.recv_group()
				.now_or_never()
				.unwrap()
				.unwrap()
				.unwrap()
				.sequence,
			0
		);
		assert!(
			empty.recv_group().now_or_never().is_none(),
			"local empty cap must not ride the unbounded aggregate"
		);

		empty.set_groups(..2);
		assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 0);
		assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 1);
		assert!(empty.recv_group().now_or_never().is_none(), "still capped at 2");

		empty.set_groups(..);
		assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 2);
	}

	/// A frame-limited exclusive end includes the last group and stops before that frame.
	#[tokio::test]
	async fn end_at_frame_limited_last_group() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay()).ordered();
		let end = Position::after(1, 1).unwrap();
		consumer.set_groups((Bound::Unbounded, end.group_end()));

		for s in 0..3u64 {
			let mut group = producer.create_group(group::Info { sequence: s }).unwrap();
			for i in 0..3u8 {
				group.write_frame(Timestamp::ZERO, vec![i]).unwrap();
			}
			group.finish().unwrap();
		}

		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		let mut last = consumer.next_group().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(last.sequence, 1);
		last.set_frames(..end.frame);
		assert_eq!(
			last.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			0
		);
		assert_eq!(
			last.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			1
		);
		assert!(
			last.read_frame().now_or_never().unwrap().unwrap().is_none(),
			"frame cap is exclusive"
		);
		assert!(
			consumer.next_group().now_or_never().is_none(),
			"group 2 is past the exclusive group cap"
		);
	}

	/// An inclusive bound at the last group withholds nothing.
	#[tokio::test]
	async fn end_at_maximum_group_is_unbounded() {
		let producer = track_producer("test", None);
		let mut consumer = producer.subscribe(replay()).ordered();
		consumer.set_groups(..=u64::MAX);

		producer.create_group(group::Info { sequence: 0 }).unwrap();
		producer.create_group(group::Info { sequence: u64::MAX }).unwrap();

		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			0
		);
		assert_eq!(
			consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
			u64::MAX
		);
	}

	#[test]
	fn write_frame_rejects_an_oversized_frame_before_appending_its_group() {
		let mut producer = track_producer("test", None);
		let frame = bytes::Bytes::from(vec![0; group::MAX_CACHE_BYTES as usize + 1]);

		assert!(matches!(
			producer.write_frame(Timestamp::ZERO, frame),
			Err(Error::FrameTooLarge)
		));
		assert_eq!(producer.latest(), None, "the rejected frame did not publish a group");
	}

	#[test]
	fn append_group_returns_bounds_exceeded_on_sequence_overflow() {
		let producer = track_producer("test", None);
		{
			let mut state = producer.state.write().ok().unwrap();
			state.max_sequence = Some(u64::MAX);
		}

		assert!(matches!(producer.append_group(), Err(Error::BoundsExceeded(_))));
	}

	#[tokio::test]
	async fn fetch_cache_hit() {
		let producer = track_producer("test", None);

		// Produce a cached group.
		let mut group = producer.append_group().unwrap(); // seq 0
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hello"))
			.unwrap();
		group.finish().unwrap();

		// A cached group resolves immediately and never queues a request. `peek_group`
		// also returns it synchronously.
		let dynamic = producer.dynamic();
		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_some());
		let mut g = consumer.fetch_group(0, None).await.unwrap();
		assert_eq!(g.sequence, 0);
		assert_eq!(&g.read_frame().await.unwrap().unwrap().payload[..], b"hello");

		// Nothing was queued for the dynamic handler to serve.
		assert!(dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending());
	}

	#[tokio::test]
	async fn fetch_miss_signals_dynamic() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		// A cache miss isn't in `peek_group`, but a dynamic handler exists, so
		// `fetch_group` stays pending and queues a request. `*pending` derefs the
		// wrapper to the inner `Fetching` (a `kio::Pollable`).
		assert!(consumer.peek_group(5).is_none());
		let pending = consumer.fetch_group(5, group::Fetch::default().with_priority(7));
		assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());

		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		assert_eq!(req.sequence(), 5);
		assert_eq!(req.priority(), 7);

		// Serve it by accepting the request; the fetch then resolves.
		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hi"))
			.unwrap();
		group.finish().unwrap();

		let mut g = pending.await.unwrap();
		assert_eq!(g.sequence, 5);
		assert_eq!(&g.read_frame().await.unwrap().unwrap().payload[..], b"hi");
	}

	#[tokio::test]
	async fn fetch_miss_rejects() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		let pending = consumer.fetch_group(5, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();

		req.reject(Error::Cancel);
		assert!(matches!(pending.await, Err(Error::Cancel)));
		let fetch = producer.state.read().fetch.clone();
		assert!(fetch.read().is_empty());
	}

	#[tokio::test]
	async fn fetch_miss_drop_rejects() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		let pending = consumer.fetch_group(5, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();

		drop(req);
		assert!(matches!(pending.await, Err(Error::Dropped)));
	}

	#[tokio::test]
	async fn fetch_reject_does_not_poison_retry() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		let pending = consumer.fetch_group(5, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		req.reject(Error::Cancel);
		assert!(matches!(pending.await, Err(Error::Cancel)));

		let retry = consumer.fetch_group(5, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"retry"))
			.unwrap();
		group.finish().unwrap();

		let mut group = retry.await.unwrap();
		assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"retry");
	}

	/// A group cached from a frame-bounded subscription starts partway in. Serving it to
	/// someone who asked for the whole group would silently hand back a tail, so it is a
	/// miss and the fetch goes upstream instead.
	#[tokio::test]
	async fn fetch_ignores_a_group_that_starts_too_late() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		// The live subscription resumed mid-group, so only the tail is cached.
		let mut group = producer.create_group(group::Info { sequence: 0 }).unwrap();
		group.start_at(3).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"tail"))
			.unwrap();
		group.finish().unwrap();

		// A fetch for the tail is covered and resolves from the cache.
		let fetch = consumer.fetch_group(0, group::Fetch::default().with_frame_start(3));
		let cached = fetch.now_or_never().expect("covered by the cache").unwrap();
		assert_eq!(cached.index(), 3);

		// A fetch for the whole group is not, so it queues for a handler rather than
		// resolving to the tail.
		let mut fetch = std::pin::pin!(consumer.fetch_group(0, None));
		assert!(
			futures::poll!(fetch.as_mut()).is_pending(),
			"must not answer from the tail"
		);

		let request = dynamic.requested_group().await.unwrap();
		assert_eq!((request.sequence(), request.frame_start()), (0, 0));

		// Serving it replaces the too-narrow entry rather than colliding with it.
		let mut whole = request.accept(None).unwrap();
		whole
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
			.unwrap();
		whole.finish().unwrap();

		let mut served = fetch.await.unwrap();
		assert_eq!(served.index(), 0);
		assert_eq!(
			served.read_frame().await.unwrap().unwrap().payload,
			bytes::Bytes::from_static(b"head")
		);
	}

	/// Joining a queued fetch widens its range, and a caller that arrives once the range
	/// is already on the wire fails cleanly rather than being handed the narrower group.
	#[tokio::test]
	async fn fetch_widens_or_fails_cleanly() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		let _narrow = consumer.fetch_group(0, group::Fetch::default().with_frame_start(5));
		let mut narrow = std::pin::pin!(_narrow);
		assert!(futures::poll!(narrow.as_mut()).is_pending());

		// Still queued: widening is honored, because nothing has read the range yet.
		let _wider = consumer.fetch_group(0, group::Fetch::default().with_frame_start(2));
		let mut wider = std::pin::pin!(_wider);
		assert!(futures::poll!(wider.as_mut()).is_pending());

		let request = dynamic.requested_group().await.unwrap();
		assert_eq!(request.frame_start(), 2, "widened while queued");

		// Handed off now: the handler holds its own copy of the range, so a later wider
		// caller cannot move what is already being served.
		let _widest = consumer.fetch_group(0, group::Fetch::default().with_frame_start(0));
		let mut widest = std::pin::pin!(_widest);
		assert!(futures::poll!(widest.as_mut()).is_pending());
		assert_eq!(request.frame_start(), 2, "the in-flight range is already on the wire");

		let mut group = request.accept(None).unwrap();
		// A handler numbers the frames from where it was asked to start, as `serve_fetch`
		// does; that offset is what makes the cached group too narrow for `widest`.
		group.start_at(2).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"from2"))
			.unwrap();
		group.finish().unwrap();

		// The two callers it covers resolve; the one it doesn't fails cleanly rather
		// than being handed a group that starts above what it asked for.
		assert_eq!(narrow.await.unwrap().index(), 5);
		assert_eq!(wider.await.unwrap().index(), 2);
		assert!(matches!(widest.await, Err(Error::NotFound)));
	}

	#[tokio::test]
	async fn fetch_coalesces_concurrent() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		// Two fetches for the same uncached group produce ONE handler request,
		// carrying the higher of the two priorities.
		let first = consumer.fetch_group(5, group::Fetch::default().with_priority(1));
		let second = consumer.fetch_group(5, group::Fetch::default().with_priority(7));
		assert!(kio::Pollable::poll(&*first, &kio::Waiter::noop()).is_pending());

		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		assert_eq!(req.sequence(), 5);
		assert_eq!(req.priority(), 7);
		assert!(
			dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending(),
			"the second fetch queued a duplicate request"
		);

		// A fetch arriving while the request is already in flight joins it too.
		let third = consumer.fetch_group(5, None);

		// One accept resolves all of them.
		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hi"))
			.unwrap();
		group.finish().unwrap();

		assert_eq!(first.await.unwrap().sequence, 5);
		assert_eq!(second.await.unwrap().sequence, 5);
		assert_eq!(third.await.unwrap().sequence, 5);
	}

	#[tokio::test]
	async fn fetch_coalesced_reject_fails_all() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		let first = consumer.fetch_group(5, None);
		let second = consumer.fetch_group(5, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		req.reject(Error::Cancel);

		assert!(matches!(first.await, Err(Error::Cancel)));
		assert!(matches!(second.await, Err(Error::Cancel)));

		// The rejected attempt is gone: a retry starts a fresh one.
		let retry = consumer.fetch_group(5, None);
		assert!(kio::Pollable::poll(&*retry, &kio::Waiter::noop()).is_pending());
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		assert_eq!(req.sequence(), 5);
	}

	#[tokio::test]
	async fn fetch_queued_fails_when_handlers_leave() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		// Queued but never popped: the last handler leaving fails it fast.
		let pending = consumer.fetch_group(5, None);
		assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());
		drop(dynamic);
		assert!(matches!(pending.await, Err(Error::NotFound)));

		// And the attempt didn't leak.
		let fetch = producer.state.read().fetch.clone();
		assert!(fetch.read().is_empty());
	}

	#[tokio::test]
	async fn fetch_miss_no_dynamic_not_found() {
		// A track with no `Dynamic` can't serve old content, so a cache miss
		// resolves to NotFound instead of blocking forever.
		let producer = track_producer("test", None);
		producer.append_group().unwrap(); // seq 0, but we miss on seq 5
		let consumer = producer.consume();
		assert!(matches!(consumer.fetch_group(5, None).await, Err(Error::NotFound)));
	}

	#[tokio::test]
	async fn fetch_past_final_not_found() {
		let producer = track_producer("test", None);
		producer.append_group().unwrap(); // seq 0
		producer.finish().unwrap(); // final_sequence = 1

		// A group at or past the final sequence can never exist, even with a handler,
		// so it resolves to NotFound.
		let dynamic = producer.dynamic();
		let consumer = producer.consume();
		assert!(matches!(consumer.fetch_group(5, None).await, Err(Error::NotFound)));

		// And it doesn't signal the dynamic handler.
		assert!(dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending());
	}

	/// Mint a track whose groups charge into a bounded [`cache::Pool`].
	fn pooled_producer(capacity: u64) -> (Producer, cache::Pool) {
		let config = cache::Config::default()
			.with_capacity(capacity)
			.with_expiry(cache::DEFAULT_EXPIRY);
		let pool = cache::Pool::new(config);
		let broadcast = broadcast::Info {
			pool: pool.clone(),
			..Default::default()
		};
		let producer = Producer::new(Arc::new(broadcast), "test", None);
		(producer, pool)
	}

	fn finished_group(producer: &mut Producer, size: usize) -> u64 {
		let mut group = producer.append_group().unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; size]))
			.unwrap();
		group.finish().unwrap();
		group.sequence
	}

	/// While the pool is over capacity, every append accrues debt and pays it by
	/// evicting this track's own oldest groups, so the newest content survives.
	#[tokio::test]
	async fn debt_evicts_oldest_group() {
		// Fits one 10k group; each additional group pushes the pool over budget.
		let (mut producer, pool) = pooled_producer(10_000);

		finished_group(&mut producer, 10_000); // seq 0
		finished_group(&mut producer, 10_000); // seq 1: over budget, debt starts accruing
		finished_group(&mut producer, 10_000); // seq 2: pays by evicting seq 0

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_none(), "oldest group is evicted");
		assert!(consumer.peek_group(2).is_some(), "latest group survives");
		// Steady state carries the protected live edge plus the just-demoted group
		// (debt is charged before the demotion, so eviction lags one append).
		assert!(
			pool.used() <= 2 * (10_000 + cache::ENTRY_OVERHEAD),
			"usage hovers near capacity: {}",
			pool.used()
		);

		// A fresh subscriber skips the evicted groups entirely.
		let mut subscriber = producer.subscribe(replay());
		assert!(subscriber.assert_group().sequence > 0, "evicted group is not delivered");
	}

	/// The latest group is never in the eviction order, so it survives any budget.
	#[tokio::test]
	async fn latest_group_never_evicted() {
		// Far too small for even one group: the latest survives anyway.
		let (mut producer, pool) = pooled_producer(100);
		finished_group(&mut producer, 1000); // seq 0
		assert!(pool.used() > 100, "the latest may exceed the budget");

		// Later writes evict the demoted seq 0; each new latest is untouchable in turn.
		finished_group(&mut producer, 1000); // seq 1: demotes seq 0
		finished_group(&mut producer, 1000); // seq 2: pays by evicting seq 0

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_none());
		let mut group = consumer.peek_group(2).expect("latest survives");
		assert_eq!(group.read_frame().await.unwrap().unwrap().payload.len(), 1000);
	}

	/// A FETCH cache hit refreshes the group's access time: anything accessed more
	/// recently than the pool-wide average is protected, so the eviction walk skips
	/// it and evicts a never-read group instead, even one that arrived later.
	#[tokio::test]
	async fn fetch_refresh_survives_eviction() {
		let (mut producer, _pool) = pooled_producer(10_000);
		let consumer = producer.consume();

		finished_group(&mut producer, 3_000); // seq 0
		crate::model::clock::advance(Duration::from_secs(1));
		finished_group(&mut producer, 3_000); // seq 1
		crate::model::clock::advance(Duration::from_secs(1));
		finished_group(&mut producer, 3_000); // seq 2
		crate::model::clock::advance(Duration::from_millis(500));

		// FETCH seq 0: the cache hit lifts its access time above the average.
		let mut fetched = consumer.fetch_group(0, None).await.unwrap();
		assert_eq!(fetched.read_frame().await.unwrap().unwrap().payload.len(), 3_000);
		crate::model::clock::advance(Duration::from_millis(500));

		// Pressure: seq 0 is first in eviction order but freshly accessed, so it
		// rotates to the back and the never-read seq 1 dies instead.
		finished_group(&mut producer, 3_000); // seq 3
		crate::model::clock::advance(Duration::from_secs(1));
		finished_group(&mut producer, 3_000); // seq 4

		assert!(consumer.peek_group(0).is_some(), "refreshed group survives");
		assert!(consumer.peek_group(1).is_none(), "unread group is evicted instead");
	}

	/// A consumer holding an evicted group surfaces the eviction, not a hang or a
	/// truncated clean end.
	#[tokio::test]
	async fn eviction_aborts_readers() {
		let (mut producer, _pool) = pooled_producer(10_000);
		let mut subscriber = producer.subscribe(None);

		finished_group(&mut producer, 10_000); // seq 0
		let mut group0 = subscriber.assert_group();

		finished_group(&mut producer, 10_000); // seq 1: demotes seq 0
		finished_group(&mut producer, 10_000); // seq 2: pays by evicting seq 0

		let read = group0.read_frame().await;
		assert!(matches!(read, Err(Error::Evicted)), "expected Evicted, got {read:?}");
	}

	/// A write smaller than the next victim carries debt instead of evicting: a
	/// large group dies only once enough debt accumulates, never to pay off a
	/// far smaller write.
	#[tokio::test]
	async fn small_writes_carry_debt() {
		// Payloads dwarf the fixed per-group charge, so the budget arithmetic below is
		// about bytes written rather than bookkeeping.
		let unit = 100 * cache::ENTRY_OVERHEAD;
		let (mut producer, pool) = pooled_producer(22 * unit);
		let consumer = producer.consume();

		finished_group(&mut producer, 20 * unit as usize); // seq 0, the large victim-to-be

		// The first few small writes owe far less than seq 0's size: the debt
		// carries over instead of evicting it.
		for _ in 0..3 {
			finished_group(&mut producer, unit as usize);
		}
		assert!(consumer.peek_group(0).is_some(), "debt smaller than the victim carries");

		// Enough small writes accumulate the debt to finally evict it.
		for _ in 0..20 {
			finished_group(&mut producer, unit as usize);
		}
		assert!(
			consumer.peek_group(0).is_none(),
			"accumulated debt evicts the large group"
		);
		// Steady state hovers within about one group of capacity: a victim smaller
		// than the outstanding debt is never evicted, so the excess stays bounded.
		assert!(pool.used() <= 24 * unit, "usage hovers near capacity: {}", pool.used());
	}

	/// One write pays at most twice what it produced, so a capacity shrink (or one
	/// track's burst) drains gradually instead of one writer dumping its whole
	/// backlog in a single call.
	#[tokio::test]
	async fn payment_capped_per_write() {
		let (mut producer, pool) = pooled_producer(1 << 40);
		for _ in 0..10 {
			finished_group(&mut producer, 1_000);
		}

		// The governor slashes the target; nothing is reclaimed synchronously.
		pool.resize(100);
		let before = pool.used();

		// One 1k write may evict at most ~2k of backlog, not all ten groups.
		finished_group(&mut producer, 1_000);

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_none(), "the oldest groups are evicted");
		assert!(consumer.peek_group(1).is_none());
		assert!(consumer.peek_group(2).is_some(), "the backlog drains gradually");
		assert!(pool.used() > before - 4_000, "one write must not dump the backlog");
	}

	/// Accepting a track after pre-accept backfill must keep the same write
	/// counter: the counter is owned by the track state, so replacing the info
	/// can't strand the bytes already-created groups keep charging.
	#[tokio::test]
	async fn accept_preserves_write_accounting() {
		let config = cache::Config::default()
			.with_capacity(12_000)
			.with_expiry(cache::DEFAULT_EXPIRY);
		let pool = cache::Pool::new(config);
		let broadcast = broadcast::Info {
			pool: pool.clone(),
			..Default::default()
		};
		let request = Request::new(Arc::new(broadcast), "test");
		let dynamic = request.dynamic();
		let consumer = request.consume();

		// Serve a backfill before the track is accepted, then grow it.
		let pending = consumer.fetch_group(0, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		let mut backfill = req.accept(None).unwrap();
		pending.await.unwrap();
		backfill
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 30_000]))
			.unwrap();

		// Accept with a fresh Info: the pre-accept group's writes must still be
		// drained by this track's future charges.
		let producer = request.accept(None);
		producer.append_group().unwrap().finish().unwrap();
		producer.append_group().unwrap().finish().unwrap();

		assert!(
			producer.consume().peek_group(0).is_none(),
			"pre-accept backfill growth is reclaimed after accept"
		);
		assert!(pool.used() <= 13_000, "usage converges: {}", pool.used());
	}

	/// Re-serving a sequence many times must not accumulate eviction hints: stale
	/// hints die on stamp mismatch and compaction reclaims them.
	#[tokio::test]
	async fn recreated_sequence_bounds_eviction_hints() {
		let (producer, _pool) = pooled_producer(1 << 40);
		producer.create_group(5u64.into()).unwrap().finish().unwrap();

		for _ in 0..200 {
			let group = producer.create_group(1u64.into()).unwrap();
			group.abort(Error::Cancel).unwrap();
		}

		let state = producer.state.read();
		assert!(
			state.evict.len() <= 2 * state.lookup.len() + EVICT_SLACK,
			"stale hints are compacted: {} entries for {} slots",
			state.evict.len(),
			state.lookup.len()
		);
	}

	/// A frame write within the same coarse tick still outranks merely-inserted
	/// content, so the freshly-written group survives and the empty one pays.
	#[tokio::test]
	async fn same_tick_write_outranks_inserted() {
		// Payloads dwarf the fixed per-group charge, so the budget arithmetic below is
		// about bytes written rather than bookkeeping.
		let unit = 100 * cache::ENTRY_OVERHEAD;
		// No time advances: every stamp lands in the same tick.
		let (mut producer, _pool) = pooled_producer(10 * unit);

		producer.append_group().unwrap().finish().unwrap(); // seq 0: empty
		finished_group(&mut producer, 3 * unit as usize); // seq 1: written
		finished_group(&mut producer, 3 * unit as usize); // seq 2
		finished_group(&mut producer, 3 * unit as usize); // seq 3
		finished_group(&mut producer, 3 * unit as usize); // seq 4: over budget, pays

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_none(), "insert-only content pays first");
		assert!(consumer.peek_group(1).is_some(), "same-tick written content survives");
	}

	/// A track that only appends frames to an open group, never inserting another
	/// group, still settles its eviction debt once enough bytes accumulate.
	#[tokio::test]
	async fn frame_only_writer_pays() {
		let (producer, pool) = pooled_producer(2_000);
		let mut demoted = producer.append_group().unwrap(); // seq 0
		producer.append_group().unwrap().finish().unwrap(); // seq 1 demotes seq 0

		// One large frame crosses the charge threshold: the write itself pays,
		// with no further group insert on this track.
		demoted
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 300_000]))
			.unwrap();

		assert!(
			pool.used() <= 5_000,
			"the frame write settled the debt: {}",
			pool.used()
		);
		assert!(matches!(demoted.finish(), Err(Error::Evicted)));
	}

	/// One `Info` describing several tracks must not join their eviction accounting:
	/// each track opens its own account against the pool.
	#[tokio::test]
	async fn each_track_owns_its_account() {
		let broadcast = Arc::new(broadcast::Info::default());
		let info = Info::default();
		let a = Producer::new(broadcast.clone(), "a", info.clone());
		let b = Producer::new(broadcast, "b", info);

		let a = a.state.read().cache.clone();
		let b = b.state.read().cache.clone();
		assert!(!Arc::ptr_eq(&a, &b), "each track owns its account");
	}

	/// A `Dynamic` still serving fetches keeps the track alive, so the publisher
	/// letting go isn't an abrupt teardown: the handler can still serve the cache.
	#[tokio::test]
	async fn a_dynamic_defers_teardown() {
		let (mut producer, pool) = pooled_producer(1 << 40);
		let dynamic = producer.dynamic();
		finished_group(&mut producer, 100);

		drop(producer);
		assert!(pool.used() > 0, "the handler still serves the cache");

		drop(dynamic);
		assert_eq!(pool.used(), 0, "the last handle tears it down");
	}

	/// A finished track releases everything once every handle is gone.
	///
	/// Its groups hold the cache account, and the account links back here, so that link
	/// has to be weak: anything stronger makes the state (and every cached frame in it)
	/// immortal, even with no producer or consumer left.
	#[tokio::test]
	async fn finished_track_frees_its_cache() {
		let (mut producer, pool) = pooled_producer(1 << 40);
		finished_group(&mut producer, 100);
		producer.finish().unwrap();

		let state = producer.state.downgrade();
		drop(producer);

		assert!(state.upgrade().is_none(), "the track state is freed");
		assert_eq!(pool.used(), 0, "so are its cached bytes");
	}

	/// A group settling its eviction debt upgrades the account's weak handle, which
	/// counts as a producer on the track state. Teardown must not mistake that for a
	/// surviving publisher, or an abrupt drop silently behaves like a clean finish.
	#[tokio::test]
	async fn teardown_ignores_a_settling_group() {
		let (mut producer, pool) = pooled_producer(1 << 40);
		finished_group(&mut producer, 100);

		// Stand in for a concurrent `cache::Track::settle`, mid-upgrade.
		let settling = producer.state.downgrade().upgrade().expect("open");
		drop(producer);

		assert_eq!(pool.used(), 0, "the abrupt teardown still released the cache");
		drop(settling);
	}

	/// A subscriber holding one cached group must not pin the whole track: a group
	/// carries the track's properties by value, not a handle back to its state.
	#[tokio::test]
	async fn cached_group_outlives_its_track() {
		let (mut producer, pool) = pooled_producer(1 << 40);
		let sequence = finished_group(&mut producer, 100);
		let group = producer.consume().peek_group(sequence).expect("cached");
		producer.finish().unwrap();

		let state = producer.state.downgrade();
		drop(producer);
		assert!(state.upgrade().is_none(), "the track state is freed");
		assert!(pool.used() > 0, "the retained group keeps its own bytes");

		drop(group);
		assert_eq!(pool.used(), 0, "which it releases when dropped");
	}

	/// A backfill served before the track was accepted settles its own debt: the
	/// account exists from the moment the state does, so acceptance replacing the
	/// `Info` can't leave already-created groups writing for free.
	#[tokio::test]
	async fn pre_accept_backfill_settles_late_writes() {
		let config = cache::Config::default()
			.with_capacity(2_000)
			.with_expiry(cache::DEFAULT_EXPIRY);
		let pool = cache::Pool::new(config);
		let broadcast = broadcast::Info {
			pool: pool.clone(),
			..Default::default()
		};
		let request = Request::new(Arc::new(broadcast), "test");
		let dynamic = request.dynamic();
		let consumer = request.consume();

		// Serve backfill seq 0 before the track is accepted.
		let pending = consumer.fetch_group(0, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		let mut backfill = req.accept(None).unwrap();
		pending.await.unwrap();

		// Accept, then demote the backfill with a live group.
		let producer = request.accept(None);
		producer.append_group().unwrap().finish().unwrap();

		// No further insert: the late write into the demoted backfill is the only
		// thing that can pay the debt it just took on.
		backfill
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 300_000]))
			.unwrap();

		assert!(
			pool.used() <= 5_000,
			"the frame write settled the debt: {}",
			pool.used()
		);
	}

	/// A late frame write restarts the LRU clock (the window measures time since
	/// last written or fetched), so an actively-growing group is not expired as
	/// idle mid-write.
	#[tokio::test]
	async fn write_restarts_retention_clock() {
		let (producer, _pool) = pooled_producer(1 << 40);
		let mut straggler = producer.append_group().unwrap(); // seq 0
		producer.append_group().unwrap().finish().unwrap(); // seq 1 demotes seq 0

		// Idle past the window, then the straggler receives a late frame.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
		straggler
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
			.unwrap();
		producer.append_group().unwrap().finish().unwrap(); // seq 2 runs expiry

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_some(), "the write restarted the clock");

		// Once the writes stop, the group ages out normally.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
		producer.append_group().unwrap().finish().unwrap(); // seq 3 runs expiry
		assert!(consumer.peek_group(0).is_none(), "idle content still expires");
	}

	/// Continuously refreshed entries at the front of the eviction order must not
	/// starve expiry of entries behind them: the scan cursor rotates.
	#[tokio::test]
	async fn refreshed_front_does_not_starve_expiry() {
		let (producer, _pool) = pooled_producer(1 << 40);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		producer.create_group(10u64.into()).unwrap().finish().unwrap();
		for sequence in 1..=5u64 {
			let pending = consumer.fetch_group(sequence, None);
			let req = dynamic
				.requested_group()
				.now_or_never()
				.expect("should not block")
				.unwrap();
			let mut group = req.accept(None).unwrap();
			group
				.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
				.unwrap();
			group.finish().unwrap();
			pending.await.unwrap();
		}

		// Age everything out, then refresh the first four backfills so they sit
		// fresh at the front of the eviction order, hiding the expired fifth.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
		for sequence in 1..=4u64 {
			consumer.fetch_group(sequence, None).await.unwrap();
		}

		// The rotating cursor reaches the fifth entry within a few writes.
		for _ in 0..3 {
			producer.append_group().unwrap().finish().unwrap();
		}
		assert!(consumer.peek_group(5).is_none(), "expired backfill is reclaimed");
		assert!(consumer.peek_group(1).is_some(), "refreshed backfill survives");
	}

	/// A publisher re-creating an aborted sequence is delivered exactly once, at
	/// its actual arrival position: the historical arrival entry is dead.
	#[tokio::test]
	async fn recreated_sequence_delivered_once() {
		let (producer, _pool) = pooled_producer(1 << 40);

		producer.create_group(0u64.into()).unwrap().finish().unwrap();
		let aborted = producer.create_group(1u64.into()).unwrap();
		aborted.abort(Error::Cancel).unwrap();
		producer.create_group(2u64.into()).unwrap().finish().unwrap();
		producer.create_group(1u64.into()).unwrap().finish().unwrap();

		let mut subscriber = producer.subscribe(replay());
		assert_eq!(subscriber.assert_group().sequence, 0);
		assert_eq!(subscriber.assert_group().sequence, 2);
		assert_eq!(
			subscriber.assert_group().sequence,
			1,
			"replacement arrives at its own position"
		);
		subscriber.assert_no_group();
	}

	/// Datagrams share `max_sequence` but must not break group demotion: the live
	/// edge is tracked per group, so interleaving datagrams can't strand groups
	/// outside the eviction order and bypass the budget.
	#[tokio::test]
	async fn datagrams_do_not_block_eviction() {
		let (mut producer, pool) = pooled_producer(1_000);
		for _ in 0..10 {
			finished_group(&mut producer, 1_000);
			producer.append_datagram(Timestamp::ZERO, &b"beat"[..]).unwrap();
		}

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_none(), "old groups still evict");
		assert!(
			pool.used() < 4 * 1_256,
			"interleaved datagrams must not bypass the budget: {}",
			pool.used()
		);
	}

	/// An aborted group releases its access sample along with its bytes, from any
	/// handle: ghost samples must not linger in the pool mean where they'd hold it
	/// in the past and over-protect every live group.
	#[tokio::test]
	async fn aborted_group_leaves_no_ghost_sample() {
		let (producer, pool) = pooled_producer(1 << 40);
		let group0 = producer.append_group().unwrap();
		producer.append_group().unwrap(); // demotes seq 0 into the mean

		assert!(pool.average().is_some(), "demoted group is sampled");
		group0.abort(Error::Cancel).unwrap();
		assert_eq!(pool.average(), None, "the abort must remove the sample");
	}

	/// Empty groups still carry fixed overhead; they must repay the budget when
	/// evicted rather than being unevictable freeloaders.
	#[tokio::test]
	async fn empty_groups_repay_overhead() {
		let (producer, pool) = pooled_producer(1_000);
		for _ in 0..100 {
			let group = producer.append_group().unwrap();
			group.finish().unwrap();
		}

		assert!(
			pool.used() <= 3_000,
			"empty-group overhead must stay near the budget: {}",
			pool.used()
		);
	}

	/// Late growth on an already-demoted group is billed: the gross-write counter
	/// feeds debt on the next append, so a straggler can't grow unbounded.
	#[tokio::test]
	async fn growth_on_demoted_group_is_billed() {
		let (producer, pool) = pooled_producer(2_000);
		let mut straggler = producer.append_group().unwrap(); // seq 0
		producer.append_group().unwrap().finish().unwrap(); // seq 1 demotes seq 0

		// The demoted group balloons: no eviction yet (nothing ran), but billed.
		straggler
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000]))
			.unwrap();

		// The next append observes the growth and evicts the straggler.
		producer.append_group().unwrap().finish().unwrap(); // seq 2

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_none(), "the ballooned group is evicted");
		assert!(pool.used() <= 3_000, "growth is reclaimed: {}", pool.used());
	}

	/// A stale arrival entry whose sequence was later re-served by fetched backfill
	/// must not leak the replacement into arrival-order subscriptions.
	#[tokio::test]
	async fn refilled_sequence_stays_out_of_subscriptions() {
		let (producer, _pool) = pooled_producer(1 << 40);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		producer.create_group(0u64.into()).unwrap().finish().unwrap();
		let aborted = producer.create_group(1u64.into()).unwrap();
		aborted.abort(Error::Cancel).unwrap();
		producer.create_group(2u64.into()).unwrap().finish().unwrap();

		// Re-serve seq 1 as backfill; its slot replaces the aborted one, and the
		// old arrival entry for seq 1 now resolves to it.
		let pending = consumer.fetch_group(1, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"backfill"))
			.unwrap();
		group.finish().unwrap();
		pending.await.unwrap();

		// The backfill serves by sequence, but never in arrival order.
		assert!(consumer.peek_group(1).is_some());
		let mut subscriber = producer.subscribe(replay());
		assert_eq!(subscriber.assert_group().sequence, 0);
		assert_eq!(subscriber.assert_group().sequence, 2);
		subscriber.assert_no_group();
	}

	/// An expired backfill can't hide behind a refreshed one: the eviction-order
	/// expiry scans a bounded prefix instead of stopping at the first fresh entry.
	#[tokio::test]
	async fn expired_backfill_behind_refreshed_reclaimed() {
		let (producer, _pool) = pooled_producer(1 << 40);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		producer.create_group(5u64.into()).unwrap().finish().unwrap();
		for sequence in [2u64, 3u64] {
			let pending = consumer.fetch_group(sequence, None);
			let req = dynamic
				.requested_group()
				.now_or_never()
				.expect("should not block")
				.unwrap();
			let mut group = req.accept(None).unwrap();
			group
				.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
				.unwrap();
			group.finish().unwrap();
			pending.await.unwrap();
		}

		// Keep seq 2 fresh while seq 3 (behind it in eviction order) expires.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2 + Duration::from_secs(1));
		consumer.fetch_group(2, None).await.unwrap();
		crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2 + Duration::from_secs(1));
		producer.create_group(6u64.into()).unwrap().finish().unwrap();

		let consumer = producer.consume();
		assert!(consumer.peek_group(2).is_some(), "refreshed backfill survives");
		assert!(consumer.peek_group(3).is_none(), "expired backfill is reclaimed");
	}

	/// A FETCH hit within the same coarse clock tick still protects the group: the
	/// refresh stamps one tick ahead, so it reads strictly newer than the mean.
	#[tokio::test]
	async fn same_tick_fetch_protects() {
		// No time advances at all: every timestamp lands in the same tick.
		let (mut producer, _pool) = pooled_producer(10_000);
		let consumer = producer.consume();

		finished_group(&mut producer, 3_000); // seq 0
		finished_group(&mut producer, 3_000); // seq 1
		finished_group(&mut producer, 3_000); // seq 2

		consumer.fetch_group(0, None).await.unwrap();

		finished_group(&mut producer, 3_000); // seq 3
		finished_group(&mut producer, 3_000); // seq 4

		assert!(consumer.peek_group(0).is_some(), "same-tick refresh protects");
		assert!(consumer.peek_group(1).is_none(), "the unread group dies instead");
	}

	/// A refetched group that reclaims max_sequence is the live edge again: it must
	/// not re-enter the eviction order, or memory pressure could evict the newest
	/// content.
	#[tokio::test]
	async fn refetched_latest_stays_protected() {
		let (producer, _pool) = pooled_producer(10_000);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		let straggler = producer.append_group().unwrap(); // seq 0

		// The publisher aborts its own latest group; the sequence stays at the live edge.
		let latest = producer.append_group().unwrap(); // seq 1
		latest.abort(Error::Cancel).unwrap();

		// Re-fetch it: the replacement takes over max_sequence.
		let pending = consumer.fetch_group(1, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 1000]))
			.unwrap();
		group.finish().unwrap();
		pending.await.unwrap();

		// The refetched latest is protected by omission: it has no entry in the
		// eviction order, so no amount of debt can select it.
		{
			let state = producer.state.read();
			assert!(state.lookup.contains_key(&1), "refetched group is cached");
			assert!(
				state.evict.iter().all(|(sequence, _)| *sequence != 1),
				"the live edge must not be an eviction candidate"
			);
		}
		drop(straggler);
	}

	/// An evicted group is a cache miss, so a fetch re-fetches it and the accepted
	/// replacement serves the sequence again (not `Error::Duplicate`).
	#[tokio::test]
	async fn eviction_allows_refetch() {
		let (mut producer, _pool) = pooled_producer(10_000);
		let dynamic = producer.dynamic();

		finished_group(&mut producer, 10_000); // seq 0
		finished_group(&mut producer, 10_000); // seq 1: demotes seq 0
		finished_group(&mut producer, 10_000); // seq 2: pays by evicting seq 0

		let consumer = producer.consume();
		assert!(consumer.peek_group(0).is_none());
		let pending = consumer.fetch_group(0, None);

		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		assert_eq!(req.sequence(), 0);

		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"refetched"))
			.unwrap();
		group.finish().unwrap();

		let mut group = pending.await.unwrap();
		assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"refetched");
	}

	/// An aborted group is dead whatever frame it starts at: the sequence is claimable
	/// again, and a cache lookup misses rather than handing back a slot that can no
	/// longer serve it.
	///
	/// The abort and the group's first frame decide this together, so they are read
	/// under one guard. Read separately, the abort can land between them and the slot
	/// answers as a live duplicate on the strength of an offset it only still has
	/// because it died. That interleaving is what the single guard rules out; this
	/// pins the committed semantics it has to preserve.
	#[test]
	fn an_aborted_group_releases_its_sequence() {
		let producer = track_producer("test", None);
		let consumer = producer.consume();

		let mut group = producer.create_group(group::Info { sequence: 3 }).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
			.unwrap();

		// While it lives the slot answers, so the sequence is taken.
		assert!(matches!(
			producer.create_group(group::Info { sequence: 3 }),
			Err(Error::Duplicate)
		));
		assert!(consumer.peek_group(3).is_some());

		group.abort(Error::Cancel).unwrap();

		assert!(consumer.peek_group(3).is_none(), "an aborted slot is a cache miss");
		producer
			.create_group(group::Info { sequence: 3 })
			.expect("an aborted slot releases its sequence")
			.finish()
			.unwrap();
	}

	/// A fetched (backfill) group is served by sequence but never replayed to
	/// arrival-order subscribers.
	#[tokio::test]
	async fn fetched_backfill_not_subscribed() {
		let (producer, _pool) = pooled_producer(1 << 40);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		// The publisher starts at seq 5; earlier groups exist only upstream.
		producer.create_group(5u64.into()).unwrap().finish().unwrap();
		producer.create_group(6u64.into()).unwrap().finish().unwrap();

		// Fetch the gap: it lands in the cache and resolves the fetch...
		let pending = consumer.fetch_group(2, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"backfill"))
			.unwrap();
		group.finish().unwrap();
		let mut fetched = pending.await.unwrap();
		assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"backfill");
		assert!(consumer.peek_group(2).is_some(), "backfill is cached for later fetches");

		// ...but an arrival-order subscriber only sees the live groups.
		let mut subscriber = producer.subscribe(replay());
		assert_eq!(subscriber.assert_group().sequence, 5);
		assert_eq!(subscriber.assert_group().sequence, 6);
		subscriber.assert_no_group();
	}

	/// Fetched backfill isn't in arrival order, so it ages out through the eviction
	/// order instead of lingering until the track closes.
	#[tokio::test]
	async fn expired_backfill_reclaimed() {
		let (producer, pool) = pooled_producer(1 << 40);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		producer.create_group(5u64.into()).unwrap().finish().unwrap();

		// Serve a backfill fetch for an old sequence.
		let pending = consumer.fetch_group(2, None);
		let req = dynamic
			.requested_group()
			.now_or_never()
			.expect("should not block")
			.unwrap();
		let mut group = req.accept(None).unwrap();
		group
			.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 1000]))
			.unwrap();
		group.finish().unwrap();
		pending.await.unwrap();
		let used = pool.used();

		// Age past the pool's LRU window; the next write reclaims the backfill.
		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
		producer.create_group(6u64.into()).unwrap().finish().unwrap();

		assert!(consumer.peek_group(2).is_none(), "expired backfill is reclaimed");
		assert!(pool.used() < used, "its bytes are released");
	}

	#[tokio::test]
	async fn fetch_aborts_with_track() {
		let producer = track_producer("test", None);
		let dynamic = producer.dynamic();
		let consumer = producer.consume();

		let pending = consumer.fetch_group(3, None);
		assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());

		producer.abort(Error::Cancel).unwrap();
		assert!(pending.await.is_err());
		drop(dynamic);
	}
}