moq_net/model/

group.rs

//! A group is a stream of frames, split into a [Producer] and [Consumer] handle.
//!
//! A [Producer] writes an ordered stream of frames.
//! Frames can be written all at once ([Producer::write_frame]), or in chunks
//! ([Producer::create_frame]).
//!
//! A [Consumer] reads an ordered stream of frames.
//! The reader can be cloned, in which case each reader receives a copy of each frame. (fanout)
//!
//! Frames are numbered from 0 in write order. A group can be short at its front or its
//! back but never in the middle: [Producer::start_at] starts it later, so a handle can
//! carry the tail of a group whose leading frames came from somewhere else, and
//! [Producer::finish] ends it wherever writing stopped. [Consumer::set_frames] bounds a reader to a sub-range the same way [`track::Subscriber`]
//! bounds group sequences.
//!
//! The stream is closed with [Error] when all writers or readers are dropped.
use crate::cache;
use crate::frame::{self, Frame, FrameBuf};
use crate::{Cap, Timescale, stats, track};
use std::collections::VecDeque;
use std::mem::MaybeUninit;
use std::ops::{Bound, RangeBounds};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Poll, ready};

use crate::{Error, IntoBytes, Result, Timestamp};

/// Maximum total size of frames in a group.
///
/// A write that would exceed this aborts the group with [`Error::GroupTooLarge`].
/// Doubles as the per-frame size cap: a larger declared size is [`Error::FrameTooLarge`]
/// before allocating, so one maximum-size frame can fill a group.
pub const MAX_CACHE_BYTES: u64 = 32 * 1024 * 1024; // 32 MB

/// Maximum number of frames in a group.
///
/// 8192 is the largest legal group; the 8193rd write returns [`Error::GroupTooLarge`]
/// and aborts the group.
pub const MAX_GROUP_FRAMES: usize = 8192;

/// Slots `VecDeque` rounds a group's first frame up to.
///
/// A `RawVec` detail rather than a knob, so it is asserted rather than trusted: std
/// handing out more would silently undercharge every cached group.
const FRAME_SLOTS: usize = 4;

/// Heap one cached group costs beyond its frame payloads, excluding the track-side
/// bookkeeping in [`track::CACHE_OVERHEAD`].
///
/// A group is one kio channel (allocated whether or not anything ever parks on it), the
/// `Arc<Alive>` its producer clones share, and the frame slots the first write rounds up
/// to. Half of [`cache::ENTRY_OVERHEAD`]; see it for why this is derived rather than
/// measured.
pub(crate) const CACHE_OVERHEAD: u64 = (kio::Producer::<GroupState>::HEAP
	// `Alive` behind an `Arc`'s two reference counts, which it is pointer-aligned to sit
	// straight after.
	+ 2 * size_of::<usize>()
	+ size_of::<Alive>()
	+ FRAME_SLOTS * size_of::<Frame>()) as u64;

/// A group contains a sequence number because they can arrive out of order.
///
/// You can use [track::Producer::append_group] if you just want to +1 the sequence number.
#[derive(Clone, Copy, Debug, Hash, Eq, PartialEq, Ord, PartialOrd)]
pub struct Info {
	/// Per-track sequence number used to detect ordering and gaps. Higher numbers
	/// supersede lower ones; consumers may skip late arrivals.
	pub sequence: u64,
}

impl Info {
	/// Create an untimed producer for this group.
	///
	/// Test-only: real groups are created via [`track::Producer`], which
	/// supplies the parent track's [`track::Info`]. This helper exists for in-crate
	/// tests that don't exercise timestamps.
	#[cfg(test)]
	pub(crate) fn produce(self) -> Producer {
		Producer::new(self, track::Info::default(), Default::default())
	}
}

impl From<usize> for Info {
	fn from(sequence: usize) -> Self {
		Self {
			sequence: sequence as u64,
		}
	}
}

impl From<u64> for Info {
	fn from(sequence: u64) -> Self {
		Self { sequence }
	}
}

impl From<u32> for Info {
	fn from(sequence: u32) -> Self {
		Self {
			sequence: sequence as u64,
		}
	}
}

impl From<u16> for Info {
	fn from(sequence: u16) -> Self {
		Self {
			sequence: sequence as u64,
		}
	}
}

/// The in-flight (tail) frame being written. At most one exists at a time, since a
/// group is a single ordered stream.
pub(crate) struct Partial {
	timestamp: Timestamp,
	buf: FrameBuf,
}

/// Shared group state. `pub(crate)` so [`frame`] handles can observe the abort flag
/// while streaming a partial frame.
#[derive(Default)]
pub(crate) struct GroupState {
	// Completed frames, each a contiguous payload. `offset` is the first frame this
	// handle holds, raised by [`Producer::start_at`].
	pub(crate) frames: VecDeque<Frame>,

	// The single in-flight frame, if one is open.
	pub(crate) partial: Option<Partial>,

	// Index of the first frame this handle holds: any the group deliberately started
	// past (see [`Producer::start_at`]). Reading below it is [`Error::Lagged`]; the
	// frames are not here.
	pub(crate) offset: usize,

	// The index the next frame written will get. Tracked separately from `frames` so it
	// survives the cache being released: a route taking the track over needs to know where
	// production stopped, and an abort is exactly when it asks.
	next_index: usize,

	// One past the last frame that was fully written. Trails `next_index` while a chunked
	// frame is in flight, which is the frame a replacement route has to redeliver: only
	// its opener saw the payload, and only partly.
	committed: usize,

	// The total size (in bytes) of all cached frames plus any in-flight frame.
	pub(crate) cache: u64,

	// Mirrors `cache` into the track's shared cache pool, so the group's bytes count
	// against the byte budget tracks evict toward.
	charge: cache::Charge,

	// The first frame's timestamp, recorded once and never revised: the group's
	// presentation start. Kept here rather than read off `frames` so an abort
	// doesn't erase where the group sat in time. `None` until the first frame is
	// written, which is the only honest answer: an empty group has not presented
	// anything yet.
	timestamp: Option<Timestamp>,

	// The newest frame's timestamp: the group's presentation end so far. A reader
	// that has taken every frame sits here, which is what a drift budget measures it
	// against. Kept alongside `timestamp` for the same reasons.
	latest: Option<Timestamp>,

	// Once finalized, the total number of frames the group will ever contain. Recorded
	// at finish so the count outlives an abort that clears the cache.
	pub(crate) fin: Option<usize>,

	// The error that caused the group to be aborted, if any. Mirrored into
	// `Alive::aborted`, so [`Producer::abort`] stays the only writer: anything else
	// setting this would leave track scans reading a group as live.
	pub(crate) abort: Option<Error>,
}

impl GroupState {
	/// Content still available to a reader of this group.
	fn content(&self) -> stats::Content {
		stats::Content {
			bytes: self.cache,
			frames: self.next_index.saturating_sub(self.offset) as u64,
			groups: 1,
			datagrams: 0,
		}
	}

	/// Content in the half-open frame range that is still cached here.
	pub(crate) fn content_range(&self, start: usize, end: usize) -> stats::Content {
		let start = start.max(self.offset);
		let end = end.min(self.next_index);
		if start >= end {
			return stats::Content::default();
		}

		let local_start = start.saturating_sub(self.offset).min(self.frames.len());
		let local_end = end.saturating_sub(self.offset).min(self.frames.len());
		let mut bytes = self
			.frames
			.range(local_start..local_end)
			.map(|frame| frame.payload.len() as u64)
			.sum();
		if start <= self.committed
			&& self.committed < end
			&& let Some(partial) = &self.partial
		{
			bytes += partial.buf.capacity() as u64;
		}

		stats::Content {
			bytes,
			frames: (end - start) as u64,
			groups: 0,
			datagrams: 0,
		}
	}

	/// Resolve the source for the frame at `index`: a completed frame (whole) or the
	/// in-flight tail (streamed). Used by [`Consumer::poll_next_frame`].
	fn poll_frame_source(&self, index: usize) -> Poll<Result<Option<(frame::Info, frame::Source)>>> {
		if index < self.offset {
			return Poll::Ready(Err(Error::Lagged));
		}
		let local = index - self.offset;
		if let Some(f) = self.frames.get(local) {
			// A frame read is a cache access: stamp it so expiry and the eviction
			// walk spare a group a consumer is actively draining.
			self.charge.refresh();
			let info = frame::Info {
				size: f.payload.len() as u64,
				timestamp: f.timestamp,
			};
			return Poll::Ready(Ok(Some((info, frame::Source::Complete(f.payload.clone())))));
		}
		if local == self.frames.len()
			&& let Some(p) = &self.partial
		{
			self.charge.refresh();
			let info = frame::Info {
				size: p.buf.capacity() as u64,
				timestamp: p.timestamp,
			};
			return Poll::Ready(Ok(Some((info, frame::Source::Partial(p.buf.clone())))));
		}
		ready!(self.poll_terminal(index))?;
		Poll::Ready(Ok(None))
	}

	/// Resolve the group's terminal state for a reader positioned at `index`.
	///
	/// A finished group is still aborted once its frames are released to free memory
	/// (aged out of the track's max age window, or evicted by the cache pool). A reader
	/// that already consumed every frame is missing nothing, so it gets the clean end of
	/// group; one that fell short sees the abort rather than a silently truncated stream.
	fn poll_terminal(&self, index: usize) -> Poll<Result<()>> {
		match (self.fin, &self.abort) {
			(Some(total), Some(err)) if index < total => Poll::Ready(Err(err.clone())),
			(Some(_), _) => Poll::Ready(Ok(())),
			(None, Some(err)) => Poll::Ready(Err(err.clone())),
			(None, None) => Poll::Pending,
		}
	}

	/// Resolve whether a reader at `index` can still make progress, answering the same
	/// question as a read without consuming anything.
	fn poll_end(&self, index: usize) -> Poll<Result<()>> {
		if index < self.offset {
			return Poll::Ready(Err(Error::Lagged));
		}
		self.poll_terminal(index)
	}

	/// Record where the group starts and currently ends in presentation time.
	/// `timestamp` keeps the first frame only; `latest` follows every frame.
	fn stamp(&mut self, timestamp: Timestamp) {
		self.timestamp.get_or_insert(timestamp);
		self.latest = Some(timestamp);
	}

	/// Whether adding `extra_frames` totaling `extra_bytes` would exceed the group budget.
	fn would_overflow(&self, extra_frames: usize, extra_bytes: u64) -> bool {
		self.next_index.saturating_sub(self.offset).saturating_add(extra_frames) > MAX_GROUP_FRAMES
			|| self.cache.saturating_add(extra_bytes) > MAX_CACHE_BYTES
	}

	/// Drop the cached frames (and any in-flight tail) and release their pool charge.
	fn release(&mut self) {
		self.frames.clear();
		self.partial = None;
		self.cache = 0;
		self.charge.clear();
	}
}

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

/// Writes frames to a group in order.
///
/// Each group is delivered independently over a QUIC stream.
/// Use [Self::write_frame] for simple single-buffer frames,
/// or [Self::create_frame] for multi-chunk streaming writes.
pub struct Producer {
	// Mutable stream state.
	state: kio::Producer<GroupState>,

	// The group header containing the sequence number. A small `Copy` value,
	// inherited by each frame (see [`Self::create_frame`]).
	info: Info,

	// The parent track's properties, inherited rather than passed piecemeal. Its
	// `timescale` is used by [`Self::create_frame`] to normalize every frame's
	// timestamp into the track scale before it enters the stream. Threaded down by
	// value from [`track::Producer::create_group`] / `append_group`.
	track: track::Info,

	// The parent track's account against the shared cache pool. Held here as well as
	// in the group's `cache::Charge` so a frame write can settle the track's eviction
	// debt with the group lock released.
	cache: Arc<cache::Track>,

	// Ingress payload meter, set by a tagged [`track::Producer`] via
	// [`Self::with_meter`]. Empty (no-op) for an untagged group.
	stats: stats::Meter,

	// Shared by every clone: its `Drop` is the abrupt-teardown, running exactly once
	// when the last of them goes.
	alive: Arc<Alive>,
}

/// Ends the group when the last [`Producer`] clone drops, including the clone the
/// parent track holds in its cache.
///
/// 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 can invalidate it. 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 {
	info: Info,
	state: kio::Producer<GroupState>,
	// Monotone mirror of `GroupState::abort` for track scans that already hold the
	// track lock. A stale false only hands out a group that is concurrently aborting;
	// true is stored after the abort exists, so it can never hide a live group.
	//
	// Only ever read on its own. A decision that pairs the abort with something else
	// out of `GroupState` has to read both under one guard, or the two halves can
	// straddle the abort: see `Producer::live_first_frame`.
	aborted: AtomicBool,
	// The cache stamp `GroupState::charge` maintains, held here as well so the
	// eviction and expiry walks can weigh a candidate without taking the group lock
	// they already hold the track lock over.
	access: Arc<cache::Access>,
}

impl Drop for Alive {
	fn drop(&mut self) {
		// See track::Alive: the last producer dropping without a clean finish releases
		// the cached frames so a stale consumer can't pin their buffers forever. A
		// finished group keeps its cache so consumers can drain.
		//
		// Check Ok and Err: Ok is unreachable after a deliberate close.
		match self.state.write() {
			Ok(mut state) => {
				if state.fin.is_some() || state.abort.is_some() {
					return;
				}
				tracing::warn!(
					sequence = self.info.sequence,
					"group::Producer dropped without finish() or abort()"
				);
				state.release();
			}
			Err(state) => {
				if state.fin.is_some() || state.abort.is_some() {
					return;
				}
				tracing::warn!(
					sequence = self.info.sequence,
					"group::Producer dropped without finish() or abort()"
				);
			}
		}
	}
}

impl std::ops::Deref for Producer {
	type Target = Info;

	fn deref(&self) -> &Self::Target {
		&self.info
	}
}

impl Producer {
	/// Create a group producer bound to its parent track's [`track::Info`] and cache
	/// account.
	///
	/// Crate-private: groups are only constructed via [`track::Producer`], which
	/// threads both down so properties like the timescale are inherited rather than
	/// passed in. Every frame added to this group is normalized to the track's
	/// timescale by [`Self::create_frame`].
	///
	/// Charges the group into `cache`, so its cached bytes count against the budget the
	/// track evicts toward under memory pressure.
	pub(crate) fn new(info: Info, track: track::Info, cache: Arc<cache::Track>) -> Self {
		let state = kio::Producer::<GroupState>::default();
		let charge = cache.charge();
		let access = charge.access();
		state.write().ok().expect("a new group is open").charge = charge;
		let alive = Arc::new(Alive {
			info,
			state: state.clone(),
			aborted: AtomicBool::new(false),
			access,
		});
		Self {
			info,
			state,
			track,
			cache,
			stats: stats::Meter::default(),
			alive,
		}
	}

	/// Attach an ingress payload meter, counting this as one delivered group.
	/// Called by a tagged [`track::Producer`] when it creates the group.
	pub(crate) fn with_meter(mut self, meter: stats::Meter) -> Self {
		meter.group();
		self.stats = meter;
		self
	}

	/// The group header.
	pub(crate) fn info(&self) -> Info {
		self.info
	}

	/// The parent track's timescale.
	pub fn timescale(&self) -> Timescale {
		self.track.timescale
	}

	/// Start the group at frame `index` rather than 0, so the first frame written lands
	/// there.
	///
	/// A group can be short at its front or its back, never in the middle: this trims
	/// the front, and simply stopping (then [`finish`](Self::finish)ing) trims the back.
	/// The frames below `index` are not a gap this handle will ever fill, so a reader
	/// positioned below it gets [`Error::Lagged`]. They belong to whoever produced the
	/// head of the group, typically another route serving the same track (see
	/// [`crate::track::Subscriber`]).
	///
	/// The counterpart of [`Consumer::set_frames`], which positions a *reader* the same
	/// way. Where the group begins is part of its shape, so this must come before the
	/// first frame; afterwards it returns [`Error::Closed`].
	pub fn start_at(&mut self, index: u64) -> Result<()> {
		let index = usize::try_from(index).map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))?;
		if index == usize::MAX {
			return Err(Error::BoundsExceeded(crate::coding::BoundsExceeded));
		}

		let mut state = modify(&self.state)?;
		// Every write advances `next_index` past `offset`, so this is "nothing written
		// yet".
		if state.fin.is_some() || state.next_index != state.offset {
			return Err(Error::Closed);
		}
		state.offset = index;
		state.next_index = index;
		state.committed = index;
		Ok(())
	}

	/// A helper method to write a frame from a single byte buffer.
	///
	/// If you want to write multiple chunks, use [Self::create_frame] to get a frame producer.
	/// But an upfront size is required.
	///
	/// `timestamp` is converted into the parent track's timescale. For data without
	/// a presentation time, pass [`Timestamp::now`] explicitly.
	pub fn write_frame<B: IntoBytes>(&mut self, timestamp: Timestamp, data: B) -> Result<()> {
		let timestamp = timestamp
			.convert(self.track.timescale)
			.map_err(|_| Error::TimestampMismatch)?;
		let payload = data.into_bytes();
		if payload.len() as u64 > MAX_CACHE_BYTES {
			return Err(Error::FrameTooLarge);
		}

		let mut state = modify(&self.state)?;
		if state.fin.is_some() {
			return Err(Error::Closed);
		}
		if state.partial.is_some() {
			return Err(Error::FrameOpen);
		}
		let next_index = state
			.next_index
			.checked_add(1)
			.ok_or(Error::BoundsExceeded(crate::coding::BoundsExceeded))?;
		debug_assert!(state.partial.is_none(), "a frame is already open");
		let size = payload.len() as u64;
		if state.would_overflow(1, size) {
			return Err(self.abort_too_large(state));
		}
		state.cache += size;
		let now = state.charge.add(size);
		state.frames.push_back(Frame { timestamp, payload });
		state.next_index = next_index;
		state.committed = state.next_index;
		state.stamp(timestamp);
		drop(state);

		// With the group lock released (lock order is track then group), settle
		// eviction debt if enough has been written since the track last paid.
		self.cache.settle(now);

		// Ingress payload: one whole frame written.
		self.stats.frames(1);
		self.stats.bytes(size);
		Ok(())
	}

	/// Write a whole batch of frames at once, draining `frames`.
	///
	/// One lock covers the batch, so an ingest with several frames in hand pays the
	/// group mutex and the track's eviction settle once rather than per frame. Build
	/// the batch with [`frame::Buffer::push`].
	///
	/// The batch is validated before anything is written, so a rejected frame leaves
	/// both the group and the buffer exactly as they were, ready to retry or redirect.
	/// Returns [`Error::FrameOpen`] if another handle is streaming a frame into this
	/// group, since appending around it would reorder the group.
	pub fn write_frames<const N: usize>(&mut self, frames: &mut frame::Buffer<N>) -> Result<()> {
		for frame in frames.filled() {
			frame
				.timestamp
				.convert(self.track.timescale)
				.map_err(|_| Error::TimestampMismatch)?;
			if frame.payload.len() as u64 > MAX_CACHE_BYTES {
				return Err(Error::FrameTooLarge);
			}
		}

		let count = frames.len();
		let bytes: u64 = frames.filled().iter().map(|frame| frame.payload.len() as u64).sum();
		let mut state = modify(&self.state)?;
		if state.fin.is_some() {
			return Err(Error::Closed);
		}
		if state.partial.is_some() {
			return Err(Error::FrameOpen);
		}
		let next_index = state
			.next_index
			.checked_add(count)
			.ok_or(Error::BoundsExceeded(crate::coding::BoundsExceeded))?;
		if state.would_overflow(count, bytes) {
			return Err(self.abort_too_large(state));
		}

		// The last frame's tick, reused below so settling does not re-read the clock.
		let mut now = None;
		for mut frame in frames.drain() {
			frame.timestamp = frame
				.timestamp
				.convert(self.track.timescale)
				.expect("timestamp scale checked above");
			let size = frame.payload.len() as u64;
			state.cache += size;
			now = state.charge.add(size);
			state.stamp(frame.timestamp);
			state.frames.push_back(frame);
		}
		state.next_index = next_index;
		state.committed = next_index;
		drop(state);

		self.cache.settle(now);
		self.stats.frames(count as u64);
		self.stats.bytes(bytes);
		Ok(())
	}

	/// Create a frame with an upfront size and presentation timestamp, streamed in
	/// chunks. Borrows the group exclusively until the returned [`frame::Producer`]
	/// is finished or dropped, so only one frame is open at a time.
	///
	/// The `timestamp` is converted into the parent track's timescale, so the scale you
	/// build it with doesn't have to match the track. Returns [`Error::FrameTooLarge`]
	/// if the declared size exceeds the group's byte budget (refused before allocating)
	/// or [`Error::TimestampMismatch`] if the timestamp can't be converted (overflow).
	pub fn create_frame(&mut self, frame: frame::Info) -> Result<frame::Producer<'_>> {
		let timestamp = frame
			.timestamp
			.convert(self.track.timescale)
			.map_err(|_| Error::TimestampMismatch)?;
		if frame.size > MAX_CACHE_BYTES {
			return Err(Error::FrameTooLarge);
		}
		let buf = FrameBuf::new(frame.size as usize);

		let mut state = modify(&self.state)?;
		if state.fin.is_some() {
			return Err(Error::Closed);
		}
		if state.partial.is_some() {
			return Err(Error::FrameOpen);
		}
		let next_index = state
			.next_index
			.checked_add(1)
			.ok_or(Error::BoundsExceeded(crate::coding::BoundsExceeded))?;
		if state.would_overflow(1, frame.size) {
			return Err(self.abort_too_large(state));
		}
		state.cache += frame.size;
		let now = state.charge.add(frame.size);
		state.partial = Some(Partial {
			timestamp,
			buf: buf.clone(),
		});
		state.next_index = next_index;
		// Opening the frame is enough: the header carries the timestamp, so the group's
		// place in time is known before a single payload byte streams in.
		state.stamp(timestamp);
		drop(state);

		// With the group lock released (lock order is track then group), settle
		// eviction debt if enough has been written since the track last paid.
		self.cache.settle(now);

		// Ingress payload: one frame opened; its bytes are counted per chunk as the
		// frame::Producer writes them.
		self.stats.frames(1);
		let meter = self.stats.clone();

		let info = frame::Info {
			size: frame.size,
			timestamp,
		};
		Ok(frame::Producer::new(self, buf, info).with_meter(meter))
	}

	/// The owned counterpart of [`Self::create_frame`], for the wire drivers that
	/// stream a frame across polls and cannot hold the group borrowed inside their
	/// state. The one-live-frame rule the borrow normally enforces becomes the
	/// caller's promise; see [`frame::ProducerOwned`].
	pub(crate) fn create_frame_owned(&mut self, frame: frame::Info) -> Result<frame::ProducerOwned> {
		let timestamp = frame
			.timestamp
			.convert(self.track.timescale)
			.map_err(|_| Error::TimestampMismatch)?;
		if frame.size > MAX_CACHE_BYTES {
			return Err(Error::FrameTooLarge);
		}
		let buf = FrameBuf::new(frame.size as usize);

		let mut state = modify(&self.state)?;
		if state.fin.is_some() {
			return Err(Error::Closed);
		}
		if state.partial.is_some() {
			return Err(Error::FrameOpen);
		}
		let next_index = state
			.next_index
			.checked_add(1)
			.ok_or(Error::BoundsExceeded(crate::coding::BoundsExceeded))?;
		if state.would_overflow(1, frame.size) {
			return Err(self.abort_too_large(state));
		}
		state.cache += frame.size;
		let now = state.charge.add(frame.size);
		state.partial = Some(Partial {
			timestamp,
			buf: buf.clone(),
		});
		state.next_index = next_index;
		// Opening the frame is enough: the header carries the timestamp, so the group's
		// place in time is known before a single payload byte streams in.
		state.stamp(timestamp);
		drop(state);

		// With the group lock released (lock order is track then group), settle
		// eviction debt if enough has been written since the track last paid.
		self.cache.settle(now);

		// Ingress payload: one frame opened; its bytes are counted per chunk as the
		// producer writes them.
		self.stats.frames(1);
		let meter = self.stats.clone();

		let info = frame::Info {
			size: frame.size,
			timestamp,
		};
		Ok(frame::ProducerOwned::new(self.clone(), buf, info).with_meter(meter))
	}

	/// Wake consumers parked on the group channel (called after a partial write).
	pub(crate) fn frame_notify(&self) {
		// The chunk that was just written is a write access: restart the retention
		// clock so a straggler group streaming a large frame isn't expired
		// mid-write (its bytes were already charged when the frame was created).
		// `record_write` takes `&mut`, which marks the guard modified: kio only
		// notifies on a mutably-accessed guard's release, and that notify is what
		// delivers the chunk to parked readers.
		let now = self
			.state
			.write()
			.ok()
			.and_then(|mut state| state.charge.record_write());
		// The payload was charged when the frame opened, but a long streamed frame
		// still counts as track activity for the independent expiry time gate.
		self.cache.settle(now);
	}

	/// Commit the in-flight frame as a completed frame (called by [`frame::Producer::finish`]).
	pub(crate) fn frame_commit(&mut self, frame: Frame) -> Result<()> {
		let mut state = modify(&self.state)?;
		// Bytes were already counted against the cache (and the pool charge) when the
		// frame was created; committing just moves the tail into the completed set.
		state.partial = None;
		state.frames.push_back(frame);
		state.committed = state.next_index;
		// Completing the frame is a write access like any chunk, and the only one the
		// payload is guaranteed to get: the wire ingest defers its chunk notifications
		// to the poll boundary, so a tail that arrives and completes in one turn never
		// reaches [`Self::frame_notify`]. Without this, a group whose payload streamed
		// in across an idle gap would expire the instant it finished.
		let now = state.charge.record_write();
		drop(state);

		// With the group lock released (lock order is track then group), settle
		// eviction debt and age idle content out, reusing the tick above.
		self.cache.settle(now);
		Ok(())
	}

	/// Fail the group because an in-flight frame couldn't complete (called by
	/// [`frame::Producer::abort`] / its drop).
	pub(crate) fn frame_abort(&mut self, err: Error) {
		let _ = self.clone().abort(err);
	}

	/// One past the index of the last frame written (completed or in-flight), which is
	/// also the index the next frame will get.
	///
	/// Counts any frames the group [started past](Self::start_at), so it's the group's
	/// logical length rather than the number of frames this handle holds.
	pub fn frame_count(&self) -> usize {
		self.state.read().next_index
	}

	/// Mark the group as complete; no more frames will be written.
	///
	/// Borrows rather than consumes, so a later failure can still be reported through
	/// [`abort`](Self::abort). The handle also keeps the cached frames readable.
	pub fn finish(&self) -> Result<()> {
		let mut state = modify(&self.state)?;
		if state.partial.is_some() {
			return Err(Error::FrameOpen);
		}
		state.fin = Some(state.next_index);
		Ok(())
	}

	/// Abort the group with the given error.
	///
	/// Consumes the handle. Drops the cached frames so a stale [`Consumer`] can't pin
	/// their buffers in memory forever; consumers that haven't drained yet surface the
	/// abort error instead of the leftover cache.
	pub fn abort(self, err: Error) -> Result<()> {
		let mut guard = modify(&self.state)?;
		guard.abort = Some(err);
		self.alive.aborted.store(true, Ordering::Release);
		guard.release();
		guard.close();
		Ok(())
	}

	/// Abort a write that would grow the group past its budget, holding the lock already
	/// taken for that write so nothing else lands in between.
	fn abort_too_large(&self, mut state: kio::Mut<'_, GroupState>) -> Error {
		let err = Error::GroupTooLarge;
		state.abort = Some(err.clone());
		self.alive.aborted.store(true, Ordering::Release);
		state.release();
		state.close();
		err
	}

	/// Whether the group has been aborted (including pool eviction). The track's
	/// read paths treat an aborted cached group as absent.
	///
	/// Reads the mirror rather than the group's state, so a track scan holding the
	/// track lock never takes the group's. Monotone, and only ever conservative: a
	/// concurrent abort can still read as live for the length of [`Self::abort`],
	/// which hands out a group whose consumer then surfaces the abort.
	pub(crate) fn is_aborted(&self) -> bool {
		self.alive.aborted.load(Ordering::Acquire)
	}

	/// Whether the group was finished: it holds every frame it will ever have.
	pub(crate) fn is_finished(&self) -> bool {
		self.state.read().fin.is_some()
	}

	/// The index of the first frame this group still holds, or `None` once it has been
	/// aborted. Non-zero when the group started later (see [`Self::start_at`]); a reader
	/// positioned below it is [`Error::Lagged`].
	///
	/// One guard for both halves, deliberately. The track asks this to decide whether a
	/// cached slot can still answer a request, and reading the abort and the offset
	/// separately lets the abort land between them: the slot reads live, then hands
	/// back an offset it only has because it is dead. The mirror
	/// ([`Self::is_aborted`]) is for scans that ask about the abort alone.
	pub(crate) fn live_first_frame(&self) -> Option<usize> {
		let state = self.state.read();
		state.abort.is_none().then_some(state.offset)
	}

	/// One past the last frame committed to an unfinished group, when that is past its
	/// first: where a replacement route resumes. `None` once the group is finished, or
	/// while it holds nothing a replacement could splice onto.
	///
	/// The *committed* count, not the written one: a route dying midway through a
	/// chunked frame leaves that frame unusable, so the replacement has to send it
	/// again rather than start after it. Answered under one guard so the count can't be
	/// weighed against an offset from a different moment.
	///
	/// An aborted group still answers: readers that already consumed its head want the
	/// tail, and the count outlives the released cache.
	pub(crate) fn resume_frame(&self) -> Option<usize> {
		let state = self.state.read();
		if state.fin.is_some() {
			return None;
		}
		(state.committed > state.offset).then_some(state.committed)
	}

	/// Where the group starts in presentation time: its first frame's timestamp,
	/// or `None` while no frame has been opened.
	///
	/// Stamped once, when the group's first frame arrives, so it measures the group's
	/// place in the media timeline rather than when it happened to be delivered. That
	/// is what lets the track tell a burst of old content apart from live content (see
	/// [`track::Subscriber`]). On protocols whose wire can't carry a timestamp the
	/// receiver stamps frames with [`Timestamp::now`], which makes this the local
	/// receive time instead: an estimate that a burst compresses.
	pub(crate) fn timestamp(&self) -> Option<Timestamp> {
		self.state.read().timestamp
	}

	/// Where the group ends in presentation time: its newest frame's timestamp, or
	/// `None` while no frame has been opened.
	///
	/// This is what a drift budget measures an untouched group against. A group is not
	/// late because it *started* long ago: a two-second group whose tail is level with
	/// the live edge still has content nobody has read. Only once its newest frame has
	/// fallen behind is there nothing left worth delivering. Grows as the group does, so
	/// a group still receiving frames stays fresh and a stalled one ages in place.
	pub(crate) fn latest(&self) -> Option<Timestamp> {
		self.state.read().latest
	}

	/// The group's full cached footprint (payload plus fixed overhead), used by the
	/// track to size this group as an eviction victim.
	pub(crate) fn cache_size(&self) -> u64 {
		self.state.read().charge.size()
	}

	/// Tick of the group's last cache access, driving eviction protection and age
	/// expiry (see [`cache::Pool::average`]).
	pub(crate) fn cache_accessed(&self) -> u64 {
		self.alive.access.get()
	}

	/// Coarse clock tick of the group's last cache access, used by age expiry.
	pub(crate) fn cache_accessed_tick(&self, now: Option<u64>) -> Option<u64> {
		self.alive.access.tick(now)
	}

	/// Enter the group into the evictable population: demoted from the live edge,
	/// or inserted behind it. Idempotent; a no-op once the group is closed.
	pub(crate) fn cache_demote(&self) {
		if let Ok(mut state) = self.state.write() {
			state.charge.demote();
		}
	}

	/// Record a cache access (delivery to a subscriber, a FETCH hit, or a fetched
	/// backfill's birth), protecting the group from eviction and restarting its
	/// expiry clock. Stamps through a read guard, whose release never notifies, so
	/// delivery can't wake every consumer parked on the group. Harmless on a
	/// closed group: its charge is already cleared.
	pub(crate) fn cache_refresh(&self) {
		self.state.read().charge.refresh();
	}

	/// Create a new consumer for the group.
	pub fn consume(&self) -> Consumer {
		Consumer {
			info: self.info,
			track: self.track.clone(),
			inner: ConsumerKind::Plain(Plain {
				state: self.state.consume(),
				index: 0,
				end: None,
				prefetch: Prefetch::default(),
				cache: self.cache.clone(),
				access: self.alive.access.clone(),
				refreshed: self.cache.pool().now(),
			}),
			// Untagged: a tagged track attaches the egress meter via `with_meter`
			// when it hands the consumer to a subscriber/fetch.
			stats: stats::Meter::default(),
			stale_stats: stats::Meter::default(),
			expiry: None,
			expired: false,
			ended: false,
			stale_counted: Arc::default(),
		}
	}

	/// Register for the first-frame timestamp while the group is still unstamped.
	pub(crate) fn poll_timestamp(&self, waiter: &kio::Waiter) -> Poll<()> {
		match self.state.poll(waiter, |state| {
			if state.timestamp.is_some() || state.fin.is_some() || state.abort.is_some() {
				Poll::Ready(())
			} else {
				Poll::Pending
			}
		}) {
			Poll::Ready(_) => Poll::Ready(()),
			Poll::Pending => Poll::Pending,
		}
	}

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

	/// Poll until the group 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())
	}

	/// 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())
	}

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

impl Clone for Producer {
	fn clone(&self) -> Self {
		Self {
			info: self.info,
			state: self.state.clone(),
			track: self.track.clone(),
			cache: self.cache.clone(),
			stats: self.stats.clone(),
			alive: self.alive.clone(),
		}
	}
}

/// A small inline batch of completed frames, drained from the shared group state
/// under one lock and then handed out without re-locking.
///
/// Each [`Consumer::read_frame`] otherwise takes the group mutex and allocates a
/// waker just to clone one `Bytes`; draining a batch amortizes both across `CAP`
/// frames. Storage is inline and uninitialized (no heap), so a consumer that never
/// reads whole frames, or drains through a higher-level buffer, pays nothing.
struct Prefetch {
	// Initialized, not-yet-taken frames are `frames[pos..len]`; the rest are uninitialized.
	frames: [MaybeUninit<Frame>; Self::CAP],
	pos: usize,
	len: usize,
}

impl Prefetch {
	const CAP: usize = 8;

	/// Take the next buffered frame, or `None` if the batch is drained.
	fn pop(&mut self) -> Option<Frame> {
		if self.pos == self.len {
			return None;
		}
		// SAFETY: `pos < len`, so this slot was written by `fill` and not yet taken.
		let frame = unsafe { self.frames[self.pos].assume_init_read() };
		self.pos += 1;
		Some(frame)
	}

	/// Refill with up to `CAP` frames. Must be drained first (`pop` returned `None`).
	fn fill(&mut self, frames: impl Iterator<Item = Frame>) {
		debug_assert_eq!(self.pos, self.len, "fill on a non-empty batch would leak frames");
		self.pos = 0;
		self.len = 0;
		for frame in frames.take(Self::CAP) {
			self.frames[self.len].write(frame);
			self.len += 1;
		}
	}

	/// `(frame count, total payload bytes)` of the buffered, not-yet-taken frames.
	/// Read once per fill to bump the egress payload counters for the whole batch.
	fn buffered(&self) -> (u64, u64) {
		let mut bytes = 0u64;
		for slot in &self.frames[self.pos..self.len] {
			// SAFETY: slots in `pos..len` are initialized (written by `fill`, not yet popped).
			bytes += unsafe { slot.assume_init_ref() }.payload.len() as u64;
		}
		((self.len - self.pos) as u64, bytes)
	}
}

impl Default for Prefetch {
	fn default() -> Self {
		Self {
			frames: [const { MaybeUninit::uninit() }; Self::CAP],
			pos: 0,
			len: 0,
		}
	}
}

impl Drop for Prefetch {
	fn drop(&mut self) {
		for slot in &mut self.frames[self.pos..self.len] {
			// SAFETY: slots in `pos..len` are initialized and were never taken.
			unsafe { slot.assume_init_drop() };
		}
	}
}

/// Consume a group, frame-by-frame.
///
/// Usually a view of one [`Producer`], but a group served across a route change is
/// *spliced*: it reads each contributing route's copy in turn, joined at the frame the
/// takeover happened on, so the reader never sees the seam.
pub struct Consumer {
	inner: ConsumerKind,

	// Immutable stream state.
	info: Info,

	// The parent track's info, inherited from the producer. Its `timescale` lets the
	// wire publisher emit per-frame timestamps at the right scale for a fetched group.
	track: track::Info,

	// Egress payload meter, set by a tagged track via [`Self::with_meter`]. Empty
	// (no-op) for an untagged group.
	stats: stats::Meter,
	// The meter that owns unread content discarded by expiry. Route-specific
	// cursors inside a spliced group inherit this without metering delivery twice.
	stale_stats: stats::Meter,

	// Subscriber-specific drift policy. A group can become stale after the track
	// hands it out, while its reader is waiting for the first or next frame.
	expiry: Option<Arc<dyn Expiry>>,
	expired: bool,
	// Sticky: the budget gave up on a cursor that had already taken every frame, so
	// the group ends rather than fails. Recorded because `expired` alone would turn a
	// clean end into `Error::Old` on the next poll, and a caller is allowed to probe
	// again after the end.
	ended: bool,
	// Cloned cursors are parallel views of one handed-out delivery. Whichever
	// observes expiry first records its unread tail; the others must not repeat it.
	stale_counted: Arc<AtomicBool>,
}

/// Subscriber-specific policy for expiring a group after it was handed out.
pub(crate) trait Expiry: Send + Sync {
	/// Return whether the group is stale, registering `waiter` for anything that
	/// could change the answer while the group remains live.
	fn is_expired(&self, waiter: &kio::Waiter) -> bool;
}

// `Plain` is the hot path and carries an inline frame prefetch, so boxing it to even the
// variants out would cost an allocation per group to save a pointer chase on the rare one.
#[expect(clippy::large_enum_variant)]
enum ConsumerKind {
	Plain(Plain),
	// Boxed: the spliced cursor set dwarfs the plain one, and splicing is the rare case.
	Spliced(Box<super::resume::Group>),
}

/// The cursor state for a group backed by a single [`Producer`].
struct Plain {
	// Shared state with the producer.
	state: kio::Consumer<GroupState>,

	// The index of the next frame to read.
	// NOTE: Cloned readers inherit this offset, but then run in parallel.
	index: usize,

	// Exclusive cap on `index`, set by [`Consumer::set_frames`]. Reads end cleanly at it.
	end: Option<usize>,

	// A batch of completed frames drained ahead under one lock (whole-frame reads only).
	prefetch: Prefetch,

	// Record prefetched reads without entering the group's state on every frame.
	cache: Arc<cache::Track>,
	access: Arc<cache::Access>,
	refreshed: u64,
}

impl Clone for Plain {
	fn clone(&self) -> Self {
		// A clone shares the channel and inherits `index`, but starts with an empty
		// prefetch: it re-reads its batch from the shared state, in parallel.
		Self {
			state: self.state.clone(),
			index: self.index,
			end: self.end,
			prefetch: Prefetch::default(),
			cache: self.cache.clone(),
			access: self.access.clone(),
			refreshed: self.refreshed,
		}
	}
}

impl Clone for Consumer {
	fn clone(&self) -> Self {
		Self {
			inner: match &self.inner {
				ConsumerKind::Plain(plain) => ConsumerKind::Plain(plain.clone()),
				ConsumerKind::Spliced(spliced) => ConsumerKind::Spliced(Box::new((**spliced).clone())),
			},
			info: self.info,
			track: self.track.clone(),
			// Inherit the meter without re-counting the group: the original already
			// counted it when the track handed it out.
			stats: self.stats.clone(),
			stale_stats: self.stale_stats.clone(),
			expiry: self.expiry.clone(),
			expired: self.expired,
			ended: self.ended,
			stale_counted: self.stale_counted.clone(),
		}
	}
}

impl std::ops::Deref for Consumer {
	type Target = Info;

	fn deref(&self) -> &Self::Target {
		&self.info
	}
}

impl Consumer {
	/// Snapshot the content this cursor would discard if its group were skipped.
	pub(crate) fn content(&self) -> stats::Content {
		match &self.inner {
			ConsumerKind::Plain(plain) => plain.state.read().content(),
			// Drift is evaluated before a segment copy is wrapped as a spliced group.
			// Keep the group count honest if a future caller reaches this fallback.
			ConsumerKind::Spliced(_) => stats::Content {
				groups: 1,
				..Default::default()
			},
		}
	}

	/// Content not already attributed as delivered by this handed-out cursor.
	fn unread_content(&self) -> stats::Content {
		match &self.inner {
			ConsumerKind::Plain(plain) => plain.unread_content(),
			// Each route-specific plain cursor enforces expiry inside a spliced group.
			ConsumerKind::Spliced(_) => stats::Content::default(),
		}
	}

	/// Rebuild this consumer as the head of a group assembled across route changes,
	/// keeping the group's identity and its track's properties. See [`super::resume`].
	pub(crate) fn into_spliced(self, mut spliced: super::resume::Group) -> Self {
		spliced.set_stale_meter(self.stale_stats.clone());
		Self {
			inner: ConsumerKind::Spliced(Box::new(spliced)),
			info: self.info,
			track: self.track,
			stats: self.stats,
			stale_stats: self.stale_stats,
			// Each segment keeps its own route-specific expiry policy. Applying the
			// head segment's policy to the assembled group would use the wrong edge
			// after a takeover.
			expiry: None,
			expired: false,
			ended: false,
			stale_counted: self.stale_counted,
		}
	}

	/// Attach an egress payload meter, counting this as one delivered group.
	/// Called by a tagged track when it hands the consumer to a subscriber or fetch.
	pub(crate) fn with_meter(mut self, meter: stats::Meter) -> Self {
		meter.group();
		self.stats = meter.clone();
		self.set_stale_meter(meter);
		self
	}

	/// Attach only the meter that owns content discarded by expiry.
	pub(crate) fn set_stale_meter(&mut self, meter: stats::Meter) {
		if let ConsumerKind::Spliced(spliced) = &mut self.inner {
			spliced.set_stale_meter(meter.clone());
		}
		self.stale_stats = meter;
	}

	/// Keep applying this subscription's drift budget while the group is read.
	pub(crate) fn with_expiry(mut self, expiry: Arc<dyn Expiry>) -> Self {
		self.expiry = Some(expiry);
		self
	}

	/// Check the parent subscription while a wire publisher drains detached payload.
	pub(crate) fn poll_expired(&mut self, waiter: &kio::Waiter) -> bool {
		self.poll_expired_while_pending(waiter, false)
	}

	/// Apply the drift budget to a read that found nothing and is about to park.
	///
	/// A group with frames in hand is always drained to its end: the budget bounds a
	/// group that has *stalled* while the live edge moved on, not one whose reader is
	/// merely slower than the wire. Judging every read instead would truncate the tail
	/// of every group, since the arrival of the next group is exactly what makes the
	/// current one no longer newest.
	///
	/// Keeping it off the ready path also keeps it off the hot path: evaluating the
	/// policy walks the track's group cache under its lock, which is shared by every
	/// subscriber of that track.
	///
	/// `Some(false)` ends the group cleanly and `Some(true)` fails it with
	/// [`Error::Old`]; see [`Self::expired_truncates`].
	fn poll_expired_if_blocked(&mut self, waiter: &kio::Waiter) -> Option<bool> {
		if self.ended {
			return Some(false);
		}
		if !self.poll_expired(waiter) {
			return None;
		}
		let truncates = self.expired_truncates();
		self.ended = !truncates;
		Some(truncates)
	}

	/// Whether giving up on this group now loses the reader anything.
	///
	/// A frame-level read only parks once the cursor has taken every frame the group
	/// holds, so expiring there costs nothing: the reader got everything that exists,
	/// and the group ends rather than fails. What it was still waiting for was the
	/// producer's FIN, and a group abandoned at its own end is indistinguishable from
	/// one that ended. A cursor that still holds unread content (a wire publisher with
	/// buffered frames, a half-read payload) is genuinely truncated and reports it.
	fn expired_truncates(&self) -> bool {
		let unread = self.unread_content();
		unread.frames > 0 || unread.bytes > 0
	}

	/// Keep checking expiry while a wire publisher still owns buffered group data.
	pub(crate) fn poll_expired_while_pending(&mut self, waiter: &kio::Waiter, pending: bool) -> bool {
		if !self.expired
			&& (pending || self.expiry_pending())
			&& self.expiry.as_ref().is_some_and(|expiry| expiry.is_expired(waiter))
		{
			self.expired = true;
			if !self.stale_counted.swap(true, Ordering::Relaxed) {
				self.stale_stats.stale(self.unread_content());
			}
		}
		self.expired
	}

	/// Whether expiry can still discard content or unblock a group that may grow.
	fn expiry_pending(&self) -> bool {
		match &self.inner {
			ConsumerKind::Plain(plain) => plain.expiry_pending(),
			// The route-specific plain cursors own expiry for a spliced group.
			ConsumerKind::Spliced(_) => false,
		}
	}

	/// Whether this cursor failed because its subscription max age budget expired.
	pub(crate) fn latency_expired(&self) -> bool {
		self.expired
	}

	/// Whether the group has been aborted (including pool eviction); the abort
	/// dropped the cached frames, so a held consumer has nothing left to read.
	///
	/// A spliced group spans several routes, so no single abort empties it; only a
	/// plain cursor can answer.
	pub(crate) fn is_aborted(&self) -> bool {
		match &self.inner {
			ConsumerKind::Plain(plain) => plain.state.read().abort.is_some(),
			ConsumerKind::Spliced(_) => false,
		}
	}

	/// Mark the group as still being read, so a slow batch drain does not expire it.
	pub fn keep_alive(&self) {
		if let ConsumerKind::Plain(plain) = &self.inner {
			plain.state.read().charge.refresh();
		}
	}

	/// Record a cache access from the consumer side: a parked group re-offered to
	/// its subscriber. Same stamp as [`Producer::cache_refresh`].
	pub(crate) fn cache_refresh(&self) {
		self.keep_alive();
	}

	/// Park `waiter` until the group closes (finish, abort, or eviction). Spliced
	/// subscribers register on parked groups so an eviction wakes them; a group
	/// that already closed cleanly can never abort, so no waiter is needed.
	///
	/// A spliced group reads as closed without registering anything: no single abort
	/// empties it, so [`Self::is_aborted`] can never turn true and there is nothing
	/// a wakeup would change.
	pub(crate) fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<()> {
		match &self.inner {
			ConsumerKind::Plain(plain) => plain.state.poll_closed(waiter),
			ConsumerKind::Spliced(_) => Poll::Ready(()),
		}
	}

	/// The parent track's timescale.
	pub fn timescale(&self) -> Timescale {
		self.track.timescale
	}

	/// The index of the next frame this consumer will return.
	///
	/// Starts at 0, or at the group's first available frame once [`Self::set_frames`] has
	/// clamped it, and advances by one per frame read.
	pub fn index(&self) -> u64 {
		match &self.inner {
			ConsumerKind::Plain(plain) => plain.index as u64,
			ConsumerKind::Spliced(spliced) => spliced.index(),
		}
	}

	/// Limit subsequent reads to these frame indices without rewinding read progress.
	///
	/// `2..=5` includes frames 2 through 5; `2..5` excludes frame 5. An omitted
	/// start preserves read progress, and an omitted end removes the cap.
	/// Raising the cap makes unread cached frames available again.
	pub fn set_frames(&mut self, frames: impl RangeBounds<u64>) {
		let (start, end) = super::subscription::sequence_bounds(frames);
		self.start_at(start);
		self.end_at(end.map_or(Bound::Unbounded, Bound::Excluded));
	}

	/// Skip ahead so the next frame returned is `index`, discarding anything buffered
	/// below it.
	///
	/// Clamped *up* to the group's first available frame: frames the group never held
	/// (see [`Producer::start_at`]) can't be returned, so asking for one just starts at
	/// the first that exists. Read [`Self::index`] back to learn where the cursor
	/// actually landed.
	/// Only moves forward; a lower `index` is ignored, since the frames behind the
	/// cursor may already have been handed out.
	pub(crate) fn start_at(&mut self, index: u64) {
		match &mut self.inner {
			ConsumerKind::Plain(plain) => plain.start_at(index),
			ConsumerKind::Spliced(spliced) => spliced.start_at(index),
		}
	}

	/// Advance the read cursor to `index`, skipping every frame below it.
	///
	/// Unlike [`Self::set_frames`], this does not clamp past a requested frame the group
	/// never held. A [`Producer::start_at`] floor above `index` still surfaces as
	/// [`Error::Lagged`].
	pub fn skip_to(&mut self, index: u64) {
		match &mut self.inner {
			ConsumerKind::Plain(plain) => plain.skip_to(index),
			ConsumerKind::Spliced(spliced) => spliced.start_at(index),
		}
	}

	/// Stop reading at `end`, or remove the cap with `..`.
	///
	/// `..=2` reads through frame 2, `..2` stops before it, and `..0` is the empty range:
	/// no frame is delivered. Reads past the cap end cleanly (`None`), as if the group
	/// finished there. The cap can move in either direction: raising it re-offers
	/// frames that are still cached.
	pub(crate) fn end_at(&mut self, end: impl Into<Cap>) {
		let end = end.into().exclusive();
		match &mut self.inner {
			ConsumerKind::Plain(plain) => {
				plain.end = end.map(|end| usize::try_from(end).unwrap_or(usize::MAX));
			}
			ConsumerKind::Spliced(spliced) => spliced.end_at(end),
		}
	}

	/// The number of frames written so far (completed plus any in-flight), independent of
	/// how many this consumer has read. The final total once the group is finished.
	pub fn frame_count(&self) -> usize {
		match &self.inner {
			ConsumerKind::Plain(plain) => {
				let state = plain.state.read();
				state.fin.unwrap_or(state.next_index)
			}
			ConsumerKind::Spliced(spliced) => spliced.frame_count(),
		}
	}

	/// Return a consumer for the next frame for chunked reading.
	pub async fn next_frame(&mut self) -> Result<Option<frame::Consumer>> {
		kio::wait(|waiter| self.poll_next_frame(waiter)).await
	}

	/// Poll for the next frame, without blocking.
	///
	/// Returns None if the group is finished and the index is out of range, or the cursor
	/// passed the [`Self::set_frames`] cap.
	pub fn poll_next_frame(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<frame::Consumer>>> {
		if self.ended {
			return Poll::Ready(Ok(None));
		}
		if self.expired {
			return Poll::Ready(Err(Error::Old));
		}
		let stats = self.stats.clone();
		let expiry = self
			.expiry
			.as_ref()
			.map(|policy| frame::Expiry::new(policy.clone(), self.stale_stats.clone(), self.stale_counted.clone()));
		let res = match &mut self.inner {
			ConsumerKind::Plain(plain) => plain.poll_next_frame(waiter, &stats, expiry),
			ConsumerKind::Spliced(spliced) => {
				// The per-route copies underneath are untagged, so meter the spliced
				// stream here: it is the one the subscriber actually reads.
				let res = ready!(spliced.poll_next_frame(waiter))?;
				if res.is_some() {
					stats.frames(1);
				}
				Poll::Ready(Ok(res.map(|frame| frame.with_meter(stats))))
			}
		};
		match res.is_pending().then(|| self.poll_expired_if_blocked(waiter)).flatten() {
			Some(true) => Poll::Ready(Err(Error::Old)),
			Some(false) => Poll::Ready(Ok(None)),
			None => res,
		}
	}

	/// Read the next frame (timestamp and payload) all at once, without blocking.
	pub fn poll_read_frame(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<frame::Frame>>> {
		if self.ended {
			return Poll::Ready(Ok(None));
		}
		if self.expired {
			return Poll::Ready(Err(Error::Old));
		}
		let stats = self.stats.clone();
		let res = match &mut self.inner {
			ConsumerKind::Plain(plain) => plain.poll_read_frame(waiter, &stats),
			ConsumerKind::Spliced(spliced) => {
				let res = ready!(spliced.poll_read_frame(waiter))?;
				if let Some(frame) = &res {
					stats.frames(1);
					stats.bytes(frame.payload.len() as u64);
				}
				Poll::Ready(Ok(res))
			}
		};
		match res.is_pending().then(|| self.poll_expired_if_blocked(waiter)).flatten() {
			Some(true) => Poll::Ready(Err(Error::Old)),
			Some(false) => Poll::Ready(Ok(None)),
			None => res,
		}
	}

	/// Read the next frame (timestamp and payload) all at once.
	pub async fn read_frame(&mut self) -> Result<Option<frame::Frame>> {
		// A prefetched frame is already buffered, so the drift budget (which only judges
		// a read that would park) can never apply to it.
		if !self.expired
			&& let ConsumerKind::Plain(plain) = &mut self.inner
		{
			// Serve from the prefetched batch without building a future or allocating a waker.
			if !plain.capped()
				&& let Some(frame) = plain.prefetch.pop()
			{
				plain.refresh_if_stale();
				plain.index += 1;
				return Ok(Some(frame));
			}
		}
		kio::wait(|waiter| self.poll_read_frame(waiter)).await
	}

	/// Fill `out` with every frame that is ready, up to its capacity, without blocking.
	///
	/// This is a short read: it returns as soon as anything is ready rather than
	/// waiting for `out` to fill. A zero count means the group ended when the buffer
	/// has non-zero capacity.
	pub fn poll_read_frames<const N: usize>(
		&mut self,
		waiter: &kio::Waiter,
		out: &mut frame::Buffer<N>,
	) -> Poll<Result<usize>> {
		out.clear();
		if out.capacity() == 0 {
			return Poll::Ready(Ok(0));
		}

		while !out.is_full() {
			match self.poll_read_frame(waiter) {
				Poll::Ready(Ok(Some(frame))) => out.push(frame).expect("buffer capacity checked"),
				Poll::Ready(Ok(None)) => break,
				Poll::Ready(Err(err)) => {
					if out.is_empty() {
						return Poll::Ready(Err(err));
					}
					break;
				}
				Poll::Pending if !out.is_empty() => break,
				Poll::Pending => return Poll::Pending,
			}
		}

		Poll::Ready(Ok(out.len()))
	}

	/// Fill `out` with every frame that is ready, blocking until a frame arrives or
	/// the group ends. Returns the current batch, empty only at the end of the group.
	pub async fn read_frames<'a, const N: usize>(
		&mut self,
		out: &'a mut frame::Buffer<N>,
	) -> Result<&'a mut [frame::Frame]> {
		kio::wait(|waiter| self.poll_read_frames(waiter, out)).await?;
		Ok(out.filled_mut())
	}

	/// Poll until the group terminates, returning this cursor's next frame index.
	pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll<Result<u64>> {
		if self.ended {
			return Poll::Ready(Ok(self.index()));
		}
		if self.expired {
			return Poll::Ready(Err(Error::Old));
		}
		let res = match &mut self.inner {
			ConsumerKind::Plain(plain) => {
				let index = plain.index;
				plain
					.poll(waiter, |state| state.poll_end(index))
					.map(|res| res.map(|()| index as u64))
			}
			ConsumerKind::Spliced(spliced) => spliced.poll_finished(waiter),
		};
		match res.is_pending().then(|| self.poll_expired_if_blocked(waiter)).flatten() {
			Some(true) => Poll::Ready(Err(Error::Old)),
			// The group ended where the cursor stands, so that is its frame count.
			Some(false) => Poll::Ready(Ok(self.index())),
			None => res,
		}
	}

	/// Block until the group terminates, returning this cursor's next frame index.
	///
	/// This answers for the cursor, not the group: a reader that drained every frame gets the
	/// clean end even if the group was aborted afterwards to release its cache, while one that
	/// stopped short gets that abort. A prior [`Self::skip_to`] contributes to the index even
	/// though those frames were not read. Use [`Self::frame_count`] for the producer's total.
	pub async fn finished(&mut self) -> Result<u64> {
		kio::wait(|waiter| self.poll_finished(waiter)).await
	}
}

impl Plain {
	/// Whether this cursor still has unread content or may receive another frame.
	fn expiry_pending(&self) -> bool {
		if self.capped() {
			return false;
		}

		let state = self.state.read();
		state.abort.is_none() && state.fin.is_none_or(|fin| self.index < fin)
	}

	/// Content this cursor has neither returned nor already counted in a prefetch batch.
	fn unread_content(&self) -> stats::Content {
		let prefetched = self.prefetch.buffered().0 as usize;
		let start = self.index.saturating_add(prefetched);
		let end = self.end.unwrap_or(usize::MAX);
		self.state.read().content_range(start, end)
	}

	/// Record prefetched reads, updating the eviction rank once per sampled tick.
	fn refresh_if_stale(&mut self) {
		self.access.touch();
		let tick = self.cache.pool().now();
		if tick != self.refreshed {
			self.state.read().charge.refresh();
			self.refreshed = tick;
		}
	}

	// 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<'_, GroupState>) -> 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)),
		})
	}

	/// Whether the cursor has passed the `end_at` cap.
	fn capped(&self) -> bool {
		self.end.is_some_and(|end| self.index >= end)
	}

	fn start_at(&mut self, index: u64) {
		let index = usize::try_from(index).unwrap_or(usize::MAX);
		let index = index.max(self.state.read().offset);
		if index <= self.index {
			return;
		}
		self.index = index;
		// The batch was drained from below the new cursor, so it can't be reused.
		self.prefetch = Prefetch::default();
	}

	fn skip_to(&mut self, index: u64) {
		let index = usize::try_from(index).unwrap_or(usize::MAX);
		if index <= self.index {
			return;
		}
		self.index = index;
		self.prefetch = Prefetch::default();
	}

	fn poll_next_frame(
		&mut self,
		waiter: &kio::Waiter,
		stats: &stats::Meter,
		expiry: Option<frame::Expiry>,
	) -> Poll<Result<Option<frame::Consumer>>> {
		if self.capped() {
			return Poll::Ready(Ok(None));
		}
		let end = self.end.unwrap_or(usize::MAX);

		// Hand out any frames a prior read_frame prefetched before touching the tail.
		// Their bytes were already counted at the batch fill, so the frame::Consumer
		// carries no meter.
		if let Some(frame) = self.prefetch.pop() {
			self.refresh_if_stale();
			self.index += 1;
			let tail = self.index.saturating_add(self.prefetch.buffered().0 as usize)..end;
			let info = frame::Info {
				size: frame.payload.len() as u64,
				timestamp: frame.timestamp,
			};
			let source = frame::Source::Complete(frame.payload);
			let frame = frame::Consumer::new(self.state.clone(), info, source);
			return Poll::Ready(Ok(Some(match expiry {
				Some(expiry) => frame.with_expiry(expiry.for_frame(tail, false)),
				None => frame,
			})));
		}

		let index = self.index;
		let Some((info, source)) = ready!(self.poll(waiter, |state| state.poll_frame_source(index))?) else {
			return Poll::Ready(Ok(None));
		};

		self.index += 1;
		// A direct read (not prefetched): count the frame here; the frame::Consumer
		// counts its bytes per chunk as they're read out.
		stats.frames(1);
		let frame = frame::Consumer::new(self.state.clone(), info, source).with_meter(stats.clone());
		Poll::Ready(Ok(Some(match expiry {
			Some(expiry) => frame.with_expiry(expiry.for_frame(self.index..end, true)),
			None => frame,
		})))
	}

	fn poll_read_frame(&mut self, waiter: &kio::Waiter, stats: &stats::Meter) -> Poll<Result<Option<frame::Frame>>> {
		if self.capped() {
			return Poll::Ready(Ok(None));
		}

		// Fast path: serve from the prefetched batch without locking or allocating a waker.
		if let Some(frame) = self.prefetch.pop() {
			self.refresh_if_stale();
			self.index += 1;
			return Poll::Ready(Ok(Some(frame)));
		}

		// The batch is drained: refill it under a single lock, registering the waiter if
		// nothing is ready. Borrow the two fields disjointly so the closure can fill.
		let index = self.index;
		// Never buffer past the cap: `end_at` can be raised later, and those frames must
		// come from the shared state then, not from a batch drained under the old cap.
		let budget = self.end.map_or(usize::MAX, |end| end.saturating_sub(index));
		let prefetch = &mut self.prefetch;
		let res = self.state.poll(waiter, |state| {
			if index < state.offset {
				return Poll::Ready(Err(Error::Lagged));
			}
			// `local` can run past the buffered count when frames were cleared out from
			// under us (abort, unfinished drop); clamp so `range` never panics on an
			// out-of-bounds start. `fill` always resets the batch, so an empty range
			// leaves `len == 0` and the terminal checks below resolve abort/fin/pending.
			let local = (index - state.offset).min(state.frames.len());
			prefetch.fill(state.frames.range(local..).take(budget).cloned());
			if prefetch.len > 0 {
				// One stamp covers the whole batch: frames popped from the prefetch
				// don't re-stamp until the next refill, which `CAP` bounds.
				state.charge.refresh();
				return Poll::Ready(Ok(()));
			}
			// Nothing completed at `index`: an in-flight tail waits, otherwise resolve
			// the terminal state (whole-frame reads never stream the partial).
			state.poll_terminal(index)
		});

		match ready!(res) {
			Ok(Ok(())) => {}
			Ok(Err(err)) => return Poll::Ready(Err(err)),
			Err(state) => return Poll::Ready(Err(state.abort.clone().unwrap_or(Error::Dropped))),
		}

		// The refill already updated the eviction rank under the group lock.
		self.refreshed = self.cache.pool().now();

		// A fresh batch was just filled (empty only on a clean end). Count the whole
		// batch once here, under no lock, so the drained pops that follow stay free.
		let (frames, bytes) = self.prefetch.buffered();
		stats.frames(frames);
		stats.bytes(bytes);

		Poll::Ready(Ok(self.prefetch.pop().inspect(|_| {
			self.index += 1;
		})))
	}
}

/// Options for a one-shot [`track::Consumer::fetch_group`] of a past group.
#[derive(Clone, Debug, Default)]
#[non_exhaustive]
pub struct Fetch {
	/// Delivery priority for the fetched group's stream. Defaults to 0.
	pub priority: u8,

	/// Index of the first frame to fetch within the group. Defaults to 0, the whole group.
	///
	/// Use this to fill a hole left by a route change: the group's head is already
	/// cached locally and only the tail is missing.
	///
	/// There is no matching end: a fetch always runs to the end of the group, and a
	/// caller wanting less caps the returned consumer with [`Consumer::set_frames`]. Stopping
	/// the *fetch* short would put a group in the cache that is indistinguishable from a
	/// complete one, so a later fetch of the whole group would resolve from it and come
	/// up short.
	pub frame_start: u64,
}

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

	/// Set the first frame to fetch, returning `self` for chaining.
	pub fn with_frame_start(mut self, frame_start: u64) -> Self {
		self.frame_start = frame_start;
		self
	}
}

/// A consumer's request for a single past group, handed to a handler via
/// [`track::Dynamic::requested_group`].
///
/// The handler fulfills it by calling [`Self::accept`], which inserts the group
/// into the track cache (resolving every [`track::Consumer::fetch_group`] that joined the
/// attempt) and returns a [`Producer`] to fill. A relay typically opens a wire
/// FETCH, reads FETCH_OK, then accepts. The request carries its own producer handle,
/// so it works the same whether or not the track has been accepted yet.
pub struct Request {
	pub(crate) state: kio::Producer<track::TrackState>,
	pub(crate) fetch: kio::Shared<track::FetchState>,
	pub(crate) sequence: u64,
	pub(crate) priority: u8,
	pub(crate) frame_start: u64,
	pub(crate) result: kio::Producer<track::FetchOutcome>,
	pub(crate) done: bool,
}

#[cfg(test)]
mod test {
	use super::*;
	use crate::model::test_tracing::count_drop_warnings;
	use bytes::Bytes;
	use futures::FutureExt;

	/// [`FRAME_SLOTS`] is std's rounding, not ours, so measure it: a larger real value
	/// would undercharge every cached group without touching a line of this crate.
	#[test]
	fn one_frame_fits_the_charged_slots() {
		let mut frames: VecDeque<Frame> = VecDeque::new();
		frames.push_back(Frame {
			timestamp: Timestamp::ZERO,
			payload: Bytes::new(),
		});
		let capacity = frames.capacity();
		assert!(
			capacity <= FRAME_SLOTS,
			"a one-frame deque now allocates {capacity} slots"
		);
	}

	#[test]
	fn basic_frame_reading() {
		let mut producer = Info { sequence: 0 }.produce();
		producer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"frame0"))
			.unwrap();
		producer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"frame1"))
			.unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		let f0 = consumer.next_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(f0.size, 6);
		let f1 = consumer.next_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(f1.size, 6);
		let end = consumer.next_frame().now_or_never().unwrap().unwrap();
		assert!(end.is_none());
	}

	#[test]
	fn read_frame_all_at_once() {
		let mut producer = Info { sequence: 0 }.produce();
		producer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"hello"))
			.unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		let frame = consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(frame.payload, Bytes::from_static(b"hello"));
	}

	#[test]
	fn read_frame_preserves_timestamp() {
		let mut producer = Info { sequence: 0 }.produce();
		let timestamp = Timestamp::from_micros(20_000).unwrap();
		producer.write_frame(timestamp, Bytes::from_static(b"hello")).unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		let frame = consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(frame.timestamp.as_micros(), 20_000);
		assert_eq!(frame.payload, Bytes::from_static(b"hello"));
	}

	#[test]
	fn chunked_frame_reads_whole() {
		let mut producer = Info { sequence: 0 }.produce();
		{
			let mut frame = producer
				.create_frame(frame::Info {
					size: 10,
					timestamp: Timestamp::ZERO,
				})
				.unwrap();
			frame.write(Bytes::from_static(b"hello")).unwrap();
			frame.write(Bytes::from_static(b"world")).unwrap();
			frame.finish().unwrap();
		}
		producer.finish().unwrap();

		// Frame data is held in a single per-frame buffer; a whole-frame read returns
		// the full contents in one slice.
		let mut consumer = producer.consume();
		let frame = consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(frame.payload, Bytes::from_static(b"helloworld"));
	}

	#[test]
	fn chunked_frame_streams_partial() {
		let mut producer = Info { sequence: 0 }.produce();
		let mut consumer = producer.consume();

		let mut frame = producer
			.create_frame(frame::Info {
				size: 6,
				timestamp: Timestamp::ZERO,
			})
			.unwrap();
		frame.write(Bytes::from_static(b"foo")).unwrap();

		// A consumer can stream the in-flight tail before it's finished.
		let mut f = consumer.next_frame().now_or_never().unwrap().unwrap().unwrap();
		let c1 = f.read_chunk().now_or_never().unwrap().unwrap();
		assert_eq!(c1, Some(Bytes::from_static(b"foo")));
		assert!(f.read_chunk().now_or_never().is_none());

		frame.write(Bytes::from_static(b"bar")).unwrap();
		frame.finish().unwrap();

		let c2 = f.read_chunk().now_or_never().unwrap().unwrap();
		assert_eq!(c2, Some(Bytes::from_static(b"bar")));
		let c3 = f.read_chunk().now_or_never().unwrap().unwrap();
		assert_eq!(c3, None);
	}

	#[test]
	fn group_finish_returns_none() {
		let producer = Info { sequence: 0 }.produce();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		let end = consumer.next_frame().now_or_never().unwrap().unwrap();
		assert!(end.is_none());
	}

	#[test]
	fn abort_propagates() {
		let producer = Info { sequence: 0 }.produce();
		let mut consumer = producer.consume();
		producer.abort(crate::Error::Cancel).unwrap();

		let result = consumer.next_frame().now_or_never().unwrap();
		assert!(matches!(result, Err(crate::Error::Cancel)));
	}

	#[test]
	fn abort_clears_cached_frames() {
		let mut producer = Info { sequence: 0 }.produce();
		producer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"data"))
			.unwrap();

		// A stale consumer that never reads must not pin the cached frames.
		let _consumer = producer.consume();
		assert_eq!(producer.state.read().frames.len(), 1);

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

		let state = producer.state.read();
		assert!(state.frames.is_empty(), "cached frames should be dropped on abort");
		assert_eq!(state.cache, 0);
	}

	#[test]
	fn drop_unfinished_clears_cached_frames() {
		let producer = Info { sequence: 0 }.produce();
		let mut writer = producer.clone();
		writer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"data"))
			.unwrap();

		// A stale consumer keeps the channel (and thus the cache) alive.
		let mut consumer = producer.consume();
		assert_eq!(producer.state.read().frames.len(), 1);

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

		let result = consumer.next_frame().now_or_never().unwrap();
		assert!(matches!(result, Err(crate::Error::Dropped)));
	}

	#[test]
	fn drop_after_abort_does_not_warn() {
		let warns = count_drop_warnings("group::Producer dropped without finish", || {
			let producer = Info { sequence: 0 }.produce();
			let keep = producer.clone();
			let mut writer = producer.clone();
			writer
				.write_frame(Timestamp::ZERO, Bytes::from_static(b"data"))
				.unwrap();
			let _consumer = producer.consume();
			writer.abort(crate::Error::Cancel).unwrap();
			drop(keep);
		});
		assert_eq!(warns, 0, "abort-then-drop must not emit unfinished-producer WARN");
	}

	#[test]
	fn drop_unfinished_warns() {
		let warns = count_drop_warnings("group::Producer dropped without finish", || {
			let producer = Info { sequence: 0 }.produce();
			let mut writer = producer.clone();
			writer
				.write_frame(Timestamp::ZERO, Bytes::from_static(b"data"))
				.unwrap();
			let _consumer = producer.consume();
			drop(writer);
			drop(producer);
		});
		assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN");
	}

	#[test]
	fn drop_finished_keeps_cached_frames() {
		let mut producer = Info { sequence: 0 }.produce();
		producer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"data"))
			.unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		drop(producer);

		// A cleanly finished group keeps its cache so the consumer can still drain.
		let frame = consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(frame.payload, Bytes::from_static(b"data"));
	}

	#[tokio::test]
	async fn pending_then_ready() {
		let mut producer = Info { sequence: 0 }.produce();
		let mut consumer = producer.consume();

		// Consumer blocks because no frames yet.
		assert!(consumer.next_frame().now_or_never().is_none());

		producer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"data"))
			.unwrap();
		producer.finish().unwrap();

		let frame = consumer.next_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(frame.size, 4);
	}

	#[test]
	fn overflow_aborts_the_group() {
		let mut producer = Info { sequence: 0 }.produce();
		let mut consumer = producer.consume();

		let big = Bytes::from(vec![0u8; MAX_CACHE_BYTES as usize]);
		producer.write_frame(Timestamp::ZERO, big.clone()).unwrap();
		assert!(matches!(
			producer.write_frame(Timestamp::ZERO, big),
			Err(Error::GroupTooLarge)
		));

		{
			let state = producer.state.read();
			assert!(matches!(state.abort, Some(Error::GroupTooLarge)));
			assert!(state.frames.is_empty());
			assert_eq!(state.offset, 0);
		}

		let result = consumer.next_frame().now_or_never().unwrap();
		assert!(matches!(result, Err(Error::GroupTooLarge)));
	}

	#[test]
	fn no_overflow_under_budget() {
		let mut producer = Info { sequence: 0 }.produce();
		// 8192 one-byte frames is the largest legal group; they all stay cached.
		for _ in 0..MAX_GROUP_FRAMES {
			producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"x")).unwrap();
		}
		producer.finish().unwrap();

		let state = producer.state.read();
		assert_eq!(state.offset, 0);
		assert_eq!(state.frames.len(), MAX_GROUP_FRAMES);
		assert!(state.abort.is_none());
	}

	#[test]
	fn writer_sees_group_too_large_on_the_8193rd_frame() {
		let mut producer = Info { sequence: 0 }.produce();
		for _ in 0..MAX_GROUP_FRAMES {
			producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"x")).unwrap();
		}
		assert!(matches!(
			producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"x")),
			Err(Error::GroupTooLarge)
		));
		assert!(matches!(producer.state.read().abort, Some(Error::GroupTooLarge)));
	}

	#[test]
	fn clone_consumer_independent() {
		let mut producer = Info { sequence: 0 }.produce();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"a")).unwrap();

		let mut c1 = producer.consume();
		// Read one frame from c1
		let _ = c1.next_frame().now_or_never().unwrap().unwrap().unwrap();

		// Clone c1, inheriting its index (past first frame).
		let mut c2 = c1.clone();

		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"b")).unwrap();
		producer.finish().unwrap();

		// c2 should get the second frame (inherited index)
		let f = c2.next_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(f.size, 1); // "b"

		let end = c2.next_frame().now_or_never().unwrap().unwrap();
		assert!(end.is_none());
	}

	fn prefetched_consumer(pool: &cache::Pool, max_age: std::time::Duration) -> (Producer, Consumer) {
		let cache = cache::Track::new(pool.clone(), kio::Weak::new());
		let track = track::Info::default().with_max_age(max_age);
		let mut producer = Producer::new(Info { sequence: 0 }, track, cache);
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"a")).unwrap();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"b")).unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		(producer, consumer)
	}

	#[test]
	fn prefetch_refresh_honors_pool_expiry() {
		let config = cache::Config::default().with_expiry(std::time::Duration::from_secs(1));
		let pool = cache::Pool::new(config);
		let (producer, mut consumer) = prefetched_consumer(&pool, std::time::Duration::MAX);
		let before = producer.cache_accessed();

		crate::model::clock::advance(std::time::Duration::from_millis(600));
		consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();

		assert!(producer.cache_accessed() > before, "the pool cadence is used");
	}

	#[test]
	fn prefetch_refresh_honors_track_max_age() {
		let config = cache::Config::default().with_expiry(std::time::Duration::from_secs(30));
		let pool = cache::Pool::new(config);
		let (producer, mut consumer) = prefetched_consumer(&pool, std::time::Duration::from_secs(1));
		let before = producer.cache_accessed();

		crate::model::clock::advance(std::time::Duration::from_millis(600));
		consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();

		assert!(producer.cache_accessed() > before, "the track cadence remains in force");
	}

	/// Reading more than one prefetch batch drains every frame in order across the
	/// batch boundary (the refill starts exactly where the previous batch ended).
	#[test]
	fn read_frame_crosses_prefetch_batches() {
		let n = Prefetch::CAP * 3 + 5;
		let mut producer = Info { sequence: 0 }.produce();
		for i in 0..n {
			producer
				.write_frame(Timestamp::ZERO, Bytes::from(vec![i as u8; 4]))
				.unwrap();
		}
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		for i in 0..n {
			let frame = consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
			assert_eq!(frame.payload, Bytes::from(vec![i as u8; 4]));
		}
		assert!(consumer.read_frame().now_or_never().unwrap().unwrap().is_none());
	}

	/// A finished group is still aborted once its frames are released to free memory (the
	/// track's max age window, or the cache pool). A reader that already drained every frame
	/// is missing nothing, so it must see the clean end of group rather than the abort.
	#[test]
	fn abort_after_finish_keeps_the_clean_end_for_a_drained_reader() {
		let mut producer = Info { sequence: 0 }.produce();
		producer
			.write_frame(Timestamp::ZERO, Bytes::from_static(b"hello"))
			.unwrap();
		producer.finish().unwrap();

		let mut drained = producer.consume();
		let mut behind = producer.consume();
		let frame = drained.read_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(frame.payload, Bytes::from_static(b"hello"));

		producer.abort(Error::Old).unwrap();

		// Drained everything before the abort: nothing is missing.
		assert!(drained.read_frame().now_or_never().unwrap().unwrap().is_none());
		assert!(drained.next_frame().now_or_never().unwrap().unwrap().is_none());

		// Never read the frame, and its bytes are gone: a truncated stream, not a clean end.
		assert!(matches!(behind.read_frame().now_or_never().unwrap(), Err(Error::Old)));
	}

	/// `finished` answers for the cursor: a drained reader gets the clean end even after the
	/// abort that released the cache, and one that stopped short gets that abort. The
	/// producer's total stays available on `frame_count`.
	#[test]
	fn finished_answers_for_the_cursor() {
		let mut producer = Info { sequence: 0 }.produce();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"a")).unwrap();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"b")).unwrap();
		producer.finish().unwrap();

		let mut drained = producer.consume();
		let mut behind = producer.consume();
		while drained.read_frame().now_or_never().unwrap().unwrap().is_some() {}
		behind.read_frame().now_or_never().unwrap().unwrap().unwrap();

		producer.abort(Error::Old).unwrap();

		assert_eq!(drained.finished().now_or_never().unwrap().unwrap(), 2);
		assert!(matches!(behind.finished().now_or_never().unwrap(), Err(Error::Old)));
		assert_eq!(behind.frame_count(), 2);
	}

	/// A cursor on a group aborted for overflowing its budget can never reach the end,
	/// so `finished` reports that abort instead of parking forever.
	#[test]
	fn finished_reports_a_group_too_large() {
		let mut producer = Info { sequence: 0 }.produce();
		let mut consumer = producer.consume();

		let big = Bytes::from(vec![0u8; MAX_CACHE_BYTES as usize]);
		producer.write_frame(Timestamp::ZERO, big.clone()).unwrap();
		assert!(matches!(
			producer.write_frame(Timestamp::ZERO, big),
			Err(Error::GroupTooLarge)
		));

		assert!(matches!(
			consumer.finished().now_or_never().unwrap(),
			Err(Error::GroupTooLarge)
		));
	}

	/// `next_frame` drains frames a prior `read_frame` prefetched, preserving order.
	#[test]
	fn interleave_read_and_next_frame() {
		let mut producer = Info { sequence: 0 }.produce();
		for i in 0..5u8 {
			producer.write_frame(Timestamp::ZERO, Bytes::from(vec![i; 1])).unwrap();
		}
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		// The first whole-frame read prefetches all five frames into the batch.
		let f0 = consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		assert_eq!(f0.payload, Bytes::from(vec![0u8; 1]));

		// next_frame must continue from the batch, not skip ahead or repeat.
		for i in 1..5u8 {
			let mut f = consumer.next_frame().now_or_never().unwrap().unwrap().unwrap();
			let data = f.read_all().now_or_never().unwrap().unwrap();
			assert_eq!(data, Bytes::from(vec![i; 1]));
		}
		assert!(consumer.next_frame().now_or_never().unwrap().unwrap().is_none());
	}

	/// A `read_frame` whose index sits past the buffered frames (cleared by an abort)
	/// must surface the error, not panic on an out-of-range `range(local..)`.
	#[test]
	fn read_frame_past_cleared_frames_does_not_panic() {
		let mut producer = Info { sequence: 0 }.produce();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"a")).unwrap();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"b")).unwrap();

		let mut consumer = producer.consume();
		consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();

		// Abort clears the cached frames but leaves the consumer's index (2) past them, so the
		// refill's `local` (2) exceeds `frames.len()` (0).
		producer.abort(Error::Cancel).unwrap();

		let result = consumer.read_frame().now_or_never().unwrap();
		assert!(matches!(result, Err(Error::Cancel)), "expected Cancel, got {result:?}");
	}

	/// Dropping a consumer mid-batch must drop the buffered-but-untaken frames
	/// (exercises the `MaybeUninit` Drop path; run under miri to catch leaks/UB).
	#[test]
	fn drop_with_partial_batch() {
		let mut producer = Info { sequence: 0 }.produce();
		for _ in 0..Prefetch::CAP {
			producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"x")).unwrap();
		}
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		// Take one frame so the batch is filled but only partially drained.
		let _ = consumer.read_frame().now_or_never().unwrap().unwrap().unwrap();
		drop(consumer);
	}

	/// A parked chunk reader is woken by each chunk write. kio only notifies when
	/// a write guard was mutably accessed, so `frame_notify` must mark the guard
	/// modified; a guard dropped untouched wakes nobody and the reader would
	/// stall until the frame completed.
	#[tokio::test]
	async fn chunk_write_wakes_parked_reader() {
		let mut producer = Info { sequence: 0 }.produce();
		let mut consumer = producer.consume();
		let mut frame = producer
			.create_frame(frame::Info {
				size: 6,
				timestamp: Timestamp::ZERO,
			})
			.unwrap();
		let mut f = consumer.next_frame().await.unwrap().unwrap();
		let handle = tokio::spawn(async move { f.read_chunk().await });
		// Let the reader park on the empty partial before the chunk lands.
		tokio::time::sleep(std::time::Duration::from_millis(50)).await;
		frame.write(Bytes::from_static(b"foo")).unwrap();
		let chunk = tokio::time::timeout(std::time::Duration::from_secs(2), handle)
			.await
			.expect("parked chunk reader was never woken by the chunk write")
			.unwrap()
			.unwrap();
		assert_eq!(chunk, Some(Bytes::from_static(b"foo")));
	}

	/// A frame whose timestamp is at a different scale is converted to the group's
	/// scale by `create_frame`.
	#[test]
	fn create_frame_converts_mismatched_scale() {
		use crate::{Timescale, Timestamp};

		let mut producer = Producer::new(
			Info { sequence: 0 },
			track::Info::default().with_timescale(Timescale::MICRO),
			Default::default(),
		);
		let frame = frame::Info {
			size: 3,
			timestamp: Timestamp::from_millis(1).unwrap(), // 1ms -> 1000µs
		};
		let writer = producer.create_frame(frame).unwrap();
		assert_eq!(writer.timestamp.scale(), Timescale::MICRO);
		assert_eq!(writer.timestamp.value(), 1000);
	}

	/// An explicit current timestamp is converted to the group's scale.
	#[tokio::test]
	async fn create_frame_converts_current_timestamp() {
		use crate::Timescale;

		let mut producer = Producer::new(
			Info { sequence: 0 },
			track::Info::default().with_timescale(Timescale::MICRO),
			Default::default(),
		);
		let writer = producer
			.create_frame(frame::Info {
				size: 3,
				timestamp: Timestamp::now(),
			})
			.unwrap();
		assert_eq!(writer.timestamp.scale(), Timescale::MICRO);
		assert!(!writer.timestamp.is_zero(), "local clock should be non-zero");
	}

	/// A group can start partway in, so a route can serve the tail of a group whose
	/// head came from somewhere else.
	#[test]
	fn start_at_starts_the_group_later() {
		let mut producer = Info { sequence: 0 }.produce();
		producer.start_at(3).unwrap();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"d")).unwrap();
		producer.finish().unwrap();

		// The frame landed at index 3, so the group's length counts the missing head.
		assert_eq!(producer.frame_count(), 4);

		let mut consumer = producer.consume();
		assert_eq!(consumer.frame_count(), 4);

		// A reader positioned at the start is missing the head, and `finished` answers
		// for that cursor.
		assert!(matches!(
			consumer.finished().now_or_never().unwrap(),
			Err(Error::Lagged)
		));
		assert!(matches!(
			consumer.read_frame().now_or_never().unwrap(),
			Err(Error::Lagged)
		));
	}

	/// Seeking to the group's first available frame is how a spliced reader picks up
	/// the tail; a lower index clamps up rather than failing.
	#[test]
	fn start_at_clamps_up_to_the_first_frame() {
		let mut producer = Info { sequence: 0 }.produce();
		producer.start_at(3).unwrap();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"d")).unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		consumer.start_at(1);
		assert_eq!(consumer.index(), 3, "clamped up to the first frame that exists");
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload,
			Bytes::from_static(b"d")
		);
	}

	/// `end_at` ends the read cleanly at the cap, and raising it re-offers the frames
	/// still cached behind it.
	#[test]
	fn end_at_caps_and_reopens() {
		let mut producer = Info { sequence: 0 }.produce();
		for i in 0..4u8 {
			producer.write_frame(Timestamp::ZERO, Bytes::from(vec![i])).unwrap();
		}
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		consumer.set_frames(..2);
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			0
		);
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			1
		);
		assert!(
			consumer.read_frame().now_or_never().unwrap().unwrap().is_none(),
			"capped reads end cleanly"
		);

		consumer.set_frames(..);
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			2
		);
	}

	#[test]
	fn frame_ranges_preserve_progress_and_make_inclusion_explicit() {
		let mut producer = Info { sequence: 0 }.produce();
		for i in 0..4u8 {
			producer.write_frame(Timestamp::ZERO, Bytes::from(vec![i])).unwrap();
		}
		producer.finish().unwrap();
		let mut consumer = producer.consume();
		consumer.set_frames(1..=1);
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			1
		);
		assert!(consumer.read_frame().now_or_never().unwrap().unwrap().is_none());
		consumer.set_frames(..3);
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			2
		);
		assert!(consumer.read_frame().now_or_never().unwrap().unwrap().is_none());
		consumer.set_frames(0..=3);
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
			3
		);
	}

	/// An exclusive cap at 0 is the empty range: no frame is delivered, and raising
	/// it re-offers the held frames.
	#[test]
	fn end_at_zero_is_empty() {
		let mut producer = Info { sequence: 0 }.produce();
		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"x")).unwrap();
		producer.finish().unwrap();

		let mut consumer = producer.consume();
		consumer.set_frames(..0);
		assert!(
			consumer.read_frame().now_or_never().unwrap().unwrap().is_none(),
			"empty cap delivers nothing"
		);

		consumer.set_frames(..1);
		assert_eq!(
			consumer.read_frame().now_or_never().unwrap().unwrap().unwrap().payload,
			Bytes::from_static(b"x")
		);
	}

	/// Where the group begins is part of its shape, so it can't move once frames exist.
	#[test]
	fn start_at_rejected_after_a_frame() {
		let mut producer = Info { sequence: 0 }.produce();
		// Re-declaring before the first frame is fine; the shape isn't committed yet.
		producer.start_at(2).unwrap();
		producer.start_at(3).unwrap();

		producer.write_frame(Timestamp::ZERO, Bytes::from_static(b"a")).unwrap();
		assert!(matches!(producer.start_at(4), Err(Error::Closed)));
		assert_eq!(producer.frame_count(), 4, "the frame landed at index 3");

		// Finishing likewise settles the shape.
		let mut producer = Info { sequence: 1 }.produce();
		producer.finish().unwrap();
		assert!(matches!(producer.start_at(1), Err(Error::Closed)));
	}

	/// The start must leave room for at least one frame index.
	#[test]
	fn start_at_rejects_the_largest_index() {
		let mut producer = Info { sequence: 0 }.produce();
		assert!(matches!(
			producer.start_at(usize::MAX as u64),
			Err(Error::BoundsExceeded(_))
		));
	}

	/// The per-frame size cap (the group byte budget) is enforced before allocating.
	#[test]
	fn create_frame_rejects_oversized() {
		let mut producer = Info { sequence: 0 }.produce();
		let result = producer.create_frame(frame::Info {
			size: MAX_CACHE_BYTES + 1,
			timestamp: Timestamp::ZERO,
		});
		assert!(matches!(result, Err(Error::FrameTooLarge)));
	}
}