/**
 * Track role handles: a live stream of groups within a broadcast.
 *
 * @module
 */

import { Duration, Instant, now, sleepUntil } from "@moq/clock";
import { Derived, type Dispose, type GetPromise, type Getter, Once, Signal } from "@moq/signals";
import * as cache from "./cache.ts";
import { Delivered } from "./delivered.ts";
import { DropSet, Runs } from "./drops.ts";
import {
	GroupTooLarge,
	ProtocolViolation,
	Skipped,
	StreamCode,
	StreamError,
	TimestampMismatch,
	TooFarBehind,
	toStreamCode,
} from "./error.ts";
import { type Asker, FillState, type Requests, requestsOf } from "./fill.ts";
import { type Frame, type Consumer as GroupConsumer, Producer as GroupProducer } from "./group.ts";
import type { IetfVersion } from "./ietf/version.ts";
import {
	type DropRun,
	dropAt,
	type FillStarts,
	finalSequence,
	groupBounds,
	type Hole,
	hooks,
	insertDrop,
	insertDrops,
	insertRun,
	type Land,
	maxDelayOf,
	NO_MAX_DELAY,
	type Recv,
	type ResumeSource,
	type Run,
	removeRun,
	runGaps,
	type StagedTrack,
	sameHole,
	skipHole,
	staleHole,
	subtractRuns,
	type TrackRequestOptions,
} from "./internal.ts";
import { RUN_OVERHEAD as ARRIVAL_OVERHEAD, grace, Ledger } from "./ledger.ts";
import { Timescale, type Timestamp } from "./time.ts";
import { checkEnd, checkSequence, max, min, UNBOUNDED } from "./util/sequence.ts";
import {
	bounded,
	type Holdings,
	hasTime,
	isWire,
	joinsLive,
	later,
	rangesOf,
	sameRanges,
	UNHELD,
	union,
	Window,
	wireSpan,
} from "./window.ts";
import { type Broadcast as BroadcastWire, registerTrackConsumer } from "./wire.ts";

export { Fill, Requests } from "./fill.ts";

// The group a non-blocking read takes, skipping drops and throwing the track's error;
// `undefined` when nothing is deliverable now.
function taken(read: () => Recv, ended: () => undefined): GroupConsumer | undefined {
	for (;;) {
		const recv = read();
		if (recv.kind === "group") return recv.group;
		if (recv.kind === "error") throw recv.error;
		if (recv.kind === "done") return ended();
		if (recv.kind !== "drop") return undefined;
	}
}

// The unset priority, the midpoint of the signed range: moq-lite carries it as 128 and
// moq-transport as 127, one urgency on both.
const DEFAULT_PRIORITY = 0;

// A group ended with one of these is a deliberate hole, never a cut readers see: cache policy,
// a reader that fell behind, or a cancel such as a relay's own unsubscribe. Mirrors Rust's
// `group::Handle::commit_abort`.
const HOLES: ReadonlySet<number> = new Set([
	StreamCode.Cancel,
	StreamCode.DeliveryTimeout,
	StreamCode.TooFarBehind,
	StreamCode.Old,
	StreamCode.Evicted,
]);

// The errors this cache closes a group with when it evicts it unfinished: an aged-out group,
// not a failed one. Mirrors Rust's cache-policy aborts in `group::Handle::commit_abort`.
const EVICTED = new WeakSet<Error>();

// Whether `successor`'s copy failed: closed with an error other than its eviction from this
// cache. Read from the copy, since a reader's own verdict (an eviction's gap, a stale group)
// says nothing about whether the group arrived.
function failed(successor: Successor): boolean {
	const abort = successor.fate();
	return abort !== undefined && !EVICTED.has(abort);
}

/**
 * A position within a track: a group's sequence and a frame's index inside it.
 *
 * As a track's end it is exclusive: `{ group: 90, frame: 0 }` ends before group 90, and
 * `{ group: 89, frame: 5 }` ends group 89 after its frames 0 to 4. Mirrors the Rust
 * `track::Position`.
 */
export interface Position {
	/** The group's sequence number within the track. */
	group: bigint;
	/** The frame's index within that group. */
	frame: number;
}

/**
 * How a track's frames are compressed, mirroring the Rust `track::Compression`: `"deflate"` makes
 * each group one DEFLATE stream (draft-lcurley-moq-flate), which suits small, repetitive frames
 * like JSON.
 */
export type Compression = "none" | "deflate";

/**
 * A track's immutable publisher properties, fixed for the lifetime of the track.
 *
 * A producer declares these once from an {@link InfoInit} (via {@link Request.accept} or
 * {@link Producer.accept}); a consumer awaits them via {@link Subscriber.info}
 * (resolved from the wire TRACK_INFO on lite-05+). They map 1:1 onto TRACK_INFO.
 */
export interface Info {
	/** How frame payloads are compressed per group on lite-07; other sessions receive plain frames. */
	compression: Compression;
	/**
	 * Units per second for this track's frame timestamps, or `null` for an untimed track, whose
	 * frames carry no timestamp.
	 *
	 * A timed track declares its scale, such as {@link Timescale | Timescale.MILLI}, or
	 * {@link Timescale | Timescale.MICRO} for media that needs sub-millisecond timing. Reported in
	 * TRACK_INFO on Lite05+ and as the TIMESCALE property on moq-transport. A track received
	 * without one (moq-lite before Lite05, or moq-transport without TIMESCALE) is untimed,
	 * and stays untimed when served onward.
	 */
	timescale: Timescale | null;
	/**
	 * Publisher Max Age: how long, in milliseconds, a superseded group stays fresh, or `null` for
	 * no publisher limit. It is stale once either clock reaches the budget: the wall-clock time
	 * since its successor arrived at this hop, or how far its successor's first timestamp trails
	 * the live edge. The newest group is never stale, and an untimed track is judged by the wall
	 * clock alone. Reported in TRACK_INFO (Lite05+) so relays re-serve with the same bound. A
	 * track received without one has no limit; zero serves only the live edge.
	 * This clamps each subscriber budget. A freshness bound, not retention: a hop's cache bounds
	 * the bytes it keeps with its own window.
	 */
	maxAge: Duration | null;
	/** Tie-break priority between subscriptions of equal subscriber priority (`-128..=127`, higher first). */
	priority: number;
	/**
	 * How many times its first frame a group's later stored bytes may grow before a joining
	 * subscriber asks for a new group, or `null` to never ask. A join against a newest group
	 * whose bytes after its first frame exceed this many times the first frame moves
	 * {@link Demand.wanted} past it. The joiner still receives the cached group while the new
	 * one is pending. `0` asks on any join after a delta. An integer in `0..2^32`.
	 */
	refresh: number | null;
	/** The original moq-transport publisher's draft, whose timestamps may rewind. @internal */
	ietf?: IetfVersion;
	/**
	 * Whether this copy's groups skip the rewind check, since its timestamps may be lite-05/06 send
	 * times. An {@link Info.ietf} track skips it too. Mirrors the Rust `track::Info::unchecked`.
	 * @internal
	 */
	unchecked?: boolean;
}

/**
 * What an author supplies to create a track: any {@link Info} field. `compression` defaults to
 * `"none"`, `timescale` and `refresh` to `null`, `priority` to `0`, and `maxAge` to zero. A
 * received {@link Info} serves onward unchanged.
 */
export interface InfoInit extends Partial<Info> {
	/**
	 * The publisher max age; see {@link Info.maxAge}. Omitted is zero, so readers get only the
	 * live edge; `null` sets no publisher limit (the wire omits it). A track whose readers reach
	 * back for superseded groups sets it explicitly. Rounded up to a whole millisecond; a
	 * negative or non-finite value, or one past `Number.MAX_SAFE_INTEGER`, is refused.
	 */
	maxAge?: Duration | null;
}

/**
 * Resolve an author's {@link InfoInit} into the {@link Info} a producer commits, validating each field.
 * @internal
 */
export function authoredInfo(init: InfoInit = {}): Info {
	if (init.compression !== undefined && init.compression !== "none" && init.compression !== "deflate")
		throw new TypeError(`unknown compression: ${init.compression}`);
	const maxAge = init.maxAge === undefined ? Duration.ZERO : init.maxAge;
	return {
		compression: init.compression ?? "none",
		timescale: init.timescale == null ? null : Timescale(init.timescale),
		maxAge: maxAge === null ? null : wireMillis("maxAge", maxAge),
		priority: priorityValue(init.priority ?? DEFAULT_PRIORITY),
		refresh: init.refresh == null ? null : refreshRatio(init.refresh),
		...(init.ietf === undefined ? {} : { ietf: init.ietf }),
		...(init.unchecked ? { unchecked: true } : {}),
	};
}

// Normalize a duration for the wire, which carries it as an unsigned varint.
//
// Callers derive it from measurements (a jitter estimate scaled off RTT), so a fractional
// millisecond is expected; ceil rather than round, because a budget shortened by rounding
// skips a group the subscriber still wants. Anything that is not a duration is refused
// here, where the field is named, rather than deep in the encoder.
function wireMillis(field: string, value: Duration): Duration {
	const millis = Math.ceil(value.asMillis());
	if (!Number.isSafeInteger(millis)) {
		throw new RangeError(`${field} exceeds the safe integer millisecond range: ${value}`);
	}
	return Duration.millis(millis);
}

function priorityValue(value: number): number {
	if (!Number.isInteger(value) || value < -128 || value > 127) {
		throw new RangeError(`priority must be an integer in -128..=127: ${value}`);
	}
	return value;
}

function refreshRatio(value: number): number {
	if (!Number.isInteger(value) || value < 0 || value > 0xffff_ffff) {
		throw new RangeError(`refresh must be an integer in 0..2^32: ${value}`);
	}
	return value;
}

/** An explicitly included or excluded group sequence. */
export type Bound = { included: bigint; excluded?: never } | { excluded: bigint; included?: never };

/** Group limits for {@link Subscriber.withRange}; an omitted endpoint is unbounded. */
export interface Groups {
	/** The first group boundary. */
	start?: Bound;
	/** The last group boundary. */
	end?: Bound;
}

/**
 * Where one end of a {@link Range} falls: a fixed group (`absolute`), a count of groups back from
 * the latest one (`relative`, clamped to group 0, so `{ relative: 0n }` is the latest group), or a
 * presentation time (`time`, the group containing it: the newest one starting at or before it).
 * Mirrors the Rust `Anchor`.
 *
 * Every anchor names a whole group: a subscription always starts at a group's first frame. A
 * relative or time anchor resolves at the hop that serves it, when it starts serving and again
 * whenever the subscriber replaces its ranges. Before any group exists a relative anchor resolves
 * at the first one, so `{ start: { relative: 0n } }` starts there.
 *
 * A time anchor resolves from where the groups the hop holds start. A start before every group
 * held reaches back to the oldest one, and one resolved before any group has started waits for
 * one that has, reading nothing of its range meanwhile. An end settles once a group starting
 * after it is known and every sequence below that one is accounted for; until then reads stop
 * at the first held group that may still start, so at most that one group past the end is read
 * before its first frame shows where it starts. A time anchor on an untimed track is refused
 * with {@link TimestampMismatch}, and on a moq-transport copy, whose timestamps may rewind.
 *
 * Older moq-lite versions carry only absolute anchors, the live start, and the oldest start,
 * refusing any other.
 */
export type Anchor =
	| { absolute: bigint; relative?: never; time?: never }
	| { relative: bigint; absolute?: never; time?: never }
	| { time: Timestamp; absolute?: never; relative?: never };

/**
 * A span of a track to deliver: from `start` up to `end`. Mirrors the Rust `Range`.
 *
 * An omitted start is the oldest group still around (Rust's `Range::oldest`), and an omitted end
 * keeps following the live edge. An absolute end is exclusive, so `{absolute: 2}..{absolute: 6}`
 * delivers groups 2 through 5; relative and time ends include their group, so
 * `{absolute: 0}..{relative: 0n}` delivers through the latest group.
 */
export interface Range {
	/** Where delivery starts, or the oldest group when omitted. */
	start?: Anchor;
	/** Where delivery stops, or onward when omitted. */
	end?: Anchor;
}

/**
 * How far behind the live edge a subscriber plays, in milliseconds: where it starts, and how
 * late it may fall. Mirrors the Rust `track::Delay`. Both default to zero, which starts at the
 * newest group and skips a superseded one at once. Each value is rounded up to a whole
 * millisecond, so one derived from a measurement is never shortened; a negative or non-finite
 * value, or one past `Number.MAX_SAFE_INTEGER` after rounding, is refused.
 */
export interface Delay {
	/**
	 * The target latency, the playout or jitter-buffer delay held behind the live edge. Omitted
	 * means `0`; `null` means not stated, which only a peer whose wire cannot carry it (moq-lite
	 * before 07, moq-transport) sends. A media consumer reads an unstated minimum as `0`.
	 *
	 * The aggregate across subscribers is the smallest stated one, so an unstated minimum never
	 * pulls it to zero, and it is forwarded upstream as is. A publisher may use it, such as to
	 * start a cold join that far back; nothing reads it there yet.
	 */
	min?: Duration | null;
	/**
	 * How long a non-latest group stays worth delivering, by the same two clocks as
	 * {@link Info.maxAge}: it is skipped once its successor arrived this long ago or starts this
	 * far behind the live edge, and never past the publisher's {@link Info.maxAge}. Omitted
	 * means `0`, the live edge only, timed or not. `null` sets no subscriber limit, so the
	 * publisher's {@link Info.maxAge} alone bounds it, which is what a reader reaching back for
	 * history wants; read in order, see {@link Ordered} for where that can block.
	 */
	max?: Duration | null;
}

/**
 * Per-subscription options, requested when a subscription opens and adjustable later via
 * {@link Subscriber.update}. Mirrors the Rust `Subscription`.
 * The obsolete `maxAge` and `maxDelay` fields are refused at runtime; use `delay`.
 */
export interface Subscription {
	/** Delivery priority relative to this session's other subscriptions (`-128..=127`, higher first). Defaults to `0`. */
	priority?: number;
	/** How far behind the live edge this subscriber plays; see {@link Delay}. Omitted is all zero. */
	delay?: Delay;
	/**
	 * The spans of the track to deliver; see {@link Range}.
	 *
	 * Defaults to `[{}]`: everything {@link Delay.max} still considers fresh.
	 * `[{ start: { relative: 0n } }]` starts at the latest group instead, and an empty list asks
	 * for nothing. {@link Delay.max} caps every range: a range names what may be delivered, and
	 * the budget skips what inside it has gone stale. A read whose ranges all end finishes once
	 * nothing in them is left to deliver, as in Rust, while one left open follows the live edge.
	 *
	 * A request to the publisher, aggregated across every live subscriber as the union of their
	 * ranges, and this subscriber's read filter: it reads only the groups its own ranges cover,
	 * whatever another subscriber widens the union to. {@link Subscriber.withRange} narrows the
	 * reads further.
	 */
	ranges?: Range[];
	/**
	 * Which pending group the publisher sends first. Defaults to `"descending"`.
	 *
	 * Carried by moq-lite 07 and moq-transport (fixed at subscribe time there); older
	 * moq-lite versions are served newest first whatever this says.
	 *
	 * Aggregated across every live subscriber: any descending subscriber makes the aggregate
	 * descending, so a relay fanning one upstream request out keeps its live subscribers
	 * fresh. Each subscriber's own publisher still serves it in its own order.
	 */
	order?: Order;
}

/**
 * Which of a subscription's pending groups the publisher sends first: `"descending"` (newest
 * first, so congestion sheds the backlog rather than the live edge) or `"ascending"` (oldest
 * first, draining a backlog in sequence order).
 *
 * A transmission order only: it never decides whether a group is delivered, which is
 * {@link Delay.max}'s job, so an ascending subscriber that wants a backlog needs a
 * budget that covers it.
 */
export type Order = "ascending" | "descending";

/** Constructors for {@link Subscription}. */
export const Subscription: {
	/**
	 * The default subscription, used when none is passed: priority `0`, descending order, and
	 * a {@link Delay} of zero that skips a superseded group at once. Mirrors the Rust
	 * `Subscription::default()`. Spread it to change one field and keep the rest:
	 * `{ ...Subscription.default(), priority: 2 }`.
	 */
	default(): Subscription;
} = {
	default: () => ({
		priority: DEFAULT_PRIORITY,
		delay: { min: Duration.ZERO, max: Duration.ZERO },
		order: "descending",
	}),
};

// Materialize the defaults at the model boundary so every layer observes a complete
// subscription rather than interpreting an omitted field differently.
function subscriptionDefaults(subscription: Subscription = Subscription.default()): Subscription {
	if ("maxAge" in subscription || "maxDelay" in subscription) {
		throw new TypeError("Subscriber maxAge and maxDelay are unsupported; use delay.max");
	}
	if ("groups" in subscription) throw new TypeError("Subscription groups is unsupported; use ranges");
	const delay = subscription.delay ?? {};
	return {
		priority: priorityValue(subscription.priority ?? DEFAULT_PRIORITY),
		delay: {
			min: delay.min === null ? null : wireMillis("delay.min", delay.min ?? Duration.ZERO),
			max:
				delay.max === null
					? Duration.millis(NO_MAX_DELAY)
					: wireMillis("delay.max", delay.max ?? Duration.ZERO),
		},
		order: orderOf(subscription.order),
		ranges: subscription.ranges === undefined ? [{}] : rangesOf(subscription.ranges),
	};
}

// A plain JS caller can pass anything, and silently serving a typo as descending would hide it.
function orderOf(order: Order | undefined): Order {
	if (order === undefined) return "descending";
	if (order !== "ascending" && order !== "descending") throw new TypeError(`unknown order: ${String(order)}`);
	return order;
}

// The smaller of two minimums, ignoring one that went unstated (`null`).
function smallest(a: Duration | null | undefined, b: Duration | null | undefined): Duration | null {
	if (a === null || a === undefined) return b ?? null;
	if (b === null || b === undefined) return a;
	return a.min(b);
}

// Aggregate the preferences of `states`, matching Rust's Subscription::poll_combined.
function combineSubscriptions(states: Iterable<TrackState>): Subscription | undefined {
	let combined: Subscription | undefined;
	for (const state of states) {
		const subscription = state.update.peek();
		if (!subscription) continue;
		if (!combined) {
			combined = { ...subscription };
			continue;
		}

		combined.priority = Math.max(combined.priority ?? 0, subscription.priority ?? 0);
		combined.delay = {
			max: Duration.millis(Math.max(maxDelayOf(combined), maxDelayOf(subscription))),
			// Only stated minimums count, so one that went unstated never pulls the smallest to zero.
			min: smallest(combined.delay?.min, subscription.delay?.min),
		};
		// One live subscriber held behind another's backlog is worse than a backlog sent late.
		combined.order =
			combined.order === "ascending" && subscription.order === "ascending" ? "ascending" : "descending";

		// A range above the live edge never hides the latest group from a merged live subscriber.
		combined.ranges = [...(combined.ranges ?? []), ...(subscription.ranges ?? [])];
	}
	if (combined) combined.ranges = union(combined.ranges ?? []);
	return combined;
}

/**
 * A request for a track the peer wants, delivered to the publishing wire layer.
 *
 * Created internally by the broadcast when a subscription (or info lookup) needs a track
 * served; answer it with {@link accept} or {@link reject}.
 *
 * @public
 */
export class Request {
	/** The requested track name. */
	readonly name: string;

	#producer: Producer;
	#pending: Set<Request>;
	#lookups?: TrackRequestOptions["lookups"];
	#wanted?: Getter<boolean>;

	private constructor(options: TrackRequestOptions) {
		this.name = options.name;
		this.#producer = options.producer;
		this.#pending = options.pending;
		this.#lookups = options.lookups;
		this.#pending.add(this);
	}

	static {
		hooks.makeRequest = (options) => new Request(options);
		hooks.stageRequest = (request, info) => request.#stage(info);
		hooks.requestWanted = (request) => {
			const wanted = request.#producer.demand().wanted;
			request.#wanted ??= request.#lookups
				? new Derived(
						[wanted, request.#lookups.pending],
						(group, lookups) => group !== undefined || lookups > 0,
					)
				: new Derived([wanted], (group) => group !== undefined);
			return request.#wanted;
		};
		hooks.keepRequest = (request) => {
			if (request.#lookups) request.#lookups.opened = false;
		};
	}

	/** Watch the subscribers waiting for this request. */
	demand(): Demand {
		return this.#producer.demand();
	}

	/** The aggregate subscription requested for this track, one-shot fetches included. */
	get subscription(): Readonly<Subscription> {
		return hooks.requested(this.#producer).peek() ?? subscriptionDefaults();
	}

	/** The subscriber's priority for this track. */
	get priority(): number {
		return this.subscription.priority ?? 0;
	}

	/**
	 * Accept the request, committing the track's immutable {@link Info}, resolved as
	 * {@link Producer.accept} does. Throws, rejecting the request, if `info` is invalid.
	 */
	accept(info: InfoInit = {}): Producer {
		return this.#stage(info).commit();
	}

	#stage(info: InfoInit): StagedTrack {
		this.#pending.delete(this);
		const producer = this.#producer;
		return {
			producer,
			commit: () => {
				try {
					return producer.accept(info);
				} catch (err) {
					// Refused, so its subscribers see the error instead of waiting on an open request.
					producer.abort(err instanceof Error ? err : new Error(String(err)));
					throw err;
				}
			},
		};
	}

	/** Reject the request with the error its subscribers receive. */
	reject(err: Error): void {
		this.#pending.delete(this);
		this.#producer.abort(err);
	}
}

/** Options for {@link Consumer.info}. */
export interface InfoOptions {
	/** Abandons the lookup, rejecting with the signal's reason; aborting after it resolves has no effect. */
	signal?: AbortSignal;
}

/** Options for {@link Consumer.fetchGroup}. */
export interface FetchGroupOptions {
	/** Delivery priority for the fetch stream. Defaults to `0`. */
	priority?: number;

	/** Maximum age behind the live edge this caller accepts. Omission leaves the publisher's {@link Info.maxAge} as the only bound. */
	maxDelay?: Duration;

	/**
	 * Abandons this fetch, rejecting with the signal's reason. Concurrent fetches of the same
	 * group share one stream, cancelled only once every caller has left. An already-aborted
	 * signal rejects before anything is sent, and aborting after the group resolves has no
	 * effect; close the group instead.
	 */
	signal?: AbortSignal;
}

/**
 * A lazy handle to a track on a consumed broadcast.
 *
 * @public
 */
export class Consumer {
	/** The track name. */
	readonly name: string;

	#broadcast: BroadcastWire;

	private constructor(name: string, broadcast: BroadcastWire) {
		this.name = name;
		this.#broadcast = broadcast;
	}

	static {
		registerTrackConsumer((name, broadcast) => new Consumer(name, broadcast));
	}

	/**
	 * Open a live subscription to the track.
	 *
	 * It reads only the groups its {@link Subscription.ranges} cover.
	 * {@link Delay.max} is what asks for data: delivery skips everything in range that the budget
	 * convicts. Without options this is {@link Subscription.default}, whose budget of zero
	 * delivers only the latest group; a larger one reaches back over what it can still use, and
	 * `delay: { max: null }` reaches back over every group the publisher's {@link Info.maxAge} still
	 * considers fresh.
	 */
	subscribe(options?: Subscription): Subscriber {
		return this.#broadcast.subscribe(this.name, options);
	}

	/**
	 * Fetch the track's immutable publisher properties without subscribing. It waits for as long
	 * as the publisher takes to answer: abort `options.signal` to give up.
	 */
	info(options?: InfoOptions): Promise<Info> {
		const signal = options?.signal;
		if (!signal) return this.#broadcast.resolveTrackInfo(this.name);
		if (signal.aborted) return Promise.reject(signal.reason);
		// Held only until the answer, so the signal abandons the lookup but never keeps the track.
		const answered = new AbortController();
		return this.#broadcast
			.resolveTrackInfo(this.name, AbortSignal.any([signal, answered.signal]))
			.finally(() => answered.abort());
	}

	/** Fetch a single group by sequence without holding a live subscription. */
	fetchGroup(sequence: bigint, options?: FetchGroupOptions): Promise<GroupConsumer> {
		try {
			checkSequence(sequence);
		} catch (err) {
			return Promise.reject(err);
		}
		if (options?.maxDelay !== undefined) {
			options = { ...options, maxDelay: wireMillis("maxDelay", options.maxDelay) };
		}
		if (!this.#broadcast.fetchGroup) {
			return Promise.reject(new Error("fetch group is not supported by this session's version"));
		}
		return this.#broadcast.fetchGroup(this.name, sequence, options);
	}
}

// The parts of sorted, disjoint `runs` inside `[start, end)`.
function intersect(runs: readonly Run[], start: bigint, end: bigint): Run[] {
	return runs
		.map((run) => ({ start: max(run.start, start), end: min(run.end, end) }))
		.filter((run) => run.start < run.end);
}

// The index of the first timeline group at or after `sequence`.
function timelineIndex(timeline: readonly { sequence: bigint }[], sequence: bigint): number {
	let lo = 0;
	let hi = timeline.length;
	while (lo < hi) {
		const mid = (lo + hi) >>> 1;
		if (timeline[mid].sequence < sequence) lo = mid + 1;
		else hi = mid;
	}
	return lo;
}

// Add a group to a sorted timeline, replacing any group with the same sequence. Groups
// usually arrive in order, so the append is checked first.
function timelineInsert(timeline: GroupConsumer[], group: GroupConsumer): void {
	const last = timeline.at(-1);
	if (!last || last.sequence < group.sequence) {
		timeline.push(group);
		return;
	}
	const index = timelineIndex(timeline, group.sequence);
	if (timeline[index]?.sequence === group.sequence) timeline[index] = group;
	else timeline.splice(index, 0, group);
}

// Add a group to a sink's sequence-sorted unread buffer, after any group with the same
// sequence, as a stable sort would. Checked for the usual in-order append first, so neither a
// replayed cache nor a growing backlog re-sorts the buffer on every arrival.
function bufferInsert(groups: GroupConsumer[], group: GroupConsumer): void {
	const last = groups.at(-1);
	if (!last || last.sequence <= group.sequence) groups.push(group);
	else groups.splice(timelineIndex(groups, group.sequence + 1n), 0, group);
}

// The shared state behind a Producer / Subscriber pair. Package-internal
// wiring, unexported so it never appears in the published type declarations.
class TrackState {
	/** The producer fanning into this sink, so a subscriber can mint a sibling of itself. */
	producer?: Producer;
	/** This sink's reader as the producer's fills know it, once it asked one. */
	asker?: Asker;
	/**
	 * Whether this sink's reader will still read group `sequence`: only a buffered group it will
	 * read keeps its source wanted. Set once the reader exists.
	 */
	reads?: (sequence: bigint) => boolean;
	groups = new Signal<GroupConsumer[]>([]);
	// Sequences the upstream publisher gave up, with why: an incoming SUBSCRIBE_SKIP or
	// SUBSCRIBE_DROP, or a producer cut. Kept apart from the cache so reclaim cannot
	// erase a disposition a downstream publisher still has to forward, and as runs so a wide
	// DROP costs one entry. A sink shares its producer's, which bills it to the track's account.
	drops = new DropSet();
	// The runs this sink's own fills refused, which it reads as given up beside `drops`.
	refused = new Signal<readonly DropRun[]>([]);
	// Every group the ranges of a received copy's current upstream terms name, which its stale runs
	// were judged against; undefined when unknown, which counts as every group.
	judged?: readonly Run[];
	// Every group still in the producer's replay cache, including groups this
	// subscriber already consumed, sorted by sequence. Drift anchors have the same
	// lifetime as content. Sorted so the latency guard, evaluated per group and per
	// arrival, searches it instead of scanning the whole retained window.
	timeline: GroupConsumer[] = [];
	// First timestamps mutate group state rather than track state, so held groups
	// watch this revision as well as arrivals when enforcing latency after handoff.
	timelineChanged = new Signal(0);
	// A copy received over a wire without ranges: see `hooks.dropAnchors`.
	dropsAnchors = new Signal(false);
	// Frames wake the sequence cursor only while an owed hole holds it.
	waitingOrdered = false;
	// One-shot group reads need FETCH, even when metadata has not reported a live edge.
	fetch = false;
	// Bumped when a wire subscription declares or resolves where its live feed starts, so a
	// publisher holding its first group for that answer wakes.
	startChanged = new Signal(0);
	// Bumped when paying eviction debt folds an arrival gap before its grace passes, so a reader
	// parked on a missing group there, with no timer to wake it, judges it again.
	folded = new Signal(0);
	latest?: bigint;
	// One past the highest group received, including evicted content.
	received = new Signal(0n);
	/**
	 * The declared end, by {@link Producer.finishAt} or a clean close at the head of the
	 * sequence after the highest produced, with the error the track ended with there. One
	 * value, so a wake on it sees both. Mirrors the Rust `End`.
	 */
	end = new Signal<End | undefined>(undefined);
	// A wire subscription's pending tail still owes a group below the end, so reaching the
	// end does not end readers yet. Mirrors the Rust `awaits_tail`.
	owed = new Signal(false);
	closed = new Once<Error | null>();
	// The producer's outcome, settled by its close or by a reached and settled end, whichever
	// is first. Only the producer's own state uses it.
	outcome = new Once<Error | null>();
	update: Signal<Subscription | undefined>;
	/** Resolved once the producer commits the immutable properties. */
	info = new Signal<Info | undefined>(undefined);
	/** Newest group head, frame, or location reported by the publisher. */
	largest = new Signal<Position | undefined>(undefined);
	// The group {@link largest} first named, where a relative anchor resolved before any group
	// existed starts.
	first?: bigint;
	// Set once {@link largest} can no longer advance while the track stays open.
	largestEnded = new Signal(false);

	constructor(subscription?: Subscription) {
		this.update = new Signal(subscription === undefined ? undefined : subscriptionDefaults(subscription));
	}
}

// A track's declared end. Mirrors the Rust `End`.
type End = {
	// The exclusive end: no frame at or past it exists.
	at: Position;
	// The error the track ended with at `at`, if it did not end cleanly: a reader reaching the
	// end gets it instead of the end of stream, while everything below stays readable.
	error?: Error;
	// A received end, declared ahead of the group streams below it that QUIC may deliver in any
	// order: readers wait for a hole below it to fill, or for the track to close, rather than end
	// at the boundary with groups still to arrive. Once the subscription feeding the copy ends (see
	// `hooks.cacheTrack`), nothing is owed and readers end at the boundary. Mirrors the Rust
	// `End::tail`.
	tail: boolean;
};

// The wire subscription feeding a received copy, and what it declared. Ending the subscription
// drops it whole, so nothing it declared outlives it.
type Wire = {
	// How its start bounds the fills readers ask below it.
	starts: FillStarts;
	// The first group its live feed serves (see `hooks.startTrack`), unset until its START
	// resolves it, null for a resolved-empty range, and the floor its SUBSCRIBE requested.
	served?: { start?: bigint | null; requested?: bigint };
};

// Settle a track state's closed Once. Idempotent: Once.set throws on a second settle, and a
// Producer closing its sinks races the Subscriber closing itself (the sink is only removed from
// #sinks a microtask later, via the closed subscription below).
function closeTrackState(state: TrackState, abort?: Error): boolean {
	if (state.closed.peek() !== undefined) return false;
	state.closed.set(abort ?? null);
	return true;
}

// Resolve the track's immutable publisher properties, or reject if it closes first.
// On a producer this resolves once info is committed (at accept time); on a consumer
// once the wire layer commits the properties it received (TRACK_INFO on lite-05+,
// SUBSCRIBE_OK on moq-transport) or an untimed default (older lite drafts), so awaiting it
// never yields a placeholder.
async function resolveInfo(state: TrackState): Promise<Info> {
	for (;;) {
		const info = state.info.peek();
		if (info) return info;

		const closed = state.closed.peek();
		if (closed instanceof Error) throw closed;
		if (closed !== undefined) throw new Error("track closed before info was known");

		await Signal.race(state.info, state.closed);
	}
}

// Whether `a` is past `b`.
function after(a: Position, b: Position): boolean {
	return a.group > b.group || (a.group === b.group && a.frame > b.frame);
}

// A source group retained in the producer cache, with the mirror handed to each sink
// so eviction can drop them together, and what it all bills the cache pool.
// `arrived` is when the group joined this track, which its mirrors age a predecessor from.
// `bound` is its first frame's timestamp once checked to fit between its neighbors', which
// bounds theirs; `unverified` marks one accepted below the eviction watermark instead.
// `fetched` marks one a fill answered, which the feed never promised.
type CachedGroup = {
	group: GroupProducer;
	charge: cache.Charge;
	mirrors: Map<TrackState, GroupConsumer>;
	arrived: Instant;
	bound?: Timestamp;
	unverified?: boolean;
	fetched?: boolean;
};

// A group's checked start, kept by sequence.
type Start = { sequence: bigint; start: Timestamp };

// A fresh read handle on a group the producer still caches; assigned in its static block.
let cachedMirror: (producer: Producer, sequence: bigint, from: number) => GroupConsumer | undefined;
// Whether the producer still caches a group that has not failed; assigned in its static block.
let cachedHolds: (producer: Producer, sequence: bigint) => boolean;
// The sequences in `[start, end)` the producer caches, unsorted, costing the smaller of the span
// and the cache; assigned in its static block.
let cachedIn: (producer: Producer, start: bigint, end: bigint) => bigint[];
// Mirror the cached groups in `[start, end)` that `offer` accepts into a sink again, cut ones
// included as a fresh sink gets them, for a reader whose floor moved below them; assigned in
// the producer's static block.
let reofferCached: (
	producer: Producer,
	sink: TrackState,
	start: bigint,
	end: bigint,
	offer: (sequence: bigint) => boolean,
) => void;
// Mirror the cached groups at `sequences` into a sink again, for a reader whose budget widened
// past the one it skipped them under; assigned in the producer's static block.
let reofferSequences: (producer: Producer, sink: TrackState, sequences: readonly bigint[]) => void;
// Whether a group may still land on the producer's live feed; assigned in its static block.
let mayLand: (producer: Producer, land: Land) => boolean;
// Recompute the producer's aggregates now, for a sink that just became a fetch; assigned in its
// static block.
let resubscribe: (producer: Producer) => void;
let owedRuns: (producer: Producer, start: bigint, end: bigint) => Iterable<Run & { folds?: Instant }>;
let owedGroup: (producer: Producer, start: bigint, end: bigint) => { sequence: bigint; folds?: Instant } | undefined;
let fillEdge: (producer: Producer) => bigint | undefined;
// The producer's cache account, which its readers' records bill to.
let accountOf: (producer: Producer) => cache.Track<CachedGroup>;
let sourceStart: (producer: Producer) => bigint | "pending" | undefined;
// The lowest group a reader naming `named` asks a fill for; see `Producer.#fillFloor`.
let fillFloor: (producer: Producer, named: bigint | undefined) => bigint | undefined;
// The producer's checked group starts, sorted by sequence; assigned in its static block.
let startsOf: (producer: Producer) => readonly Start[];
// The exclusive end of the run of sequences from `sequence` the producer's feed can no longer
// deliver, undefined while it may; assigned in its static block.
let goneAt: (producer: Producer, sequence: bigint) => bigint | undefined;

/** The groups a reader reads, and the exclusive end it stops at; undefined before its ranges resolve. */
type Coverage = { runs: readonly Run[]; cap?: bigint } | undefined;

/** Whether two coverages hold the same groups and cap. */
function sameCoverage(a: Coverage, b: Coverage): boolean {
	if (a === undefined || b === undefined) return a === b;
	return (
		a.cap === b.cap &&
		a.runs.length === b.runs.length &&
		a.runs.every((run, i) => run.start === b.runs[i].start && run.end === b.runs[i].end)
	);
}

/**
 * Whether the live edge `received` reached `final`: no sequence below it is left that was neither
 * written nor given up, as a declared skip or a drop. Mirrors the Rust `TrackState::reaches`.
 */
// A run of the drop set and the pool tick it outlives the idle window at.
interface Born extends Run {
	expires: number;
}

// Whether `a` and `b` give a sequence up alike, however each was folded in.
function sameVerdict(a: DropRun, b: DropRun): boolean {
	return sameHole({ ...a, received: undefined }, { ...b, received: undefined });
}

// The runs of `drops` and a reader's `refused` overlapping `[start, end)`, lowest first. A drop
// that `settles` the reader, judged from where it meets `start`, hides the refusals it covers:
// the track's verdict wins, as in Rust, but a stale one the reader rejects is no verdict for it.
function givenUp(
	drops: DropSet,
	refused: readonly DropRun[],
	start: bigint,
	end: bigint,
	settles: (run: DropRun) => boolean,
): DropRun[] {
	const found = drops.overlapping(start, end);
	if (refused.length === 0) return found;
	const own = refused.filter((run) => run.start < end && start < run.end);
	const hiding = found.filter((run) => settles({ ...run, start: max(run.start, start) }));
	found.push(...subtractRuns(own, hiding));
	return found.sort((a, b) => (a.start < b.start ? -1 : a.start > b.start ? 1 : 0));
}

// Whether the live edge reached `end`, counting a gap up to it that was wholly given up, or a
// received end's tail decides it instead: what it misses below is owed only while the feed may
// still deliver it. Mirrors the Rust `TrackState::reaches`.
function reaches(received: bigint, drops: DropSet, end: End, refused: readonly Run[] = []): boolean {
	if (end.tail) return true;
	const final = finalSequence(end.at);
	if (received >= final) return true;
	const given: Run[] = [];
	for (const run of [...drops.overlapping(received, final), ...refused]) insertRun(given, run.start, run.end);
	return runGaps(given, received, final).length === 0;
}

/**
 * Whether `run` settles a reader judging under `budget` with `coverage`, on a track whose upstream
 * judged stale runs against ranges naming the groups in `judged` (undefined for every group). Any
 * run but a received stale one does. A stale verdict holds for a reader no wider than the budget it
 * was judged under that reads every group the upstream's ranges name above the run's start: its
 * own successor and edge are then no older than those the verdict was judged against, since a
 * group's verdict looks only above it. Mirrors the Rust `settles`.
 */
function settles(run: DropRun, budget: number, coverage: Coverage, judged: readonly Run[] | undefined): boolean {
	if (run.stale === undefined) return true;
	if (budget > run.stale || coverage === undefined) return false;
	const above = run.start + 1n;
	// Whether the reader reads every group from `start` up to the exclusive `end`, `UNBOUNDED`
	// past the live edge.
	const reads = (start: bigint, end: bigint): boolean => {
		if (coverage.cap !== undefined && (end === UNBOUNDED || end > coverage.cap)) return false;
		const held = coverage.runs.find((run) => run.start <= start && start < run.end);
		return held !== undefined && held.end >= end;
	};
	for (const range of judged ?? [{ start: 0n, end: UNBOUNDED }]) {
		if (range.end <= above) continue;
		if (!reads(max(range.start, above), range.end)) return false;
	}
	return true;
}
// Actual object metadata outlives cached payloads, unlike empty group heads in model largest.
let objectLargest: (producer: Producer) => Position | undefined;
// The fills serving the producer's misses; assigned in its static block.
let fillsOf: (producer: Producer) => FillState;

// The sequence-order cursor lives inside `Subscriber` (it shares the group buffer and the
// drift anchor with the arrival cursor), so `Ordered` reaches it through this bridge,
// assigned in the class's static block.
let makeOrdered: (subscriber: Subscriber) => Ordered;
let ordered_: {
	nextGroup(subscriber: Subscriber): Promise<GroupConsumer | undefined>;
	tryNextGroup(subscriber: Subscriber): Recv;
	ended(subscriber: Subscriber): undefined;
	readFrame(subscriber: Subscriber): Promise<({ group: bigint; frame: number } & Frame) | undefined>;
	readString(subscriber: Subscriber): Promise<string | undefined>;
	readJson(subscriber: Subscriber): Promise<unknown | undefined>;
};

// Constructs a Demand from within this module; assigned in the class's static block.
let makeDemand: (name: string, wanted: Getter<bigint | undefined>, state: TrackState) => Demand;

/**
 * A watch-only view of a track's subscriber demand, from {@link Producer.demand} or {@link Request.demand}. Mirrors the
 * Rust `Demand`.
 *
 * Lets a publisher gate work (on-demand capture or encoding) on whether anyone is subscribed,
 * without the ability to write groups or close the track. The consumer wire watches it to tear an
 * idle upstream subscription down instead of downloading to nobody.
 */
export class Demand {
	/** The track name. */
	readonly name: string;

	#wanted: Getter<bigint | undefined>;
	#state: TrackState;

	private constructor(name: string, wanted: Getter<bigint | undefined>, state: TrackState) {
		this.name = name;
		this.#wanted = wanted;
		this.#state = state;
	}

	static {
		makeDemand = (name, wanted, state) => new Demand(name, wanted, state);
		hooks.demandPriority = (demand) => demand.#state.info.peek()?.priority ?? DEFAULT_PRIORITY;
		hooks.demandReleased = (demand) => demand.#state.closed;
	}

	/**
	 * The group subscribers want: `undefined` while nobody subscribes, otherwise the sequence
	 * the next group should have at least. A publisher opens a group whenever it passes its
	 * largest. The first subscriber of a track with no group wants `0`, and a join judged by
	 * {@link Info.refresh} wants the next free sequence until a group lands there. It never goes back. Watch it
	 * (`effect.get` / `.peek()`) to follow demand.
	 */
	get wanted(): Getter<bigint | undefined> {
		return this.#wanted;
	}

	/**
	 * Settles once the track's outcome is final: `null` once its declared end is reached and
	 * settled (its error, if declared with one), or on a clean release, else the abort
	 * {@link Error}. Never because readers left.
	 */
	get closed(): GetPromise<Error | null> {
		return this.#state.outcome;
	}
}

// Constructs a Subscriber from within this module without exposing a public
// constructor that would leak the unexported TrackState. Assigned in the class's
// static block.
let makeSubscriber: (name: string, state: TrackState) => Subscriber;

/**
 * The write side of a track, mirroring the Rust `Producer`.
 *
 * A producer is a fan-out source: every {@link subscribe} (including each wire
 * subscription the publisher serves from it) gets an independent
 * {@link Subscriber} that receives a full copy of the groups, each with its own
 * read cursor. Groups are mirrored into every live subscriber and retained for the
 * idle cache window so a late subscriber can replay groups within its media budget.
 *
 * Obtained from {@link Request.accept} (the wire asks the application for a track to
 * serve) or constructed directly for an in-process track.
 */
export class Producer {
	/** The track name. */
	readonly name: string;

	// The producer's own state is the source of truth (info/closed); subscribers
	// read mirrored sinks, never this state directly.
	#state = new TrackState();
	// One past the highest group this producer received, like the Rust
	// `max_sequence`, and where the next append lands.
	#received = 0n;
	#subscribing = new Signal(false);
	// The aggregate across live subscribers only. A fetch's preferences judge just the group it
	// asks for, so a publisher sizing what it serves (such as how much retained media to replay)
	// must not see them. The wire's aggregate, `#state.update`, keeps them.
	#live = new Signal<Subscription | undefined>(undefined);

	// Recently written source groups, in arrival order, retained for replay to late subscribers
	// until the cache pool evicts them, for its budget or once idle. A set, so evicting a run of
	// them is linear. Each entry tracks the mirror it handed to every sink so eviction can drop
	// them too: otherwise a slow consumer that never reads would pin old groups (and their frame
	// bytes) forever.
	#cache = new Set<CachedGroup>();
	// The same entries by sequence, so a write finds a duplicate without a scan.
	#cached = new Map<bigint, CachedGroup>();
	// The cached groups' checked starts, sorted by sequence, so a first frame finds its
	// neighbors' with a search.
	#bounds: Start[] = [];
	// The highest-sequence checked start evicted. Eviction is not a sequence prefix, so a
	// group above it is checked against it as well as its cached neighbors; one at or below
	// it may have lost its lower bound. One entry, so nothing grows with the cache.
	#evicted?: Start;
	// This track's account against its broadcast's cache pool (a pool of its own when it has no
	// broadcast), which bills the cached groups and evicts them under its budget and idle window.
	#account: cache.Track<CachedGroup>;
	// The pool tick past which each run of the drop set outlives the pool's idle window, as of when
	// it was last given up; empty when the pool has no idle window. See `#setDrops`.
	#dropsBorn = new Runs<Born>((a, b) => a.expires === b.expires);
	// At most the lowest tick in `#dropsBorn`, so an idle scan finds nothing to forget without
	// walking it.
	#dropsOldest?: number;
	// What the drop set and its expiry index bill to the account.
	#dropsBilled = 0;

	// One independent downstream state per live subscriber.
	#sinks = new Set<TrackState>();

	// The group subscribers want while any is attached, watched through {@link demand}.
	#wanted = new Signal<bigint | undefined>(undefined);
	// The lowest group they want while no refresh is pending, never lowered.
	#floor = 0n;
	// A join over the refresh ratio asked for a group past every sequence written, and none
	// has landed yet.
	#refresh = false;
	#demand: Demand;
	// The largest group may still grow until it closes cleanly; an aborted one can be recreated.
	#largestOpen = false;
	// Only real objects or upstream metadata advance this, never the head of an empty group.
	#objectLargest?: Position;

	// The wire subscription feeding this copy, from `hooks.receiveTrack` until `hooks.cacheTrack`
	// ends it: only a copy one feeds owes missing groups. Unset on a track's own publisher.
	#wire?: Wire;
	// The misses its readers asked of a {@link Requests} handle.
	#fills: FillState;
	// The sequences the live feed delivered, as a group or a drop, and the stale runs a reclaim
	// forgot: what a pending tail is judged by. Eviction never touches it. A gap folds away a
	// subscription's grace after a later arrival opened it, as the session's tail does, so this
	// holds only the gaps still worth waiting for.
	#arrivals = new Ledger();
	// What `#arrivals` bills to the account, `ARRIVAL_OVERHEAD` per gap. Paying eviction debt folds
	// its oldest gaps early, so an unbounded grace holds no more gaps than the pool would.
	#arrivalsBilled = 0;
	// What this track's own publisher produced since its last append, which skips every
	// sequence below it that never arrived and was not given up: the producer moved on. The
	// arrivals cannot say so, since a gap folds into them once nobody waits for it. Only
	// sequences at or above `#appended` are kept. Mirrors the Rust `produced`.
	#produced: Run[] = [];
	// One past the last append: every sequence below it arrived, was given up, or was skipped.
	#appended?: bigint;

	constructor(name: string) {
		this.name = name;
		this.#demand = makeDemand(name, this.#wanted, this.#state);
		this.#fills = new FillState({
			name,
			accept: (sequence, info) => {
				if (info !== undefined && this.#state.info.peek() === undefined) this.accept(info);
				const group = this.#createGroup(sequence, 0);
				const entry = this.#cached.get(sequence);
				if (entry) entry.fetched = true;
				return group;
			},
			drop: (start, end, error) => {
				this.#drop(start, end, { error });
				this.#chargeDebt();
			},
			holds: (sequence) => cachedHolds(this, sequence),
			closed: () => this.#state.closed.peek() !== undefined,
			released: this.#state.closed,
		});
		this.#account = this.#openAccount(new cache.Pool({ expiry: cache.DEFAULT_EXPIRY }));
	}

	// Open this track's account against `pool`, evicting through this producer's cache.
	#openAccount(pool: cache.Pool): cache.Track<CachedGroup> {
		return new cache.Track(pool, {
			protects: (entry) => this.#protects(entry),
			edge: () => {
				const entry = this.#edgeEntry();
				return entry && this.#protects(entry) ? entry : undefined;
			},
			evict: (entry, reason) => {
				this.#evict(entry, reason);
				this.#uncache(entry, true);
			},
			shed: (entry, debt, cap) => hooks.shedGroup(entry.group, debt, cap, entry.mirrors.values()),
			settle: () => this.#settle(),
			shedDrops: (debt, cap) => this.#shedDrops(debt, cap),
			shedArrivals: (debt, cap) => this.#shedArrivals(debt, cap),
			expireDrops: (now) => this.#expireDrops(now),
			holdsDrops: () => (this.#state.drops.all[0]?.start ?? UNBOUNDED) < this.#keptDrops(),
		});
	}

	// Give up `[start, end)` as `hole` says, or forget it without one, billing the drop set to the
	// track's account and dating each run against the pool's idle window. Mirrors the Rust
	// `TrackState::change_drops`.
	#setDrops(start: bigint, end: bigint, hole?: Hole): void {
		this.#state.drops.set(start, end, hole);
		const pool = this.#account.pool;
		const window = pool.expiryTicks;
		if (!hole) this.#dropsBorn.set(start, end);
		else if (window !== undefined) {
			// Rounded up to a quarter of the window, so runs given up together stay one entry and a
			// run is forgotten no sooner than the window after it was given up.
			const quarter = Math.max(1, Math.floor(window / 4));
			const expires = (Math.floor(pool.now() / quarter) + 1) * quarter + window;
			this.#dropsBorn.set(start, end, { start, end, expires });
			this.#dropsOldest = Math.min(this.#dropsOldest ?? expires, expires);
		}
		this.#billDrops();
	}

	// Bill the drop set and its expiry index to the account as they now stand.
	#billDrops(): void {
		const billed = this.#state.drops.size * cache.RUN_OVERHEAD + this.#dropsBorn.size * cache.BORN_OVERHEAD;
		if (billed > this.#dropsBilled) this.#account.bill(billed - this.#dropsBilled);
		else this.#account.unbill(this.#dropsBilled - billed);
		this.#dropsBilled = billed;
	}

	// Forget the lowest runs of the drop set toward `debt`, at most about `cap` bytes, returning the
	// bytes freed. Their sequences are unknown again: a reader judges them by its budget, or fills
	// them by range.
	#shedDrops(debt: number, cap: number): number {
		let paid = 0;
		const kept = this.#keptDrops();
		while (paid < debt && paid < cap) {
			const run = this.#state.drops.all[0];
			if (!run || run.start >= kept) break;
			const billed = this.#dropsBilled;
			const end = min(run.end, kept);
			this.#setDrops(run.start, end);
			this.#forgot([{ start: run.start, end }]);
			paid += billed - this.#dropsBilled;
		}
		return paid;
	}

	// Bill the arrivals' gaps to the account as they now stand. The run they split is the track's
	// own, so an in-order feed bills nothing. A closed track bills none: nothing arrives to add a
	// gap, and its account may outlive it in the pool's sweep, which would keep them billed.
	#billArrivals(): void {
		const closed = this.#state.closed.peek() !== undefined;
		const billed = closed ? 0 : Math.max(0, this.#arrivals.size - 1) * ARRIVAL_OVERHEAD;
		if (billed > this.#arrivalsBilled) this.#account.bill(billed - this.#arrivalsBilled);
		else this.#account.unbill(this.#arrivalsBilled - billed);
		this.#arrivalsBilled = billed;
	}

	// Fold the arrivals' oldest gaps toward `debt`, at most about `cap` bytes, returning the bytes
	// freed: nothing waits on them any longer, as when their grace passes, so a reader parked on one
	// judges it missing. Mirrors the Rust `TrackState::shed_arrivals`.
	#shedArrivals(debt: number, cap: number): number {
		let paid = 0;
		let folded = false;
		while (paid < debt && paid < cap && this.#arrivals.foldOldest()) {
			const billed = this.#arrivalsBilled;
			this.#billArrivals();
			paid += billed - this.#arrivalsBilled;
			folded = true;
		}
		if (folded) {
			for (const state of [this.#state, ...this.#sinks]) state.folded.update((revision) => revision + 1);
			this.#owe();
		}
		return paid;
	}

	// Forget the runs given up longer than the pool's idle window before tick `now`; walks the set
	// only once its oldest run is due, which a quarter of the window separates from the next.
	#expireDrops(now: number): boolean {
		if (this.#dropsOldest === undefined || now < this.#dropsOldest) return false;
		const kept = this.#keptDrops();
		const expired = this.#dropsBorn.all
			.filter((run) => now >= run.expires && run.start < kept)
			.map(({ start, end }) => ({ start, end: min(end, kept) }));
		// In one pass: runs given up in the same quarter of the window expire together.
		if (expired.length > 0) {
			this.#state.drops.clear(expired);
			this.#dropsBorn.clear(expired);
			this.#billDrops();
			this.#forgot(expired);
		}
		this.#dropsOldest = undefined;
		for (const run of this.#dropsBorn.all) {
			if (run.start < kept) this.#dropsOldest = Math.min(this.#dropsOldest ?? run.expires, run.expires);
		}
		return expired.length > 0;
	}

	// Account `forgotten`, runs a reclaim took out of the drop set, in the arrivals as an evicted
	// group stays: the publisher accounted for them, so neither the tail nor a reader waits on
	// them again, and a reader that wants one asks a fill or judges it by its budget. A reclaim
	// never adds a run to the arrivals: a run touching no accounted one takes the gap below it
	// too, given up early as a fold would. Nothing below the tail's floor is accounted, since
	// accounting it would lower the floor onto gaps nobody waited for. Mirrors the Rust
	// `TrackState::forget_drops`.
	#forgot(forgotten: readonly Run[]): void {
		const feed = this.#feedStart();
		for (const run of forgotten) {
			const floor = (feed === "pending" ? this.#wire?.served?.requested : feed) ?? this.#arrivals.first() ?? 0n;
			let start = max(run.start, floor);
			if (start >= run.end) continue;
			// The highest run starting at or below the run's end touches it when it ends at or past
			// its start; otherwise it is the run below.
			const below = this.#arrivals.endBelow(run.end);
			if (below === undefined || below < start) start = below === undefined ? floor : max(below, floor);
			this.#arrive(start, run.end);
		}
	}

	// Where the runs a reclaim may forget end: every run on a received copy, but only those below an
	// open local track's edge, since one above it is what lets its declared end be reached (see
	// `reaches`). A local producer gives up only what it chose to. Mirrors the Rust `kept_drops`,
	// except that a closed track's runs all go here too: Rust frees them with the state's last
	// handle, but here the pool holds the account until the sweep forgets them.
	#keptDrops(): bigint {
		return this.#wire || this.#state.closed.peek() !== undefined ? UNBOUNDED : this.#received;
	}

	// The runs from `from` up to `#keptDrops` were kept and are now reclaimable: date the oldest so
	// the next idle scan finds it, and have the sweep visit the track for them.
	#rearmDrops(from: bigint): void {
		const runs = this.#dropsBorn.overlapping(from, this.#keptDrops());
		for (const run of runs) this.#dropsOldest = Math.min(this.#dropsOldest ?? run.expires, run.expires);
		if (this.#state.drops.overlapping(from, this.#keptDrops()).length > 0)
			this.#account.pool.register(this.#account);
	}

	static {
		hooks.poolTrack = (producer, pool) => {
			if (!producer.#account.empty || producer.#state.drops.size > 0)
				throw new Error("track already caches groups");
			producer.#account = producer.#openAccount(pool);
		};
		hooks.arrivalRuns = (producer) => producer.#arrivals.size;
		accountOf = (producer) => producer.#account;
		hooks.billTrack = (producer) => producer.#account.reader();
		hooks.subscribing = (producer) => producer.#subscribing;
		hooks.requested = (producer) => producer.#state.update;
		resubscribe = (producer) => producer.#updateSubscription();
		hooks.cacheTrack = (producer) => {
			producer.#wire = undefined;
			// Nothing feeds the tail anymore, so it owes nothing and readers end at the boundary.
			for (const state of [producer.#state, ...producer.#sinks]) state.owed.set(false);
			producer.#endLargest();
		};
		hooks.refuseLive = (producer, edge, error) => {
			for (const sink of producer.#sinks) {
				if (sink.fetch) continue;
				const ranges = sink.update.peek()?.ranges ?? [{}];
				const span = wireSpan(ranges);
				if (bounded(ranges) && span.end !== undefined && span.end <= edge) continue;
				closeTrackState(sink, error);
			}
		};
		fillEdge = (producer) =>
			!producer.#wire && producer.#state.info.peek()?.ietf !== undefined
				? producer.#state.largest.peek()?.group
				: undefined;
		hooks.receiveTrack = (producer, starts = {}) => {
			producer.#wire = { starts, served: producer.#wire?.served };
		};
		hooks.receiveGroup = (producer, { sequence, start, fill }) => {
			const group = producer.#createGroup(sequence, start);
			const entry = fill ? producer.#cached.get(sequence) : undefined;
			if (entry) entry.fetched = true;
			return group;
		};
		hooks.arrivingGroup = (producer, sequence) => {
			const closed = producer.#cached.get(sequence)?.group.closed;
			return closed?.peek() === undefined ? closed : undefined;
		};
		sourceStart = (producer) => producer.#wire?.served && producer.#feedStart();
		fillFloor = (producer, named) => producer.#fillFloor(named);
		startsOf = (producer) => producer.#bounds;
		goneAt = (producer, sequence) => {
			// Without a declared start, the first arrival stands in for it, as the tail's does.
			const floor = producer.#wire?.served?.start ?? producer.#arrivals.first() ?? 0n;
			if (sequence < floor) return floor;
			return producer.#state.drops.at(sequence)?.end ?? producer.#arrivals.runEnd(sequence);
		};
		hooks.producerLargest = (producer) => producer.#state.largest;
		hooks.joinProducer = (producer) => producer.#join();
		hooks.endTrack = (producer, end, error) => producer.#endPending(end, error);
		hooks.tailGrace = (producer) => producer.#grace();
		hooks.releaseTrack = (producer) => producer.#close(undefined);
		hooks.trackReleased = (producer) => producer.#state.closed;
		hooks.trackFinished = (producer) => producer.#state.end.peek() !== undefined;
		hooks.dropTrackGroups = (producer, start, end, hole) => {
			const runs = producer.#drop(start, end, hole, true);
			// A peer's drops bill the pool as its groups do, so they pay their debt the same way.
			producer.#chargeDebt();
			return runs;
		};
		hooks.skipTrackGroups = (producer, start, end) => producer.#skipReceived(start, end);
		hooks.supersedeStale = (producer, judged) => {
			const held = producer.#state.drops.all
				.filter((run) => run.stale !== undefined)
				.map(({ start, end }) => ({ start, end }));
			for (const state of [producer.#state, ...producer.#sinks]) state.judged = judged;
			for (const run of held) producer.#setDrops(run.start, run.end);
			return held;
		};
		hooks.dropAnchors = (producer) => {
			for (const state of [producer.#state, ...producer.#sinks]) state.dropsAnchors.set(true);
		};
		hooks.startTrack = (producer, served) => {
			if (!producer.#wire) throw new Error("only a wire subscription declares a start");
			producer.#settle();
			producer.#wire.served = served;
			producer.#owe();
			for (const state of [producer.#state, ...producer.#sinks]) state.startChanged.update((rev) => rev + 1);
		};
		hooks.cachedLargest = (producer) => producer.#cachedLargest();
		hooks.reportedLargest = (producer, largest) => producer.#advanceObject(largest);
		hooks.reportedGroup = (producer, group) => producer.#advanceLargest({ group, frame: 0 });
		objectLargest = (producer) => producer.#objectLargest;
		fillsOf = (producer) => producer.#fills;
		cachedHolds = (producer, sequence) => {
			const entry = producer.#cached.get(sequence);
			return entry !== undefined && !(entry.group.closed.peek() instanceof Error);
		};
		cachedIn = (producer, start, end) => {
			if (end - start <= BigInt(producer.#cached.size)) {
				const found: bigint[] = [];
				for (let sequence = start; sequence < end; sequence++)
					if (producer.#cached.has(sequence)) found.push(sequence);
				return found;
			}
			return [...producer.#cached.keys()].filter((sequence) => sequence >= start && sequence < end);
		};
		cachedMirror = (producer, sequence, from) => {
			const entry = producer.#cached.get(sequence);
			if (!entry || entry.group.closed.peek() instanceof Error) return undefined;
			const mirror = entry.group.mirror();
			if (hooks.groupIndex(mirror) > from) {
				mirror.close();
				return undefined;
			}
			hooks.arriveGroup(mirror, entry.arrived);
			// A cache hit is an access: the group outranks those nobody read.
			entry.charge.refresh();
			return mirror;
		};
		reofferCached = (producer, sink, start, end, offer) => {
			const held = new Set(sink.groups.peek().map((group) => group.sequence));
			for (const entry of producer.#cache) {
				const sequence = entry.group.sequence;
				if (sequence < start || sequence >= end || held.has(sequence) || !offer(sequence)) continue;
				producer.#mirror(entry, sink);
			}
		};
		reofferSequences = (producer, sink, sequences) => {
			const held = new Set(sink.groups.peek().map((group) => group.sequence));
			for (const sequence of sequences) {
				const entry = producer.#cached.get(sequence);
				if (entry && !held.has(sequence)) producer.#mirror(entry, sink);
			}
		};
		mayLand = (producer, land) => producer.#landable(land) !== undefined;
		owedRuns = (producer, start, end) => {
			if (!producer.#wire) return [];
			const feed = producer.#feedStart();
			const floor = feed === "pending" ? producer.#wire.served?.requested : feed;
			return producer.#landing({
				start: max(start, floor ?? 0n),
				end,
				floor: floor !== undefined,
				grace: producer.#grace(),
			});
		};
		owedGroup = (producer, start, end) => {
			for (const run of owedRuns(producer, start, end)) return { sequence: run.start, folds: run.folds };
			return undefined;
		};
	}

	// The first group that may still land from the live feed, by the rule a pending tail is judged
	// by; see {@link Land}. Mirrors the Rust `TrackState::may_land`.
	#landable(land: Land): { sequence: bigint; folds?: Instant } | undefined {
		for (const run of this.#landing(land)) {
			if (!land.delivered?.covers(run.start, run.end)) return { sequence: run.start, folds: run.folds };
		}
		return undefined;
	}

	*#landing(land: Land): Generator<Run & { folds?: Instant }> {
		if (this.#state.closed.peek() !== undefined) return;
		let start = land.start;
		if (!land.floor) {
			const first = this.#arrivals.first();
			if (first === undefined) return;
			start = max(start, first);
		}
		start = max(start, this.#wire?.served?.start ?? 0n);
		const declared = this.#state.end.peek()?.at;
		const end = declared === undefined ? land.end : min(land.end, finalSequence(declared));
		const at = now();
		const limit = land.grace;
		const drops = this.#state.drops;
		for (const gap of this.#arrivals.gaps(start, end)) {
			const folds = gap.since?.saturatingAdd(limit);
			if (folds !== undefined && at.ge(folds)) continue;
			for (const open of runGaps(drops.overlapping(gap.start, gap.end), gap.start, gap.end)) {
				yield { ...open, folds };
			}
		}
	}

	// Record `[start, end)` given up as `hole` says, returning the runs recorded. A skip is final
	// and the most authoritative verdict, so a drop never replaces one. An upstream drop (`open`)
	// also ends the groups still arriving in it with its error, so their readers settle now
	// rather than once the stream's reset lands; one that finished here stays whole. A skip names
	// nothing produced.
	#drop(start: bigint, end: bigint, hole: Hole, open = false): Run[] {
		if (start >= end) return [];
		const skipped = hole.skip ? [] : this.#state.drops.overlapping(start, end).filter((run) => run.skip);
		let runs = runGaps(skipped, start, end);
		// A received stale verdict leaves a group that arrived and was not aborted: a lite-07
		// publisher may serve a group after its stale drop, and the group's stream may land first.
		// The group wins, still arriving or not.
		if (open && hole.stale !== undefined) {
			for (const [sequence, entry] of this.#cached) {
				if (sequence >= start && sequence < end && !(entry.group.closed.peek() instanceof Error)) {
					runs = removeRun(runs, sequence, sequence + 1n);
				}
			}
		}
		// A fill no longer owes what the track gave up for good. A stale verdict settles only
		// the readers it covers, so a fill asked by any other is still owed.
		if (hole.stale === undefined) this.#fills.dropped(start, end);
		for (const run of runs) this.#setDrops(run.start, run.end, hole);
		if (open && !hole.skip && hole.stale === undefined) {
			for (const [sequence, entry] of this.#cached) {
				if (sequence >= start && sequence < end && entry.group.closed.peek() === undefined) {
					entry.group.abort(hole.error);
				}
			}
		}
		// A pending tail owes nothing the publisher gave up. A stale verdict is not receipt: new
		// terms may supersede it, leaving those groups unknown again, as in Rust.
		if (hole.stale === undefined) this.#arrive(start, end);
		// Giving up the last sequences below the end reaches it, as writing them would.
		this.#endLargest();
		return runs;
	}

	// A publisher may re-create a group it cut, and a group a reader's fill refused may still
	// arrive; the sequence is no longer given up.
	#undrop(sequence: bigint): void {
		if (this.#state.drops.at(sequence)) this.#setDrops(sequence, sequence + 1n);
		for (const sink of this.#sinks) {
			const refused = sink.refused.peek();
			if (dropAt(refused, sequence)) sink.refused.set(removeRun(refused, sequence, sequence + 1n));
		}
	}

	/**
	 * Resolve this track's immutable publisher properties, committed at accept time.
	 * Rejects if the track is closed before the properties are known.
	 */
	info(): Promise<Info> {
		return resolveInfo(this.#state);
	}

	/**
	 * Publisher priority from the committed {@link Info}, or the default before {@link accept}.
	 *
	 * Higher is served first. Hang publishers set this from `Catalog.PRIORITY` so
	 * audio outranks video on the wire and in the bandwidth allocator.
	 */
	get priority(): number {
		return this.#state.info.peek()?.priority ?? DEFAULT_PRIORITY;
	}

	/**
	 * Settles once the track's outcome is final: `null` once its declared end is reached and
	 * settled (every group below it finished; its error, if declared with one) or on a clean
	 * release, else the abort {@link Error}. The producer stays open after a settled end, serving its cache.
	 * Peek it synchronously (`undefined` while open), observe it reactively, or `await` it.
	 */
	get closed(): GetPromise<Error | null> {
		return this.#state.outcome;
	}

	/** The aggregate subscription across live subscribers, or `undefined` when there are none. */
	get subscription(): Getter<Subscription | undefined> {
		return this.#live;
	}

	/**
	 * Commit the immutable publisher properties, resolving {@link info}. An omitted `maxAge` is
	 * zero, the live edge only; see {@link InfoInit.maxAge}. Returns `this`.
	 */
	accept(info: InfoInit = {}): this {
		if (this.#state.closed.peek() !== undefined) return this;
		const resolved = authoredInfo(info);
		// Anything written before accept() must match too. A mismatch closes the track, so
		// nothing waiting on info() hangs on a half-bound cache.
		try {
			for (const { group } of this.#cache) {
				hooks.bindGroupTimed(group, resolved);
				hooks.bindGroupCompression(group, resolved.compression === "deflate");
			}
		} catch (err) {
			this.abort(err instanceof Error ? err : new Error(String(err)));
			throw err;
		}
		this.#state.info.set(resolved);
		// Propagate to any sink handed out before accept (the on-demand path).
		for (const sink of this.#sinks) sink.info.set(resolved);
		this.#updateSubscription();
		return this;
	}

	/**
	 * An independent {@link Subscriber} reading this track's groups.
	 *
	 * It reads only the groups its {@link Subscription.ranges} cover.
	 * {@link Delay.max} is what asks for data: delivery skips everything in range that the budget
	 * convicts. Without options this is {@link Subscription.default}, whose budget of zero
	 * delivers only the latest group; a larger one reaches back over what it can still use, and
	 * `delay: { max: null }` reaches back over every group the publisher's {@link Info.maxAge} still
	 * considers fresh.
	 */
	subscribe(options: Subscription = Subscription.default()): Subscriber {
		const sink = new TrackState(options);
		this.#addSink(sink);
		return makeSubscriber(this.name, sink);
	}

	/** A watch-only view of this track's subscriber demand. */
	demand(): Demand {
		return this.#demand;
	}

	/**
	 * Open a {@link Requests} handle serving the groups readers miss: ones their ranges name that
	 * the cache no longer holds, or never did, below the newest group written. Most producers
	 * never need one; a publisher backed by storage opens one to serve its history. Mirrors the
	 * Rust `Producer::requests`.
	 *
	 * While one is open, each run of misses queues a {@link Fill}, which readers missing the same
	 * groups share; with none open, a miss is given up and {@link Consumer.fetchGroup} of an
	 * uncached group fails with `NotFound`. A reader only asks for what its budget still wants,
	 * so a reader at the live edge never asks. The track stays one track: answered groups land
	 * in its cache, and every reader of it sees them.
	 */
	requests(): Requests {
		return requestsOf(this.#fills);
	}

	// A subscriber joined at the live edge: ask for a group past every sequence written when
	// the newest group's bytes after its first frame exceed the refresh ratio times the first
	// frame.
	#join(): void {
		const ratio = this.#state.info.peek()?.refresh;
		const latest = this.#state.latest;
		if (ratio == null || latest === undefined) return;
		const group = this.#cached.get(latest)?.group;
		if (!group || !hooks.overRefresh(group, ratio)) return;
		this.#refresh = true;
		this.#reportWanted();
	}

	// Report the wanted group while anyone subscribes: the next free sequence while a refresh
	// is pending.
	#reportWanted(): void {
		const wanted = this.#refresh ? max(this.#received, this.#floor) : this.#floor;
		this.#wanted.set(this.#sinks.size > 0 ? wanted : undefined);
	}

	// Register a downstream sink: seed its info, replay the retained window, and (while
	// the track is open) mirror future groups into it. A late subscriber to a closed
	// track still drains the buffered groups before seeing the end.
	#addSink(sink: TrackState): void {
		sink.producer = this;
		sink.drops = this.#state.drops;
		const info = this.#state.info.peek();
		if (info) sink.info.set(info);
		sink.largest.set(this.#state.largest.peek());
		sink.first = this.#state.largest.peek()?.group;
		sink.largestEnded.set(this.#state.largestEnded.peek());
		sink.dropsAnchors.set(this.#state.dropsAnchors.peek());

		const closed = this.#state.closed.peek();
		if (closed === undefined) {
			if (joinsLive(sink.update.peek()?.ranges)) this.#join();
			this.#sinks.add(sink);
			this.#reportWanted();

			// Forward subscription updates from the sink's Subscriber to the producer's own
			// state, which the wire layer (or the serving application) watches.
			const forward = sink.update.subscribe(() => this.#updateSubscription());
			this.#updateSubscription();

			// Drop the sink once its consumer goes away, closing its mirrors so source
			// groups stop teeing into them, so a long-lived producer doesn't leak. This
			// covers mirrors already handed out via recvGroup (no longer in sink.groups)
			// by closing them through the cache's per-sink tracking.
			const dispose = sink.closed.subscribe((c) => {
				if (c === undefined) return;
				const abort = c instanceof Error ? c : undefined;
				forward();
				this.#sinks.delete(sink);
				if (sink.asker) this.#fills.leave(sink.asker);
				this.#updateSubscription();
				// Update demand: once the last subscriber leaves, the consumer wire (watching
				// {@link Demand.wanted}) tears the upstream down instead of downloading to nobody.
				this.#reportWanted();
				// A closed track leaves a sink's buffered mirrors readable; they age out
				// with the cache, so a stale consumer can't pin them.
				if (this.#state.closed.peek() !== undefined) {
					dispose();
					return;
				}
				for (const entry of this.#cache) {
					const mirror = entry.mirrors.get(sink);
					if (mirror) {
						// A group its recovery moved to another copy no longer reads this one.
						if (hooks.readsMirrored(mirror)) hooks.releaseGroupReader(mirror, abort);
						entry.mirrors.delete(sink);
						entry.charge.sub(cache.MIRROR_OVERHEAD);
					}
				}
				for (const group of sink.groups.peek()) hooks.releaseGroupReader(group, abort);
				dispose();
			});
		}

		// Full, unlike a write's bounded scan, so the subscriber gets no group already idled out.
		this.#evictExpired(true);
		for (const entry of this.#cache) this.#mirror(entry, sink);
		sink.judged = this.#state.judged;

		sink.received.set(this.#received);
		sink.end.set(this.#state.end.peek());
		sink.owed.set(this.#state.owed.peek());
		if (closed !== undefined) {
			closeTrackState(sink, closed instanceof Error ? closed : undefined);
		}
	}

	// Recompute from every sink because an update or close can narrow as well as widen the
	// aggregates. The wire layer observes `#state.update` and emits SUBSCRIBE_UPDATE.
	//
	// The max delay is the subscribers' own, not clamped to the track's maxAge: the publisher
	// applies its own, as each reader's budget does, so the demand asked upstream is the same
	// before and after the info lands.
	#updateSubscription(): void {
		const live = [...this.#sinks].filter((sink) => !sink.fetch);
		this.#subscribing.set(live.some((sink) => sink.closed.peek() === undefined));
		const combined = combineSubscriptions(this.#sinks);
		this.#settle();
		this.#state.update.set(combined);
		this.#live.set(combineSubscriptions(live));
		// The requested floor bounds what a pending tail owes.
		this.#owe();
	}

	// Mirror a cached source group into a sink. The mirror fills synchronously as the
	// source is written and keeps its own read cursor; frame bytes are shared by
	// reference. Tracked on the entry so eviction can drop it from the sink.
	#mirror(entry: CachedGroup, sink: TrackState): void {
		const dst = entry.group.mirror();
		hooks.arriveGroup(dst, entry.arrived);
		if (entry.unverified) hooks.unverifyGroup(dst);
		if (!entry.mirrors.has(sink)) entry.charge.bill(cache.MIRROR_OVERHEAD);
		entry.mirrors.set(sink, dst);
		if (sink.reads?.(dst.sequence) === false) hooks.groupReads(dst, false);
		timelineInsert(sink.timeline, dst);
		void dst.readable().then(() => sink.timelineChanged.update((revision) => revision + 1));
		sink.latest = max(sink.latest ?? 0n, dst.sequence);
		sink.received.set(max(sink.received.peek(), dst.sequence + 1n));
		sink.groups.mutate((groups) => bufferInsert(groups, dst));
	}

	// Drop a cached group's mirror from every sink so no consumer can pin it, the cache pool
	// having reclaimed it for its bytes (`evicted`) or for idling (`expired`).
	#evict(entry: CachedGroup, reason: "evicted" | "expired" = "expired"): void {
		// Reclaiming a group the publisher never closed ends it as the gap it is, before
		// the mirrors below would otherwise report a clean finish to a reader that had
		// drained it. The usual case, an already-closed group aging out, keeps its own
		// terminal state.
		const lost =
			reason === "evicted"
				? new StreamError(StreamCode.Evicted, { message: "group evicted from the cache" })
				: new TooFarBehind();
		EVICTED.add(lost);
		if (!entry.group.isClosed) entry.group.abort(lost);
		const mirrors = [...entry.mirrors.values()].filter((mirror) => hooks.evictGroup(mirror, lost));
		this.#unlink(entry);
		for (const mirror of mirrors) mirror.close();
	}

	// Take a cached group's mirrors out of every sink, so a subscriber can no longer
	// receive them. A reader already holding one keeps it as is.
	#unlink(entry: CachedGroup): void {
		for (const [sink, mirror] of entry.mirrors) {
			sink.groups.mutate((groups) => {
				const i = groups.indexOf(mirror);
				if (i >= 0) groups.splice(i, 1);
			});
			const index = timelineIndex(sink.timeline, mirror.sequence);
			if (sink.timeline[index] === mirror) sink.timeline.splice(index, 1);
		}
		entry.charge.sub(entry.mirrors.size * cache.MIRROR_OVERHEAD);
		entry.mirrors.clear();
	}

	// Take a cached group out of the cache lookups, releasing what it billed the pool.
	#uncache(entry: CachedGroup, evicted = false): void {
		this.#cache.delete(entry);
		this.#cached.delete(entry.group.sequence);
		this.#unbound(entry, evicted);
		this.#account.remove(entry);
		hooks.chargeGroup(entry.group, undefined);
	}

	// The starts a first frame of group `sequence` must fall between: the nearest checked
	// starts cached on either side, since groups land out of sequence order, and the eviction
	// watermark. Throws {@link TimestampMismatch} on a rewind; an equal start is allowed. Returns
	// whether the group is late, at or below the watermark, where its floor may be gone, so
	// the watermark caps it instead and it is accepted unverified.
	#admit(sequence: bigint, start: Timestamp): boolean {
		const bounds = this.#bounds;
		const index = timelineIndex(bounds, sequence);
		let floor = bounds[index - 1]?.start;
		let ceiling = bounds[bounds[index]?.sequence === sequence ? index + 1 : index]?.start;
		const evicted = this.#evicted;
		const late = evicted !== undefined && sequence <= evicted.sequence;
		if (evicted && late) {
			if (ceiling === undefined || evicted.start.isBefore(ceiling)) ceiling = evicted.start;
		} else if (evicted) {
			if (floor === undefined || floor.isBefore(evicted.start)) floor = evicted.start;
		}
		if ((floor && start.isBefore(floor)) || ceiling?.isBefore(start)) throw new TimestampMismatch();
		return late;
	}

	// Record a cached group's checked start: a bound for its neighbors, or unverified.
	#record(entry: CachedGroup, start: Timestamp, late: boolean): void {
		if (late) {
			entry.unverified = true;
			for (const mirror of entry.mirrors.values()) hooks.unverifyGroup(mirror);
			return;
		}
		entry.bound = start;
		const sequence = entry.group.sequence;
		const last = this.#bounds.at(-1);
		if (!last || last.sequence < sequence) this.#bounds.push({ sequence, start });
		else this.#bounds.splice(timelineIndex(this.#bounds, sequence), 0, { sequence, start });
	}

	// Drop a group's checked start as it leaves the cache, raising the watermark to it when
	// it was evicted.
	#unbound(entry: CachedGroup, evicted = false): void {
		const start = entry.bound;
		if (start === undefined) return;
		const sequence = entry.group.sequence;
		const index = timelineIndex(this.#bounds, sequence);
		if (this.#bounds[index]?.sequence === sequence) this.#bounds.splice(index, 1);
		if (evicted && (this.#evicted === undefined || sequence > this.#evicted.sequence)) {
			this.#evicted = { sequence, start };
		}
	}

	// The cached group at the newest sequence written, if still cached.
	#edgeEntry(): CachedGroup | undefined {
		const latest = this.#state.latest;
		return latest === undefined ? undefined : this.#cached.get(latest);
	}

	// The newest group remains replayable while the track is live, even after the group
	// closes: a catalog can publish one snapshot for the broadcast's whole lifetime. An
	// aborted one holds nothing worth replaying, so it ages out like any other.
	#protects(entry: CachedGroup): boolean {
		return (
			this.#state.closed.peek() === undefined &&
			entry.group.sequence === this.#state.latest &&
			!(entry.group.closed.peek() instanceof Error)
		);
	}

	// The newest frame in the cache, as {@link Subscriber.largest} reads it from a sink:
	// unknown once the newest group is evicted, since what is left cannot say where the edge is.
	#cachedLargest(): Position | undefined {
		const latest = this.#state.latest;
		if (latest === undefined || !this.#cached.has(latest)) return undefined;

		let largest: Position | undefined;
		for (const { group } of this.#cache) {
			const count = hooks.groupFrameCount(group);
			if (count > hooks.groupOffset(group) && (largest === undefined || group.sequence > largest.group)) {
				largest = { group: group.sequence, frame: count - 1 };
			}
		}
		return largest;
	}

	#advanceLargest(largest: Position | undefined): void {
		if (!largest) return;
		const current = this.#state.largest.peek();
		if (current && !after(largest, current)) return;
		this.#largestOpen = true;
		for (const state of [this.#state, ...this.#sinks]) {
			state.first ??= largest.group;
			state.largest.set(largest);
		}
	}

	#advanceObject(largest: Position | undefined): void {
		if (largest && (!this.#objectLargest || after(largest, this.#objectLargest))) this.#objectLargest = largest;
		this.#advanceLargest(largest);
	}

	/** The track's largest location: its newest group at the newest frame, or undefined before any. */
	largest(): Position | undefined {
		return this.#state.largest.peek();
	}

	// Accrue and pay the cache pool's eviction debt for what this track wrote.
	#chargeDebt(): void {
		if (this.#account.chargeDebt()) this.#settle();
	}

	// Evict the groups idle past the pool's window: every one when `full`, else a bounded scan.
	#evictExpired(full: boolean): void {
		if (this.#account.evictExpired(full)) this.#settle();
	}

	// Retain a source group and fan it out to every live sink; its frame 0 is checked
	// against this track when it lands.
	#publish(group: GroupProducer): void {
		// Before the group is cached, so the write creating it can never evict it.
		this.#chargeDebt();
		this.#settle();
		const entry: CachedGroup = {
			group,
			charge: this.#account.charge(),
			mirrors: new Map<TrackState, GroupConsumer>(),
			arrived: now(),
		};
		hooks.chargeGroup(group, entry.charge);
		this.#undrop(group.sequence);
		group.closed.then((closed) => {
			// An aborted live edge is no longer protected, so it may expire like any other.
			if (closed instanceof Error) this.#account.demote(entry);
			if (!(closed instanceof Error) || HOLES.has(toStreamCode(closed))) return;
			// A re-created group already replaced this one.
			const current = this.#cached.get(group.sequence);
			if (current && current.group !== group) return;
			this.#drop(group.sequence, group.sequence + 1n, { error: closed });
		});
		const sequence = group.sequence;
		if (this.#unchecked()) {
			// Its timestamps may rewind, so they are not checked.
		} else {
			hooks.admitGroupStart(group, (start) => {
				const late = this.#admit(sequence, start);
				return () => this.#record(entry, start, late);
			});
		}
		this.#cache.add(entry);
		this.#cached.set(group.sequence, entry);
		this.#fills.arrived(group.sequence);
		this.#advance(group.sequence, true);
		this.#arrive(group.sequence);
		this.#produce(group.sequence);
		// A newer group demotes the live edge to an evictable one.
		const edge = this.#edgeEntry();
		this.#state.latest = max(this.#state.latest ?? 0n, group.sequence);
		if (edge && edge !== entry && !this.#protects(edge)) this.#account.demote(edge);
		this.#account.insert(entry);
		for (const sink of this.#sinks) this.#mirror(entry, sink);
		// Follow the group's frames as they land, past the head `#advance` already named.
		const stop = hooks.groupProgress(group, (total) => {
			if (total > hooks.groupOffset(group)) this.#advanceObject({ group: group.sequence, frame: total - 1 });
			// A frame can change an owed hole's reach or drift edge without adding a group.
			for (const sink of this.#sinks)
				if (sink.waitingOrdered) sink.timelineChanged.update((revision) => revision + 1);
		});
		void group.closed.then((closed) => {
			stop();
			for (const sink of this.#sinks)
				if (sink.waitingOrdered) sink.timelineChanged.update((revision) => revision + 1);
			if (closed === null) this.#finishLargest(group.sequence);
		});
		// Give held mirrors the new live edge before pruning their timeline entry,
		// so their latency guard can preserve a terminal expiry verdict.
		this.#evictExpired(false);
	}

	// Record a written sequence. It raises the largest too, at the head of an empty group, so
	// a later group below it never changes it. A group (`answers`) at or past the next free
	// sequence answers a pending refresh; anything else written there moves the sequence it
	// asks for.
	#advance(sequence: bigint, answers: boolean): void {
		const next = this.#received;
		const kept = this.#keptDrops();
		this.#received = max(next, sequence + 1n);
		if (this.#keptDrops() > kept) this.#rearmDrops(kept);
		this.#advanceLargest({ group: sequence, frame: 0 });
		this.#endLargest();
		if (!this.#refresh || sequence < next) return;
		if (answers) {
			this.#refresh = false;
			this.#floor = max(this.#floor, sequence);
		}
		this.#reportWanted();
	}

	// The sequence the next append takes: one past this producer's own writes.
	#nextSequence(): bigint {
		const next = this.#received;
		this.#writable(next);
		return next;
	}

	// Hold a group to this track's timedness, once the track has committed it.
	#bind(group: GroupProducer): void {
		const info = this.#state.info.peek();
		if (info) {
			hooks.bindGroupTimed(group, info);
			hooks.bindGroupCompression(group, info.compression === "deflate");
		}
	}

	// Refuse a write once the track is closed, or at or past its declared end.
	#writable(sequence: bigint): void {
		if (this.#state.closed.peek() !== undefined) throw new Error("track is closed");
		const end = this.#state.end.peek()?.at;
		if (end !== undefined && sequence >= finalSequence(end)) {
			throw new Error(`sequence ${sequence} is at or past the track's end ${end.group}/${end.frame}`);
		}
	}

	// Cap a group the end cuts, so it never holds a frame past the end.
	#cap(group: GroupProducer): void {
		const end = this.#state.end.peek()?.at;
		if (end !== undefined && end.frame !== 0 && end.group === group.sequence) hooks.limitGroup(group, end.frame);
	}

	/**
	 * Append a new group with the next sequence number: one past this producer's own writes.
	 *
	 * Appending says the producer will not go back, so it skips every sequence below the new
	 * group that it never produced, from just past its previous append (before the first, from
	 * the lowest group it created), as {@link skipGroups} would, replacing any drop there. Those
	 * sequences are then final: creating one throws {@link Skipped}. Appending alone never leaves
	 * such a gap; one left by out-of-order {@link createGroup} calls is what it names.
	 *
	 * The same broadcast, track, and group sequence always name the same content, and nothing
	 * checks it: the broadcast remembers nothing of an earlier producer of the name, so a
	 * replacement starts at 0. A recreated track, or a restarted publisher keeping its instance,
	 * continues past its predecessor's groups with {@link createGroup}, or uses a new track name
	 * or instance. Reusing a sequence only corrupts this broadcast for its readers. Datagram
	 * sequences name no cached content and are exempt.
	 *
	 * Its frames must match the track's {@link Info.timescale}: a timestamp on every frame of
	 * a timed track and none on an untimed one, or the write throws {@link TimestampMismatch}.
	 * So does a first frame that starts before the group below it or after the one above it.
	 */
	appendGroup(): GroupProducer {
		const sequence = this.#nextSequence();
		this.#refuseSkipped(sequence);
		const group = new GroupProducer(sequence);
		this.#bind(group);
		this.#cap(group);
		this.#publish(group);
		this.#settleAppend(sequence);

		return group;
	}

	/**
	 * Declare that this producer will never produce `[start, end)`: none will arrive, so a reader
	 * waiting on one moves past it, the declared end does not wait for them, and a relay serving
	 * this track forwards the skip (lite-07 SUBSCRIBE_SKIP). Unlike a drop, it ends no reader.
	 *
	 * This and {@link appendGroup} are the only sources of a skip: a gap between created groups
	 * says nothing, since groups may be created out of order, so only the producer knows a
	 * sequence will never exist. A skip is final: creating a group at a skipped sequence throws
	 * {@link Skipped}, and a later drop of it is ignored.
	 *
	 * A skip replaces any drop of the same sequences, whatever its error: a reader the drop
	 * already ended stays ended, and every later reader and relay sees the skip. A group already
	 * cached keeps what it is, since it was produced after all. Mirrors the Rust `skip_groups`.
	 */
	skipGroups(start: bigint, end: bigint): void {
		checkSequence(start);
		checkEnd(end);
		this.#skipReceived(start, end);
	}

	// `skipGroups`, returning the cached sequences it left alone: a skip naming a group that
	// arrived contradicts it, which a subscriber reports.
	#skipReceived(start: bigint, end: bigint): Run[] {
		if (this.#state.closed.peek() !== undefined) return [];
		const cached: Run[] = [];
		for (const sequence of this.#cached.keys()) {
			if (sequence >= start && sequence < end) insertRun(cached, sequence, sequence + 1n);
		}
		this.#skipGaps(cached, start, end);
		this.#chargeDebt();
		return cached;
	}

	// Skip what of `[start, end)` neither `kept` nor an earlier skip names, replacing any drop.
	#skipGaps(kept: readonly Run[], start: bigint, end: bigint): void {
		const skipped = this.#state.drops.overlapping(start, end).filter((run) => run.skip);
		for (const gap of runGaps(kept, start, end)) {
			for (const open of runGaps(skipped, gap.start, gap.end)) {
				this.#drop(open.start, open.end, skipHole());
			}
		}
	}

	// Record a sequence this track's own publisher produced, for the next append to skip around.
	#produce(sequence: bigint): void {
		if (this.#wire || (this.#appended !== undefined && sequence < this.#appended)) return;
		insertRun(this.#produced, sequence, sequence + 1n);
	}

	// Settle an append at `sequence`: skip every sequence below it this track's own publisher
	// never produced, from just past its previous append (before the first, from the lowest it
	// produced), since appending says it will not go back. The skip replaces a drop given there,
	// as `skipGroups` does.
	#settleAppend(sequence: bigint): void {
		const from = this.#appended ?? this.#produced[0]?.start;
		if (from !== undefined) {
			const kept = [...this.#produced];
			for (const cached of this.#cached.keys()) {
				if (cached >= from && cached < sequence) insertRun(kept, cached, cached + 1n);
			}
			this.#skipGaps(kept, from, sequence);
		}
		this.#produced = [];
		this.#appended = sequence + 1n;
	}

	// Refuse a sequence the publisher declared skipped: a skip is final.
	#refuseSkipped(sequence: bigint): void {
		if (this.#state.drops.at(sequence)?.skip) throw new Skipped(sequence);
	}

	/**
	 * Create a new group with the given sequence number.
	 *
	 * It says nothing about the sequences around it: groups may be created out of order, and an
	 * aborted or evicted sequence created again. A gap it leaves is skipped only once the
	 * producer says it will not fill it, with {@link skipGroups} or {@link appendGroup}. A
	 * sequence already skipped throws {@link Skipped}, and the reserved sequence 2^64-1 a
	 * `RangeError`.
	 *
	 * Throws on a sequence that is still cached: a live duplicate would fan out to every
	 * subscriber twice. An aborted incarnation is evicted so a fresh group can serve the
	 * sequence again. Best effort (mirrors Rust): nothing remembers a sequence whose cache
	 * entry is already gone, so a long-evicted sequence is accepted as new. Its frames are
	 * checked like {@link appendGroup}'s.
	 */
	createGroup(sequence: bigint): GroupProducer {
		return this.#createGroup(sequence, 0);
	}

	#createGroup(sequence: bigint, start: number): GroupProducer {
		this.#writable(sequence);
		// A skip says the sequence will never be produced, so it is final. A drop is not: a
		// publisher may re-create a group it cut or aborted.
		this.#refuseSkipped(sequence);
		const existing = this.#cached.get(sequence);
		if (existing) {
			if (!(existing.group.closed.peek() instanceof Error)) {
				throw new Error(`duplicate group: sequence=${sequence}`);
			}
			// Subscribers may have seen the aborted incarnation's start, so it stays a watermark.
			this.#evict(existing);
			this.#uncache(existing, true);
		}

		const group = new GroupProducer(sequence);
		// Mirrors copy the absolute indices, so set them before publishing to any sink.
		hooks.startGroup(group, start);
		this.#bind(group);
		this.#cap(group);
		this.#publish(group);
		return group;
	}

	// Whether this track's timestamps may rewind (see {@link Info.unchecked} and {@link Info.ietf}).
	#unchecked(): boolean {
		const info = this.#state.info.peek();
		return info?.unchecked === true || info?.ietf !== undefined;
	}

	/**
	 * Declare the track's exclusive end, possibly ahead of the live edge, mirroring the Rust
	 * `finish_at`.
	 *
	 * `end` is the first position that will never be produced: a track whose last group is
	 * 89 finishes at `{ group: 90, frame: 0 }`, and `{ group: 89, frame: 5 }` also caps group
	 * 89 at frames 0 to 4. Groups and frames below it are still accepted; anything
	 * at or past it is refused. Unlike {@link close} it is not terminal: call `close()` once
	 * the remaining groups are written. Readers end once the live edge reaches the end's
	 * group, independently of closure. Throws if the track is closed, already has an end, or
	 * `end` would orphan a group or frame already produced.
	 *
	 * With an `error`, the track ended that way rather than cleanly, yet keeps its history, unlike
	 * {@link abort}: readers still get every group below the end, and a reader reaching the end
	 * gets the error instead of the end of stream.
	 */
	finishAt(end: Position, error?: Error): void {
		if (this.#state.closed.peek() !== undefined) throw new Error("track is closed");
		const declared = this.#state.end.peek()?.at;
		if (declared !== undefined) throw new Error(`track already ends at ${declared.group}/${declared.frame}`);
		this.#declareEnd(end, error, false);
	}

	// The wire's end: repeating it is a no-op, and a mid-group end may refine the group
	// boundary it falls just before. Either must carry the same error code. Mirrors the Rust
	// `finish_at` on a `received()` copy. Only a wire subscription owes a tail below the end: a
	// FETCH's end leaves nothing in flight, as the Rust `set_end` a fill declares.
	#endPending(end: Position, error?: Error): void {
		const declared = this.#state.end.peek();
		const code = (err?: Error) => (err === undefined ? undefined : toStreamCode(err));
		const tail = this.#wire !== undefined;
		if (declared === undefined) {
			if (this.#state.closed.peek() !== undefined) throw new Error("track is closed");
			this.#declareEnd(end, error, tail);
		} else if (code(declared.error) !== code(error)) {
			throw new ProtocolViolation(`conflicting track end error: ${error}`);
		} else if (declared.at.group !== end.group || declared.at.frame !== end.frame) {
			if (declared.at.frame !== 0 || end.frame === 0 || finalSequence(end) !== declared.at.group) {
				throw new ProtocolViolation(`conflicting track end: ${end.group}/${end.frame}`);
			}
			this.#declareEnd(end, error, declared.tail || tail);
		}
		this.#owe();
	}

	// Record that the live feed accounted for `[start, end)`, then fold the gaps that aged out.
	#arrive(start: bigint, end = start + 1n): void {
		const at = now();
		this.#arrivals.account(start, end, at);
		// Only a gap folds, so an in-order feed never reads the subscription.
		if (this.#arrivals.hasGaps()) this.#arrivals.expire(at, this.#grace());
		this.#billArrivals();
		this.#owe();
	}

	// The grace the session's tail uses, from the max delay the track's subscribers ask for,
	// bounded by how long the publisher keeps a superseded group fresh. The demand asked upstream is not.
	#grace(): Duration {
		const update = this.#state.update.peek();
		const maxDelay = Duration.millis(update ? maxDelayOf(update) : 0);
		const maxAge = this.#state.info.peek()?.maxAge;
		return grace(maxAge == null ? maxDelay : maxDelay.min(maxAge));
	}

	// Tell every reader whether the pending tail still owes a group below the end.
	#owe(): void {
		const end = this.#state.end.peek();
		if (!end?.tail) return;
		const owed = this.#owes(finalSequence(end.at));
		for (const state of [this.#state, ...this.#sinks]) state.owed.set(owed);
		this.#endLargest();
	}

	// Settle the outcome cleanly once the declared end is reached and settled. Also run before a
	// write that could reopen it (a late group below the end, a fresh grace for the tail): that
	// write is still served, but the clean outcome stays final. Mirrors the Rust `latch_settled`.
	#settle(): void {
		if (this.#state.outcome.peek() === undefined && this.#settled())
			this.#state.outcome.set(this.#state.end.peek()?.error ?? null);
	}

	// Whether the pending tail misses a sequence from where the feed starts up to `final`: one
	// no live group delivered, in a gap younger than the grace. Judged by the
	// arrivals, not the cache, so an evicted group counts as delivered.
	//
	// The feed starts at the declared start while the demand keeps the floor its SUBSCRIBE asked
	// for, since an update gets no fresh START: a moved floor stands on its own, and moving
	// back restores the START. Without either, the first arrival stands in for it.
	// Until the START arrives, the feed start is pending; the requested floor still bounds it.
	#feedStart(): bigint | "pending" | undefined {
		const ranges = this.#state.update.peek()?.ranges;
		const requested = ranges && wireSpan(ranges).start;
		const served = this.#wire?.served;
		if (!served || served.requested !== requested) return requested;
		if (served.start === null) return requested;
		if (served.start !== undefined) return served.start;
		// A closed track never gets its START.
		return this.#state.closed.peek() === undefined ? "pending" : requested;
	}

	// The lowest group a reader whose ranges name `named` (undefined for one reaching back to
	// the oldest group) asks a fill for, or undefined for no bound. Below a start resolved from
	// the budget, each reader fills what its own budget reaches; otherwise only what its ranges
	// name; and never on a wire whose fills cannot judge a group by that budget. Mirrors the
	// Rust `Misses::judge`. Fetches are never bounded by it.
	#fillFloor(named: bigint | undefined): bigint | undefined {
		const wire = this.#wire;
		if (!wire) return undefined;
		const feed = this.#feedStart();
		const start = feed === "pending" ? wire.served?.requested : feed;
		if (start === undefined || start === 0n) return undefined;
		if (wire.starts.fillsFromStart) return start;
		// The START applies only while the demand keeps the floor it answered, as `#feedStart`.
		const ranges = this.#state.update.peek()?.ranges;
		const served = wire.served;
		const resolved =
			wire.starts.resolvesStart === true &&
			typeof served?.start === "bigint" &&
			served.requested === (ranges && wireSpan(ranges).start);
		if (resolved) return undefined;
		return named === undefined || named >= start ? start : undefined;
	}

	#owes(final: bigint): boolean {
		if (!this.#wire || !this.#state.end.peek()?.tail) return false;
		const feed = this.#feedStart();
		const floor = (feed === "pending" ? this.#wire?.served?.requested : feed) ?? this.#arrivals.first();
		if (floor === undefined) return final > 0n;
		return !this.#arrivals.covers(min(floor, final), final);
	}

	#declareEnd(end: Position, error: Error | undefined, tail: boolean): void {
		const { group, frame } = end;
		if (typeof group !== "bigint" || group < 0n || !Number.isSafeInteger(frame) || frame < 0) {
			throw new RangeError(`invalid track end: ${group}/${frame}`);
		}
		const final = finalSequence(end);
		if (final < this.#received) {
			throw new ProtocolViolation(`track end ${group}/${frame} is below the next sequence ${this.#received}`);
		}
		const cut = frame === 0 ? undefined : this.#cached.get(group)?.group;
		if (cut) hooks.limitGroup(cut, frame);
		const declared: End = { at: { group, frame }, error, tail };
		for (const state of [this.#state, ...this.#sinks]) state.end.set(declared);
		this.#endLargest();
	}

	// Group `sequence` holds every frame it will have: it closed cleanly, its stream finished.
	// A group name fixes its content, so no legal write raises the largest past it.
	#finishLargest(sequence: bigint): void {
		if (this.#state.largest.peek()?.group === sequence) this.#largestOpen = false;
		this.#endLargest();
	}

	// End the largest stream while the track stays open, once it can no longer advance: the
	// declared end is reached, and the largest group's content is final. Every sequence below
	// the end raised it already, so neither a late group below it nor a recreated copy of it
	// moves it after this.
	#endLargest(): void {
		this.#settle();
		if (this.#state.largestEnded.peek() || this.#largestOpen) return;
		const end = this.#state.end.peek();
		if (end === undefined || !reaches(this.#received, this.#state.drops, end) || this.#owes(finalSequence(end.at)))
			return;
		for (const state of [this.#state, ...this.#sinks]) state.largestEnded.set(true);
	}

	/**
	 * Declare the track's end after the highest sequence written, returning it as {@link Finished}.
	 * A no-op once an end is declared or the track closed, so a cleanup path can finish whatever
	 * the track's state. A track that closed first keeps its outcome and declares no end, so its
	 * readers' `finished()` rejects.
	 *
	 * Not terminal: groups below the end may still be written. Readers end once the live edge
	 * reaches it.
	 */
	finish(): Finished {
		if (this.#state.end.peek() === undefined && this.#state.closed.peek() === undefined)
			this.finishAt({ group: this.#received, frame: 0 });
		return this;
	}

	/**
	 * Abort the track and every subscriber, mirroring the error to their open groups. Terminal
	 * and idempotent.
	 *
	 * Subscribers still draining get the finished groups first, then the error. An abort after
	 * the declared end settled (reached, with every group below it finished) only releases the
	 * track: readers end cleanly. The groups left behind still age out on the idle cache
	 * window, so a stale subscriber can't pin them.
	 */
	abort(error: Error): void {
		this.#close(error);
	}

	/**
	 * Release this producer, like dropping the last Rust producer. Idempotent.
	 *
	 * With an end declared ({@link finish} or {@link finishAt}), the track ends there and its
	 * subscribers drain to it. Without one, the track aborts: nothing finished it.
	 */
	close(): void {
		if (this.#state.end.peek() !== undefined) this.#close(undefined);
		else this.#close(new StreamError(StreamCode.Cancel, { message: `track ${this.name} closed without finish` }));
	}

	// `abort` is undefined for a clean close, which keeps whatever end was declared. A received
	// copy released because nobody wants it closes cleanly without one: a dead route, not finished
	// content.
	#close(abort: Error | undefined) {
		if (this.#state.closed.peek() !== undefined) return;
		// A gap past the grace was lost before the abort rather than cut off by it.
		if (abort && this.#state.end.peek()?.tail) {
			this.#arrivals.expire(now(), this.#grace());
			this.#billArrivals();
		}
		if (abort && (this.#state.outcome.peek() === null || this.#settled())) abort = undefined;
		// Nobody will finish these, so a subscriber that has not taken one yet never sees it.
		// Not evicted: a reader already holding one keeps its frames and sees the abort.
		const open = [...this.#cache].filter((entry) => entry.group.closed.peek() === undefined);
		const kept = this.#keptDrops();
		closeTrackState(this.#state, abort);
		this.#billArrivals();
		if (this.#keptDrops() > kept) this.#rearmDrops(kept);
		if (this.#state.outcome.peek() === undefined)
			this.#state.outcome.set(abort ?? this.#state.end.peek()?.error ?? null);
		// Closing a copy never supplies the missing tail of a group. Already finished groups
		// ignore the abort and stay readable; unfinished ones retain their truncation cause.
		const groupError =
			abort ??
			new StreamError(StreamCode.Cancel, { message: `track ${this.name} closed with unfinished groups` });
		for (const { group } of this.#cache) group.abort(groupError);
		for (const entry of open) {
			this.#unlink(entry);
			this.#uncache(entry);
		}
		for (const sink of this.#sinks) {
			for (const group of sink.groups.peek()) hooks.releaseGroupReader(group, abort);
			closeTrackState(sink, abort);
		}
		this.#sinks.clear();
		// The former live edge becomes evictable too, its idle window restarted, so the close
		// never reclaims a group its subscribers have yet to drain.
		const edge = this.#edgeEntry();
		if (edge) {
			edge.charge.recordWrite();
			this.#account.demote(edge);
		}
		this.#evictExpired(true);
	}

	// Whether the declared end was reached, the pending tail owes nothing below it, and every
	// cached group below it finished, so the track already holds everything it promised. An
	// abort over a group the tail still owed cut it off, so readers get the error rather than
	// a clean end without it. A failed fill answer holds nothing open, since the feed never
	// promised it. Mirrors the Rust `is_settled`.
	#settled(): boolean {
		const end = this.#state.end.peek();
		if (end === undefined) return false;
		const final = finalSequence(end.at);
		if (!reaches(this.#received, this.#state.drops, end) || this.#owes(final)) return false;
		for (const { group, fetched } of this.#cache) {
			if (group.sequence >= final) continue;
			const closed = group.closed.peek();
			if (closed !== null && !(fetched && closed instanceof Error)) return false;
		}
		return true;
	}

	/** Append a frame as its own single-frame group. */
	writeFrame(frame: Frame) {
		const group = this.appendGroup();
		try {
			group.writeFrame(frame);
		} catch (err) {
			// Abort the group so no subscriber waits on a group that will never get its frame.
			group.abort(err instanceof Error ? err : new Error(String(err)));
			throw err;
		}
		group.finish();
	}
}

// One read pass's view of the budget and the live edge; see Subscriber.#drift.
type Drift = {
	budget: number;
	// The budget is the track's max age, which no widening passes: its verdicts are final.
	bounded: boolean;
	presentation?: { sequence: bigint; timestamp: Timestamp };
	end?: bigint;
};

// The successor a handed-out group was last judged by, so the group stays judgeable once the
// cache evicts both. Each reader keeps its own, so the memory is bounded by the groups held.
type Superseded = { successor?: Successor };

// A successor, when it arrived, and its copy's fate (see `hooks.groupFate`). Not the consumer:
// a handed-out successor's own record would retain the next one, chaining every later group to
// the oldest held.
type Successor = { sequence: bigint; arrived: Instant; fate: () => Error | undefined };

// A run a reader already reported, and how: dropped for good (or stale until its group was handed
// out, so no update reopens it), dropped as stale, which an update reopens, or skipped, which
// replaces a drop. `received` when an upstream's stale run decided it, which is the upstream's to
// judge again then. Mirrors the Rust `Report`.
/**
 * How a reader reported a run, and what its wire was told. Mirrors the Rust `record::Report`:
 * `settled` is a group served, or a drop on a version whose drops are all final.
 *
 * @internal
 */
export interface Reported extends Run {
	kind: "settled" | "dropped" | "stale" | "skip";
	/** A stale run an upstream decided, which is the upstream's to judge again. */
	received?: boolean;
	/** A stale run whose Group Stream expired, or that went out OLD for good: its publisher drops it again under each update. */
	expired?: boolean;
	/** A stale run whose group was handed out after the drop, which the group won: no update reopens it. */
	taken?: boolean;
}

const sameReport = (a: Reported, b: Reported) =>
	a.kind === b.kind && !a.received === !b.received && !a.expired === !b.expired && !a.taken === !b.taken;

// Whether `reported` settles its run for good, so a fold may forget it.
const finalReport = (reported: Reported) => reported.kind !== "stale" || reported.taken === true;

// Whether an update reopens `reported`: a stale drop nothing settled since.
const reopens = (reported: Reported) => reported.kind === "stale" && !reported.expired && !reported.taken;

// Whether `reported` stops `hole` going out over it: a skip goes out over a drop, and a final
// drop over a stale one, since each replaces what it goes out over. An OLD drop never goes out
// over a stale one, which already said OLD. Mirrors the Rust `Report::blocks`.
function blocksHole(reported: Reported, hole: Hole): boolean {
	switch (reported.kind) {
		case "settled":
		case "skip":
			return true;
		case "dropped":
			return !hole.skip;
		case "stale":
			return reported.taken ? !hole.skip : !hole.skip && toStreamCode(hole.error) === StreamCode.Old;
	}
}

/**
 * What a lite publisher told its wire about a reader's sequences, kept in the reader's own
 * record so the cursor's drops and the publisher's groups consult one map. Mirrors the
 * publisher half of the Rust `record::Record`.
 *
 * @internal
 */
export interface WireRecord {
	/** Whether the wire was told of `sequence`: served, or given up. */
	told(sequence: bigint): boolean;
	/** Whether the pool folded `sequence` into the record's floor, final for this reader. */
	folded(sequence: bigint): boolean;
	/** Whether `sequence` went out as a stale drop, which a group still replaces. */
	toldStale(sequence: bigint): boolean;
	/** The parts of `[start, end)` the wire was told nothing of. */
	untold(start: bigint, end: bigint): Run[];
	/** Record that the wire was told `report` of `[start, end)`. */
	tell(start: bigint, end: bigint, report: Omit<Reported, "start" | "end">): void;
	/** Forget what the wire was told of `[start, end)`, which is owed again. */
	untell(start: bigint, end: bigint): void;
	/** Forget the drops of `[start, end)` the wire was told, which a verdict or group replaces. */
	replaceDrops(start: bigint, end: bigint): void;
	/** The runs each update drops again: stale ones whose stream expired, or that went out OLD for good. */
	expired(): Run[];
}

/**
 * A track its producer finished: its end is declared unless it closed first, and it can still
 * take an older group below that end, be read, or be aborted or closed, but not append. {@link Producer.finish} returns it,
 * mirroring the Rust `track::Finished`; the producer itself still refuses a group at or past the
 * end.
 *
 * @public
 */
export interface Finished {
	/** The track name. */
	readonly name: string;
	/** Settles once the track's outcome is final; see {@link Producer.closed}. */
	readonly closed: GetPromise<Error | null>;
	/** The track's info, once accepted. */
	info(): Promise<Info>;
	/** Create an older group below the end, which arrived late; see {@link Producer.createGroup}. */
	createGroup(sequence: bigint): GroupProducer;
	/** Serve the groups readers miss below the end; see {@link Producer.requests}. */
	requests(): Requests;
	/** Subscribe to this in-process track; see {@link Producer.subscribe}. */
	subscribe(options?: Subscription): Subscriber;
	/** Abort the track; see {@link Producer.abort}. */
	abort(error: Error): void;
	/** Release the track, like dropping the last Rust handle; see {@link Producer.close}. */
	close(): void;
}

/**
 * The read side of a live track subscription, mirroring the Rust `Subscriber`.
 *
 * Obtained from `Broadcast.Consumer.subscribe` / `Track.Consumer.subscribe`, or from
 * {@link Producer.subscribe} for an in-process track. Reads the groups a
 * {@link Producer} on the same underlying state writes.
 */
export class Subscriber {
	/** The track name. */
	readonly name: string;

	#state: TrackState;
	// Reads its routes' copies instead of `#state`'s buffer: the counterpart of the Rust
	// `Inner::Resume`. `#state` then only carries the reader's own preferences, info, and close.
	#resume?: ResumeSource;
	// Every group sequence this reader handed out, so a re-created group or one a second route
	// delivers is handed out once.
	#delivered = new Delivered();
	// Cached groups the arrival cursor skipped as stale, each with the budget it was judged
	// under, so a wider budget offers them again. Only a widening reverses a stale verdict, and
	// an entry is kept only while its group is cached, since nothing could offer it again.
	#skipped = new Map<bigint, number>();
	// At most the narrowest budget in #skipped, so a read whose budget did not widen skips the scan.
	#narrowest = Number.POSITIVE_INFINITY;
	// This reader closed itself, so lowering its floor offers nothing: a producer's clean close
	// alone leaves the cache readable, but the reader leaving must stay terminal. Leaving also
	// settles the shared `#state.closed`, so a guard for either close reads that alone.
	#left = false;
	// What {@link largest} last reported, and whether it ended: moving the bounds afterwards
	// does not bring it back.
	#largestSeen?: Position;
	#largestDone = false;
	#nextSequence = 0n;
	// A wire publisher's reader parks past the end of a bounded window rather than end there:
	// its peer would read the end of the stream as the track's. Set through `hooks.parked`.
	#parks = false;
	// The local bounds {@link withRange} set.
	#local: { start: bigint; end?: bigint } = { start: 0n };
	// This reader's own ranges resolved to the groups it reads, whatever another subscriber
	// widens the aggregate to. Assigned in the constructor.
	#window!: Window;
	// Bumped whenever the window moves, so the drop scan knows to look again.
	#revision = 0;
	// The hull of what this reader reads now: its local bounds narrowed to the window's floor
	// and end. A group inside it is read only if the window contains it too.
	#cursor = new Signal<{ start: bigint; end?: bigint }>({ start: 0n });
	// A FETCH's own budget in place of the subscription's; unlike SUBSCRIBE's zero
	// default, an omitted one is unbounded, leaving the publisher's maxAge as the only bound.
	#fetchBudget?: number;
	// Which cursor owns this subscription. Both cursors draw from one buffer, so the
	// first group read commits the track to arrival order and {@link ordered} commits
	// it to sequence order; whichever wins is the only one allowed from here on.
	#mode?: "arrival" | "ordered";
	// What this cursor reports as drops: sequences given up and not reported yet,
	// the track's drops last folded in, and every sequence already reported. Drops are
	// persistent track state, so they are tracked here rather than removed for everyone.
	#pending: DropRun[] = [];
	#dropsSeen?: {
		// The track's drops revision folded in: later changes fold in alone.
		drops: number;
		refused: readonly DropRun[];
		start: bigint;
		end?: bigint;
		revision: number;
		budget: number;
		coverage?: Coverage;
		judged?: readonly Run[];
	};
	#reported = new Runs<Reported>(sameReport);
	// Everything below was settled for good and folded away under pool pressure: handed out, or
	// reported final, so none is handed out or reported again. Mirrors the Rust `Record::floor`.
	#folded = 0n;
	// What the record bills to the track's account, opened on its first run.
	#bill?: cache.Reader;
	// The pool needed a run this reader was promised, so it ended with TooFarBehind.
	#lagged = false;
	// The group the frame-level helpers are currently draining, acquired through the
	// sequence cursor so frame reads and {@link Ordered.nextGroup} share one floor.
	#frameGroup?: GroupConsumer;
	// The earliest wall deadline a held read armed, and its timer, which bumps `revision`
	// when it comes due. See #wakeAt. Kept apart from the subscriber so a pending timer
	// retains only this, never the subscriber or its track state, once no read waits.
	#wall: { revision: Signal<number>; at: number; timer?: AbortController } = {
		revision: new Signal(0),
		at: Number.POSITIVE_INFINITY,
	};
	// When each group the timeline held at close arrived, failed ones left out. A group handed
	// out earlier can outlive this subscriber, as one a resumed reader still writes from a copy
	// it let go of, and its budget still runs from its successor's arrival.
	#arrivals?: { sequence: bigint; arrived: Instant }[];
	// The misses this reader asked a fill for, each judged once, and those asked again after the
	// answer was evicted before this reader took it, which is never repeated: a slow reader's
	// refills cannot evict each other without bound. Mirrors the Rust `Misses`.
	#asked: Run[] = [];
	#refilled: Run[] = [];
	// Fill-owed runs this reader gave up once its budget convicted them: never wanted again, so
	// the fills withdraw them and their handlers hear, as the Rust `Misses` gives up a stale run.
	#givenUp: Run[] = [];
	// The received stale runs that do not settle this reader, and the terms they were judged
	// under; see #unknownRuns.
	#unknown?: {
		drops: number;
		judged: readonly Run[] | undefined;
		budget: number;
		coverage: Coverage;
		runs: DropRun[];
	};
	// The last read waited on a fill, so the staleness deadline it armed wakes the next one: a
	// fill that never answers cannot pin a reader in either order.
	#fillWait = false;
	// Bumped whenever the fills this reader joined are rejudged, which may withdraw groups it reads
	// again by the time it looks (a narrowing undone in one turn moves no signal), so it asks anew.
	#fillMoves = new Signal(0);

	#drift(): Drift {
		// A fetch judges its group against the whole track, as the Rust `fetch_group` does: a
		// newer group outside its one-group window still supersedes it.
		const end = this.#fetchBudget === undefined ? this.#cursor.peek().end : undefined;
		const timeline = this.#state.timeline;
		let presentation: { sequence: bigint; timestamp: Timestamp } | undefined;
		const info = this.#state.info.peek();
		// An accepted untimed track cannot contain a presentation timestamp.
		if (info === undefined || info.timescale != null) {
			// The edge wants the newest content that exists, so it takes the newest
			// stamped group's latest frame: walk back from the cap to the first one.
			for (let i = (end === undefined ? timeline.length : timelineIndex(timeline, end)) - 1; i >= 0; i--) {
				const group = timeline[i];
				if (hooks.groupFailed(group) || !this.#judges(group.sequence)) continue;
				const timestamp = hooks.groupTimestamp(group);
				if (timestamp === undefined) continue;
				presentation = { sequence: group.sequence, timestamp: hooks.groupLatest(group) ?? timestamp };
				break;
			}
		}

		const update = this.#state.update.peek();
		const requested = this.#fetchBudget ?? (update ? maxDelayOf(update) : 0);
		const fresh = info?.maxAge?.asMillis();
		return {
			budget: fresh === undefined ? requested : Math.min(requested, fresh),
			bounded: fresh !== undefined && requested >= fresh,
			presentation,
			end,
		};
	}

	// The group after `sequence` that bounds it: the first servable one below `end` that
	// counts as content (its first frame landed, or it ended), stamped or not.
	//
	// Its first timestamp is the furthest presentation time `sequence` could still reach.
	// 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. A group whose header
	// landed without a frame is not content yet, so the next one is the bound; an unstamped
	// successor leaves the reach unbounded.
	#successor(sequence: bigint, end?: bigint): GroupConsumer | undefined {
		return this.#after(sequence, end, hooks.groupStarted);
	}

	// The group after `sequence` that starts its wall clock: like {@link #successor}, but
	// counting a group before its first frame. Its header arrived, so a relay ages the group
	// before it as the origin does, even if its first frame never comes.
	#arrival(sequence: bigint, end?: bigint): GroupConsumer | undefined {
		return this.#after(sequence, end, () => true);
	}

	// Whether group `sequence` counts toward the drift edge: inside the window, or anywhere for a
	// fetch.
	#judges(sequence: bigint): boolean {
		return this.#fetchBudget !== undefined || this.#window.contains(sequence);
	}

	// The first servable group after `sequence` below `end` that `counts`.
	#after(
		sequence: bigint,
		end: bigint | undefined,
		counts: (group: GroupConsumer) => boolean,
	): GroupConsumer | undefined {
		const timeline = this.#state.timeline;
		for (let i = timelineIndex(timeline, sequence + 1n); i < timeline.length; i++) {
			const successor = timeline[i];
			if (end !== undefined && successor.sequence >= end) break;
			if (hooks.groupFailed(successor) || !this.#judges(successor.sequence)) continue;
			if (counts(successor)) return successor;
		}
		return undefined;
	}

	// Whether the drift budget says to give up on `group`.
	//
	// 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. A group is abandoned only once everything it could still present
	// falls outside the budget.
	//
	// A group's reach is bounded by its nearest successor: it cannot present past where the
	// next group begins. Its own frames say nothing, since a frame's duration is not on the
	// wire, and the candidate needs no timestamp of its own: an empty group is bounded by
	// its stamped successor the same way. The bound is exclusive, so the comparison is `>=`:
	// 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. A zero budget
	// falls out for free.
	//
	// The wall clock judges it too: a group is also stale once its successor arrived a full
	// budget ago, so a peer whose timestamps stop advancing, an untimed track, or an empty
	// successor that never presents cannot pin a superseded group. The successor is re-read,
	// not latched, like the reach, except by a reader holding the group: it passes `held`, so
	// the wall clock outlives the successor's eviction (see {@link #supersededAt}). The newest
	// group has no successor, so neither clock ever convicts it. `wake` arms a timer for the
	// wall deadline, for a held read that waits on it.
	//
	// Judged by its successor alone, so the group need not be in this track: evicted, held
	// from another copy, or outliving this subscriber. Once closed, only the wall clock judges
	// it, from the arrivals kept at close.
	#staleAt(sequence: bigint, drift: Drift, wake = false, held?: Superseded): boolean {
		const arrived = this.#arrivals
			? this.#keptSupersededAt(sequence, drift, held)
			: this.#supersededAt(this.#arrival(sequence, drift.end), drift, held);
		if (arrived === undefined) return false;
		// Plain arithmetic: an unenforced budget is infinite, which no duration brand holds.
		const deadline = arrived.since(Instant.ZERO).asNanos() + drift.budget * 1_000_000;
		if (now().since(Instant.ZERO).asNanos() >= deadline) return true;
		if (wake) this.#wakeAt(deadline);
		// A closed subscriber keeps no timeline to bound the reach with.
		if (this.#arrivals) return false;

		const successor = this.#successor(sequence, drift.end);
		if (!successor) return false;
		const reach = hooks.groupTimestamp(successor)?.asMillis();
		return (
			drift.presentation !== undefined &&
			drift.presentation.sequence > sequence &&
			reach !== undefined &&
			drift.presentation.timestamp.asMillis() - reach >= drift.budget
		);
	}

	// When the wall clock starts: the cached `arrival`'s arrival, or that of the one `held`
	// recorded, unless a group landed between them since. The record outlives the successor's
	// eviction: otherwise an evicted successor restarts the clock at whichever newer group the
	// cache still holds, which a budget past the cache window never reaches. A recorded
	// successor that failed hands the clock on, as a re-read does, whether or not the cache
	// has evicted it yet.
	#supersededAt(arrival: GroupConsumer | undefined, drift: Drift, held?: Superseded): Instant | undefined {
		const arrived = arrival && hooks.groupArrived(arrival);
		if (!held) return arrived;
		const recorded = held.successor;
		if (
			recorded &&
			!failed(recorded) &&
			(drift.end === undefined || recorded.sequence < drift.end) &&
			(!arrival || arrival.sequence > recorded.sequence)
		) {
			return recorded.arrived;
		}
		if (!arrival || arrived === undefined) {
			held.successor = undefined;
		} else if (recorded?.sequence !== arrival.sequence || !recorded.arrived.eq(arrived) || failed(recorded)) {
			// Re-recorded only on a change, so judging the same successor again allocates nothing.
			held.successor = { sequence: arrival.sequence, arrived, fate: hooks.groupFate(arrival) };
		}
		return arrived;
	}

	// {@link #supersededAt} for a closed subscriber, with the arrivals kept at close standing in
	// for the timeline it cleared.
	#keptSupersededAt(sequence: bigint, drift: Drift, held?: Superseded): Instant | undefined {
		const covered = (candidate: bigint) => drift.end === undefined || candidate < drift.end;
		const kept = this.#arrivals?.find((entry) => entry.sequence > sequence && covered(entry.sequence));
		const recorded = held?.successor;
		if (
			recorded &&
			!failed(recorded) &&
			covered(recorded.sequence) &&
			(!kept || kept.sequence > recorded.sequence)
		) {
			return recorded.arrived;
		}
		return kept?.arrived;
	}

	// Bump {@link #wall} at `at`, unless a wakeup is already armed sooner. Each held read
	// re-arms on the bump, so one timer covers every deadline this subscriber waits on.
	#wakeAt(at: number): void {
		const wall = this.#wall;
		if (!Number.isFinite(at) || at >= wall.at) return;
		wall.timer?.abort();
		const timer = new AbortController();
		// A staleness wakeup is never a reason to hold a Node/Bun process open.
		void sleepUntil(Instant.ZERO.saturatingAdd(Duration.nanos(at)), { signal: timer.signal, ref: false }).then(
			() => {
				if (timer.signal.aborted) return;
				wall.timer = undefined;
				wall.at = Number.POSITIVE_INFINITY;
				wall.revision.update((revision) => revision + 1);
			},
			() => {},
		);
		wall.timer = timer;
		wall.at = at;
	}

	#guard(group: GroupConsumer): GroupConsumer {
		// Handed out, it is read, whatever this reader judged it while buffered. Taking it is the
		// cache access buffering it is not, as in Rust: it restarts the group's idle window.
		hooks.groupReads(group, true);
		hooks.refreshGroup(group);
		const held: Superseded = {};
		hooks.onShedRead(group, () => {
			if (this.#state.latest === group.sequence && this.#state.producer) hooks.joinProducer(this.#state.producer);
		});
		hooks.expireGroup(group, {
			expired: () => this.#staleAt(group.sequence, this.#drift(), true, held),
			changed: this.#expiryChanged(),
		});
		return group;
	}

	// What can change a drift verdict.
	#expiryChanged(): Getter<unknown>[] {
		return [
			this.#state.groups,
			this.#state.timelineChanged,
			this.#wall.revision,
			this.#state.update,
			this.#state.info,
			this.#cursor,
			this.#state.closed,
		];
	}

	// Record `sequence` as handed out. An ordered read never goes back, and any reader stops
	// expecting what is past its budget, so only the gaps within it cost memory.
	#handOut(sequence: bigint, ordered: boolean): void {
		if (ordered) this.#delivered.forgetBelow(sequence);
		this.#delivered.add(sequence);
		this.#skipped.delete(sequence);
		// The group won over any stale report on it, so no update reopens it. A resumed reader
		// retires its own.
		const stale = this.#reported.at(sequence);
		if (stale?.kind === "stale")
			this.#reported.set(sequence, sequence + 1n, { ...stale, start: sequence, end: sequence + 1n, taken: true });
		// Another copy of it buffered here is never handed out.
		this.#judge(sequence, sequence + 1n);
		if (this.#resume) this.#resume.forgetPastBudget(this.#delivered);
		else this.#forgetPastBudget(this.#delivered);
		// What it hands out, and forgets below it, it no longer wants from a fill: tell the fills
		// it joined, as a move of its bounds does, so their handlers stop serving it.
		this.#fillsMoved();
	}

	// What this reader wants from the fills it joined may have narrowed: rejudge them, so their
	// handlers hear it.
	#fillsMoved(): void {
		const producer = this.#state.producer;
		if (!this.#state.asker || !producer) return;
		// At once, before the fills withdraw what it no longer reads: a widening back asks those
		// groups afresh, never through the one refill an evicted answer is allowed.
		this.#clipAsked();
		fillsOf(producer).moved(this.#state.asker);
		this.#fillMoves.update((moves) => moves + 1);
	}

	// Keep what this reader asked within what it still reads, as the Rust `Misses` clips what it
	// judged to the window: a group handed out, forgotten past the budget, passed, or outside the
	// window is never asked again, so these stay within the live span. A window moving back over a
	// clipped group asks it once more, as in Rust.
	#clipAsked(): void {
		if (this.#givenUp.length > 0) this.#givenUp = this.#givenUp.flatMap((run) => this.#reads(run.start, run.end));
		if (this.#asked.length > 0) this.#asked = this.#asked.flatMap((run) => this.#wants(run.start, run.end));
		if (this.#refilled.length > 0) {
			this.#refilled = this.#refilled.flatMap((run) => this.#wants(run.start, run.end));
		}
	}

	// Raise the sequence cursor to `floor`: what lies below it is passed for good.
	#raiseNext(floor: bigint | undefined): void {
		if (floor === undefined || floor <= this.#nextSequence) return;
		const passed = this.#nextSequence;
		this.#nextSequence = floor;
		this.#judge(passed, floor);
		this.#fillsMoved();
	}

	// Forget the gaps in `delivered` below each group past this subscriber's budget now, whether
	// this track holds it or not: one landing there would be written off. A group skipped as
	// stale is not forgotten, since a wider budget offers it again.
	#forgetPastBudget(delivered: Delivered): void {
		if (delivered.aboveGap() === undefined) return;
		// One edge for every group judged, so forgetting N gaps never resolves it N times.
		const drift = this.#drift();
		// Found once a gap is stale, so a hand-out that forgets nothing never scans #skipped. A
		// skip whose group left the cache can never be offered again, so it is forgotten here
		// rather than holding the gaps above it forever.
		let skipped: bigint | undefined;
		let scanned = false;
		let forgot: bigint | undefined;
		for (let oldest = delivered.aboveGap(); oldest !== undefined; oldest = delivered.aboveGap()) {
			if (!this.#staleAt(oldest, drift)) break;
			if (!scanned) {
				scanned = true;
				const producer = this.#state.producer;
				for (const sequence of this.#skipped.keys()) {
					if (producer && !cachedHolds(producer, sequence)) this.#skipped.delete(sequence);
					else skipped = min(skipped ?? sequence, sequence);
				}
			}
			if (skipped !== undefined && skipped < oldest) {
				delivered.forgetBelow(skipped);
				forgot = skipped;
				break;
			}
			delivered.forgetBelow(oldest);
			forgot = oldest;
		}
		// A buffered group in a gap forgotten here is never handed out.
		if (forgot !== undefined && delivered === this.#delivered) this.#judge(0n, forgot);
	}

	// Record `sequence` as skipped stale under `budget`, so a wider one offers it again.
	#skip(sequence: bigint, budget: number): void {
		this.#skipped.set(sequence, budget);
		this.#narrowest = Math.min(this.#narrowest, budget);
		// Each kept entry holds a distinct cached group, so pruning once they outnumber the
		// cache twice over costs O(1) per skip amortized.
		const producer = this.#state.producer;
		if (!producer || this.#skipped.size <= 2 * this.#state.timeline.length) return;
		for (const skipped of this.#skipped.keys()) {
			if (!cachedHolds(producer, skipped)) this.#skipped.delete(skipped);
		}
	}

	// Offer again the cached groups skipped as stale under a budget narrower than `budget`, now
	// that it widened. Each stays recorded, judged under `budget`, until it is handed out or
	// skipped again, so a newer one handed out first never folds it into #delivered. Costs the
	// skipped groups, never the cache.
	#reoffer(budget: number): void {
		if (this.#skipped.size === 0 || budget <= this.#narrowest) return;
		const sequences: bigint[] = [];
		let narrowest = Number.POSITIVE_INFINITY;
		for (const [sequence, judged] of this.#skipped) {
			if (judged >= budget) {
				narrowest = Math.min(narrowest, judged);
				continue;
			}
			this.#skipped.set(sequence, budget);
			narrowest = budget;
			sequences.push(sequence);
			// Judged again, so a stale drop not reported yet is withdrawn.
			if (dropAt(this.#pending, sequence) instanceof Error) {
				this.#pending = this.#pending.flatMap((run) =>
					run.start <= sequence && sequence < run.end && run.stale !== undefined
						? removeRun([run], sequence, sequence + 1n)
						: [run],
				);
			}
		}
		this.#narrowest = narrowest;
		const producer = this.#state.producer;
		if (producer && !this.#left) reofferSequences(producer, this.#state, sequences);
	}

	// The publisher's view of this reader's record.
	#wireRecord(): WireRecord {
		const reported = this.#reported;
		return {
			told: (sequence) => sequence < this.#folded || reported.at(sequence) !== undefined,
			folded: (sequence) => sequence < this.#folded,
			toldStale: (sequence) => reported.at(sequence)?.kind === "stale",
			untold: (start, end) => {
				const from = max(start, this.#folded);
				return from < end ? runGaps(reported.overlapping(from, end), from, end) : [];
			},
			tell: (start, end, report) => {
				if (start >= end) return;
				// A stale report keeps the parts an upstream decided, which stay its to judge again.
				const received =
					report.kind === "stale" && !report.received
						? reported.overlapping(start, end).filter((run) => run.kind === "stale" && run.received)
						: [];
				reported.set(start, end, { ...report, start, end } as Reported);
				for (const run of received) {
					const from = max(run.start, start);
					const to = min(run.end, end);
					reported.set(from, to, { ...report, start: from, end: to, received: true } as Reported);
				}
			},
			untell: (start, end) => {
				reported.set(start, end);
			},
			replaceDrops: (start, end) => {
				const drops = reported
					.overlapping(start, end)
					.filter((run) => run.kind === "dropped" || run.kind === "stale");
				for (const run of drops) reported.set(max(run.start, start), min(run.end, end));
			},
			expired: () =>
				reported.all
					.filter((run) => run.kind === "stale" && run.expired)
					.map(({ start, end }) => ({ start, end })),
		};
	}

	// The bill on the account of the track this reads: `current` while it bills that account, or
	// else a new one. A resumed reader's follows its serving copy across route changes, so the pool
	// that holds what it reads is the one its debt comes from.
	#billOn(current: cache.Reader | undefined): cache.Reader | undefined {
		if (this.#resume) return this.#resume.bill(current);
		const producer = this.#state.producer;
		if (!producer) return current;
		const account = accountOf(producer);
		return current?.bills(account) ? current : account.reader();
	}

	// How many runs the record holds above its floor, each billed `cache.READER_RUN_OVERHEAD`:
	// the handed-out prefix is one run that bills nothing.
	#recordRuns(): number {
		return (
			Math.max(0, this.#delivered.runs - 1) +
			this.#skipped.size +
			this.#pending.length +
			this.#reported.size +
			this.#givenUp.length +
			this.#state.refused.peek().length
		);
	}

	// Bill this reader's record, and pay what it can of the debt its track left its readers by
	// folding what it settled for good into its floor. Returns TooFarBehind once that needed a
	// run it was promised and its record is one to end; `handed` stays above the fold (see
	// #fold). Mirrors the Rust `Record::settle`.
	#settleRecord(handed?: bigint): TooFarBehind | undefined {
		if (this.#lagged) return new TooFarBehind();
		const bytes = this.#recordRuns() * cache.READER_RUN_OVERHEAD;
		if (!this.#bill && bytes === 0) return undefined;
		const bill = this.#billOn(this.#bill);
		if (bill !== this.#bill) {
			// The serving route changed: the replaced route's pool stops carrying this record.
			this.#bill?.close();
			this.#bill = bill;
		}
		if (!bill) return undefined;
		bill.set(bytes);
		if (bill.debt === 0) return undefined;
		this.#fold(handed);
		bill.set(this.#recordRuns() * cache.READER_RUN_OVERHEAD);
		if (bill.debt > 0 && bill.ends) {
			this.#lagged = true;
			bill.close();
			this.#bill = undefined;
			this.#pending = [];
			this.#reported = new Runs<Reported>(sameReport);
			this.#skipped.clear();
			this.#givenUp = [];
			return new TooFarBehind();
		}
		return undefined;
	}

	// Raise the floor over the final prefix: each run from it up to the lowest one that is not
	// final. Final are the groups handed out, the reports nothing reopens, and this reader's own
	// stale skips, which a wider budget would offer again only while the pool still held their
	// groups. Not final are the reports an update reopens, the runs given up and not reported
	// yet, and the sequences nothing settled. The runs copied from the track's drop set go
	// first: the next scan folds in again whatever the track still holds. Mirrors the Rust
	// `Record::fold`. It stops below `handed`, a group this read hands out that its publisher has
	// yet to tell its wire of.
	#fold(handed?: bigint): void {
		if (this.#pending.some((run) => run.received)) {
			this.#pending = this.#pending.filter((run) => !run.received);
			this.#dropsSeen = undefined;
		}
		let floor = this.#folded;
		for (;;) {
			const reported = this.#reported.at(floor);
			let next: bigint;
			if (reported) {
				if (!finalReport(reported)) break;
				next = reported.end;
			} else {
				if (this.#pending.some((run) => run.start <= floor && floor < run.end)) break;
				let end = this.#delivered.runEnd(floor);
				if (end === undefined && this.#skipped.has(floor)) end = floor + 1n;
				if (end === undefined) break;
				// A report or a run given up inside it is judged on its own.
				const [inside] = this.#reported.overlapping(floor, end);
				if (inside) end = min(end, max(inside.start, floor));
				for (const run of this.#pending) if (run.start > floor && run.start < end) end = run.start;
				next = end;
			}
			if (handed !== undefined && floor <= handed && handed < next) next = handed;
			if (next <= floor) break;
			floor = next;
		}
		if (floor === this.#folded) return;
		this.#folded = floor;
		this.#delivered.forgetBelow(floor);
		for (const sequence of this.#skipped.keys()) if (sequence < floor) this.#skipped.delete(sequence);
		this.#pending = this.#pending.flatMap((run) => removeRun([run], 0n, floor));
		this.#reported.set(0n, floor);
		this.#givenUp = removeRun(this.#givenUp, 0n, floor);
		// The fills it refused, as the Rust `Misses::given_up` is forgotten below the floor.
		if (this.#state.refused.peek().some((run) => run.start < floor))
			this.#state.refused.update((runs) => removeRun(runs, 0n, floor));
	}

	// Owe again every run this cursor gave up as stale, reported or not: an update reopens the
	// verdict. A group still cached is offered again as a widened skip is, a missing one is missed
	// again (asked of a fill, or dropped as not found when none can serve it), and whatever is
	// still stale under the new terms is given up and reported again. Returns the runs taken back. A
	// resumed reader takes back what it queued and what its serving route's copy reported.
	#reofferStale(): Run[] {
		if (this.#resume) return this.#resume.reofferStale();
		const taken: Run[] = [];
		const received: Run[] = [];
		const stale = this.#reported.all.filter(reopens);
		for (const run of stale) {
			insertRun(taken, run.start, run.end);
			if (run.received) insertRun(received, run.start, run.end);
		}
		this.#reported.clear(stale);
		for (const run of this.#pending) {
			if (run.stale === undefined) continue;
			insertRun(taken, run.start, run.end);
			if (run.received) insertRun(received, run.start, run.end);
		}
		this.#pending = this.#pending.filter((run) => run.stale === undefined);
		// A received stale run is folded in again by the next scan, if the track still holds it.
		this.#dropsSeen = undefined;
		const producer = this.#state.producer;
		const sequences: bigint[] = [];
		for (const sequence of this.#skipped.keys()) {
			if (!taken.some((run) => run.start <= sequence && sequence < run.end)) continue;
			// A skip whose group left the cache can never be offered again: forgotten, so the
			// run is judged again below, as the Rust `prune_skipped` forgets it first.
			if (producer && !cachedHolds(producer, sequence)) this.#skipped.delete(sequence);
			else sequences.push(sequence);
		}
		// What no skip offers again, such as a group written off at the track's max age bound,
		// which no budget reverses, is still stale under the new terms: reported again, as is a
		// missing group they still convict. What an
		// upstream's stale run decided is the upstream's to judge again: the scan above folds it
		// in again if the track still holds it, and it stays owed if new terms superseded it.
		// Mirrors Rust.
		let kept: Run[] = taken.map((run) => ({ ...run }));
		for (const sequence of sequences) kept = removeRun(kept, sequence, sequence + 1n);
		for (const decided of received) kept = removeRun(kept, decided.start, decided.end);
		if (kept.length > 0) {
			const drift = this.#drift();
			const stale = staleHole(drift.budget);
			for (const run of kept) {
				for (const part of this.#missed(run)) {
					// A miss the new terms still convict is stale again; one a fill can still bring
					// stays owed, asked afresh; the rest no copy left here can serve.
					if (!part.missed || this.#staleAt(part.end - 1n, drift)) {
						this.#pending = insertDrop(this.#pending, part.start, part.end, stale);
					} else if (!this.#retry(part)) {
						const hole = { error: new StreamError(StreamCode.NotFound, { message: "group not found" }) };
						this.#pending = insertDrop(this.#pending, part.start, part.end, hole);
					}
				}
			}
		}
		if (sequences.length > 0 && producer && !this.#left) reofferSequences(producer, this.#state, sequences);
		return taken;
	}

	// `run` split into its cached groups, which no skip offers again and so stay stale, and the
	// gaps between them, which the cache no longer holds: misses, judged as the Rust `Misses`
	// judges a gap, by its newest sequence.
	#missed(run: Run): (Run & { missed: boolean })[] {
		const producer = this.#state.producer;
		if (!producer) return [{ ...run, missed: false }];
		const parts: (Run & { missed: boolean })[] = [];
		let next = run.start;
		// A failed copy holds nothing to read, so it is missed like an evicted one.
		const held = cachedIn(producer, run.start, run.end).filter((sequence) => cachedHolds(producer, sequence));
		for (const sequence of held.sort((a, b) => (a < b ? -1 : a > b ? 1 : 0))) {
			if (next < sequence) parts.push({ start: next, end: sequence, missed: true });
			parts.push({ start: sequence, end: sequence + 1n, missed: false });
			next = sequence + 1n;
		}
		if (next < run.end) parts.push({ start: next, end: run.end, missed: true });
		return parts;
	}

	// Ask a fill again for `run`, which an update reopened: no longer judged, so the next read
	// asks it or gives it up afresh, as the Rust `Misses::retry`. False when no fill can answer.
	#retry(run: Run): boolean {
		const producer = this.#state.producer;
		if (!producer || this.#left) return false;
		const fills = fillsOf(producer);
		if (!fills.serving && !fills.active) return false;
		this.#givenUp = removeRun(this.#givenUp, run.start, run.end);
		this.#asked = removeRun(this.#asked, run.start, run.end);
		this.#refilled = removeRun(this.#refilled, run.start, run.end);
		return true;
	}

	// Fresh reads of the cached groups in `[start, end)` this cursor admits and no longer judges
	// stale, each expiring against the budget, and the runs still given up. A resumed reader asks
	// its serving route's copy. Mirrors the Rust `reoffer_groups`.
	#reofferGroups(start: bigint, end: bigint): { groups: GroupConsumer[]; givenUp: Run[] } {
		if (this.#resume) return this.#resume.reofferGroups(start, end);
		const groups: GroupConsumer[] = [];
		let givenUp: Run[] = [{ start, end }];
		const producer = this.#state.producer;
		if (!producer) return { groups, givenUp };
		const drift = this.#drift();
		const cap = this.#cursor.peek().end;
		// Only what the cache holds, so a long run of groups long gone costs nothing.
		const cached = cachedIn(producer, start, end).sort((a, b) => (a < b ? -1 : a > b ? 1 : 0));
		for (const sequence of cached) {
			const admitted = (cap === undefined || sequence < cap) && this.#window.contains(sequence);
			if (!admitted || this.#staleAt(sequence, drift)) continue;
			const group = cachedMirror(producer, sequence, 0);
			if (!group) continue;
			groups.push(this.#guard(group));
			givenUp = removeRun(givenUp, sequence, sequence + 1n);
		}
		return { groups, givenUp };
	}

	private constructor(name: string, state: TrackState) {
		this.name = name;
		this.#state = state;
		state.reads = (sequence) => this.#reading(sequence);
		// Each reader filters locally even when another reader widens upstream demand.
		this.#resolve();
	}

	// The latest group to count relative anchors back from. A reader following route copies
	// leaves them to the copies, which hold the groups.
	#latestGroup(): bigint | undefined {
		return this.#resume ? undefined : this.#state.largest.peek()?.group;
	}

	// Resolve this reader's ranges again, as when they are replaced.
	#resolve(): void {
		this.#window = new Window(this.#state.update.peek()?.ranges ?? [{}], this.#latestGroup(), this.#holdings());
		this.#moved();
	}

	// The window moved: narrow the bounds to it.
	#moved(): void {
		this.#revision++;
		this.#bound();
	}

	// Where the groups this reader's track holds start, for resolving its time anchors: the
	// producer's checked starts, and this reader's mirror of its cache. A reader following route
	// copies leaves them to the copies, which hold the groups.
	#holdings(): Holdings {
		const state = this.#state;
		const producer = state.producer;
		if (this.#resume || !producer) return UNHELD;
		return {
			containing: (pts) => {
				const starts = startsOf(producer);
				if (starts.length === 0) return "unknown";
				if (later(starts[0].start, pts)) return "before";
				// Checked starts never rewind, so they ascend with the sequence.
				let lo = 1;
				let hi = starts.length;
				while (lo < hi) {
					const mid = (lo + hi) >>> 1;
					if (later(starts[mid].start, pts)) hi = mid;
					else lo = mid + 1;
				}
				return { group: starts[lo - 1].sequence };
			},
			nextStart: (from) => {
				const starts = startsOf(producer);
				return starts[timelineIndex(starts, from)];
			},
			nextHeld: (from) => {
				const group = state.timeline[timelineIndex(state.timeline, from)];
				if (!group) return undefined;
				const started = hooks.groupTimestamp(group);
				if (started) return { sequence: group.sequence, held: { started } };
				return { sequence: group.sequence, held: hooks.groupStarted(group) ? "never" : "pending" };
			},
			gone: (sequence) => goneAt(producer, sequence),
			// Only once the producer is gone: a group held below a reached end may still start.
			complete: () => state.closed.peek() !== undefined,
		};
	}

	// Resolve what the groups now held settle: a relative anchor waiting on a first group once
	// one exists, and a time end as groups start. Returns the refusal of ranges this copy cannot
	// resolve, which every read reports.
	#settle(): Error | undefined {
		const ranges = this.#window.ranges;
		if (this.#state.dropsAnchors.peek() && !ranges.every(isWire)) {
			return new Error("a relative or time anchor other than the live start needs moq-lite-07");
		}
		const info = this.#state.info.peek();
		if (info && info.timescale == null && hasTime(ranges)) return new TimestampMismatch();
		// Its timestamps may rewind, so its group starts are never checked or indexed.
		if (info?.ietf !== undefined && hasTime(ranges)) {
			return new Error("a time anchor needs timestamps that never rewind, which moq-transport does not promise");
		}
		if (this.#resume || this.#window.settled) return undefined;
		// The live feed had no group when the ranges resolved, so the newest one then was the first.
		const latest = this.#window.pending ? this.#state.first : this.#latestGroup();
		if (this.#window.settle(latest, this.#holdings())) this.#moved();
		return undefined;
	}

	// Narrow the local bounds to the window into the hull every read starts from.
	#bound(): void {
		const end = this.#window.end();
		const local = this.#local;
		this.#cursor.set({
			start: max(local.start, this.#window.floor() ?? 0n),
			end: end === undefined ? local.end : min(local.end ?? end, end),
		});
		this.#judge(0n, UNBOUNDED);
		// A fill this reader alone waited on for a group it no longer reads is withdrawn.
		this.#fillsMoved();
	}

	// The parts of `[start, end)` this reader still reads: inside its cursor and window, not
	// handed out, and not passed by the sequence cursor, which never goes back.
	#reads(start: bigint, end: bigint): Run[] {
		if (this.#state.closed.peek() !== undefined) return [];
		return this.#unread(start, end);
	}

	// The parts of `[start, end)` this reader's bounds and cursor still leave unread, even once
	// the track closed, since what it buffered stays readable.
	#unread(start: bigint, end: bigint): Run[] {
		const cursor = this.#cursor.peek();
		const from = max(max(start, cursor.start), this.#nextSequence);
		const to = cursor.end === undefined ? end : min(end, cursor.end);
		return this.#window.runs(from, to).flatMap((run) => this.#delivered.gaps(run.start, run.end));
	}

	// Tell each group buffered in `[start, end)` whether this reader will still read it, as the
	// Rust `Join` holds a run until its reader takes it, passes it, or stops reading it. A budget
	// judges a group only once a read reaches it, as there.
	#judge(start: bigint, end: bigint): void {
		const groups = this.#state.groups.peek();
		for (let i = timelineIndex(groups, start); i < groups.length && groups[i].sequence < end; i++) {
			hooks.groupReads(groups[i], this.#reading(groups[i].sequence));
		}
	}

	// Whether this reader will still read group `sequence` once buffered: what it wants from a
	// fill, judged even once the track closed.
	#reading(sequence: bigint): boolean {
		if (this.#left) return false;
		const unread = this.#unread(sequence, sequence + 1n);
		return unread.length > 0 && runGaps(this.#givenUp, sequence, sequence + 1n).length > 0;
	}

	// What this reader still wants from a fill: what it reads, less what it gave up as stale.
	#wants(start: bigint, end: bigint): Run[] {
		const reads = this.#reads(start, end);
		if (this.#givenUp.length === 0) return reads;
		return reads.flatMap((run) => runGaps(this.#givenUp, run.start, run.end));
	}

	// This reader as the producer's fills know it.
	#askerOf(): Asker {
		this.#state.asker ??= {
			wants: (start, end) => this.#wants(start, end),
			priority: () => this.#state.update.peek()?.priority ?? 0,
			maxDelay: () => this.#drift().budget,
			order: () => this.#state.update.peek()?.order ?? "descending",
			refuse: (start, end, error, stale) =>
				this.#state.refused.update((runs) => insertDrop(runs, start, end, { error, stale })),
		};
		return this.#state.asker;
	}

	// Ask the producer's fills for the misses in `[start, end)`: groups this reader reads below
	// the newest one received that the cache does not hold and nobody gave up. A run whose newest
	// group the budget already convicts is not asked, so a reader at the live edge never asks.
	// Returns the first sequence a fill still owes this reader whose budget still wants it, arming
	// the wakeup for when that changes. Mirrors the Rust `Misses::judge`.
	#askFills(start: bigint, end: bigint, drift?: Drift): bigint | undefined {
		const producer = this.#state.producer;
		if (this.#resume || !producer || this.#left) return undefined;
		const fills = fillsOf(producer);
		// A fill already out can be joined even once its handler's handle closed.
		if (fills.serving || fills.active) {
			drift ??= this.#drift();
			const asker = this.#askerOf();
			this.#clipAsked();
			const timeline = this.#state.timeline;
			const accounted: Run[] = [];
			// Reordered streams still owed by the live feed are not cache misses.
			for (const run of owedRuns(producer, start, end)) {
				insertRun(accounted, run.start, run.end);
				if (run.folds !== undefined) this.#wakeAt(run.folds.since(Instant.ZERO).asNanos());
			}
			// Below where the feed starts, only the history the copy's start rule leaves fillable.
			const floor = fillFloor(producer, this.#window.start());
			// A received stale run that does not settle this reader is unknown to it, so it is
			// still asked for, as the Rust `Misses::unclaimed` judges it.
			const budget = drift.budget;
			const coverage = this.#coverage();
			const { drops, refused } = this.#state;
			const low = floor === undefined ? start : max(start, floor);
			const edge = min(end, max(this.#state.received.peek(), (fillEdge(producer) ?? -1n) + 1n));
			const spans: Run[] = [{ start: low, end: edge }];
			// The upstream accounted for a stale run, so the live feed brings none of it unless an
			// update supersedes it: asked even above the newest group received. Only the runs that
			// do not settle this reader are walked, never every run above the edge.
			const above = max(low, edge);
			for (const dropped of this.#unknownRuns(budget, coverage)) {
				if (dropped.end <= above || dropped.start >= end) continue;
				const span = { ...dropped, start: max(dropped.start, above), end: min(dropped.end, end) };
				if (span.start < span.end && !settles(span, budget, coverage, this.#state.judged)) spans.push(span);
			}
			for (const run of spans.flatMap((span) => this.#wants(span.start, span.end))) {
				const gone = [...accounted];
				for (const dropped of givenUp(drops, refused.peek(), run.start, run.end, (run) =>
					settles(run, budget, coverage, this.#state.judged),
				)) {
					// Judged from where it meets the asked run, as the reader would read it.
					const clipped = { ...dropped, start: max(dropped.start, run.start) };
					if (settles(clipped, budget, coverage, this.#state.judged))
						insertRun(gone, clipped.start, dropped.end);
				}
				for (const open of runGaps(gone, run.start, run.end)) {
					let next = open.start;
					for (let i = timelineIndex(timeline, open.start); i < timeline.length; i++) {
						const group = timeline[i];
						if (group.sequence >= open.end) break;
						// A failed copy holds nothing to read, so it is missed like an evicted one.
						if (hooks.groupFailed(group)) continue;
						this.#askMiss(next, group.sequence, drift, asker);
						next = group.sequence + 1n;
					}
					this.#askMiss(next, open.end, drift, asker);
				}
			}
		}
		const asker = this.#state.asker;
		if (!asker) return undefined;
		// Every group of an owed run waits on the same successor, so its newest judges the run;
		// a stale run is given up, and gives way to a later one the budget still wants.
		let owed: bigint | undefined;
		const stale: Run[] = [];
		for (const run of fills.covered(start, end, asker)) {
			drift ??= this.#drift();
			if (this.#staleAt(run.end - 1n, drift, true)) {
				stale.push(run);
				continue;
			}
			this.#fillWait = true;
			owed = run.start;
			break;
		}
		if (stale.length > 0) {
			for (const run of stale) {
				insertRun(this.#givenUp, run.start, run.end);
				this.#judge(run.start, run.end);
			}
			this.#fillsMoved();
		}
		return owed;
	}

	// The received stale runs that do not settle this reader under `budget` and `coverage`,
	// lowest first. While the terms hold, only what changed since is judged again: a run the
	// track changed or forgot leaves the memo, and the runs the change left are judged anew, so a
	// drop costs each probe the runs it touches. A later start only narrows what a run asks the
	// reader to read, so a run that settles here settles clipped too. Mirrors the Rust
	// `Misses::unknown`.
	#unknownRuns(budget: number, coverage: Coverage): readonly DropRun[] {
		const { drops, judged } = this.#state;
		const unsettled = (run: DropRun) => !settles(run, budget, coverage, judged);
		const memo = this.#unknown;
		let runs: DropRun[];
		if (memo && memo.judged === judged && memo.budget === budget && sameCoverage(memo.coverage, coverage)) {
			if (memo.drops === drops.revision) return memo.runs;
			runs = memo.runs.filter((run) => drops.at(run.start) === run);
			for (const changed of drops.since(memo.drops)) {
				for (const run of drops.overlapping(changed.start, changed.end)) {
					if (!unsettled(run) || runs.includes(run)) continue;
					const at = runs.findIndex((held) => held.start > run.start);
					runs.splice(at < 0 ? runs.length : at, 0, run);
				}
			}
		} else {
			runs = drops.all.filter(unsettled);
		}
		this.#unknown = { drops: drops.revision, judged, budget, coverage, runs };
		return runs;
	}

	// Ask a fill for a fetch's group at once, written yet or not, as the Rust `fetch_group` queues
	// one; a group the live feed writes meanwhile answers it. `false` with no handler to ask; `"pending"` until the track is accepted.
	#askFetch(sequence: bigint): boolean | "pending" {
		const producer = this.#state.producer;
		if (!producer) return false;
		if (this.#state.info.peek() === undefined && this.#state.closed.peek() === undefined) return "pending";
		if (cachedHolds(producer, sequence)) return true;
		// Joins a fill already bringing it, handler or not; asks a new one only while one is open.
		const asked = fillsOf(producer).ask(sequence, sequence + 1n, this.#askerOf());
		if (asked.length === 0) return false;
		insertRun(this.#asked, sequence, sequence + 1n);
		return true;
	}

	// The lowest group a fill still brings this reader, asking for what it misses first.
	#filling(): bigint | undefined {
		const producer = this.#state.producer;
		if (this.#resume || !producer) return undefined;
		const fills = fillsOf(producer);
		if (!fills.serving && !fills.active) return undefined;
		const drift = this.#drift();
		return this.#askFills(this.#cursor.peek().start, UNBOUNDED, drift);
	}

	// Ask a fill for the missed run `[start, end)`, unless its newest group is stale: what was
	// never asked, and once more what was asked but is missed again.
	#askMiss(start: bigint, end: bigint, drift: Drift, asker: Asker): void {
		if (start >= end || this.#staleAt(end - 1n, drift)) return;
		const fills = fillsOf(this.#state.producer as Producer);
		// Asked before and missed again, unless a fill still brings it: its answer was evicted
		// before this reader took it.
		const again = intersect(this.#asked, start, end).flatMap((run) =>
			runGaps(this.#refilled, run.start, run.end).flatMap((refill) =>
				runGaps(fills.covered(refill.start, refill.end, asker), refill.start, refill.end),
			),
		);
		// Only what a fill took counts as asked: with no handler open, a miss no fill covers is
		// left to ask once one opens.
		for (const gap of runGaps(this.#asked, start, end)) {
			for (const run of fills.ask(gap.start, gap.end, asker)) insertRun(this.#asked, run.start, run.end);
		}
		// Refilled and missed again, unless a fill still brings it: given up and named, never
		// asked a third time, as the Rust `Misses` gives up a refill evicted again.
		const lost = intersect(this.#refilled, start, end).flatMap((run) =>
			runGaps(fills.covered(run.start, run.end, asker), run.start, run.end),
		);
		for (const refill of again) {
			for (const run of fills.ask(refill.start, refill.end, asker)) insertRun(this.#refilled, run.start, run.end);
		}
		for (const run of lost) {
			insertRun(this.#givenUp, run.start, run.end);
			asker.refuse(run.start, run.end, new StreamError(StreamCode.NotFound, { message: "group evicted again" }));
		}
		if (lost.length > 0) this.#fillsMoved();
	}

	// Whether a held group waits on the local cap rising: one past it may still be read. A
	// group outside the window alone never holds a finished track open.
	#parked(groups: readonly GroupConsumer[]): boolean {
		const cap = this.#local.end;
		return cap !== undefined && (groups.at(-1)?.sequence ?? -1n) >= cap;
	}

	// Whether a held group from `from` on may still start, and so settle a time anchor the window
	// waits on, which a reached end does not decide: until it starts or the track closes, as
	// {@link Holdings.complete} says, the window may still take it in.
	#starting(groups: readonly GroupConsumer[], from = 0n): boolean {
		if (this.#window.settled || this.#state.closed.peek() !== undefined) return false;
		return groups.some((group) => group.sequence >= from && !hooks.groupStarted(group));
	}

	// Take the group at `index` off the ordered buffer, closing the out-of-range ones below it,
	// which sequence order passes for good.
	#pass(groups: GroupConsumer[], index: number): void {
		const [group] = groups.splice(index, 1);
		for (const passed of groups.splice(0, index)) passed.close();
		this.#nextSequence = group.sequence + 1n;
		this.#judge(group.sequence, this.#nextSequence);
	}

	// Whether this read ends because its bounded window has nothing left to deliver; see
	// {@link #drained}. A wire publisher's reader never ends there.
	#exhausted(): boolean {
		if (this.#parks) return false;
		if (this.#resume) return this.#resume.exhausted(this.#delivered, this.#window);
		return this.#drained([this.#delivered]);
	}

	// Whether a bounded window has nothing left to deliver: every group in its ranges was handed
	// out (in any of `taken`), dropped, or never will be, as one skipped as stale, one below where
	// the feed started, or a hole the feed no longer owes. An open window always has more, and so
	// does one the cap keeps a group out of, since the reader may raise it again. Mirrors the
	// Rust `Cursor::is_exhausted`.
	#drained(taken: readonly Delivered[]): boolean {
		const end = this.#window.end();
		if (end === undefined) return false;
		// A fill still owes a group it reads.
		if (this.#askFills(this.#window.floor() ?? 0n, end) !== undefined) return false;
		const held = this.#state.groups.peek();
		const producer = this.#state.producer;
		const received = max(this.#state.received.peek(), ((producer && fillEdge(producer)) ?? -1n) + 1n);
		// A received stale run that does not settle this reader is unknown to it (see `settles`):
		// the reader still wants those groups, however the feed accounted for them.
		const budget = this.#drift().budget;
		const coverage = this.#coverage();
		const { drops: all, refused } = this.#state;
		for (const run of this.#window.runs(this.#window.floor() ?? 0n, end)) {
			let gaps = [run];
			for (const delivered of taken) gaps = gaps.flatMap((gap) => delivered.gaps(gap.start, gap.end));
			for (const gap of gaps) {
				// Each run judged from where it meets the gap, as the reader would read it.
				const overlapping = givenUp(all, refused.peek(), gap.start, gap.end, (run) =>
					settles(run, budget, coverage, this.#state.judged),
				).map((run) => ({
					...run,
					start: max(run.start, gap.start),
					end: min(run.end, gap.end),
				}));
				const drops = overlapping.filter((run) => settles(run, budget, coverage, this.#state.judged));
				const unknown = overlapping.filter((run) => !settles(run, budget, coverage, this.#state.judged));
				for (const open of runGaps(drops, gap.start, gap.end)) {
					if (unknown.some((run) => run.start < open.end && open.start < run.end)) return false;
					// Not received yet, so it may still arrive.
					if (open.end > received) return false;
					// Held, so a read still hands it out, skips it, or waits for the cap to rise.
					const index = timelineIndex(held, open.start);
					if (held[index] && held[index].sequence < open.end) return false;
					const owed = producer && owedGroup(producer, open.start, open.end);
					if (owed) {
						if (owed.folds !== undefined) this.#wakeAt(owed.folds.since(Instant.ZERO).asNanos());
						return false;
					}
				}
			}
		}
		return true;
	}

	static {
		makeSubscriber = (name, state) => new Subscriber(name, state);
		hooks.subscriberEnd = (subscriber) =>
			subscriber.#resume ? subscriber.#resume.end() : subscriber.#state.end.peek()?.at;
		hooks.subscriberEndError = (subscriber) =>
			subscriber.#resume ? subscriber.#resume.endError() : subscriber.#state.end.peek()?.error;
		hooks.joinSubscriber = (subscriber) => {
			if (subscriber.#resume) subscriber.#resume.join();
			else if (subscriber.#state.producer) hooks.joinProducer(subscriber.#state.producer);
		};
		hooks.sourceStart = (subscriber) => {
			if (subscriber.#resume) return subscriber.#resume.start();
			// A relative anchor still pending reads every group, so a fill asked below would claim
			// the history it never names. Mirrors the Rust `cursor.sync()` in `poll_start`.
			subscriber.#settle();
			const producer = subscriber.#state.producer;
			const start = producer && sourceStart(producer);
			if (start === "pending") return start;
			const floor = start === undefined ? undefined : max(start, subscriber.#cursor.peek().start);
			// The lowest group a fill brings this reader starts it too, as the Rust `poll_start`
			// has it: a group arriving first must not give up the history being filled below it.
			const filling = subscriber.#filling();
			if (filling === undefined) return floor;
			return floor === undefined ? filling : min(floor, filling);
		};
		hooks.tryRecvGroup = (subscriber, dispositions) => subscriber.#tryRecvGroup(dispositions);
		hooks.parked = (subscriber) => {
			subscriber.#parks = true;
			return subscriber;
		};
		// A route copy judged for the logical reader that handed out `delivered`: what the copy
		// handed out itself counts too. A copy that is itself a front judges across its own copies.
		hooks.exhaustedFor = (subscriber, delivered) =>
			subscriber.#resume
				? subscriber.#resume.exhausted(subscriber.#delivered.union(delivered), subscriber.#window)
				: subscriber.#drained([subscriber.#delivered, delivered]);
		hooks.tryDrop = (subscriber) => subscriber.#tryDrop();
		// A reader following route copies leaves its ranges to the serving copy, which resolves
		// their anchors against the groups it holds, then narrows them to its own local bounds.
		hooks.servingRuns = (subscriber, start, end) => {
			const cursor = subscriber.#cursor.peek();
			const from = max(start, cursor.start);
			const to = min(end, cursor.end ?? end);
			return subscriber.#resume?.runs(from, to) ?? subscriber.#window.runs(from, to);
		};
		hooks.servingEnd = (subscriber) => {
			const cursor = subscriber.#cursor.peek();
			if (!subscriber.#resume) return subscriber.#window.end() === undefined ? undefined : cursor.end;
			const end = subscriber.#resume.servingEnd();
			return end === undefined ? undefined : min(end, cursor.end ?? end);
		};
		hooks.servingStarts = (subscriber, from) => {
			const resumed = subscriber.#resume?.starts(from);
			if (resumed) return resumed;
			subscriber.#settle();
			return subscriber.#window.starts(from);
		};
		hooks.servingResolved = (subscriber) => {
			const resumed = subscriber.#resume?.resolved();
			if (resumed) return resumed;
			subscriber.#settle();
			return subscriber.#window.resolved();
		};
		hooks.tryNextGroup = (subscriber, floor) => {
			subscriber.#raiseNext(floor);
			return subscriber.#tryNextGroup();
		};
		hooks.peekGroup = (subscriber, ordered, floor) => {
			if (ordered) subscriber.#raiseNext(floor);
			return subscriber.#peekGroup(ordered);
		};
		hooks.forgetPastBudget = (subscriber, delivered) => subscriber.#forgetPastBudget(delivered);
		hooks.wire = (subscriber) => subscriber.#wireRecord();
		hooks.reofferStale = (subscriber) => subscriber.#reofferStale();
		hooks.reofferGroups = (subscriber, start, end) => subscriber.#reofferGroups(start, end);
		hooks.deliveredRuns = (subscriber) => subscriber.#delivered.runs;
		hooks.handedOut = (subscriber, sequence) => subscriber.#delivered.has(sequence);
		hooks.hasUnreadGroup = (subscriber, start, end, delivered) =>
			subscriber.#state.groups
				.peek()
				.some(
					(group) =>
						group.sequence >= start &&
						(end === undefined || group.sequence < end) &&
						!delivered.has(group.sequence) &&
						!hooks.groupFailed(group),
				);
		hooks.spent = (subscriber, start, end, delivered) =>
			(subscriber.#state.closed.peek() !== undefined || subscriber.#complete()) &&
			!hooks.hasUnreadGroup(subscriber, start, end, delivered);
		hooks.awaitsFill = (subscriber) => subscriber.#fillWait;
		hooks.staleAt = (subscriber, sequence) => {
			const held: Superseded = {};
			return {
				expired: () => subscriber.#staleAt(sequence, subscriber.#drift(), true, held),
				changed: subscriber.#expiryChanged(),
			};
		};
		hooks.cachedGroup = (subscriber, sequence, from = 0) => {
			if (subscriber.#resume) return subscriber.#resume.cachedGroup(sequence, from);
			const producer = subscriber.#state.producer;
			return producer && cachedMirror(producer, sequence, from);
		};
		hooks.holdsGroup = (subscriber, sequence) => {
			const producer = subscriber.#state.producer;
			return producer !== undefined && cachedHolds(producer, sequence);
		};
		hooks.staleHeld = (subscriber, start, end) => {
			const held: Run[] = [];
			for (const run of subscriber.#state.drops.overlapping(start, end)) {
				if (run.stale !== undefined) insertRun(held, max(run.start, start), min(run.end, end));
			}
			return held;
		};
		hooks.givenUp = (subscriber, sequence) => {
			const { drops, refused, judged } = subscriber.#state;
			const drop = drops.at(sequence);
			const own = dropAt(refused.peek(), sequence);
			// As in `givenUp`: only a drop that settles the reader hides its own refusal.
			const wins =
				drop &&
				(own === undefined ||
					settles({ ...drop, start: sequence }, subscriber.#drift().budget, subscriber.#coverage(), judged));
			return {
				error: wins ? drop.error : own,
				changed: [drops.version, refused],
			};
		};
		hooks.mayLand = (subscriber, land) => {
			const producer = subscriber.#state.producer;
			return producer !== undefined && mayLand(producer, land);
		};
		hooks.billSubscriber = (subscriber, current) => subscriber.#billOn(current);
		hooks.resumeSubscriber = (name, subscription, source) => {
			const state = new TrackState(subscription ?? Subscription.default());
			const subscriber = new Subscriber(name, state);
			subscriber.#resume = source({
				subscription: state.update,
				cursor: subscriber.#cursor,
				closed: state.closed,
				info: () => state.info.peek(),
				commit: (info) => {
					if (state.info.peek() === undefined) state.info.set(info);
				},
				close: (abort) => {
					closeTrackState(state, abort);
				},
			});
			return subscriber;
		};
		hooks.watchSubscriberError = (subscriber, fn) => {
			if (subscriber.#resume) return subscriber.#resume.watchError(fn);
			const check = (error: Error | null | undefined) => {
				if (error instanceof Error) fn(error);
			};
			const stop = subscriber.closed.subscribe(check);
			check(subscriber.closed.peek());
			return stop;
		};
		hooks.latestSequence = (subscriber) =>
			subscriber.#resume ? subscriber.#resume.latest() : subscriber.#state.latest;
		hooks.bufferedLargest = (subscriber) => subscriber.#bufferedLargest();
		hooks.objectLargest = (subscriber) => {
			if (subscriber.#resume) return subscriber.#resume.objectLargest();
			const producer = subscriber.#state.producer;
			return producer && objectLargest(producer);
		};
		hooks.peekLargest = (subscriber) => {
			if (subscriber.#resume) return subscriber.#resume.peekLargest();
			const state = subscriber.#state;
			return {
				largest: state.largest.peek(),
				ended: state.closed.peek() !== undefined || state.largestEnded.peek(),
			};
		};
		hooks.largestChanged = (subscriber, fn) => {
			const state = subscriber.#state;
			let fired = false;
			const once = () => {
				if (fired) return;
				dispose();
				fn();
			};
			const disposes = [
				state.largest.changed(once),
				state.largestEnded.changed(once),
				state.closed.changed(once),
			];
			const dispose = () => {
				fired = true;
				for (const close of disposes) close();
			};
			return dispose;
		};
		hooks.groupChanged = (subscriber, fn) => subscriber.#groupChanged(fn);
		hooks.askFetch = (subscriber, sequence) => subscriber.#askFetch(sequence);
		hooks.fetchBudget = (subscriber, maxDelay) => {
			subscriber.#fetchBudget = maxDelay?.asMillis() ?? Number.POSITIVE_INFINITY;
			subscriber.#state.fetch = true;
			// The sink joined as a live subscriber; leave the live aggregate before anyone sees it.
			if (subscriber.#state.producer) resubscribe(subscriber.#state.producer);
			const update = subscriber.#state.update.peek();
			if (update) subscriber.#state.update.set(update, true);
		};
		// The sequence cursor lives here (it shares the buffer and the drift anchor with
		// the arrival cursor); `Ordered` is the handle that reaches it.
		ordered_ = {
			nextGroup: (subscriber) => subscriber.#nextGroup(),
			tryNextGroup: (subscriber) => subscriber.#tryNextGroup(),
			ended: (subscriber) => subscriber.#ended(),
			readFrame: (subscriber) => subscriber.#readFrame(),
			readString: (subscriber) => subscriber.#readString(),
			readJson: (subscriber) => subscriber.#readJson(),
		};
	}

	// Refuse a read on this handle once `ordered()` has taken the subscription.
	#live() {
		if (this.#mode === "ordered") throw new Error("track is read in sequence order; use the Ordered handle");
	}

	/**
	 * Resolve this track's immutable publisher properties.
	 *
	 * Resolves once the wire layer commits the properties it received (TRACK_INFO on
	 * lite-05+, SUBSCRIBE_OK on moq-transport) or an untimed default (older lite drafts),
	 * so awaiting it never yields a placeholder. Rejects if
	 * the track is closed before the properties are known (e.g. a rejected subscription).
	 */
	info(): Promise<Info> {
		return resolveInfo(this.#state);
	}

	/** Settles once the track closes; see {@link Producer.closed}. */
	get closed(): GetPromise<Error | null> {
		return this.#state.closed;
	}

	/** This subscriber's current options, including defaults and the last {@link update}. */
	get subscription(): Getter<Subscription | undefined> {
		return this.#state.update;
	}

	/**
	 * Resolve with the track's exclusive end once it is known, mirroring the Rust `finished`.
	 *
	 * Resolves as soon as the end is declared, which may be ahead of the live edge, so it
	 * says nothing about every group having arrived: read until the cursor returns
	 * `undefined` for that. Rejects with the abort, or if the track closes without an end.
	 */
	async finished(): Promise<Position> {
		if (this.#resume) return this.#resume.finished();
		for (;;) {
			const end = this.#state.end.peek()?.at;
			if (end !== undefined) return end;

			const closed = this.#state.closed.peek();
			if (closed instanceof Error) throw closed;
			if (closed !== undefined) throw new Error("track closed before its end was known");

			await Signal.race(this.#state.end, this.#state.closed);
		}
	}

	/**
	 * The lowest group this subscriber's ranges name, as resolved here, raised to the start of
	 * {@link withRange}: undefined while one reaches back to the oldest group still around, or a
	 * relative start still waits on the first group from the live feed.
	 *
	 * Where a reader that waits on each group in sequence starts, rather than at whatever arrives
	 * first, which newest-first delivery makes the latest. Mirrors the Rust `Subscriber::floor`.
	 */
	floor(): bigint | undefined {
		let start: bigint | undefined;
		if (this.#resume) {
			for (const range of this.#resume.resolved() ?? [undefined]) {
				if (!range?.floored) return undefined;
				start = start === undefined ? range.start : min(start, range.start);
			}
		} else {
			this.#settle();
			start = this.#window.start();
		}
		return start === undefined ? undefined : max(start, this.#local.start);
	}

	/**
	 * Where the source's feed starts once its publisher resolved it, raised to this reader's own
	 * floor: the first group a publisher tells its peer it serves, or lower where a fill this
	 * reader waits on starts. Undefined when the source declares none and no fill is out.
	 *
	 * Where a reader that waits on each group in sequence starts when {@link floor} names none:
	 * groups arrive newest first, so the first to arrive is not the oldest served. A copy says
	 * only once subscribed upstream, so ask once {@link info} resolves or a group arrives. Mirrors
	 * the Rust `Subscriber::poll_start`.
	 */
	async start(): Promise<bigint | undefined> {
		for (;;) {
			const start = hooks.sourceStart(this);
			if (start !== "pending") return start;
			if (this.#state.closed.peek() !== undefined) return undefined;
			await this.#wait();
		}
	}

	/**
	 * The lowest group at or above `from` this subscriber's ranges and {@link withRange} floor
	 * cover, undefined when none does: the groups between are never read, so a reader waiting on
	 * each group in sequence need not wait for them. `from` itself while a relative anchor waits
	 * on the first group, which can't say yet. Mirrors the Rust `Subscriber::next_in_range`.
	 */
	nextInRange(from: bigint): bigint | undefined {
		from = max(from, this.#local.start);
		if (this.#resume) {
			// Without a serving copy nothing says yet.
			const runs = this.#resume.runs(from, UNBOUNDED);
			return runs ? runs[0]?.start : from;
		}
		this.#settle();
		return this.#window.next(from);
	}

	/**
	 * The largest location within {@link withRange}: the current one at first, then each newer
	 * one (skipping any in between), and `undefined` once the track ends.
	 *
	 * The largest is the track's newest group at the newest frame (0 while empty), as
	 * the publisher reported it and its feed delivered since, so it can name a frame before its
	 * group arrives. A largest past the cap or below the floor reports nothing: the cache does not
	 * stand in for it, so a cap moved below the largest stays pending. Moving the bounds wakes a
	 * pending read, which reports the largest once it is within them and newer than the last report.
	 */
	async largest(): Promise<Position | undefined> {
		if (this.#largestDone) return undefined;
		if (this.#resume) {
			const largest = await this.#resume.largest(this.#largestSeen);
			if (largest) this.#largestSeen = largest;
			else this.#largestDone = true;
			return largest;
		}
		for (;;) {
			const { start, end } = this.#cursor.peek();
			const largest = this.#state.largest.peek();
			const within =
				largest && largest.group >= start && (end === undefined || largest.group < end) ? largest : undefined;
			const seen = this.#largestSeen;
			if (within && (!seen || after(within, seen))) {
				this.#largestSeen = within;
				return within;
			}
			if (this.#state.closed.peek() !== undefined || this.#state.largestEnded.peek()) {
				this.#largestDone = true;
				return undefined;
			}
			await Signal.race(this.#state.largest, this.#state.closed, this.#state.largestEnded, this.#cursor);
		}
	}

	// The newest frame buffered for this subscriber, which a wire answer names as its edge:
	// a newest group with no frames yet falls back to the newest older group that does, and it
	// reads as undefined once the newest group was evicted or received, since what is left can
	// no longer say where the edge is.
	#bufferedLargest(): Position | undefined {
		if (this.#resume) return this.#resume.bufferedLargest();
		const latest = this.#state.latest;
		if (latest === undefined) return undefined;

		const groups = this.#state.groups.peek();
		const newest = groups[groups.length - 1];
		if (!newest || newest.sequence !== latest) return undefined;

		for (let i = groups.length - 1; i >= 0; i--) {
			const count = groups[i].frameCount;
			if (count > 0) return { group: groups[i].sequence, frame: count - 1 };
		}
		return undefined;
	}

	/**
	 * An independent subscriber to the same track, with its own cursor and its own replay of
	 * the retained window.
	 *
	 * Reads here do not consume anything this one would deliver, which is what lets a
	 * publisher read the cache alongside the cursor it is serving from. Resolving the track
	 * again through the broadcast would not do: a dynamic serve is one request per peer
	 * subscription, so that mints a second producer the application has to answer separately.
	 * A routed track's fork follows the same logical track across route swaps.
	 *
	 * @internal Wire layers only.
	 */
	fork(options?: Subscription): Subscriber {
		if (this.#resume) return this.#resume.fork(options);
		const producer = this.#state.producer;
		if (!producer) throw new Error("track has no producer to fork from");
		return producer.subscribe(options);
	}

	/**
	 * Limit subsequent reads to these groups and return this reader for chaining. Mirrors the
	 * Rust `with_range`.
	 *
	 * Both bounds are assigned, in either direction: an omitted start is group 0 and an
	 * omitted end removes the cap. They narrow the subscription's ranges, never widen them.
	 * Lowering the floor or raising the cap offers the groups it newly covers that the track
	 * still caches and this reader never handed out; a group is never handed out twice. A local
	 * filter, not a request: it leaves the wire subscription alone, so to change what is asked
	 * for, replace {@link Subscription.ranges} with {@link update}. One exception: on a track
	 * served across route changes, lowering the floor asks the serving route again for the groups
	 * it newly covers within the subscription's ranges, since the route that delivered them may
	 * be gone. {@link largest} reports within these bounds.
	 */
	withRange(groups: Groups): this {
		const before = this.#cursor.peek();
		this.#local = groupBounds(groups);
		this.#bound();
		const { start } = this.#cursor.peek();
		const producer = this.#state.producer;
		if (!this.#resume && !this.#left && producer && start < before.start) {
			// The ordered cursor let go of what it passed below the old floor; the arrival
			// cursor keeps it buffered, which this skips.
			reofferCached(
				producer,
				this.#state,
				start,
				before.start,
				(sequence) => !this.#delivered.has(sequence) && this.#window.contains(sequence),
			);
		}
		return this;
	}

	/** Close the track (optionally with an error), closing any pending groups. Idempotent. */
	close(abort?: Error) {
		if (!this.#left) {
			const arrivals = this.#state.timeline.filter((group) => !hooks.groupFailed(group));
			this.#arrivals = arrivals.map((group) => ({
				sequence: group.sequence,
				arrived: hooks.groupArrived(group),
			}));
		}
		this.#left = true;
		this.#bill?.close();
		this.#bill = undefined;
		this.#state.waitingOrdered = false;
		// Settle if we're first (the producer may already have); either way drop anything
		// still buffered. Groups parked at the withRange cap deliberately outlive a clean
		// producer close, so the subscriber leaving is what must release them: closing
		// and clearing wakes a pending read to observe the end instead of hanging.
		closeTrackState(this.#state, abort);
		this.#resume?.close(abort);
		this.#state.groups.mutate((groups) => {
			for (const group of groups) hooks.releaseGroupReader(group, abort);
			groups.length = 0;
		});
		if (this.#frameGroup) hooks.releaseGroupReader(this.#frameGroup, abort);
		this.#frameGroup = undefined;
		this.#state.timeline.length = 0;
		this.#wall.timer?.abort();
		this.#wall.timer = undefined;
		// A deadline kept in #arrivals may still arm a wakeup.
		this.#wall.at = Number.POSITIVE_INFINITY;
	}

	/**
	 * Receive every group on this track exactly once, as it becomes available.
	 *
	 * Groups may arrive out of order or with gaps due to network conditions; unlike
	 * {@link Ordered.nextGroup}, one that arrives after a newer group was already returned
	 * is still delivered. When several groups are buffered, the lowest sequence is
	 * returned first.
	 *
	 * Honors the range set by {@link withRange}: a group
	 * beyond the cap stays buffered (not dropped) and is offered once the cap rises, even
	 * after a clean close, without blocking in-range groups that arrive behind it.
	 * Once the track closes cleanly, a reader with nothing left in range finishes. So does a
	 * read whose {@link Subscription.ranges} all end, once every group in them was handed out,
	 * dropped, skipped as stale, or will never arrive, while the track goes on, as in Rust;
	 * later reads return `undefined` immediately too, and widening the ranges with {@link update}
	 * reads on without replaying groups already returned. A cap keeping a group out holds it
	 * open, since the cap may rise.
	 * A group whose presentation time is further behind the live edge than this
	 * subscriber's `delay.max` is skipped. The default subscription's zero takes the live edge.
	 * The budget remains attached after return, so a pending frame read rejects if a stalled
	 * group becomes stale while newer data advances.
	 *
	 * The first call commits this track to arrival order: {@link ordered} throws afterwards.
	 */
	async recvGroup(): Promise<GroupConsumer | undefined> {
		this.#live();
		this.#mode = "arrival";
		for (;;) {
			const recv = this.#tryRecvGroup();
			switch (recv.kind) {
				case "group":
					return recv.group;
				case "drop":
					continue;
				case "done":
					return this.#ended();
				case "error":
					throw recv.error;
			}

			// While open, wait for more groups or a cap raise.
			await this.#wait();
		}
	}

	// A drained read: the end of stream, or the error the track ended with once this reader's
	// range reaches that end. A range short of the end never reaches the error.
	#ended(): undefined {
		const error = hooks.subscriberEndError(this);
		if (error === undefined) return undefined;
		const end = hooks.subscriberEnd(this);
		const cap = this.#cursor.peek().end;
		if (end !== undefined && cap !== undefined && cap < finalSequence(end)) return undefined;
		throw error;
	}

	// Park until a group read may go differently.
	#wait(): Promise<void> {
		return new Promise((resolve) => {
			const dispose = this.#groupChanged(() => {
				dispose();
				resolve();
			});
		});
	}

	// Package-internal synchronous half of recvGroup. The lite publisher uses this so applying
	// control state, popping the group, and positioning its frames are one JavaScript turn.
	#tryRecvGroup(dispositions = false): Recv {
		const recv = this.#tryRecvUnsettled(dispositions);
		// What the read settled is billed now, and pays what it can of its readers' debt. The group
		// it hands out stays above the fold until the next read, so the publisher tells its wire of
		// it first rather than taking it for one already served.
		const lagged = this.#settleRecord(recv.kind === "group" ? recv.group.sequence : undefined);
		if (!lagged) return recv;
		// Nobody takes the group the lag replaced, so it lets go of its source here.
		if (recv.kind === "group") recv.group.close();
		return { kind: "error", error: lagged };
	}

	#tryRecvUnsettled(dispositions: boolean): Recv {
		if (this.#lagged) return { kind: "error", error: new TooFarBehind() };
		if (this.#resume) return this.#tryResumed(false, dispositions);
		const refused = this.#settle();
		if (refused) return { kind: "error", error: refused };
		this.#fillWait = false;
		const groups = this.#state.groups.peek();
		const { start, end } = this.#cursor.peek();

		if (dispositions) {
			const dropped = this.#tryDrop();
			if (dropped) return dropped;
		}
		const drift = this.#drift();
		this.#reoffer(drift.budget);
		// Keep unread groups below the floor until cache eviction: replacing the
		// subscription can lower it, just as raising the cap re-offers held groups. Found
		// after the re-offer, which may insert below the floor.
		let index = timelineIndex(groups, start);

		for (;;) {
			// The buffer is sequence-sorted, so an in-range group that arrives behind a
			// beyond-cap one sorts in front of it and is never blocked by it.
			const group = groups[index];
			if (!group || (end !== undefined && group.sequence >= end)) break;
			// Kept, like a group below the floor: replacing the ranges can take it in.
			if (!this.#window.contains(group.sequence)) {
				index++;
				continue;
			}
			groups.splice(index, 1);
			const failed = hooks.groupFailed(group);
			if (dispositions && failed) {
				// A group expired under this subscription's budget is a stale verdict.
				const hole =
					toStreamCode(failed) === StreamCode.Old
						? { ...staleHole(drift.budget), error: failed }
						: { error: failed };
				const dropped = this.#report(group.sequence, group.sequence + 1n, hole);
				if (dropped) return dropped;
				continue;
			}
			if (this.#delivered.has(group.sequence)) {
				group.close();
				continue;
			}
			if (this.#staleAt(group.sequence, drift)) {
				group.close();
				// A re-offered group stale at the bound is final.
				if (drift.bounded) this.#skipped.delete(group.sequence);
				else this.#skip(group.sequence, drift.budget);
				if (dispositions) {
					const dropped = this.#report(group.sequence, group.sequence + 1n, staleHole(drift.budget));
					if (dropped) return dropped;
				} else if (this.#dropsSeen) {
					// A cursor #tryDrop reads reports it there.
					this.#pending = insertDrop(
						this.#pending,
						group.sequence,
						group.sequence + 1n,
						staleHole(drift.budget),
					);
				}
				continue;
			}
			this.#handOut(group.sequence, false);
			return { kind: "group", group: this.#guard(group) };
		}

		const closed = this.#state.closed.peek();
		if (closed instanceof Error) return { kind: "error", error: closed };
		// A clean close completes this range, retaining groups a later cap raise can read.
		if (closed !== undefined) return { kind: "done" };
		// Nothing buffered: ask a fill for what this reader misses, and wait while one owes it.
		if (this.#askFills(start, end ?? UNBOUNDED, drift) !== undefined) return { kind: "idle" };
		// A bounded window ends once nothing in it is left to deliver, as in Rust.
		if (this.#exhausted()) return { kind: "done" };
		if (!this.#complete() || this.#starting(groups)) return { kind: "idle" };
		// An open track can still raise the cap. Keep the groups beyond it.
		return this.#parked(groups) ? { kind: "boundary" } : { kind: "done" };
	}

	// The first part of `[from, to)` this cursor has not reported yet within its window, now
	// reported, with why.
	#report(from: bigint, to: bigint, hole: Hole & { received?: boolean }): (Recv & { kind: "drop" }) | undefined {
		const { start, end } = this.#cursor.peek();
		for (const run of this.#window.runs(max(from, start), min(to, end ?? to))) {
			const dropped = this.#reportRun(run.start, run.end, hole);
			if (dropped) return dropped;
		}
		return undefined;
	}

	// The first part of `[from, to)` nothing already reported blocks, now reported, with why,
	// replacing what it goes out over. Mirrors the Rust `Record::report`.
	#reportRun(from: bigint, to: bigint, hole: Hole & { received?: boolean }): (Recv & { kind: "drop" }) | undefined {
		const start = max(from, this.#folded);
		if (start >= to) return undefined;
		const blocked = this.#reported.overlapping(start, to).filter((reported) => blocksHole(reported, hole));
		const [gap] = runGaps(blocked, start, to);
		if (!gap) return undefined;
		const kind = hole.skip ? "skip" : hole.stale !== undefined ? "stale" : "dropped";
		this.#reported.set(gap.start, gap.end, {
			...gap,
			kind,
			...(kind === "stale" && hole.received ? { received: true } : {}),
		});
		return { kind: "drop", ...gap, error: hole.error, skip: hole.skip, stale: hole.stale };
	}

	// The next run this cursor gave up and has not reported, without taking a group. From the
	// first call on, a read without dispositions keeps its budget skips for it too. A resumed
	// one reports what its routes' copies gave up.
	#tryDrop(): (Recv & { kind: "drop" }) | undefined {
		// What a publisher told its wire since is billed now. A reader the pool ended reports
		// nothing more: its reads fail with the reason.
		if (this.#settleRecord()) return undefined;
		if (this.#resume) {
			for (;;) {
				const queued = this.#nextResumedDrop();
				if (queued) return queued;
				const next = this.#resume.tryDrop(this.#delivered);
				if (!next) return undefined;
				this.#queueResumedDrop(next);
			}
		}
		this.#settle();
		const { start, end } = this.#cursor.peek();
		// The whole span is folded in again only after the window, the budget, or what settles a
		// stale run moved, since out-of-window and unsettling runs were skipped and may be owed
		// now; otherwise only the runs that changed are, so a drop costs each reader the runs it
		// touches.
		const { drops, refused } = this.#state;
		const revision = this.#revision;
		const { budget } = this.#drift();
		const coverage = this.#coverage();
		const judged = this.#state.judged;
		const seen = this.#dropsSeen;
		const full =
			!seen ||
			seen.refused !== refused.peek() ||
			seen.start !== start ||
			seen.end !== end ||
			seen.revision !== revision ||
			seen.budget !== budget ||
			!sameCoverage(seen.coverage, coverage) ||
			seen.judged !== judged;
		const changed: Run[] = full ? [{ start, end: end ?? UNBOUNDED }] : drops.since(seen.drops);
		if (full) this.#pending = this.#pending.filter((run) => !run.received);
		this.#dropsSeen = {
			drops: drops.revision,
			refused: refused.peek(),
			start,
			end,
			revision,
			budget,
			coverage,
			judged,
		};
		for (const span of changed) {
			const from = max(span.start, start);
			const to = end === undefined ? span.end : min(span.end, end);
			if (from >= to) continue;
			if (!full) {
				this.#pending = this.#pending.flatMap((run) => (run.received ? removeRun([run], from, to) : [run]));
			}
			// The reader's own refusals too: clearing the span above took out those it covered.
			const folded = givenUp(drops, refused.peek(), from, to, (run) => settles(run, budget, coverage, judged));
			const received: DropRun[] = [];
			for (const dropped of folded) {
				// Judged from where it meets this cursor's span, as Rust clips it.
				const run = { ...dropped, start: max(dropped.start, from), end: min(dropped.end, to) };
				if (run.start >= run.end) continue;
				// A received stale run that does not settle this reader is unknown to it.
				if (!settles(run, budget, coverage, judged)) continue;
				received.push({ ...run, received: true });
			}
			// Sorted and disjoint: the settling drops and the refusals they leave never overlap.
			this.#pending = insertDrops(this.#pending, received);
		}
		for (let run = this.#pending.shift(); run; run = this.#pending.shift()) {
			// A received run the track has since forgotten, shed or superseded, is unknown again:
			// only the part it still holds as it was folded in is owed.
			if (run.received) {
				const folded = run;
				const held = givenUp(drops, refused.peek(), run.start, run.end, (run) =>
					settles(run, budget, coverage, judged),
				).find((given) => sameVerdict(given, folded));
				if (!held) continue;
				if (held.end < run.end) this.#pending.unshift({ ...run, start: held.end });
				run = { ...run, start: max(run.start, held.start), end: min(run.end, held.end) };
			}
			const dropped = this.#report(run.start, run.end, run);
			if (!dropped) continue;
			if (dropped.end < run.end) this.#pending.unshift({ ...run, start: dropped.end });
			return dropped;
		}
		return undefined;
	}

	// The groups this reader reads, and its cap; see `Coverage`.
	#coverage(): Coverage {
		if (this.#window.pending) return undefined;
		return { runs: this.#window.runs(), cap: this.#cursor.peek().end };
	}

	// A group from the routes' copies; see #resume.
	#tryResumed(ordered: boolean, dispositions = false): Recv {
		const resume = this.#resume as ResumeSource;
		// A clean end leaves what the copies hold readable, as a direct reader's buffer is.
		const closed = this.#state.closed.peek();
		if (closed instanceof Error) return { kind: "error", error: closed };
		for (;;) {
			if (dispositions) {
				const queued = this.#nextResumedDrop();
				if (queued) return queued;
			}
			const recv = resume.tryGroup(ordered, this.#delivered, dispositions);
			if (recv.kind === "drop") {
				this.#queueResumedDrop(recv);
				continue;
			}
			if (recv.kind === "group") this.#handOut(recv.group.sequence, ordered);
			else if (recv.kind === "idle" && this.#exhausted()) return { kind: "done" };
			return recv;
		}
	}

	// Queue a run a resumed reader's copy gave up. Each copy reports to its own route; this
	// reader's record keeps one route's run from going out over what another's already settled.
	#queueResumedDrop(dropped: Recv & { kind: "drop" }): void {
		const { start, end, error, skip, stale } = dropped;
		this.#pending.push({ start, end, error, skip, stale });
	}

	// The lowest queued run's first part nothing already reported blocks, now reported; the rest
	// of that run waits its turn. Mirrors the Rust `Record::next_pending`.
	#nextResumedDrop(): (Recv & { kind: "drop" }) | undefined {
		for (let run = this.#pending.shift(); run; run = this.#pending.shift()) {
			const dropped = this.#reportRun(run.start, run.end, run);
			if (!dropped) continue;
			if (dropped.end < run.end) this.#pending.unshift({ ...run, start: dropped.end });
			return dropped;
		}
		return undefined;
	}

	// The live edge reached the end, with no hole a wire subscription's pending tail still
	// owes below it. Reader completion does not settle the tail: open streams remain
	// accounted for by the wire layer, and an abort still wins on the next read.
	#complete(): boolean {
		const end = this.#state.end.peek();
		return (
			end !== undefined &&
			reaches(this.#state.received.peek(), this.#state.drops, end, this.#state.refused.peek()) &&
			!this.#state.owed.peek()
		);
	}

	// Package-internal readiness half of recvGroup. Each registration fires at most once, and
	// the caller disposes the losers after whichever source wakes it. A declared end wakes it
	// too, so a publisher can forward the end before the live edge reaches it.
	#groupChanged(fn: () => void): Dispose {
		if (this.#resume) {
			const resumed = [this.#resume.changed(fn), this.#cursor.changed(fn), this.#state.closed.changed(fn)];
			return () => {
				for (const close of resumed) close();
			};
		}
		const producer = this.#state.producer;
		const dispose = [
			this.#state.groups.changed(fn),
			this.#state.drops.version.changed(fn),
			this.#state.refused.changed(fn),
			this.#fillMoves.changed(fn),
			// A handler opening takes a miss nobody could ask before.
			...(producer ? [fillsOf(producer).opened.changed(fn)] : []),
			this.#state.startChanged.changed(fn),
			this.#state.folded.changed(fn),
			// A group starting may settle a time end, or move the cap it holds reads back with.
			...(this.#window.settled ? [] : [this.#state.timelineChanged.changed(fn)]),
			// Accepting the track may refuse a time anchor, with no group arriving to say so.
			this.#state.info.changed(fn),
			this.#state.dropsAnchors.changed(fn),
			this.#cursor.changed(fn),
			// A widened budget re-offers the groups it skipped as stale.
			this.#state.update.changed(fn),
			this.#state.closed.changed(fn),
			this.#state.end.changed(fn),
			this.#state.received.changed(fn),
			this.#state.owed.changed(fn),
		];
		if (this.#state.waitingOrdered)
			dispose.push(this.#state.timelineChanged.changed(fn), this.#wall.revision.changed(fn));
		// A bounded window ends once the feed stops owing its last hole, which only time decides,
		// and a fill's hole stops holding a read once the budget convicts it.
		else if (this.#window.end() !== undefined || this.#fillWait) dispose.push(this.#wall.revision.changed(fn));
		return () => {
			for (const close of dispose) close();
		};
	}

	/**
	 * Take the next buffered group without blocking, honoring the same cursor bounds as
	 * {@link recvGroup}.
	 *
	 * Returns `undefined` when nothing is deliverable right now, which is not by itself the
	 * end of the track: a group may still arrive, or one may be parked beyond the
	 * {@link withRange} cap. Use it to drain what the retained window already holds, where
	 * waiting for a sequence nothing will republish would park forever.
	 */
	tryRecvGroup(): GroupConsumer | undefined {
		this.#live();
		this.#mode = "arrival";
		return taken(
			() => this.#tryRecvGroup(),
			() => this.#ended(),
		);
	}

	// The sequence cursor behind {@link Ordered}, which owns the only public door to it.
	async #nextGroup(): Promise<GroupConsumer | undefined> {
		for (;;) {
			const recv = this.#tryNextGroup();
			switch (recv.kind) {
				case "group":
					return recv.group;
				case "done":
					return this.#ended();
				case "error":
					throw recv.error;
			}
			await this.#wait();
		}
	}

	// A missing group the subscription still owes holds sequence order until it lands,
	// is accounted for, or the same drift budget as a buffered group gives up on it.
	#waitOwed(start: bigint, end: bigint, drift: Drift): boolean {
		const producer = this.#state.producer;
		if (start >= end || !producer) return false;
		// Only a hole in this reader's ranges is owed to it.
		let owed: { sequence: bigint; folds?: Instant } | undefined;
		for (const run of this.#window.runs(start, end)) {
			owed = owedGroup(producer, run.start, run.end);
			if (owed) break;
		}
		if (!owed) {
			// A fill still owes a group below, which lands as a group does. Its staleness deadline
			// wakes the cursor too, so a fill that never answers cannot pin it.
			if (this.#askFills(start, end, drift) === undefined) return false;
			this.#state.waitingOrdered = true;
			return true;
		}
		if (this.#staleAt(owed.sequence, drift, true)) return false;
		if (owed.folds !== undefined) this.#wakeAt(owed.folds.since(Instant.ZERO).asNanos());
		this.#state.waitingOrdered = true;
		return true;
	}

	// The lowest buffered candidate in range that the budget spares. An ordered take may
	// still wait on an owed hole below it, so a resumed reader must keep its place.
	#peekGroup(ordered: boolean): bigint | undefined {
		this.#settle();
		const cursor = this.#cursor.peek();
		const start = ordered ? max(cursor.start, this.#nextSequence) : cursor.start;
		const drift = this.#drift();
		for (const group of this.#state.groups.peek()) {
			if (group.sequence < start || !this.#window.contains(group.sequence)) continue;
			if (cursor.end !== undefined && group.sequence >= cursor.end) return undefined;
			if (!this.#delivered.has(group.sequence) && !this.#staleAt(group.sequence, drift)) {
				return group.sequence;
			}
		}
		return undefined;
	}

	// The synchronous half of #nextGroup.
	#tryNextGroup(): Recv {
		if (this.#lagged) return { kind: "error", error: new TooFarBehind() };
		let recv = this.#resume ? this.#tryResumed(true) : this.#tryNextBuffered();
		const lagged = this.#settleRecord();
		if (lagged) {
			// Nobody takes the group the lag replaced, so it lets go of its source here.
			if (recv.kind === "group") recv.group.close();
			recv = { kind: "error", error: lagged };
		}
		// One cursor: the frame helpers must not keep draining a group this read just
		// moved past, or interleaved reads would run backwards.
		if (recv.kind === "group" && this.#frameGroup && this.#frameGroup.sequence < recv.group.sequence) {
			this.#frameGroup.close();
			this.#frameGroup = undefined;
		}
		return recv;
	}

	#tryNextBuffered(): Recv {
		const refused = this.#settle();
		if (refused) return { kind: "error", error: refused };
		this.#state.waitingOrdered = false;
		this.#fillWait = false;
		const groups = this.#state.groups.peek();
		const cursor = this.#cursor.peek();
		let start = max(cursor.start, this.#nextSequence);
		while (groups.length > 0 && groups[0].sequence < start) groups.shift()?.close();
		// One anchor for the whole pass, so walking a backlog off stays linear.
		const drift = this.#drift();

		for (let index = 0; ; ) {
			const group = groups[index];
			if (!group || (cursor.end !== undefined && group.sequence >= cursor.end)) break;
			// Out of range for now, but kept until the cursor passes it: an anchor still
			// resolving, or an update widening the ranges, may take it in.
			if (!this.#window.contains(group.sequence)) {
				index++;
				continue;
			}
			// Every frame this group could still hold is past the budget, so keep
			// scanning: one pass walks a whole backlog off rather than replaying it.
			if (this.#staleAt(group.sequence, drift)) {
				this.#pass(groups, index);
				// The cursor passed what lies below it, which a fill no longer owes this reader.
				this.#fillsMoved();
				// Holes below a skipped group are undeliverable, even ones the wall clock
				// still spares because their successor arrived late.
				start = this.#nextSequence;
				index = 0;
				group.close();
				continue;
			}
			if (this.#waitOwed(start, group.sequence, drift)) return { kind: "idle" };
			this.#pass(groups, index);
			this.#handOut(group.sequence, true);
			return { kind: "group", group: this.#guard(group) };
		}

		const closed = this.#state.closed.peek();
		if (closed instanceof Error) return { kind: "error", error: closed };
		// A clean close completes this range, retaining groups a later cap raise can read.
		if (closed !== undefined) return { kind: "done" };
		// Nothing buffered: ask a fill for what this reader misses, and wait while one owes it.
		if (this.#askFills(start, cursor.end ?? UNBOUNDED, drift) !== undefined) {
			// Its staleness deadline wakes the cursor, as for any owed hole.
			this.#state.waitingOrdered = true;
			return { kind: "idle" };
		}
		const end = this.#state.end.peek()?.at;
		if (this.#exhausted()) return { kind: "done" };
		if (
			!this.#parked(groups) &&
			!this.#starting(groups, start) &&
			(this.#complete() || (end !== undefined && start >= finalSequence(end)))
		) {
			return { kind: "done" };
		}
		return { kind: "idle" };
	}

	/**
	 * Reads the next frame across groups, in sequence order, with its group and frame numbers.
	 * Treat the returned frame bytes as read-only; they are shared with other consumers.
	 *
	 * Groups are acquired through the same sequence cursor as {@link Ordered.nextGroup},
	 * so frames never run backwards: a late lower-sequence group is skipped, and so is
	 * one every frame of which `delay.max` proves is too old. A group the budget abandons
	 * mid-stall ends cleanly and the cursor resyncs from the next group; a gap inside a
	 * group still surfaces as `Evicted`, {@link TooFarBehind}, or {@link GroupTooLarge}.
	 */
	async #readFrame(): Promise<({ group: bigint; frame: number } & Frame) | undefined> {
		for (;;) {
			if (!this.#frameGroup) {
				this.#frameGroup = await this.#nextGroup();
				if (!this.#frameGroup) return undefined;
			}
			const group = this.#frameGroup;

			let next: ({ sequence: number } & Frame) | undefined;
			try {
				next = await group.readFrameSequence();
			} catch (err) {
				// The group failed underneath us: resync from the next one, surfacing
				// only what the caller can act on (a gap, or the track's own abort).
				this.#frameGroup = undefined;
				group.close();
				if (
					err instanceof TooFarBehind ||
					err instanceof GroupTooLarge ||
					(err instanceof StreamError && err.code === StreamCode.Evicted)
				)
					throw err;
				const closed = this.#state.closed.peek();
				if (closed instanceof Error) throw closed;
				continue;
			}

			if (next) {
				return {
					group: group.sequence,
					frame: next.sequence,
					payload: next.payload,
					timestamp: next.timestamp,
				};
			}

			// The group is exhausted (or the budget abandoned its stall); move on.
			this.#frameGroup = undefined;
			group.close();
		}
	}

	/** Reads the next frame and decodes it as a UTF-8 string. */
	async #readString(): Promise<string | undefined> {
		const next = await this.#readFrame();
		if (!next) return undefined;
		return new TextDecoder().decode(next.payload);
	}

	/** Reads the next frame and parses it as JSON. */
	async #readJson(): Promise<unknown | undefined> {
		const next = await this.#readString();
		if (next === undefined) return undefined;
		return JSON.parse(next);
	}

	/**
	 * Read this track's groups in sequence order instead of arrival order.
	 *
	 * Both cursors draw from the same buffer, so a track is read one way or the other: this
	 * hands the subscription to the returned {@link Ordered} and leaves this handle inert.
	 * {@link recvGroup} throw afterwards. Throws once a
	 * {@link recvGroup} call has already committed the track to arrival order.
	 *
	 */
	ordered(): Ordered {
		if (this.#mode === "ordered") throw new Error("track is already read in sequence order");
		if (this.#mode === "arrival") throw new Error("track is already read in arrival order");
		this.#mode = "ordered";
		return makeOrdered(this);
	}

	/**
	 * Update this subscription's options (e.g. priority), triggering a SUBSCRIBE_UPDATE to the
	 * publisher. Replacing the ranges resolves them again, without redelivering groups already
	 * read; an arrival reader can receive the unread groups it still holds that the new ranges
	 * take in.
	 * Mirrors the Rust `Subscriber::update`.
	 */
	update(options: Subscription) {
		const subscription = subscriptionDefaults(options);
		const replaced = !sameRanges(subscription.ranges ?? [], this.#window.ranges);
		const reordered = subscription.order !== this.#state.update.peek()?.order;
		this.#state.update.set(subscription);
		if (replaced) this.#resolve();
		// A fill's order follows its readers, so its handler wakes to serve the new first group.
		else if (reordered) this.#fillsMoved();
	}
}

/**
 * A {@link Subscriber} that reads groups in sequence order.
 *
 * Created by {@link Subscriber.ordered}, which takes the subscription over: the two
 * cursors draw from one buffer, so a track is read one way or the other and never both.
 * Every group this returns has a higher sequence than the last, so a late arrival is
 * skipped rather than delivered out of turn.
 *
 * `delay.max` 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, since the reach is
 * the only proof that every frame it could still hold is past the budget, so a backlog
 * inside the budget is still delivered whole, as a burst in order. The budget follows a
 * group already handed out too: a stalled group's pending frame read rejects once newer
 * content has pulled that far ahead.
 *
 * A history or VOD read sets `delay: { max: null }` (no limit). A subscribed track waits for a
 * missing group the feed still owes, within the budget, or until it resolves without one.
 * Local tracks skip missing groups. An unlimited or long budget
 * only avoids blocking where the feed signals its gaps: moq-lite, whose publisher sends
 * `SUBSCRIBE_DROP` for every group it won't deliver, or an ordered moq-transport FETCH.
 * Over a moq-transport SUBSCRIBE, which signals no gaps, that wait can last until the
 * track ends.
 */
export class Ordered {
	/** The track name. */
	readonly name: string;

	#subscriber: Subscriber;

	private constructor(subscriber: Subscriber) {
		this.name = subscriber.name;
		this.#subscriber = subscriber;
	}

	static {
		makeOrdered = (subscriber) => new Ordered(subscriber);
	}

	/** Resolve this track's immutable publisher properties; see {@link Subscriber.info}. */
	info(): Promise<Info> {
		return this.#subscriber.info();
	}

	/** Settles once the track closes; see {@link Producer.closed}. */
	get closed(): GetPromise<Error | null> {
		return this.#subscriber.closed;
	}

	/** This subscriber's current options, including defaults and the last {@link update}. */
	get subscription(): Getter<Subscription | undefined> {
		return this.#subscriber.subscription;
	}

	/** The largest location within {@link withRange}; see {@link Subscriber.largest}. */
	largest(): Promise<Position | undefined> {
		return this.#subscriber.largest();
	}

	/** Resolve with the track's exclusive end once known; see {@link Subscriber.finished}. */
	finished(): Promise<Position> {
		return this.#subscriber.finished();
	}

	/**
	 * Limit subsequent reads to these groups and return this reader for chaining; see
	 * {@link Subscriber.withRange}. An ordered read never goes back, so lowering the floor
	 * below a group it already returned offers nothing.
	 */
	withRange(groups: Groups): this {
		this.#subscriber.withRange(groups);
		return this;
	}

	/** Update this subscription's options; see {@link Subscriber.update}. */
	update(options: Subscription) {
		this.#subscriber.update(options);
	}

	/** Close the track (optionally with an error), closing any pending groups. Idempotent. */
	close(abort?: Error) {
		this.#subscriber.close(abort);
	}

	/**
	 * Return the next group with a strictly-greater sequence number than the last returned.
	 *
	 * Late arrivals (sequence at or below the last returned) are silently skipped, as is a
	 * group whose every frame is further behind the live edge than `delay.max` (the default
	 * subscription's zero keeps only what nothing newer has superseded). Honors the bounds set by
	 * {@link withRange}. Like {@link Subscriber.recvGroup}, a read whose ranges all end
	 * finishes once nothing in them is left to deliver. Later reads return `undefined`
	 * immediately too; widening the ranges with {@link update} reads on above the last sequence returned.
	 */
	nextGroup(): Promise<GroupConsumer | undefined> {
		return ordered_.nextGroup(this.#subscriber);
	}

	/**
	 * Take the next group as {@link nextGroup} would, without waiting; see
	 * {@link Subscriber.tryRecvGroup}. Mirrors the Rust `poll_next_group`.
	 */
	tryNextGroup(): GroupConsumer | undefined {
		return taken(
			() => ordered_.tryNextGroup(this.#subscriber),
			() => ordered_.ended(this.#subscriber),
		);
	}

	/**
	 * Read the next frame across groups, in sequence order, with its group and frame numbers.
	 *
	 * Rides the same cursor as {@link nextGroup} and shares this handle's contract: a
	 * buffered backlog is drained in full up to the point `delay.max` proves it useless.
	 * Treat the returned frame bytes as read-only; they are shared with other consumers.
	 */
	readFrame(): Promise<({ group: bigint; frame: number } & Frame) | undefined> {
		return ordered_.readFrame(this.#subscriber);
	}

	/** Read the next frame and decode it as a UTF-8 string. */
	readString(): Promise<string | undefined> {
		return ordered_.readString(this.#subscriber);
	}

	/** Read the next frame and parse it as JSON. */
	readJson(): Promise<unknown | undefined> {
		return ordered_.readJson(this.#subscriber);
	}
}