media-plane 0.10.0

//! [`Trunk`] — the sample ring, [`TrunkWriter`], and [`SampleCursor`] (plan
//! step 3b-i); the segment log, [`SegmentCursor`], and the
//! lossless-by-retention pinning mechanism (plan step 3b-ii); the 90 kHz
//! event log, [`EventCursor`], and [`EventAnchor`] (plan step 3b-iii); the
//! live-part log and the [`Trunk::listen`] reader-wake primitive (plan step
//! 3b-iv), closing the two gaps step 3d found while reading
//! `hls-runtime/src/server/` before writing the egress traits — see
//! [The live-part log](#the-live-part-log-parts-before-their-segment-closes)
//! and
//! [The reader-wake primitive](#the-reader-wake-primitive-listen-not-one-registration-per-remote-peer)
//! below; and now [`SegmentWriter`], splitting the single write handle 3b-i
//! introduced by **ring group** so a segmenter can exist at all — see
//! [One writer per ring group, not one writer per `Trunk`](#one-writer-per-ring-group-not-one-writer-per-trunk)
//! below — per
//! `docs/superpowers/specs/2026-07-26-media-plane-architecture.md` §1.2.
//!
//! This is four bounded rings behind two write handles, split by ring group
//! (see the section linked just above), and the cursors/queries that read
//! them: the sample path, the segment log, the event log, and the live-part
//! log. See
//! [The event log: 90 kHz absolute, and the B1 crux](#the-event-log-90-khz-absolute-and-the-b1-crux)
//! below for why the event log needed a third, genuinely different shape —
//! not just a third copy of `ClassLog`/`SegmentLog` (this module's internal
//! per-class/per-segment logs) — to resolve the architecture audit's
//! blocking finding B1.
//!
//! # One writer per ring group, not one writer per `Trunk`
//!
//! [`Trunk::writer`] used to be the *only* way to publish anything, and its
//! doc said so in terms broader than the actual reason it exists: "a second
//! concurrent writer would silently interleave two unrelated publish
//! sequences into one ring with no way for a reader to tell them apart." That
//! sentence is true, but it over-generalises from "one ring" to "one
//! `Trunk`" — and a real consumer wiring this crate up (a segmenter: it reads
//! samples via a [`SampleCursor`] and, from what it reads, produces segments
//! and parts) hit the gap that over-generalisation opened: with exactly one
//! writer for the entire `Trunk`, whichever component takes it for ingest
//! samples makes it **structurally impossible** for anything else to ever
//! call [`SegmentWriter::publish_segment`] — no segmenter can exist, and a
//! segment log/live-part log that can never be filled is not a real feature,
//! just an unreachable one.
//!
//! **The actual invariant, restated correctly**: *within a given ring, there
//! is exactly one appender* — that is what prevents the interleave the
//! original sentence worried about, because an interleave requires two
//! writers racing to append to the *same* ring. It says nothing about a
//! *different* ring. So the write capability is split by **ring group**,
//! each still taken at most once via the same `compare_exchange` pattern
//! [`Trunk::writer`] always used, just with one flag per group instead of
//! one flag for the whole `Trunk`:
//!
//! - [`TrunkWriter`] (via [`Trunk::writer`]) — the **samples + events**
//!   group: [`TrunkWriter::publish`] (the two [`RetentionClass`] sample
//!   rings) and [`TrunkWriter::publish_event`] (the event ring's only
//!   *appending* operation). Held by the ingest driver — the entity that
//!   actually produces both: a demuxed sample, or an inband SCTE-35/`emsg`
//!   event lifted straight off the incoming stream.
//! - [`SegmentWriter`] (via [`Trunk::segment_writer`]) — the
//!   **segments + parts** group: [`SegmentWriter::publish_segment`],
//!   [`SegmentWriter::publish_part`], plus
//!   [`SegmentWriter::note_segment_start`]/[`SegmentWriter::set_time_anchor`].
//!   Held by whoever owns segmentation.
//!
//! The last two methods are grouped here on purpose, not by accident of
//! naming: neither **appends** an entry to the event ring the way
//! [`TrunkWriter::publish_event`] does — both *resolve an already-stored*
//! [`EventAnchor::Segment`]/[`EventAnchor::Utc`] entry **in place** (see
//! [The event log](#the-event-log-90-khz-absolute-and-the-b1-crux) below).
//! Since neither is an append, giving them to [`SegmentWriter`] does not
//! create a second appender for the event ring — [`TrunkWriter::publish_event`]
//! remains the event ring's only one — while `note_segment_start` in
//! particular *has* to live wherever segmentation lives: only the segmenter
//! knows where a segment boundary actually falls (this is the literal B1 fix
//! — "it cannot be finalised until the segmenter owns a boundary"), so it is
//! the one entity that can honestly report `note_segment_start`. Placing
//! `set_time_anchor` alongside it, rather than on [`TrunkWriter`], keeps this
//! crate's set of "resolve a pending anchor" entry points in one place
//! instead of splitting a single conceptual capability (anchor resolution)
//! across two handles for no test or caller that needs it split further; if
//! a future caller's wall-clock mapping genuinely comes from the ingest side
//! instead, adding it to [`TrunkWriter`] alongside `publish_event` is
//! additive, not a breaking re-split of what is here today.
//!
//! **Still exactly one writer per group, enforced the same way**: both
//! [`Trunk::writer`] and [`Trunk::segment_writer`] return `None` while a
//! previously-issued handle is still alive, via their own `AtomicBool`
//! (released when the handle is dropped, so a source reconnect can re-take
//! it) — two concurrent
//! *sample* writers remain exactly as impossible as before this split; what
//! changed is that a *segment* writer and a *sample* writer are no longer
//! forced to be the same handle.
//!
//! **The cross-ring ordering question this split raises, answered rather
//! than left implicit.** A segment is derived from samples the segmenter has
//! already consumed via its own [`SampleCursor`] — causally, those samples
//! exist first. With one shared writer, that causal fact was also a
//! *program-order* fact (one thread called `publish` some number of times,
//! then called `publish_segment`). With the split, ingest and segmentation
//! are ordinarily two different threads — could a consumer ever observe
//! [`SegmentWriter::publish_segment`]'s entry in the segment log *before* the
//! samples that produced it are visible in the sample ring? **No** — and not
//! by luck: every ring here (`timed`, `sparse`, `segments`, `events`,
//! `parts`) still lives inside the *one* `Mutex<TrunkState>` this module has
//! always used (see [the benchmark verdict](#the-benchmark-verdict-this-design-is-built-around)
//! below) — splitting the *write handle* did not split the *lock*. The
//! segmenter can only have samples to build a segment from because its own
//! `SampleCursor::poll` already returned them, which requires those samples
//! to have been committed to `state.timed`/`state.sparse` under an *earlier*
//! acquisition of that same `Mutex`; `SegmentWriter::publish_segment` is then
//! called afterward, in the segmenter's own program order, under a *later*
//! acquisition of the identical `Mutex`. Any third party that subsequently
//! acquires that lock — to poll any ring, from any thread — is guaranteed by
//! the transitivity of the `Mutex`'s release/acquire ordering to observe at
//! least everything the segmenter itself had already observed before it
//! published, samples included. So the specific ordering a consumer must
//! never see reversed (a segment's constituent samples appearing to lag
//! behind the segment itself) cannot happen. What *is* true, and is exactly
//! the existing [`RetentionClass::Timed`]/[`RetentionClass::Sparse`]
//! precedent extended one layer: the *global* order across unrelated
//! ring-group activity is not fixed by any one thread's program order any
//! more — a new, unrelated sample the ingest thread publishes concurrently
//! may land before or after a segment close the segmenter thread publishes,
//! in either order, depending on which wins the lock race. Nothing
//! downstream needs that unrelated cross-ring interleave to be
//! deterministic (see [Two retention classes](#two-retention-classes-and-why-they-are-two-independent-rings)
//! below for why this crate already treats "no global cross-ring order" as
//! an acceptable, load-bearing property, not a defect) — only the *causal*
//! one, which is what the shared `Mutex` structurally guarantees.
//!
//! **Why the one lock is kept (issue #1082, r06-O2), and what was done
//! instead.** Splitting `TrunkState` into a sample-group lock and a
//! segment/part/event-group lock would have to give up exactly the
//! guarantee above (the shared lock *is* the proof that a segment never
//! overtakes its samples; per-group locks would replace it with an argument
//! about the segmenter's program order plus lock-ordering rules for every
//! method that touches both groups — `Trunk::writer`'s handover log spans
//! the two sample classes, and the `StallIngest` `Condvar` is paired with the
//! segment ring and its pins). The contention is now measured:
//! `benches/trunk_contention.rs` (1 publisher, N spinning [`SampleCursor`]s;
//! numbers and the decision rule in `benches/RESULTS.md`) reads publisher cost
//! per `publish` after the `parking_lot` swap of 38 ns at 1 reader, 125 ns at
//! 4, 574 ns at 16 and 2.6 µs at 64 — the O(N)-in-cursor-count rule the
//! spike found, still present. That crossed the pre-registered SPLIT
//! threshold (`t(16) >= 3 x t(1)`), so a sample-lock / segment-event-part-lock
//! split was implemented and measured: publisher cost at 16 readers went from
//! 574 ns to 1104 ns, far short of the pre-registered >= 30 % gain bar, and it
//! was **reverted**. The benchmark's publisher and readers only touch the
//! sample group, so splitting the *other* rings off the lock cannot relieve
//! it. See `benches/RESULTS.md` for both tables and the reasoning.
//! The measurable cost the finding names — the O(capacity) request scans
//! under that lock — was removed instead: `part_bytes`/`parts_in_segment`
//! are served from a per-segment index and touch only that segment's parts.
//! The event queries (`events_between`/`events_in_segment`) remain linear in
//! [`TrunkConfig::event_capacity`] on purpose: an `Event`'s media time is
//! resolved *in place* after publication (`Segment`/`Utc` anchors), so the
//! ring is not sorted by media time and an ordered index would need
//! re-keying on every resolution; the ring holds SCTE-35-rate events (tens),
//! not per-part entries, and the scan filters before cloning.
//!
//! # Why this module needs `std`, unlike its byte-layer siblings
//!
//! [`crate::byte_stage`], [`crate::byte_tap`], and [`crate::byte_merge`] are
//! `no_std` because each is driven synchronously by a single caller — there is
//! no cross-thread sharing to arrange. `Trunk` is different in kind: one
//! writer thread per ring group (ingest for samples/events; a segmenter for
//! segments/parts, once one exists) and an unbounded set of reader threads
//! (egress, analysis, DVR) must observe the *same* ring concurrently. That
//! needs a
//! shared, lockable interior — `std::sync::Arc` plus a `parking_lot::Mutex` here, matching
//! exactly the shape validated by `spikes/trunk-bench` (§3.1 of the spec).
//! Pulling in a `no_std` spinlock crate just to keep this one module
//! `no_std`-capable was considered and rejected: every real `Trunk` consumer
//! (`IngestSession`, `PushEgress`/`SegmentEgress`/`ServedEgress` impls) is
//! already `std`+`tokio` per the architecture, so there is no `no_std` caller
//! this would ever serve. This module is therefore `#[cfg(feature = "std")]`
//! — `media-plane --no-default-features` builds clean without it, exactly
//! like every other `std`-only corner of the workspace.
//!
//! # The benchmark verdict this design is built around
//!
//! `spikes/trunk-bench` (commit `acdbf3d0`) measured the naive
//! one-`Mutex`-guarded-log shape this module implements: **PASS** at the
//! specced scale (200-track MPTS × 6 readers, 999.97/1000 Mbit/s sustained,
//! publish mean 5.6 µs / p99 44.3 µs against a ~111 µs budget), but it
//! **refuted the original O(1)-fan-out premise** — writer cost is **O(N) in
//! cursor count** (956 ns → 9.98 µs from 1 → 16 readers), because writer and
//! readers all contend one shared `Mutex`.
//!
//! **Consequence, stated where it will actually be read:** see
//! [`Trunk::subscribe`]. A cursor is for a distinct *consumer of the stream*
//! — never one per peer of a one-to-many protocol. There is no tee, no
//! broadcast channel, and no per-consumer queue here, and there will not be
//! one added to chase higher fan-out: fan-out *is* `subscribe()`, and a
//! sample's payload is already [`bytes::Bytes`], so handing a clone to a
//! second, third, or sixteenth reader is a refcount bump, not a copy (see
//! [Zero-copy fan-out](#zero-copy-fan-out-honestly) below). If a route needs
//! to serve hundreds or thousands of peers (LL-HLS, WHEP), it takes **one**
//! cursor here and fans out to its peers itself, at the layer that already
//! has to hold per-peer state (congestion window, pacing epoch, SRTP
//! context) anyway.
//!
//! The segment log added in this step is a **sibling ring behind the same
//! one `Mutex`** (the internal `TrunkState`), not a second lock — a
//! [`SegmentCursor`] contends exactly the lock a [`SampleCursor`] does, so
//! the same O(N)-in-cursor-count rule and the same single-digit-reader
//! guidance apply to it verbatim; see [`Trunk::subscribe_segments`] and
//! [`Trunk::pin_segments`].
//!
//! # Two retention classes, and why they are two independent rings
//!
//! [`RetentionClass::Timed`] (regular-cadence media) and
//! [`RetentionClass::Sparse`] (irregular, semantically-critical entries — an
//! SCTE-35 splice cue, a subtitle sample) are **not** stored in one merged,
//! globally-ordered log. An earlier design considered exactly that: one
//! `VecDeque` in strict publish order, with `Sparse` entries migrated to a
//! small overflow buffer instead of being dropped when the main ring evicted
//! them. It was rejected as needless complexity for a property nothing
//! actually needs: nothing downstream reads a `Trunk` expecting a strict
//! chronological interleave of, say, video samples and SCTE-35 sections —
//! consumers correlate by PTS/DTS themselves, and the *real* requirement (see
//! [`RetentionClass::Sparse`]) is only that `Sparse` retention must never be
//! collateral damage from unrelated `Timed` churn. Two independently
//! capacity-bounded rings give that guarantee *by construction* — a flood of
//! video frames cannot evict a still-live splice cue, because there is
//! nowhere for it to reach it — while a single merged ring would have to
//! re-implement the same isolation by hand (the rejected overflow-buffer
//! design above), for no observable benefit. [`SampleCursor::poll`] merges
//! the two rings only at read time, and documents the (best-effort, not
//! globally-ordered) precedence it uses.
//!
//! # Zero-copy fan-out, honestly
//!
//! **This claim was made falsely on this project before**: an earlier
//! zero-copy fan-out claim was proven only by a test that sliced `Bytes`
//! itself, while the crate under test contained zero `.slice()` calls of its
//! own — i.e. the test manufactured the evidence it was supposed to be
//! checking for. So, stated plainly: **the production path in this module
//! achieves zero-copy fan-out, not only the test.** the internal per-class
//! log stores the [`transmux::Sample`] handed to it; [`SampleCursor::poll`] returns it to
//! a reader via [`Clone::clone`] on the whole `Sample`, which clones
//! `Sample.data: Bytes` through `Bytes`'s own `Clone` impl — an `Arc`-style
//! refcount bump, not a byte copy. There is no `.slice()`, no
//! `Bytes::copy_from_slice`, and no re-allocation anywhere on this path. The
//! test in this module (`payload_is_shared_not_copied_across_cursors`)
//! asserts `Bytes::as_ptr()` *identity* across multiple cursors reading the
//! same published entry, precisely so it cannot be satisfied by two payloads
//! that merely have equal contents — and a mutation swapping the `clone()`
//! for a real copy is recorded as run against it (see that test's doc
//! comment).
//!
//! The segment log added in this step makes the **same** claim, honestly, on
//! the **same** terms: [`SegmentEntry::bytes`] is [`bytes::Bytes`],
//! [`SegmentCursor::poll`] hands it back via `Clone` on the whole
//! [`SegmentEntry`] (which clones `Bytes` through `Bytes`'s own `Clone`), and
//! there is no `.slice()`/`copy_from_slice`/re-allocation anywhere on that
//! path either. `segment_bytes_are_shared_not_copied_across_cursors` asserts
//! the same pointer-identity property for segments that
//! `payload_is_shared_not_copied_across_cursors` asserts for samples — this
//! is the **production** path achieving zero-copy fan-out, not a test
//! manufacturing its own evidence.
//!
//! # The DVR contradiction: losslessness from retention, not back-pressure
//!
//! A DVR/archive consumer must not miss a segment — a hole in a recording is
//! a defect, not a degradation, unlike a dropped video frame. But the writer
//! must **never** block, for exactly the reason stated everywhere else in
//! this module: a stalled archive writer must not stall live ingest. Those
//! two requirements contradict each other directly if "losslessness" is
//! implemented the obvious way — by making the writer wait for a slow
//! archive reader.
//!
//! **The resolution: losslessness comes from retention, not from
//! back-pressure.** A [`SegmentCursor`] obtained via [`Trunk::pin_segments`]
//! *pins* every segment it has not yet consumed — the log will not evict a
//! pinned entry as a matter of course, the way it freely evicts for an
//! ordinary [`Trunk::subscribe_segments`] cursor. "Consumed" here means
//! "returned by [`SegmentCursor::poll`]" — the same progress counter that
//! already governs in-order delivery does double duty as the pin floor,
//! rather than adding a second, explicit acknowledge-after-durable-write API
//! call. That two-call shape (poll to receive, then a separate `ack` once
//! the archive write actually lands on disk) was considered — it is the more
//! conservative choice, since a consumer that has polled a segment but not
//! yet finished writing it to disk is not truly safe from loss if the trunk
//! evicts under it — and rejected for *this* step: it doubles the API
//! surface and the bookkeeping (two offsets per pin instead of one) for a
//! distinction (poll's delivery vs. a durable write landing) this step has
//! no test that needs, since nothing downstream is implemented yet
//! (`docs/superpowers/plans/2026-07-26-media-plane-implementation.md` step
//! 3d's `SegmentEgress`/DVR writer is what would consume it). If that step
//! needs the finer-grained split, it is additive — a second, later
//! acknowledgement point on the same pin — not a breaking change to this
//! one.
//!
//! **Pinning is bounded, and by design there is no second capacity knob for
//! it**: a pin is measured against exactly [`TrunkConfig::segment_capacity`],
//! the same bound that governs ordinary eviction for every cursor. There is
//! no independent "how far behind may a pin fall" setting to tune
//! separately and get wrong. When the segment log is at capacity and the
//! next [`SegmentWriter::publish_segment`] would evict an entry some pin has
//! not yet consumed, the bound has been hit, and something genuinely has to
//! give — the caller decided what, in advance, via the [`ArchiveOverrun`]
//! passed to [`Trunk::pin_segments`]:
//!
//! - [`ArchiveOverrun::Gap`] (**the default**) — evict the pinned entry
//!   anyway, and tell that cursor it lost data
//!   ([`SegmentCursorItem::Gap`]). The recording gets a hole; the live
//!   stream and every other cursor are unaffected.
//! - [`ArchiveOverrun::StallIngest`] — apply real back-pressure:
//!   [`SegmentWriter::publish_segment`] blocks until this cursor consumes
//!   enough to release its pin (or is dropped). **The only place in this
//!   entire design where a reader may block the writer** — opt-in,
//!   documented loudly here and on the variant itself, and never the
//!   default.
//! - [`ArchiveOverrun::Terminate`] — drop the cursor's pin outright instead
//!   of gapping the recording or stalling ingest; the cursor is done
//!   ([`SegmentCursorItem::Terminated`]) and the log continues without it.
//!
//! This is a genuine three-way trade between the recording, the live
//! stream, and the archive consumer — **no option is free**, and there is
//! deliberately no fourth "just make it work" variant: any such variant
//! would have to secretly pick one of the three trade-offs above anyway
//! (drop bytes, block the writer, or drop the consumer), just without
//! naming which — which is worse, not better.
//!
//! # The event log: 90 kHz absolute, and the B1 crux
//!
//! `TrunkState` (the shared state behind a `Trunk`) holds the two per-class
//! sample logs, the segment log, and now the event log — a **sibling ring
//! behind the same one `Mutex`**, exactly the pattern the segment log
//! established (`TrunkState::events: EventLog`, `Trunk::subscribe_events`/
//! `events_between`/`events_in_segment` shaped like `Trunk::subscribe`/
//! `Trunk::subscribe_segments`, `TrunkConfig::event_capacity` alongside
//! `timed_capacity`/`sparse_capacity`/`segment_capacity`). Where it is
//! genuinely a new shape, not a third `ClassLog`/`SegmentLog` copy, is
//! its *clock* and its *addressing* — both forced by architecture audit
//! finding B1
//! (`docs/superpowers/specs/2026-07-26-media-plane-architecture.md` §0/§1.2).
//!
//! **What B1 got wrong.** Revision 1 of the spec claimed one time model for
//! everything the plane carries: an absolute `i64` in the *producing
//! track's* timescale. That is false in two ways this project already
//! parses, and both are events, not samples:
//!
//! - `splice_schedule.utc_splice_time` (SCTE-35 §9.7.4) is **GPS-epoch
//!   UTC** — not a media timestamp in any track's timescale at all.
//! - `emsg` version 0's `presentation_time_delta` (ISO/IEC 23009-1
//!   §5.10.3.3) is **segment-relative** — its value only means something
//!   once you know which segment it lands in, and that segment's earliest
//!   presentation time is not knowable until the segmenter has actually cut
//!   the boundary. `timed_metadata::convert::emsg_convert` already encodes
//!   this exact arithmetic (`T = EPT + presentation_time_delta`) for
//!   *converting* one emsg to another; the event log's job is different —
//!   it has to hold the delta *honestly unresolved* for however long the
//!   boundary is unknown, which a stateless conversion function has no
//!   reason to model.
//!
//! Neither of those is expressible as a single struct field without either
//! (a) losing information (which timescale? relative to what?) or (b)
//! **fabricating** a resolution that has not actually happened yet — an
//! event log that stores a plausible-looking media time for a
//! `splice_schedule` cue before any wall-clock↔media-clock mapping exists,
//! or for an `emsg` v0 before its segment's start is known, has invented
//! data. **The failure mode is not a crash: it is an ad break firing at the
//! wrong wall-clock instant**, because a plausible-but-wrong media time is
//! indistinguishable from a correct one until playout.
//!
//! **Why 90 kHz absolute, not per-track timescale.** A single `Media` can
//! carry several tracks at several timescales (48 kHz audio, a 25 fps
//! video track at 90 000, a subtitle track with none at all) — there is no
//! one track whose timescale the *event* log could borrow without an
//! arbitrary, undocumented choice among them. [`EventAnchor::Media`]
//! therefore carries [`timed_metadata::MediaTime`] — 90 kHz ticks,
//! wrap-unrolled, the same clock SCTE-35's own `pts_time` already uses —
//! rather than any one track's clock. This is also why the event log is a
//! genuinely separate ring from the sample rings, not a third
//! [`RetentionClass`]: a [`transmux::Sample`] is timestamped in its
//! *track's* clock ([`transmux::Sample::pts`]/`dts`, per §4 of the spec);
//! an event lives on the trunk's own, track-independent clock.
//!
//! **Carries [`timed_metadata::TimedEvent`], not a parallel type.** It is
//! owned, lossless, `#[non_exhaustive]`, and already published (0.4.0, live
//! on crates.io) — [`EventEntry::event`] stores it verbatim rather than
//! re-deriving a second event representation this crate would then have to
//! keep in sync by hand. `mp4_emsg::EmsgBox<'a>` is *borrowed* and cannot
//! outlive the buffer it was parsed from, so it cannot sit in a `'static`
//! ring; [`timed_metadata::SourcePayload::Emsg`] is already its owned form
//! (scheme/value/verbatim `message_data`), and is what ends up inside the
//! stored `TimedEvent` for an `emsg`-sourced entry.
//!
//! **The B1 crux: [`EventAnchor`] — an unresolved event stays honestly
//! unresolved.** Every entry's addressability is one of three states, and
//! there is deliberately no path from `Segment`/`Utc` to `Media` other than
//! the specific fact each one is waiting for actually arriving:
//!
//! - [`EventAnchor::Media`] — already on the trunk's 90 kHz clock (a
//!   `splice_time` PTS post-wrap-unroll, or an already-absolute `emsg` v1).
//! - [`EventAnchor::Segment`] — an `emsg` v0's `presentation_time_delta`
//!   plus the `segment_number` it is relative to. Stays exactly this
//!   variant — addressable by segment number, **not** by media time —
//!   until [`SegmentWriter::note_segment_start`] reports that segment's
//!   start, at which point this module's internal event log resolves it
//!   **in place**, computed from *that segment's own* reported start —
//!   never "whichever segment happens to be currently open", which would
//!   silently produce *a* segment instead of *the* segment the emsg
//!   actually named.
//! - [`EventAnchor::Utc`] — a GPS/UTC instant (`splice_schedule`) with no
//!   media-timeline position at all. Stays exactly this variant — not
//!   returned by [`Trunk::events_between`] or [`Trunk::events_in_segment`],
//!   because there is no honest media time to filter on — until
//!   [`SegmentWriter::set_time_anchor`] gives the event log a
//!   [`timed_metadata::TimeAnchor`] to translate through. This is the
//!   literal B1 test: an event with only a wall-clock time and no anchor
//!   must never be handed a fabricated media time.
//!
//! `epoch_ms_to_media` (the UTC→media direction) is the mirror image of
//! [`timed_metadata::TimeAnchor::media_to_epoch_ms`] (which only goes the
//! other way) — plain affine algebra, **not** a reimplementation of
//! [`timed_metadata::Timeline`]'s 33-bit wrap-unroll, which this module
//! reuses rather than hand-rolls: every `MediaTime` this ring ever stores
//! either came out of `Timeline::push_scte35` already unrolled, or is
//! computed from one that did (`Segment`/`Utc` resolution only ever adds a
//! non-negative delta or an anchor-relative offset to an already-unrolled
//! value).
//!
//! **Dual addressing: media time *and* segment, both, not either** — because
//! a manifest renderer needs "the events in segment N" while a playback
//! scheduler needs "the events between T1 and T2", and neither is a special
//! case of the other. [`Trunk::events_between`] answers the first
//! (half-open `[from, to)` over every currently-`Media`-resolved entry);
//! [`Trunk::events_in_segment`] answers the second, by consulting
//! `EventLog::segment_starts` — a small boundary table, populated by
//! [`SegmentWriter::note_segment_start`], bounded by the **same**
//! `TrunkConfig::event_capacity` rather than a second, independent knob
//! (exactly [`TrunkConfig::segment_capacity`]'s "no second capacity knob"
//! precedent for pinning). Both queries only ever return `Media`-resolved
//! entries — an entry still `Segment`/`Utc`-anchored is not fabricated a
//! position just to satisfy either query.
//!
//! Both point-in-time queries read the same log a subscribed
//! [`EventCursor`] does (via [`Trunk::subscribe_events`]) — the same
//! single-`Mutex`, single-digit-reader-by-design, in-band-loss-reporting,
//! writer-never-blocks shape [`Trunk::subscribe`]/[`Trunk::subscribe_segments`]
//! already established, reused verbatim rather than reconsidered: an
//! `EventCursor` sees an entry (and a `Lagged` loss report, if it fell
//! behind [`TrunkConfig::event_capacity`]'s eviction) the moment it is
//! published, whether or not it has resolved yet, while the two query
//! methods are a snapshot of what has resolved *so far*.
//!
//! `SegmentEgress` and tiered `Retention` (plan steps 3d/3e — an egress
//! trait that owns one [`SegmentCursor`] and pushes to DVR/MABR/ROUTE/Smooth,
//! and a hot/cold archive store behind it) are **not** built here, and their
//! attachment point is exactly [`Trunk::pin_segments`]: a `SegmentEgress`
//! implementation is the caller this step's [`ArchiveOverrun`] was written
//! for — it takes a pinning cursor with whichever policy its durability
//! contract requires (`StallIngest` for "this archive must never have a
//! hole", `Gap` for "best-effort is fine"), drains [`SegmentCursor::poll`],
//! and writes [`SegmentEntry::bytes`] to its store. Nothing in this step's
//! shape needs to change to make room for that; it is exactly the sample
//! path's `PushEgress`-owns-one-`SampleCursor` story repeated one layer up.
//! (`SegmentEgress`/`Retention` are named here only to document the
//! attachment point per this step's brief — neither type exists in this
//! crate yet.)
//!
//! # The live-part log: parts before their segment closes
//!
//! Step 3d built `ServedEgress`/`EgressResponse::Await` and, per its own
//! brief, read `hls-runtime/src/server/` before finishing to report what
//! did **not** fit. It found the segment log alone cannot serve LL-HLS at
//! all: RFC 8216bis's entire low-latency mechanism is **part-level**
//! availability ("does part 3 of the segment currently being written
//! exist"), and before this step there was nowhere in this `Trunk` to ask
//! that — the segment log holds only *finished* segments. This step adds a
//! fourth ring, the live-part log (`TrunkState::parts: PartLog`), storing
//! [`PartEntry`] exactly the way this module's internal `SegmentLog` stores [`SegmentEntry`] —
//! same evict-then-push shape, same zero-copy-fan-out claim (see
//! [Zero-copy fan-out](self#zero-copy-fan-out-honestly)) — bounded by the
//! new [`TrunkConfig::part_capacity`].
//!
//! **Addressed the way a client actually asks**: not a moving cursor
//! position, but a direct `(segment_number, part_index)` key —
//! [`Trunk::part_bytes`] and [`Trunk::parts_in_segment`], the live-part
//! counterparts of [`Trunk::events_between`]/[`Trunk::events_in_segment`].
//! This is deliberate, not an oversight of "should there also be a
//! `PartCursor`": a `ServedEgress` implementing LL-HLS resolves *random*
//! requests ("is part 3.2 ready") against whatever is currently true, not a
//! sequential stream of every part ever produced — exactly the same
//! resolve-a-request-against-shared-state shape
//! [`crate::egress::ServedEgress::resolve`]'s own module doc already argues
//! for the event log's snapshot queries. No `PartCursor` is added because no
//! test in this step (or in `crate::egress`) needs one; streaming every part
//! as it is produced (a hypothetical future low-latency `PushEgress`) is
//! additive later, not a gap today.
//!
//! **What happens when the parent segment closes — decided, not left
//! implicit**: [`SegmentWriter::publish_segment`] does **not** touch the
//! live-part log at all. A part stays addressable via [`Trunk::part_bytes`]
//! for exactly as long as [`TrunkConfig::part_capacity`]'s ordinary
//! evict-oldest bound has not yet reclaimed it — whether its parent segment
//! is still open or has already closed makes no difference to this ring.
//! Three alternatives were considered and rejected:
//!
//! - *Roll a closed segment's parts into its [`SegmentEntry`]* — rejected:
//!   `SegmentEntry::bytes` is already the whole muxed segment; attaching its
//!   parts too would store the same encoded media twice (once whole, once
//!   split), the opposite of this crate's zero-copy-fan-out discipline, for
//!   a property ([`Trunk::part_bytes`] already answers "is this part ready")
//!   nothing needs.
//! - *Evict a segment's parts the instant it closes* — rejected: this is
//!   the exact bug `hls_runtime::server::MediaStore`'s own `recent_parts`
//!   buffer exists to prevent (documented there as "the segmenter emits a
//!   segment's final part and closes the segment in the same pipeline
//!   step... without this the part is evicted microseconds after it
//!   appears — before the blocked part request can wake"). A `ServedEgress`
//!   built on this `Trunk` needs the same guarantee, and immediate eviction
//!   on close would remove it.
//! - *A second, shorter-lived "recently closed" bound, chained after the
//!   live bound* — this is what `MediaStore` actually does
//!   (`live_parts` + a separately-capped `recent_parts`, doubling worst-case
//!   retention) — rejected here as the "second knob" this file's precedent
//!   argues against: one bound, applied uniformly regardless of open/closed
//!   status, gives the same client-visible guarantee (a just-closed part
//!   stays fetchable) without a second, independently-tunable lifetime that
//!   can disagree with the first.
//!
//! **What a client requesting a just-rolled part receives**: the same
//! answer as the instant before the segment closed — `Some(bytes)` from
//! [`Trunk::part_bytes`] — because closing did not touch this ring. It
//! becomes `None` only once ordinary `part_capacity` eviction reclaims it,
//! at which point this ring cannot distinguish "evicted" from "never
//! existed"; a `ServedEgress` wanting RFC 8216bis's sharper "will never
//! exist, stop waiting" signal for a part of an *already-closed* segment
//! (`hls_runtime::server::MediaStore::resolve_resource`'s
//! `ResourceOutcome::NotFound` case) gets that distinction the same way
//! `MediaStore` itself does: by also consulting [`Trunk::last_closed_segment`]
//! — if the requested part's `segment_number` is at or before that value
//! and [`Trunk::part_bytes`] answers `None`, the part will never arrive; if
//! it is beyond it, the segment (and the part) may still be produced.
//!
//! # The reader-wake primitive: `listen`, not one registration per remote peer
//!
//! The second gap step 3d found, recorded rather than solved: every `Trunk`
//! reader was a synchronous, non-blocking `poll()`, so a `ServedEgress`
//! implementing RFC 8216bis §6.2.5.2 blocking reload had nothing to wait
//! *on* — only a poll-with-backoff loop. [`Trunk::listen`] closes that gap
//! by handing back a [`ProgressListener`] wrapping
//! [`event_listener::EventListener`] — the exact runtime-agnostic primitive
//! `hls_runtime::server::MediaStore::listen` already returns (an already
//! std+`event-listener`-feature dependency of this crate's sibling, and now
//! of this one), not a hand-rolled parallel mechanism, so a caller ports
//! mechanically: `.await` it under any executor, or call
//! [`ProgressListener::wait_deadline`] with no executor at all — precisely
//! `MediaStore::listen`'s own two documented ways to wait.
//!
//! **The writer never blocks on this.** [`SegmentWriter::publish_part`]/
//! [`SegmentWriter::publish_segment`] call `Event::notify(usize::MAX)`, which
//! wakes every currently-registered listener without waiting for any of
//! them to actually resume running — the same non-blocking-producer
//! guarantee this module makes everywhere else
//! ([`TrunkWriter::publish`]'s doc), extended to a wake channel instead of a
//! data ring. A registered [`ProgressListener`] that nobody ever polls or
//! waits on again (a vanished HTTP peer, a wedged executor) costs the
//! writer nothing beyond that one `notify` call's O(waiter-count) fan-out —
//! it never becomes a wait.
//!
//! **Bounded, and reusing [`TrunkConfig::part_capacity`] rather than a sixth
//! knob.** [`Trunk::listen`] refuses (`None`) once `part_capacity`
//! concurrent [`ProgressListener`]s are outstanding. This is deliberately
//! **not** sized "one registration per remote viewer": that would repeat
//! exactly the O(N)-in-cursor-count mistake [`Trunk::subscribe`]'s own docs
//! warn against for data cursors, now for wake registrations instead of
//! poll positions. The intended shape mirrors `subscribe`'s "one cursor per
//! distinct consumer, never one per peer" rule: a `ServedEgress` adapter
//! serving a thousand LL-HLS viewers takes **one** (or a small, fixed
//! number of) [`Trunk::listen`] registration(s) for the route and fans the
//! single wake-up out to its own thousand blocked HTTP handlers itself,
//! using its own broadcast mechanism — exactly the same layering
//! [`crate::egress::PushEgress`] already requires for sample fan-out. Under
//! that shape, `part_capacity`-many concurrent *distinct-consumer*
//! registrations is generous headroom, not a production ceiling; if a
//! caller instead wires one HTTP request directly to one `Trunk::listen`
//! call each (mirroring how `MediaStore`'s single, uncapped `Event` is used
//! today), the cap is exactly the backstop this step exists to add — a
//! caller hitting `None` must treat it the same way it treats an already-
//! expired [`crate::egress::AwaitPolicy`]: answer the request as
//! unavailable now rather than waiting with no slot to wait in. Composing
//! with [`crate::egress::AwaitPolicy`]'s deadline is the caller's
//! conversion of `AwaitPolicy::deadline` (a [`Timestamp`]) to the
//! `std::time::Instant` it already anchors that `Timestamp` to, passed to
//! [`ProgressListener::wait_deadline`] (or wrapped in the caller's own
//! executor timeout around the `Future` impl) — see
//! [`ProgressListener::wait_deadline`]'s own doc.

use std::collections::{HashMap, VecDeque};
use std::future::Future;
use std::num::NonZeroUsize;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use transmux::TrackId;

use parking_lot::{Condvar, Mutex, MutexGuard};

use crate::ids::{PartIndex, SegmentSeq};
use broadcast_common::stage::Timestamp;
use bytes::Bytes;
use event_listener::{Event, EventListener, Listener};
use thiserror::Error;
use timed_metadata::{MediaTime, PTS_HZ, TimeAnchor, TimedEvent};

/// Re-exported so a caller of [`Trunk::waiter_slot_freed`] can name (and
/// `await`) the returned slot-freed listener without a direct `event-listener`
/// dependency of its own.
pub use event_listener::EventListener as SlotFreedFuture;
use transmux::{Sample, SegmentMeta, TrackSpec};

/// Which retention discipline a published entry follows once inside the
/// [`Trunk`]'s sample ring.
///
/// Named `RetentionClass`, not `Retention` — plan step 3e's tiered hot/cold
/// archive policy (`docs/superpowers/plans/2026-07-26-media-plane-implementation.md`)
/// owns the name `Retention` for an unrelated, later concept. This is the
/// orthogonal, in-ring question of "how eagerly can this entry be evicted",
/// decided per [`TrunkWriter::publish`] call by whoever is feeding the
/// writer — it reflects a *track's* nature (video/audio vs. an SCTE-35
/// section PID), not something intrinsic to a [`transmux::Sample`] itself,
/// so it is not a field the spec's `Sample` type carries.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum RetentionClass {
    /// Regular-cadence media samples (audio, video, ...): count-bounded, and
    /// ordinary eviction is reported to a lagging [`SampleCursor`] as
    /// [`SampleCursorItem::Lagged`] — a consumer that misses a video frame is
    /// gapped, not wrong; it resumes from the next sample.
    Timed,
    /// Irregular, semantically-critical entries — an SCTE-35 splice cue, a
    /// subtitle sample — where losing one leaves a consumer's *derived
    /// state* wrong, not merely gapped: a missed splice cue means splicing
    /// in the wrong place, or not at all.
    ///
    /// # The retention rule
    ///
    /// A `Sparse` entry lives in a ring bounded **independently** of the
    /// `Timed` ring ([`TrunkConfig::sparse_capacity`], separate from
    /// [`TrunkConfig::timed_capacity`]). It is therefore never evicted
    /// "merely because a time window rolled" on the unrelated `Timed`
    /// class: no volume of video/audio publishes can push a still-live
    /// splice cue out of the trunk, because `Timed` publishes never touch
    /// the `Sparse` ring at all. A `Sparse` entry is only evicted once
    /// `Sparse` publish volume *itself* exceeds the `Sparse` ring's own
    /// bound — and when that happens, [`SampleCursor::poll`] reports it as
    /// [`SampleCursorItem::Degraded`], not ordinary `Lagged`: a distinct,
    /// stronger signal, because the consumer's semantic state (e.g. "where
    /// the next ad break splices") is now wrong. A consumer that sees
    /// `Lagged` should simply resume from the next sample; a consumer that
    /// sees `Degraded` should treat its derived state as unsynchronised
    /// until the next authoritative signal (a fresh cue, a manifest
    /// reload) re-establishes it — resuming silently would splice on stale
    /// information.
    Sparse,
}

/// Construction parameters for a [`Trunk`].
///
/// # Why every capacity is a [`NonZeroUsize`], not a validated `usize`
///
/// A zero capacity is not a value this type rejects — it is a value this type
/// **cannot represent**. Every ring in this module evicts its oldest entry
/// when `entries.len() == capacity`, so a zero capacity would evict every
/// entry the instant it was pushed, and a zero waiter cap would make
/// [`Trunk::listen`] incapable of ever registering anybody: not a
/// configuration, a broken one.
///
/// Two weaker designs were considered and rejected:
///
/// - **Panicking on zero in [`Trunk::new`]** (what this type did before):
///   internally consistent, but a library that panics on a value which
///   arrives *from a file* is a real operational hazard, not a style
///   question — `multimux` takes its routes from a JSON config, so once
///   these capacities become operator-configurable a stray `0` would take
///   down the server process instead of producing a config error. It also
///   contradicted `transmux::ProgressiveDemux::new`'s deliberate
///   panic-to-fallible change, which is exactly the kind of
///   two-crates-apart inconsistency that makes an API feel arbitrary.
/// - **A fallible `TrunkConfig::new -> Result<Self, _>`** (the
///   `ProgressiveDemux` shape): correct, but strictly worse *here*.
///   `ProgressiveDemux` already returns `Result` as part of its `Stage`
///   contract and already has an `Error` type; `TrunkConfig` has neither, so
///   this would mean inventing a construction error type and threading
///   `?`/`unwrap` through every construction site to encode one bit of
///   information the type system can carry for free. `NonZeroUsize` puts
///   the invariant *in the signature*, where a reader learns it without
///   reading this doc — and, for the JSON-config hazard specifically, a
///   `serde` deserialize of `0` into a `NonZeroUsize` field already fails as
///   an ordinary deserialization error at the config boundary, with no
///   hand-written check and no panic.
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub struct TrunkConfig {
    /// Bound, in entry count, on the [`RetentionClass::Timed`] ring.
    pub timed_capacity: NonZeroUsize,
    /// Bound, in entry count, on the [`RetentionClass::Sparse`] ring —
    /// independent of `timed_capacity`; see [`RetentionClass::Sparse`] for
    /// why that independence is the entire point of the retention rule.
    pub sparse_capacity: NonZeroUsize,
    /// Bound, in entry count, on the segment log. **Also** the bound a
    /// pinning [`SegmentCursor`]'s retention is measured against — see
    /// [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure)
    /// for why there is deliberately no second, independent "pin depth"
    /// knob.
    pub segment_capacity: NonZeroUsize,
    /// Bound, in entry count, on the event log — **and** on its segment
    /// boundary table (`EventLog::segment_starts`). See
    /// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux)
    /// for why a segment-relative event's target boundary shares this one
    /// knob rather than getting a second, independently-tuned one —
    /// exactly [`TrunkConfig::segment_capacity`]'s "no second capacity
    /// knob" precedent for pinning.
    pub event_capacity: NonZeroUsize,
    /// Bound, in entry count, on the live-part log (step 3b-iv) — **and**
    /// on how many concurrent [`Trunk::listen`] registrations this trunk
    /// will honor at once. See
    /// [The live-part log](self#the-live-part-log-parts-before-their-segment-closes)
    /// for why a part-of-the-open-segment shares this one knob for both
    /// jobs, rather than getting a second, independently-tuned "how many
    /// waiters" setting — the third instance of this file's "no second
    /// capacity knob" precedent (after [`TrunkConfig::segment_capacity`]'s
    /// pin reuse and [`TrunkConfig::event_capacity`]'s `segment_starts`
    /// reuse).
    pub part_capacity: NonZeroUsize,
}

impl TrunkConfig {
    /// Build a config with all five ring capacities. Nothing is validated
    /// here, and nothing needs to be: [`NonZeroUsize`] makes the only
    /// invalid value unrepresentable rather than merely rejected — see
    /// [this type's own docs](TrunkConfig#why-every-capacity-is-a-nonzerousize-not-a-validated-usize)
    /// for why that beats both the panic this replaced and a fallible
    /// constructor.
    pub fn new(
        timed_capacity: NonZeroUsize,
        sparse_capacity: NonZeroUsize,
        segment_capacity: NonZeroUsize,
        event_capacity: NonZeroUsize,
        part_capacity: NonZeroUsize,
    ) -> Self {
        TrunkConfig {
            timed_capacity,
            sparse_capacity,
            segment_capacity,
            event_capacity,
            part_capacity,
        }
    }
}

/// One retention class's bounded, append-ordered log of `(track_id, Sample)`
/// entries.
///
/// Bench-identical bounding: when full, the oldest entry is evicted and
/// `base` (the count of entries ever evicted from *this* log) advances by
/// one; `published` is the count of entries ever pushed. A cursor's lag for
/// this class is computed purely from `base` vs. how much of it the cursor
/// has consumed — see [`SampleCursor::poll`].
struct ClassLog {
    entries: VecDeque<(TrackId, Sample)>,
    base: u64,
    published: u64,
    capacity: usize,
}

impl ClassLog {
    fn new(capacity: usize) -> Self {
        ClassLog {
            entries: VecDeque::with_capacity(capacity),
            base: 0,
            published: 0,
            capacity,
        }
    }

    /// Push one entry, evicting the oldest if the log is already at
    /// `capacity`. Never rejects, never blocks — this is what lets
    /// [`TrunkWriter::publish`] complete unconditionally regardless of how
    /// far behind any reader has fallen.
    fn push(&mut self, track_id: TrackId, sample: Sample) {
        if self.entries.len() == self.capacity {
            self.entries.pop_front();
            self.base += 1;
        }
        self.entries.push_back((track_id, sample));
        self.published += 1;
    }
}

/// Bound on [`WriterHandoverLog`]'s backlog — deliberately small and fixed:
/// a hand-off happens once per source reconnect, not per sample, so this is
/// nowhere near the hot path any other ring in this module bounds. A cursor
/// that falls more than this many reconnects behind before polling again
/// cannot report the exact discontinuity that was evicted — but it is never
/// silently swallowed: [`SampleCursor::poll`] reports it as ordinary
/// `Lagged` instead (this crate's existing "you fell behind, resynchronise"
/// signal), so the loss stays visible in-band rather than disappearing.
const WRITER_HANDOVER_LOG_CAPACITY: usize = 32;

/// One [`TrunkWriter`] hand-off: the `Timed`/`Sparse` publish position (each
/// class's [`ClassLog::published`] count) at the moment a *new* writer was
/// issued after a previous one was dropped — see
/// [`Trunk::writer`]. Recorded so an already-subscribed [`SampleCursor`] can
/// report the hand-off in-band, positioned exactly between the old writer's
/// last sample and the new writer's first: the same technique
/// [`SampleCursorItem::Lagged`]/[`Degraded`](SampleCursorItem::Degraded)
/// already use (a position counter compared against the cursor's own
/// progress, not a side channel a cursor could forget to check).
#[derive(Debug, Clone, Copy)]
struct WriterHandover {
    timed_at: u64,
    sparse_at: u64,
}

/// Bounded, append-ordered log of `WriterHandover` markers — identical
/// evict-then-push shape to [`ClassLog`]/[`SegmentLog`]/[`EventLog`]; `base`/
/// `published` mean exactly the same thing here as there.
struct WriterHandoverLog {
    entries: VecDeque<WriterHandover>,
    base: u64,
    published: u64,
    capacity: usize,
}

impl WriterHandoverLog {
    fn new(capacity: usize) -> Self {
        WriterHandoverLog {
            entries: VecDeque::with_capacity(capacity),
            base: 0,
            published: 0,
            capacity,
        }
    }

    fn push(&mut self, handover: WriterHandover) {
        if self.entries.len() == self.capacity {
            self.entries.pop_front();
            self.base += 1;
        }
        self.entries.push_back(handover);
        self.published += 1;
    }
}

/// One finished media segment recorded by the segment log, in playlist
/// order.
///
/// Reuses [`transmux::SegmentMeta`] for exactly what it already models — the
/// per-segment discontinuity bit [`transmux::Segmenter::take_ready_with_meta`]
/// returns — by holding the whole type rather than copying its one field out
/// into a `discontinuous: bool` of this struct's own; a field `SegmentMeta`
/// gains later is picked up here for free. It does **not** fit whole,
/// though, and this struct says so rather than pretending it does: nothing
/// in `transmux` computes a segment's wall-clock duration, its `moof`/`mfhd`
/// sequence number, or its position on *this trunk's* absolute timeline —
/// those are properties of the log a segment lands in, not of the segmenter
/// that produced its bytes, so they are new fields here, supplied by
/// whoever is feeding [`SegmentWriter::publish_segment`], exactly as
/// `track_id`/[`RetentionClass`] are supplied by whoever feeds
/// [`TrunkWriter::publish`].
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct SegmentEntry {
    /// The segment's encoded bytes. `Bytes`, not `Vec<u8>`, for the same
    /// reason as [`transmux::Sample::data`]: fan-out to every
    /// [`SegmentCursor`] reading this entry is a refcount bump, not a copy —
    /// see [Zero-copy fan-out](self#zero-copy-fan-out-honestly).
    pub bytes: Bytes,
    /// This segment's `moof`/`mfhd` sequence number (1-based, matching
    /// [`transmux::Segmenter`]'s own numbering) — what a consumer needs to
    /// name the segment in a playlist or manifest.
    pub sequence_number: SegmentSeq,
    /// This segment's duration, wall-clock — what a consumer needs for
    /// `#EXTINF`/`<S d="...">`.
    pub duration: Duration,
    /// This segment's start position on the trunk's absolute timeline.
    pub timeline_position: Timestamp,
    /// The discontinuity bit from the segmenter itself; see this struct's
    /// own doc for why it is reused by embedding the whole type, not
    /// re-derived as a field of this struct.
    pub meta: SegmentMeta,
}

impl SegmentEntry {
    /// Build one segment log entry.
    pub fn new(
        bytes: impl Into<Bytes>,
        sequence_number: SegmentSeq,
        duration: Duration,
        timeline_position: Timestamp,
        meta: SegmentMeta,
    ) -> Self {
        SegmentEntry {
            bytes: bytes.into(),
            sequence_number,
            duration,
            timeline_position,
            meta,
        }
    }
}

/// The caller-chosen policy for what happens when a **pinning**
/// [`SegmentCursor`] (from [`Trunk::pin_segments`]) has not yet consumed an
/// entry the segment log needs to evict because it is at
/// [`TrunkConfig::segment_capacity`].
///
/// See [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure)
/// for why this is a three-way, caller-chosen trade with no free option and
/// deliberately no fourth "just make it work" variant.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum ArchiveOverrun {
    /// Evict the pinned entry anyway, and report the loss to this cursor as
    /// [`SegmentCursorItem::Gap`] on its next [`SegmentCursor::poll`]. The
    /// recording gets a hole; live ingest and every other cursor are
    /// unaffected. **The default** — a pinning cursor that does not choose
    /// otherwise gets availability over completeness, the same trade
    /// [`RetentionClass::Timed`]'s ordinary `Lagged` already makes for the
    /// sample ring.
    Gap,
    /// Apply real back-pressure: [`SegmentWriter::publish_segment`] blocks
    /// until this cursor consumes far enough to release its pin (or the
    /// cursor is dropped). **The only place in this entire design where a
    /// reader may block the writer** — opt-in only, never the default;
    /// choosing it means a wedged or malicious archive consumer can stall
    /// segment publication indefinitely.
    StallIngest,
    /// Drop this cursor's pin outright instead of gapping the recording or
    /// stalling ingest: the cursor is terminated (its next `poll` returns
    /// [`SegmentCursorItem::Terminated`], and every `poll` after that
    /// returns `None`) and the log continues without it.
    Terminate,
}

impl Default for ArchiveOverrun {
    /// [`ArchiveOverrun::Gap`] — see that variant's doc for why gapping the
    /// recording, rather than stalling ingest, is the safe default.
    fn default() -> Self {
        ArchiveOverrun::Gap
    }
}

/// Per-pinning-cursor bookkeeping the segment log consults, at each
/// [`SegmentWriter::publish_segment`], to decide whether evicting the oldest
/// entry is safe.
struct PinState {
    /// This pin's own read progress: the same role [`SampleCursor`]'s local
    /// `*_consumed` fields play, made visible to the *writer* instead of
    /// staying purely cursor-local, because eviction has to consult it
    /// *before* evicting, not merely report loss after the fact.
    /// "Acknowledged" (module docs) means "returned by
    /// [`SegmentCursor::poll`]" — see the module docs' DVR section for why a
    /// separate ack-after-durable-write step was considered and rejected
    /// for this step.
    consumed: u64,
    /// The policy chosen at [`Trunk::pin_segments`] time.
    policy: ArchiveOverrun,
    /// Set once [`ArchiveOverrun::Terminate`] has fired for this pin; the
    /// next `poll` on the owning cursor reports
    /// [`SegmentCursorItem::Terminated`] and removes this entry.
    terminated: bool,
}

/// The segment log: a bounded, append-ordered log of [`SegmentEntry`]
/// values, plus the pin bookkeeping [`ArchiveOverrun`] needs.
///
/// Evict-then-push shape identical to [`ClassLog`] — `base`/`published`
/// mean exactly the same thing here as there — with one addition: a publish
/// that would evict an entry a pinning cursor has not yet consumed does not
/// evict unconditionally; [`SegmentWriter::publish_segment`] consults that
/// pin's [`ArchiveOverrun`] first.
struct SegmentLog {
    entries: VecDeque<SegmentEntry>,
    base: u64,
    published: u64,
    capacity: usize,
    pins: HashMap<u64, PinState>,
    next_pin_id: u64,
}

impl SegmentLog {
    fn new(capacity: usize) -> Self {
        SegmentLog {
            entries: VecDeque::with_capacity(capacity),
            base: 0,
            published: 0,
            capacity,
            pins: HashMap::new(),
            next_pin_id: 0,
        }
    }

    /// Unconditional evict-then-push — exactly [`ClassLog::push`]'s shape.
    /// [`ArchiveOverrun`] handling against `pins` happens *before* this is
    /// called; see [`SegmentWriter::publish_segment`].
    fn push(&mut self, entry: SegmentEntry) {
        if self.entries.len() == self.capacity {
            self.entries.pop_front();
            self.base += 1;
        }
        self.entries.push_back(entry);
        self.published += 1;
    }
}

/// How one [`EventEntry`] is currently addressable on the trunk's 90 kHz
/// absolute clock ([`timed_metadata::MediaTime`]) — the distinction
/// architecture-audit finding B1 exists to make honest. See
/// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux) for
/// why these three states cannot be collapsed into one `MediaTime` without
/// reintroducing B1's silent-wrong-instant failure.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum EventAnchor {
    /// Already expressible on this trunk's 90 kHz absolute clock — a
    /// SCTE-35 `splice_time` PTS after [`timed_metadata::Timeline`]'s
    /// 33-bit wrap-unroll, or an `emsg` v1 (already-absolute)
    /// `presentation_time` on this same clock. The only variant
    /// [`Trunk::events_between`]/[`Trunk::events_in_segment`] can ever
    /// match against.
    Media(MediaTime),
    /// Segment-relative (`emsg` v0's `presentation_time_delta`, ISO/IEC
    /// 23009-1 §5.10.3.3): this event's media time is `delta` ticks after
    /// the *start* of segment `segment_number` — a start this entry does
    /// not know yet. Resolves in place, to that segment's own reported
    /// start, the instant [`SegmentWriter::note_segment_start`] reports it;
    /// until then it stays exactly this variant — addressable by
    /// `segment_number` (once a boundary exists), never by a fabricated
    /// media time.
    Segment {
        /// The target segment's sequence number — matches
        /// [`SegmentEntry::sequence_number`].
        segment_number: SegmentSeq,
        /// `presentation_time_delta`: ticks after that segment's start.
        delta: u64,
    },
    /// GPS/UTC wall-clock only (SCTE-35 `splice_schedule.utc_splice_time`,
    /// §9.7.4): this event has **no** media-timeline position at all, only
    /// an instant on the wall clock, until
    /// [`SegmentWriter::set_time_anchor`] gives the event log a
    /// [`TimeAnchor`] to translate through. **This is the B1 case** — see
    /// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux).
    Utc {
        /// Milliseconds since the Unix epoch — matches
        /// [`TimeAnchor::utc_epoch_ms`]'s unit.
        utc_epoch_ms: i64,
    },
}

/// One entry in the event log: the owned, lossless [`TimedEvent`] this
/// trunk carries verbatim — see
/// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux) for
/// why this is *the* published `timed_metadata` type, not a parallel one —
/// plus its current [`EventAnchor`] resolution state.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct EventEntry {
    /// The canonical event, carried verbatim.
    pub event: TimedEvent,
    /// This entry's current resolution state.
    pub anchor: EventAnchor,
}

/// The event log: a bounded, append-ordered log of [`EventEntry`] values,
/// plus the two small resolution tables an [`EventAnchor::Segment`]/
/// [`EventAnchor::Utc`] entry resolves against.
///
/// Evict-then-push shape identical to [`ClassLog`]/[`SegmentLog`] —
/// `base`/`published` mean exactly the same thing here as there.
struct EventLog {
    entries: VecDeque<EventEntry>,
    base: u64,
    published: u64,
    capacity: usize,
    /// Recently-reported segment starts, in the order
    /// [`SegmentWriter::note_segment_start`] received them (playlist order in
    /// practice, since segments are announced in sequence). Bounded by the
    /// **same** `capacity` as `entries` — see [`TrunkConfig::event_capacity`]'s
    /// doc for why this deliberately is not a second, independently-tuned
    /// knob.
    segment_starts: VecDeque<(SegmentSeq, MediaTime)>,
    /// The one wall-clock↔media-clock mapping this trunk's event log
    /// knows, if any. Mirrors [`timed_metadata::Timeline`]'s own
    /// `anchor: Option<TimeAnchor>` field — one mapping per session/trunk,
    /// not one per event.
    time_anchor: Option<TimeAnchor>,
}

impl EventLog {
    fn new(capacity: usize) -> Self {
        EventLog {
            entries: VecDeque::with_capacity(capacity),
            base: 0,
            published: 0,
            capacity,
            segment_starts: VecDeque::with_capacity(capacity),
            time_anchor: None,
        }
    }

    /// Resolve `anchor` against whatever segment starts / time anchor are
    /// already known — **without** fabricating a resolution the log cannot
    /// yet justify. An anchor this call cannot resolve is returned
    /// unchanged: no anchor, no media time, per B1.
    fn try_resolve(&self, anchor: EventAnchor) -> EventAnchor {
        match anchor {
            EventAnchor::Segment {
                segment_number,
                delta,
            } => self
                .segment_starts
                // `.rev()`: a reused `segment_number` can leave more than
                // one entry with the same number; the most recently
                // recorded one is the still-current boundary — see
                // `Trunk::events_in_segment`'s identical `.rev()` note.
                .iter()
                .rev()
                .find(|(n, _)| *n == segment_number)
                .map(|(_, start)| EventAnchor::Media(MediaTime(start.0.saturating_add(delta))))
                .unwrap_or(anchor),
            EventAnchor::Utc { utc_epoch_ms } => self
                .time_anchor
                .as_ref()
                .and_then(|a| epoch_ms_to_media(a, utc_epoch_ms))
                .map(EventAnchor::Media)
                .unwrap_or(anchor),
            EventAnchor::Media(_) => anchor,
        }
    }

    /// Push one event, evicting the oldest if the log is already at
    /// `capacity`. Never rejects, never blocks — exactly [`ClassLog::push`]/
    /// [`SegmentLog::push`]'s contract.
    fn push(&mut self, event: TimedEvent, anchor: EventAnchor) {
        let anchor = self.try_resolve(anchor);
        if self.entries.len() == self.capacity {
            self.entries.pop_front();
            self.base += 1;
        }
        self.entries.push_back(EventEntry { event, anchor });
        self.published += 1;
    }

    /// Record segment `segment_number`'s start on this trunk's 90 kHz
    /// absolute clock, and resolve, **in place**, every still-pending
    /// [`EventAnchor::Segment`] entry that targets exactly this
    /// `segment_number` — not whichever segment happened to be open when
    /// the event was published (that would resolve to *a* segment, not
    /// *the* segment the `emsg` actually named, which is exactly the bug
    /// this design avoids).
    fn note_segment_start(&mut self, segment_number: SegmentSeq, start: MediaTime) {
        if self.segment_starts.len() == self.capacity {
            self.segment_starts.pop_front();
        }
        self.segment_starts.push_back((segment_number, start));
        for entry in &mut self.entries {
            if let EventAnchor::Segment {
                segment_number: n,
                delta,
            } = entry.anchor
                && n == segment_number
            {
                entry.anchor = EventAnchor::Media(MediaTime(start.0.saturating_add(delta)));
            }
        }
    }

    /// Record this trunk's wall-clock↔media-clock mapping, and resolve, in
    /// place, every still-pending [`EventAnchor::Utc`] entry through it.
    /// Before this call, a `Utc`-anchored entry stays a `Utc` entry — see
    /// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux).
    fn set_time_anchor(&mut self, anchor: TimeAnchor) {
        self.time_anchor = Some(anchor);
        for entry in &mut self.entries {
            if let EventAnchor::Utc { utc_epoch_ms } = entry.anchor {
                // `None` (instant before the timeline origin, or beyond
                // `u64`): stay honestly `Utc`, never a fabricated position.
                if let Some(media) = epoch_ms_to_media(&anchor, utc_epoch_ms) {
                    entry.anchor = EventAnchor::Media(media);
                }
            }
        }
    }
}

/// The inverse of [`TimeAnchor::media_to_epoch_ms`]: the [`MediaTime`]
/// `anchor` implies for a UTC instant (milliseconds since the Unix epoch).
///
/// Plain affine algebra — the mirror image of a function `timed_metadata`
/// already publishes — **not** a reimplementation of
/// [`timed_metadata::Timeline`]'s 33-bit wrap-unroll, a different, modular
/// arithmetic problem this module does not re-solve; see
/// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux).
/// Returns `None` — never a clamped, fabricated position — when the instant
/// maps to a negative media time (earlier than the anchor's timeline origin)
/// or beyond `u64`: a stale/replayed or adversarial `splice_schedule` entry
/// must neither crash the writer nor be handed a made-up media time (B1);
/// the caller leaves such an entry [`EventAnchor::Utc`].
fn epoch_ms_to_media(anchor: &TimeAnchor, utc_epoch_ms: i64) -> Option<MediaTime> {
    let delta_ms = i128::from(utc_epoch_ms) - i128::from(anchor.utc_epoch_ms);
    let delta_ticks = delta_ms * i128::from(PTS_HZ) / 1000;
    let media = i128::from(anchor.pts_90k) + delta_ticks;
    u64::try_from(media).ok().map(MediaTime)
}

/// One LL-HLS **partial segment** ("part") of the segment currently being
/// written — RFC 8216bis §4.4.4.9's independently-fetchable CMAF chunk,
/// addressable by `(segment_number, part_index)` the way a client actually
/// asks for one (`_HLS_msn`/`_HLS_part`, or a `part-<seq>.<idx>.m4s` URI). See
/// [The live-part log](self#the-live-part-log-parts-before-their-segment-closes).
///
/// Does **not** reuse `transmux::ll_hls::PartInfo` whole, for the same reason
/// [`SegmentEntry`] does not reuse `transmux::ll_hls::SegmentInfo` whole:
/// `PartInfo::bytes` is `Vec<u8>`, and copying it into a `Bytes` here to get
/// zero-copy fan-out ([Zero-copy fan-out](self#zero-copy-fan-out-honestly))
/// would be exactly one copy per part, on the one path this module exists to
/// keep copy-free; a caller publishing a part therefore builds a
/// `bytes::Bytes` directly (e.g. from the encoder's own output buffer)
/// instead of routing through `Vec<u8>` first.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct PartEntry {
    /// The part's encoded bytes: a bare `moof`+`mdat` CMAF fragment (no
    /// `styp`). `Bytes`, not `Vec<u8>` — fan-out to every reader of this
    /// entry is a refcount bump, not a copy; see
    /// [Zero-copy fan-out](self#zero-copy-fan-out-honestly).
    pub bytes: Bytes,
    /// The parent segment's sequence number — matches
    /// [`SegmentEntry::sequence_number`] once that segment closes.
    pub segment_number: SegmentSeq,
    /// 0-based index of this part within its parent segment.
    pub part_index: PartIndex,
    /// This part's duration, wall-clock.
    pub duration: Duration,
    /// `true` when this part's first sample is a sync sample, so it begins
    /// with an independently decodable frame (RFC 8216bis's
    /// `INDEPENDENT=YES`).
    pub independent: bool,
}

impl PartEntry {
    /// Build one part log entry.
    pub fn new(
        bytes: impl Into<Bytes>,
        segment_number: SegmentSeq,
        part_index: PartIndex,
        duration: Duration,
        independent: bool,
    ) -> Self {
        PartEntry {
            bytes: bytes.into(),
            segment_number,
            part_index,
            duration,
            independent,
        }
    }
}

/// The live-part log: a bounded, append-ordered log of [`PartEntry`] values.
///
/// Evict-then-push shape identical to [`ClassLog`]/[`SegmentLog`]/[`EventLog`]
/// — `base`/`published` mean exactly the same thing here as there. Unlike the
/// segment log, publishing a segment ([`SegmentWriter::publish_segment`]) does
/// **not** touch this ring at all — see
/// [The live-part log](self#the-live-part-log-parts-before-their-segment-closes)
/// for why a part's addressability deliberately does not change the instant
/// its parent segment closes.
/// The part ring's storage. Every *read* of a resident entry goes through
/// [`PartRing::get`]/[`PartRing::iter`], which (under `cfg(test)`) count each
/// entry examined — so a lookup that scans the ring cannot avoid the
/// complexity counter the O1 test asserts on.
struct PartRing {
    q: VecDeque<PartEntry>,
    /// Test-only work counter: ring entries examined by reads.
    #[cfg(test)]
    visits: core::sync::atomic::AtomicUsize,
}

impl PartRing {
    fn with_capacity(capacity: usize) -> Self {
        PartRing {
            q: VecDeque::with_capacity(capacity),
            #[cfg(test)]
            visits: core::sync::atomic::AtomicUsize::new(0),
        }
    }
    fn len(&self) -> usize {
        self.q.len()
    }
    fn pop_front(&mut self) -> Option<PartEntry> {
        self.q.pop_front()
    }
    fn push_back(&mut self, e: PartEntry) {
        self.q.push_back(e);
    }
    fn get(&self, idx: usize) -> Option<&PartEntry> {
        #[cfg(test)]
        self.visits
            .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
        self.q.get(idx)
    }
    /// Counting full-ring iteration (test builds only: production lookups
    /// go through the per-segment index and `get`).
    #[cfg(test)]
    fn iter(&self) -> impl DoubleEndedIterator<Item = &PartEntry> {
        self.q.iter().inspect(|_| {
            self.visits
                .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
        })
    }
}

struct PartLog {
    entries: PartRing,
    base: u64,
    published: u64,
    capacity: usize,
    /// Index: `segment_number` -> the **absolute** ring positions
    /// (`base`-relative addressing, like a cursor's) of every resident part
    /// carrying that number, in publish order. Lets `part_bytes` /
    /// `parts_in_segment` touch only that segment's own parts instead of
    /// scanning the whole ring under the trunk's mutex on every HTTP request
    /// (issue #1082, r06-O1). Bounded by `capacity` in total: every position
    /// is dropped when its entry is evicted, and an emptied key is removed.
    by_segment: HashMap<SegmentSeq, VecDeque<u64>>,
}

impl PartLog {
    fn new(capacity: usize) -> Self {
        PartLog {
            entries: PartRing::with_capacity(capacity),
            base: 0,
            published: 0,
            capacity,
            by_segment: HashMap::new(),
        }
    }

    /// Push one part, evicting the oldest if the log is already at
    /// `capacity`. Never rejects, never blocks — exactly [`ClassLog::push`]/
    /// [`SegmentLog::push`]/[`EventLog::push`]'s contract.
    fn push(&mut self, entry: PartEntry) {
        if self.entries.len() == self.capacity
            && let Some(evicted) = self.entries.pop_front()
        {
            // The oldest ring entry is, necessarily, the oldest position in
            // its own segment's list.
            if let Some(positions) = self.by_segment.get_mut(&evicted.segment_number) {
                positions.pop_front();
                if positions.is_empty() {
                    self.by_segment.remove(&evicted.segment_number);
                }
            }
            self.base += 1;
        }
        self.by_segment
            .entry(entry.segment_number)
            .or_default()
            .push_back(self.published);
        self.entries.push_back(entry);
        self.published += 1;
    }

    /// The resident entry at absolute position `abs`, if still resident.
    fn at(&self, abs: u64) -> Option<&PartEntry> {
        let idx = usize::try_from(abs.checked_sub(self.base)?).ok()?;
        self.entries.get(idx)
    }

    /// Every resident part of `segment_number`, publish order.
    fn in_segment(&self, segment_number: SegmentSeq) -> Vec<PartEntry> {
        self.by_segment
            .get(&segment_number)
            .into_iter()
            .flatten()
            .filter_map(|&abs| self.at(abs).cloned())
            .collect()
    }

    /// The most recently published resident part for the pair.
    fn part_bytes(&self, segment_number: SegmentSeq, part_index: PartIndex) -> Option<Bytes> {
        self.by_segment
            .get(&segment_number)?
            .iter()
            .rev()
            .filter_map(|&abs| self.at(abs))
            .find(|p| p.part_index == part_index)
            .map(|p| p.bytes.clone())
    }
}

/// `consumed - base` as a ring index: `None` when `consumed` is behind
/// `base` or the offset does not fit `usize` (never a wrapping cast).
fn ring_offset(consumed: u64, base: u64) -> Option<usize> {
    usize::try_from(consumed.checked_sub(base)?).ok()
}

/// The shared state behind one [`Trunk`]: the two sample [`ClassLog`]s, the
/// [`SegmentLog`], the [`EventLog`], and the [`PartLog`]. See
/// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux) for
/// why the event log needed its own shape rather than being a third copy of
/// the other two, and
/// [The live-part log](self#the-live-part-log-parts-before-their-segment-closes)
/// for the fourth.
struct TrunkState {
    timed: ClassLog,
    sparse: ClassLog,
    segments: SegmentLog,
    events: EventLog,
    parts: PartLog,
    /// Positions at which a [`TrunkWriter`] was re-issued after a previous
    /// one was dropped — see [`Trunk::writer`] and `WriterHandover`.
    handovers: WriterHandoverLog,
    /// The program's current complete track set — see
    /// [`TrunkWriter::set_tracks`] for why this is always a full replacement
    /// snapshot, never a delta. `Arc<[TrackSpec]>` rather than a bare `Vec`
    /// so [`Trunk::tracks`] hands back a clone of the *reference*, not the
    /// whole set, to every caller — cheap even for a many-track program.
    tracks: Arc<[TrackSpec]>,
    /// Bumped by exactly one on every [`TrunkWriter::set_tracks`] call — see
    /// [`Trunk::track_generation`] for why a consumer compares this instead
    /// of the track [`Vec`] itself.
    track_generation: u64,
}

/// The sample ring: bounded, dual-retention-class, single-writer,
/// multi-cursor. See the [module docs](self) for the design this
/// implements and the benchmark that shaped it.
///
/// Always held as `Arc<Trunk>` — [`Trunk::writer`] and [`Trunk::subscribe`]
/// take `self: &Arc<Self>` because a [`TrunkWriter`]/[`SampleCursor`] each
/// need to keep the shared state alive independently of the `Trunk` handle
/// that created them, exactly as `spikes/trunk-bench`'s validated shape
/// does.
pub struct Trunk {
    state: Mutex<TrunkState>,
    /// Wakes a [`SegmentWriter::publish_segment`] parked on
    /// [`ArchiveOverrun::StallIngest`] once a pin it is waiting on advances
    /// (a [`SegmentCursor::poll`] consuming further) or is released (its
    /// cursor dropped). Paired with `state` in the usual `Condvar` idiom:
    /// `wait` atomically releases the `Mutex` while parked, so a stalled
    /// segment publish does not hold the lock other `Trunk` operations
    /// (sample publish, any cursor's `poll`) need — see
    /// [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure).
    segment_pin_released: Condvar,
    /// Guards [`Trunk::writer`]'s single-take — the **samples + events** ring
    /// group. See [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk).
    writer_taken: AtomicBool,
    /// Guards [`Trunk::segment_writer`]'s single-take — the
    /// **segments + parts** ring group, taken independently of
    /// `writer_taken` so a segmenter and the ingest driver can each hold
    /// their own write handle at once. See
    /// [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk).
    segment_writer_taken: AtomicBool,
    /// `true` once [`Trunk::writer`] has succeeded at least once — lets
    /// `writer` distinguish "the first take" (nothing to report; no reader
    /// could possibly be missing anything yet) from "a re-issue after a
    /// previous `TrunkWriter` was dropped" (record a `WriterHandover` so
    /// an already-subscribed [`SampleCursor`] can report the discontinuity —
    /// see [`Trunk::writer`]).
    writer_issued_once: AtomicBool,
    /// Broad "a part or a segment close was just published, go re-check
    /// your condition" notification — see
    /// [The reader-wake primitive](self#the-reader-wake-primitive-listen-not-one-registration-per-remote-peer).
    /// Bumped by exactly [`SegmentWriter::publish_part`]/
    /// [`SegmentWriter::publish_segment`] and, since the ingress track-set
    /// plumbing (issue #781), [`TrunkWriter::set_tracks`] — never by a
    /// sample/event publish (nothing today waits on those through this
    /// channel). Track-set changes are rare compared to samples/parts, so
    /// folding them into this same broad wake is additive scope, not a new
    /// channel to reason about.
    progress: Event,
    /// Dedicated wake for a freed [`Trunk::listen`] waiter slot — the event a
    /// caller parked on [`Trunk::waiter_slot_freed`] waits on. Kept SEPARATE
    /// from `progress` on purpose: `progress` is the broad part/segment-close
    /// wake every `listen()` waiter (blocking reload, push loop, egress) is
    /// registered on, so reusing it here would let one waiter's *slot release*
    /// wake every other waiter — and each of those, dropping its own
    /// `ProgressListener` to re-listen, would release another slot and wake the
    /// rest again: a ping-pong wake-storm with no media progress at all (100%
    /// CPU for as long as >=2 callers keep re-listening). See `WaiterSlot`'s
    /// `Drop`.
    slot_freed: Event,
    /// Count of currently-registered, not-yet-dropped [`ProgressListener`]s —
    /// what bounds [`Trunk::listen`] against `part_waiter_cap`. A plain
    /// `AtomicUsize`, not part of `state`'s `Mutex`, so registering/releasing
    /// a listener never contends the same lock `publish`/`poll` do.
    waiter_count: AtomicUsize,
    /// Test-only probe: the number of publishers currently parked in
    /// [`ArchiveOverrun::StallIngest`]'s wait. Lets a test observe "blocked"
    /// directly instead of waiting a fixed time and inferring it.
    #[cfg(test)]
    stalled_publishers: AtomicUsize,
    /// Copy of [`TrunkConfig::part_capacity`], read without locking `state` —
    /// the cap [`Trunk::listen`] enforces against `waiter_count`. See
    /// [The reader-wake primitive](self#the-reader-wake-primitive-listen-not-one-registration-per-remote-peer)
    /// for why this reuses `part_capacity` rather than adding a sixth,
    /// independent knob.
    part_waiter_cap: usize,
}

impl Trunk {
    /// Construct a fresh, empty `Trunk`.
    ///
    /// Cannot fail and cannot panic on its configuration: every
    /// [`TrunkConfig`] capacity is a [`NonZeroUsize`], so the one invalid
    /// value (zero — a ring that evicts every entry the instant it is
    /// pushed) is unrepresentable rather than merely rejected. See
    /// [`TrunkConfig`]'s own docs for why that replaced this method's
    /// former five `assert!`s.
    pub fn new(config: TrunkConfig) -> Arc<Trunk> {
        Arc::new(Trunk {
            state: Mutex::new(TrunkState {
                timed: ClassLog::new(config.timed_capacity.get()),
                sparse: ClassLog::new(config.sparse_capacity.get()),
                segments: SegmentLog::new(config.segment_capacity.get()),
                events: EventLog::new(config.event_capacity.get()),
                parts: PartLog::new(config.part_capacity.get()),
                handovers: WriterHandoverLog::new(WRITER_HANDOVER_LOG_CAPACITY),
                tracks: Arc::from(Vec::new()),
                track_generation: 0,
            }),
            segment_pin_released: Condvar::new(),
            writer_taken: AtomicBool::new(false),
            segment_writer_taken: AtomicBool::new(false),
            writer_issued_once: AtomicBool::new(false),
            progress: Event::new(),
            slot_freed: Event::new(),
            waiter_count: AtomicUsize::new(0),
            #[cfg(test)]
            stalled_publishers: AtomicUsize::new(0),
            part_waiter_cap: config.part_capacity.get(),
        })
    }

    /// Lock `state`. `parking_lot::Mutex` does not poison: a panic on some
    /// other thread while it held this lock cannot turn every later writer or
    /// reader into a panic, so no recovery wrapper is needed. (Every critical
    /// section here is a bounded push/pop or counter update with no foreign
    /// code inside — see [`crate::retention`] for why the sink hand-off stays
    /// outside any lock this type takes.)
    fn lock_state(&self) -> MutexGuard<'_, TrunkState> {
        self.state.lock()
    }

    /// Take the one [`TrunkWriter`] for this `Trunk` — the write handle for
    /// the **samples + events** ring group ([`TrunkWriter::publish`]/
    /// [`TrunkWriter::publish_event`]). See
    /// [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk)
    /// for the invariant this enforces (and why it does not also cover
    /// [`Trunk::segment_writer`]'s group).
    ///
    /// Returns `None` while a [`TrunkWriter`] is currently checked out — at
    /// most one holder at a time, enforced here rather than left as a
    /// documented-only convention, because a second concurrent sample/event
    /// writer would silently interleave two unrelated publish sequences into
    /// the same ring with no way for a reader to tell them apart.
    ///
    /// **Re-issuable**: once a previously-issued `TrunkWriter` is dropped
    /// (its `Drop` releases the slot — see that impl), this succeeds again,
    /// which is exactly what a source reconnect needs: the replacement
    /// writer keeps publishing into the *same* rings the original writer's
    /// cursors are already reading, rather than orphaning every subscriber
    /// on a stream that no longer receives writes. A re-issue past the
    /// first is recorded as a `WriterHandover` at the rings' current
    /// publish position, so an already-subscribed [`SampleCursor`] reports
    /// the hand-off in-band ([`SampleCursorItem::Discontinuity`]) at exactly
    /// the point the new writer's first sample follows the old writer's
    /// last, without needing to resubscribe.
    pub fn writer(self: &Arc<Self>) -> Option<TrunkWriter> {
        self.writer_taken
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .ok()?;
        if self.writer_issued_once.swap(true, Ordering::AcqRel) {
            // Not the first take: a previous `TrunkWriter` was dropped and
            // this is its replacement. Record where in each ring the
            // hand-off happened so existing cursors can report it.
            let mut state = self.lock_state();
            let handover = WriterHandover {
                timed_at: state.timed.published,
                sparse_at: state.sparse.published,
            };
            state.handovers.push(handover);
        }
        Some(TrunkWriter {
            trunk: Arc::clone(self),
        })
    }

    /// Take the one [`SegmentWriter`] for this `Trunk` — the write handle for
    /// the **segments + parts** ring group ([`SegmentWriter::publish_segment`]/
    /// [`SegmentWriter::publish_part`]/[`SegmentWriter::note_segment_start`]/
    /// [`SegmentWriter::set_time_anchor`]), independent of [`Trunk::writer`]'s
    /// group so a segmenter can hold this while the ingest driver
    /// simultaneously holds a [`TrunkWriter`] — see
    /// [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk)
    /// for why the split is safe and what it does and does not guarantee
    /// across rings.
    ///
    /// Returns `None` while a previously-issued [`SegmentWriter`] is still
    /// alive — this ring group has exactly one live writer too, guarded by
    /// its own `AtomicBool` rather than [`Trunk::writer`]'s, for exactly the
    /// same reason: a second concurrent segment/part writer would silently
    /// interleave two unrelated publish sequences into the segment or part
    /// ring. Like [`Trunk::writer`] it is **re-issuable**: dropping the
    /// handle releases the slot, so a later call succeeds again (resume
    /// numbering from [`SegmentWriter::next_sequence_number`]).
    pub fn segment_writer(self: &Arc<Self>) -> Option<SegmentWriter> {
        self.segment_writer_taken
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .ok()
            .map(|_| SegmentWriter {
                trunk: Arc::clone(self),
            })
    }

    /// Subscribe a new [`SampleCursor`], starting from *now* — the next
    /// entry [`TrunkWriter::publish`] produces after this call, not any
    /// backlog already in either ring. See [`Trunk::subscribe_from_backlog`]
    /// for the seek-to-past variant this method's own docs used to
    /// anticipate (a consumer built *after* samples it needs already landed
    /// in the ring — e.g. a segmenter reacting to the same batch that
    /// announced its program — wants that one instead).
    ///
    /// # This call *is* fan-out — read this before calling it per connection
    ///
    /// `spikes/trunk-bench` measured writer cost as **O(N) in cursor
    /// count** (956 ns → 9.98 µs from 1 → 16 readers; spec §3.1) — every
    /// cursor contends the same shared lock every publish. **A cursor is
    /// for a distinct consumer of the stream** (a segmenter, a DVR writer,
    /// an analysis tap, one push relay) — **never** one per peer of a
    /// one-to-many protocol. Supported reader count is **single-digit by
    /// design**: LL-HLS serving a thousand viewers takes **one** cursor
    /// here and fans out to its viewers itself, at the layer that already
    /// holds per-viewer state anyway. Do not call this once per connection;
    /// there is no tee, broadcast channel, or per-consumer queue to reach
    /// for instead — a sample's payload is already [`bytes::Bytes`], so
    /// fan-out beyond this one cursor is a refcount bump the relay performs
    /// itself, not something this type needs to do for you.
    ///
    /// # A cursor created in the gap between a writer drop and its re-issue
    /// never sees a spurious `Discontinuity`
    ///
    /// [`SampleCursorItem::Discontinuity`] reports a writer hand-off
    /// **relative to what this specific cursor could have observed
    /// continuity across**, not merely "any hand-off recorded after this
    /// call": one recorded at or before this cursor's own starting position
    /// on both classes (nothing was published on either ring between this
    /// `subscribe` call and the hand-off) is consumed silently — reporting
    /// it would mean a cursor's *very first* `poll` ever is a spurious
    /// "discontinuity" from data it never had any continuity with in the
    /// first place.
    pub fn subscribe(self: &Arc<Self>) -> SampleCursor {
        let state = self.lock_state();
        SampleCursor {
            trunk: Arc::clone(self),
            timed_consumed: state.timed.published,
            sparse_consumed: state.sparse.published,
            handovers_consumed: state.handovers.published,
            subscribe_timed_at: state.timed.published,
            subscribe_sparse_at: state.sparse.published,
        }
    }

    /// Subscribe a new [`SampleCursor`], starting from the **oldest entry
    /// each ring currently retains** instead of [`Trunk::subscribe`]'s
    /// "now" — i.e. this cursor's first `poll` replays whatever backlog is
    /// still resident in the `Timed` ring and the `Sparse` ring, each
    /// independently, before catching up to the live tail.
    ///
    /// This is the "seek-to-past variant" [`Trunk::subscribe`]'s own docs
    /// anticipated ("a later step may add a seek-to-past variant... this
    /// step does not need one") — the step turned out to be issue #808's
    /// segment-bridge fix: a [`TrunkWriter::publish`] batch that lands
    /// *before* a consumer subscribes (e.g. the very same `feed` call that
    /// both announces a program and publishes its first samples) is
    /// otherwise invisible to a [`Trunk::subscribe`] cursor forever, even
    /// though the samples are sitting right there in the ring.
    ///
    /// # Replay is bounded by ring capacity, not "everything ever published"
    ///
    /// This does **not** reach further back than what each ring still
    /// holds: an entry already evicted by [`TrunkConfig::timed_capacity`]/
    /// [`TrunkConfig::sparse_capacity`] before this call is gone, exactly as
    /// it would be for any other cursor — there is no unbounded replay log
    /// behind this method, only the same fixed-size rings every other
    /// cursor reads. Concretely: this cursor starts at each ring's current
    /// `base` (the oldest index still resident), not index 0, so its first
    /// `poll` never reports a spurious `Lagged`/`Degraded` for data that was
    /// evicted *before* this call — from this cursor's point of view,
    /// "backlog" means "what the ring can show me right now", not "what was
    /// ever published". Both retention classes replay this way,
    /// independently: a `Timed` backlog and a `Sparse` backlog are each
    /// bounded by their own ring's own capacity.
    ///
    /// [`SampleCursorItem::Lagged`]/[`SampleCursorItem::Degraded`] still
    /// fire exactly as they do for a [`Trunk::subscribe`] cursor for any
    /// loss that happens **after** this call — falling behind the live tail
    /// once subscribed is reported in-band the same way for both kinds of
    /// cursor; only the starting position differs.
    ///
    /// # This call *is* fan-out — read this before calling it per connection
    ///
    /// Exactly [`Trunk::subscribe`]'s own fan-out warning, verbatim: writer
    /// cost is **O(N) in cursor count** (`spikes/trunk-bench` measured 956 ns
    /// → 9.98 µs from 1 → 16 readers; spec §3.1) — every cursor contends the
    /// same shared lock every publish. **A cursor is for a distinct
    /// consumer of the stream**, **never** one per peer of a one-to-many
    /// protocol. Supported reader count is **single-digit by design**; do
    /// not call this once per connection.
    pub fn subscribe_from_backlog(self: &Arc<Self>) -> SampleCursor {
        let state = self.lock_state();
        SampleCursor {
            trunk: Arc::clone(self),
            timed_consumed: state.timed.base,
            sparse_consumed: state.sparse.base,
            handovers_consumed: state.handovers.base,
            subscribe_timed_at: state.timed.base,
            subscribe_sparse_at: state.sparse.base,
        }
    }

    /// Diagnostic: entries currently resident in the `Timed` ring. Never
    /// exceeds [`TrunkConfig::timed_capacity`].
    pub fn timed_len(&self) -> usize {
        self.lock_state().timed.entries.len()
    }

    /// Diagnostic: entries currently resident in the `Sparse` ring. Never
    /// exceeds [`TrunkConfig::sparse_capacity`].
    #[cfg(test)]
    pub(crate) fn sparse_len(&self) -> usize {
        self.lock_state().sparse.entries.len()
    }

    /// Subscribe a new **non-pinning** [`SegmentCursor`], starting from
    /// *now* — the same "next entry only, no backlog" rule as
    /// [`Trunk::subscribe`], and the same single-digit-reader,
    /// one-cursor-per-distinct-consumer guidance from that method's docs
    /// applies here verbatim (this cursor contends exactly the lock
    /// `subscribe`'s cursors do).
    ///
    /// This cursor is **not** protected by [`ArchiveOverrun`]: if it falls
    /// behind the segment log's ordinary [`TrunkConfig::segment_capacity`]
    /// eviction, it simply sees [`SegmentCursorItem::Lagged`], exactly like
    /// an ordinary [`RetentionClass::Timed`] sample reader. Use this for a
    /// consumer that tolerates ordinary loss (LL-HLS window rendering,
    /// catch-up within the live window) — use [`Trunk::pin_segments`]
    /// instead for a consumer that must not miss a segment (DVR/archive).
    pub fn subscribe_segments(self: &Arc<Self>) -> SegmentCursor {
        let state = self.lock_state();
        SegmentCursor {
            trunk: Arc::clone(self),
            consumed: state.segments.published,
            pin_id: None,
            done: false,
        }
    }

    /// Subscribe a new **pinning** [`SegmentCursor`] for a DVR/archive
    /// consumer that must not miss a segment — see
    /// [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure)
    /// for the full design story this method is the entry point for.
    ///
    /// `on_overrun` is this cursor's chosen [`ArchiveOverrun`] for the one
    /// moment its guarantee runs out: the segment log at
    /// [`TrunkConfig::segment_capacity`], about to evict an entry this
    /// cursor has not yet consumed. There is no default parameter here on
    /// purpose — pinning is an explicit request for a stronger guarantee
    /// than [`Trunk::subscribe_segments`] gives, so the trade made when that
    /// guarantee cannot be kept is an explicit choice too, not a silent
    /// fallback (though [`ArchiveOverrun::default`] exists for a caller that
    /// affirmatively wants the same default the rest of this module uses).
    ///
    /// Also starts from *now*, and also single-digit-by-design — the same
    /// fan-out rule as [`Trunk::subscribe`] and [`Trunk::subscribe_segments`]
    /// applies; a pinning cursor is exactly as expensive per publish as any
    /// other.
    pub fn pin_segments(self: &Arc<Self>, on_overrun: ArchiveOverrun) -> SegmentCursor {
        let mut state = self.lock_state();
        let pin_id = state.segments.next_pin_id;
        state.segments.next_pin_id += 1;
        let consumed = state.segments.published;
        state.segments.pins.insert(
            pin_id,
            PinState {
                consumed,
                policy: on_overrun,
                terminated: false,
            },
        );
        SegmentCursor {
            trunk: Arc::clone(self),
            consumed: 0,
            pin_id: Some(pin_id),
            done: false,
        }
    }

    /// Diagnostic: entries currently resident in the segment log. Never
    /// exceeds [`TrunkConfig::segment_capacity`] — true even with an
    /// un-acking pinning cursor attached, which is exactly the property
    /// [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure)'s
    /// "pinning is bounded" claim means.
    pub fn segment_len(&self) -> usize {
        self.lock_state().segments.entries.len()
    }

    /// Subscribe a new [`EventCursor`] over the event log, starting from
    /// *now* — the same "next entry only, no backlog" rule as
    /// [`Trunk::subscribe`]/[`Trunk::subscribe_segments`], and the same
    /// single-digit-reader, one-cursor-per-distinct-consumer guidance
    /// applies here verbatim (this cursor contends exactly the lock every
    /// other cursor does).
    ///
    /// A streaming consumer — e.g. a playback scheduler that wants every
    /// event as it resolves — wants this. A point-in-time query — "what has
    /// resolved for segment N" (a manifest renderer), or "what resolved
    /// between T1 and T2" (that same scheduler, replaying its window) —
    /// wants [`Trunk::events_in_segment`]/[`Trunk::events_between`] instead;
    /// both read the same log, just as a snapshot rather than a moving
    /// position. See
    /// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux).
    pub fn subscribe_events(self: &Arc<Self>) -> EventCursor {
        let state = self.lock_state();
        EventCursor {
            trunk: Arc::clone(self),
            consumed: state.events.published,
        }
    }

    /// Every currently-**resolved** ([`EventAnchor::Media`]) event whose
    /// media time falls in the half-open range `[from, to)` — start
    /// inclusive, end exclusive. An entry still `Segment`/`Utc`-anchored
    /// never appears here: it has no honest media time yet, and
    /// fabricating one to satisfy this query would be exactly B1 — see
    /// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux).
    pub fn events_between(&self, from: MediaTime, to: MediaTime) -> Vec<EventEntry> {
        let state = self.lock_state();
        state
            .events
            .entries
            .iter()
            .filter(|e| matches!(e.anchor, EventAnchor::Media(t) if t.0 >= from.0 && t.0 < to.0))
            .cloned()
            .collect()
    }

    /// The wall-clock anchor this trunk's event log has been given, if any.
    ///
    /// Returns `None` until [`SegmentWriter::set_time_anchor`] has been
    /// called — exactly the same `None` that makes
    /// [`EventAnchor::Utc`]-anchored entries stay unresolved.
    pub fn time_anchor(&self) -> Option<TimeAnchor> {
        let state = self.lock_state();
        state.events.time_anchor
    }

    /// The reported start time of segment `seq` on this trunk's 90 kHz
    /// source clock (the same domain as [`TimeAnchor::pts_90k`]), if
    /// [`SegmentWriter::note_segment_start`] has reported one and it has
    /// not been evicted from the bounded `segment_starts` table. Returns
    /// `None` when the segment was never reported or has rolled off the
    /// front of the bounded table.
    pub fn segment_start_time(&self, seq: SegmentSeq) -> Option<MediaTime> {
        let state = self.lock_state();
        let log = &state.events;
        log.segment_starts
            .iter()
            .rev()
            .find(|(n, _)| *n == seq)
            .map(|(_, start)| *start)
    }

    /// Every currently-resolved event whose media time falls within segment
    /// `segment_number`'s span: `[start_N, start_{N+1})` once
    /// [`SegmentWriter::note_segment_start`] has reported the *next*
    /// segment's start too, else `[start_N, ∞)` (the segment is still open
    /// — nothing yet says where it ends). Returns nothing for a
    /// `segment_number` this trunk has never reported a start for: there is
    /// no span to contain anything, and an unresolved
    /// [`EventAnchor::Segment`] entry targeting it is not returned either,
    /// for the same B1 reason [`Trunk::events_between`] documents.
    pub fn events_in_segment(&self, segment_number: SegmentSeq) -> Vec<EventEntry> {
        let state = self.lock_state();
        let log = &state.events;
        // `rposition`: a reused `segment_number` (wraparound, or a
        // restarted segmenter) can leave more than one entry in
        // `segment_starts` with the same number — the most recently
        // recorded one is the boundary that is actually still current; a
        // plain front-to-back search would resolve to whichever one
        // happened to be pushed first, silently returning a stale boundary
        // once the number repeats.
        let Some(chosen_idx) = log
            .segment_starts
            .iter()
            .rposition(|(n, _)| *n == segment_number)
        else {
            return Vec::new();
        };
        let Some(&(_, start)) = log.segment_starts.get(chosen_idx) else {
            return Vec::new();
        };
        // The end boundary is the first start recorded AFTER the CHOSEN
        // occurrence above — positionally, not by looking up
        // `segment_number + 1`. `segment_starts` is append-ordered
        // (chronological), so the very next entry is always this
        // occurrence's true close, regardless of what number it carries;
        // looking it up by number instead can find a stale, chronologically
        // EARLIER entry once numbers repeat (e.g. segment 2's start from a
        // previous round, before segment 1 was reused) and invert the
        // range into one nothing can ever fall inside.
        let end = log.segment_starts.get(chosen_idx + 1).map(|&(_, s)| s.0);
        log.entries
            .iter()
            .filter(|e| match e.anchor {
                EventAnchor::Media(t) => t.0 >= start.0 && end.map(|e2| t.0 < e2).unwrap_or(true),
                _ => false,
            })
            .cloned()
            .collect()
    }

    /// Diagnostic: entries currently resident in the event log. Never
    /// exceeds [`TrunkConfig::event_capacity`].
    #[cfg(test)]
    pub(crate) fn event_len(&self) -> usize {
        self.lock_state().events.entries.len()
    }

    /// A live part's bytes by `(segment_number, part_index)` — the direct,
    /// `&self`-shaped query a [`ServedEgress`](crate::egress::ServedEgress)
    /// implementation needs to answer "does this part exist right now",
    /// exactly the shape [`Trunk::events_between`]/[`Trunk::events_in_segment`]
    /// already give the event log rather than forcing a caller to drain a
    /// cursor into a self-maintained cache. See
    /// [The live-part log](self#the-live-part-log-parts-before-their-segment-closes)
    /// for why a part answers `Some` here for as long as it has not been
    /// evicted by [`TrunkConfig::part_capacity`]'s ordinary bound —
    /// including after its parent segment has closed.
    pub fn part_bytes(&self, segment_number: SegmentSeq, part_index: PartIndex) -> Option<Bytes> {
        // The most recently published entry for the pair wins: a
        // `SegmentWriter` re-issue can reuse a `segment_number` (see
        // `NonMonotonicSequenceNumber`), so the newest is the current one.
        // Indexed by segment: touches only that segment's own parts.
        self.lock_state()
            .parts
            .part_bytes(segment_number, part_index)
    }

    /// Every currently-resident part of segment `segment_number`, in publish
    /// order — the part-log counterpart of [`Trunk::events_in_segment`],
    /// letting a caller derive "how many parts does the open segment have so
    /// far" (RFC 8216bis's `_HLS_part` blocking-reload condition) without a
    /// cursor.
    ///
    /// Every resident part carrying `segment_number` is returned, in publish
    /// order, regardless of what else shares the ring (served from a
    /// per-segment index, so only this segment's own parts are touched).
    ///
    /// This is deliberately **not** a from-the-back run isolation that stops
    /// at the first non-matching entry: the ring does not hold a segment's
    /// parts as one contiguous tail run. A live producer
    /// (`multimux::source::segment::ProgramSegmenter`) can publish part
    /// `(N+1, 0)` while segment `N` is still open, and a run isolation would
    /// then report *zero* parts for `N`, dropping its `#EXT-X-PART` tags from
    /// a served LL-HLS playlist.
    ///
    /// Trade-off: after a `SegmentWriter` re-issue reuses a `segment_number`
    /// (see [`SegmentWriter::next_sequence_number`]), a stale older run under
    /// the same number, if still resident, is returned alongside the current
    /// one. That needs a real number reuse (`publish_segment` otherwise
    /// rejects a non-monotonic `sequence_number`) and the stale run to have
    /// survived in the ring.
    pub fn parts_in_segment(&self, segment_number: SegmentSeq) -> Vec<PartEntry> {
        self.lock_state().parts.in_segment(segment_number)
    }

    /// Diagnostic: entries currently resident in the live-part log. Never
    /// exceeds [`TrunkConfig::part_capacity`].
    pub fn part_len(&self) -> usize {
        self.lock_state().parts.entries.len()
    }

    /// Diagnostic: currently-outstanding [`ProgressListener`] registrations
    /// (from [`Trunk::listen`], not yet dropped). Never exceeds
    /// [`TrunkConfig::part_capacity`] — see
    /// [The reader-wake primitive](self#the-reader-wake-primitive-listen-not-one-registration-per-remote-peer).
    pub fn waiter_count(&self) -> usize {
        self.waiter_count.load(Ordering::Acquire)
    }

    /// The sequence number of the most-recently-**closed** segment (the
    /// newest [`SegmentWriter::publish_segment`] call), or `None` if no
    /// segment has closed yet. Distinguishes "closed" (a whole, fetchable
    /// [`SegmentEntry`]) from merely "has live parts" — RFC 8216bis
    /// §6.2.5.2's bare-`_HLS_msn` blocking-reload condition needs exactly
    /// this distinction (mirrors
    /// `hls_runtime::server::MediaStore::last_closed_segment_seq`, which
    /// this method lets a `ServedEgress` stop duplicating).
    pub fn last_closed_segment(&self) -> Option<SegmentSeq> {
        self.lock_state()
            .segments
            .entries
            .back()
            .map(|e| e.sequence_number)
    }

    /// This program's current complete track set — see
    /// [`TrunkWriter::set_tracks`] for how it is set/replaced. Empty until
    /// the first `set_tracks` call (a freshly-minted `Trunk` announces no
    /// tracks yet). Stored as `Arc<[TrackSpec]>`, so this is a cheap `Arc`
    /// clone (a refcount bump), never a `Vec` copy, however many tracks the
    /// program carries.
    pub fn tracks(&self) -> Arc<[TrackSpec]> {
        Arc::clone(&self.lock_state().tracks)
    }

    /// Bumped by exactly one on every [`TrunkWriter::set_tracks`] call
    /// (including one that happens to set an identical set to what was
    /// already there — this counts *calls*, not distinct sets). Lets a
    /// consumer detect "the track set may have changed" by comparing two
    /// `u64`s rather than diffing two `Vec<TrackSpec>`s — cheap regardless
    /// of how many tracks a program carries. `0` until the first
    /// `set_tracks` call.
    pub fn track_generation(&self) -> u64 {
        self.lock_state().track_generation
    }

    /// Register for the next part/segment-close notification — see
    /// [The reader-wake primitive](self#the-reader-wake-primitive-listen-not-one-registration-per-remote-peer).
    ///
    /// Returns `None` once [`TrunkConfig::part_capacity`] concurrent
    /// registrations are already outstanding — the caller must not wait in
    /// that case (there is no slot to wait *in*); it should fall back to an
    /// immediate re-poll or answer its request as unavailable now, exactly
    /// as a caller must once [`crate::egress::AwaitPolicy`] itself has
    /// expired. **Register before re-checking the condition you are waiting
    /// on** — `event_listener`'s standard idiom, and the same ordering
    /// `hls_runtime::server::MediaStore::listen`'s own docs require —
    /// otherwise a `notify` racing your check can be missed.
    pub fn listen(self: &Arc<Self>) -> Option<ProgressListener> {
        loop {
            let current = self.waiter_count.load(Ordering::Acquire);
            if current >= self.part_waiter_cap {
                return None;
            }
            if self
                .waiter_count
                .compare_exchange(current, current + 1, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
            {
                break;
            }
        }
        Some(ProgressListener {
            _slot: WaiterSlot(Arc::clone(self)),
            listener: self.progress.listen(),
        })
    }

    /// A future that resolves once a [`Trunk::listen`] waiter slot is freed
    /// (some [`ProgressListener`] dropped), or `None` when a slot is free
    /// right now. A caller that got `None` from [`listen`](Self::listen) parks
    /// on this (raced against its own cancellation) instead of sleep-polling
    /// for a slot to free (W2b-1 SP1.4/B10b).
    ///
    /// Register-before-recheck, matching [`listen`](Self::listen)'s own
    /// ordering contract: the listener is registered on the dedicated
    /// `slot_freed` event FIRST, then the count is re-checked, so a slot freed
    /// between the two cannot be missed (a check-then-listen would leave the
    /// caller parked with no timeout until some unrelated part/segment wakes
    /// the shared `progress` event, or forever on an idle source).
    pub fn waiter_slot_freed(&self) -> Option<EventListener> {
        let listener = self.slot_freed.listen();
        if self.waiter_count.load(Ordering::Acquire) < self.part_waiter_cap {
            return None;
        }
        Some(listener)
    }
}

/// RAII release of one [`Trunk`] waiter slot — split out from
/// [`ProgressListener`] itself (rather than a `Drop` impl directly on
/// `ProgressListener`) specifically so [`ProgressListener::wait_deadline`]
/// can destructure `self` and move its `listener` field into
/// [`event_listener::Listener::wait_deadline`] by value: Rust forbids moving
/// a field out of a type that implements `Drop` itself, but does not forbid
/// it for a type that merely *contains* a field whose type implements
/// `Drop` — each field is then dropped independently, in this case when the
/// destructured local bindings go out of scope at the end of that method.
struct WaiterSlot(Arc<Trunk>);

impl Drop for WaiterSlot {
    /// Release this `Trunk`'s bounded waiter slot — see
    /// [The reader-wake primitive](self#the-reader-wake-primitive-listen-not-one-registration-per-remote-peer)
    /// for why this cap exists at all (an unbounded waiter set is a remote
    /// resource-exhaustion vector). Fires whether the owning
    /// [`ProgressListener`] was ever polled/waited on, woken, or simply
    /// dropped un-awaited — a caller that gives up on its own request must
    /// not leak a slot.
    fn drop(&mut self) {
        self.0.waiter_count.fetch_sub(1, Ordering::AcqRel);
        // Wake one caller parked on `Trunk::waiter_slot_freed` (a would-be
        // egress that got `None` from `listen`) that a slot is now free. Wakes
        // the DEDICATED `slot_freed` event, never the shared `progress` event
        // (which every `listen()` waiter is registered on — waking it here
        // would be a wake-storm, see the `slot_freed` field doc), and exactly
        // ONE waiter: this drop freed exactly one slot.
        self.0.slot_freed.notify(1);
    }
}

/// A registered wake-up from [`Trunk::listen`] — the
/// [`event_listener::EventListener`] a [`ServedEgress`](crate::egress::ServedEgress)
/// adapter waits on, so a blocked request need not busy-poll. See
/// [The reader-wake primitive](self#the-reader-wake-primitive-listen-not-one-registration-per-remote-peer).
///
/// Releases this `Trunk`'s bounded waiter slot when dropped (via the
/// `_slot` field's own `Drop`, see this module's internal `WaiterSlot`) — whether this listener
/// was woken, timed out, or is simply discarded — so a caller that gives up
/// does not leak a slot forever.
pub struct ProgressListener {
    _slot: WaiterSlot,
    listener: EventListener,
}

impl ProgressListener {
    /// Block the calling thread until woken or `deadline` passes, whichever
    /// comes first — `true` if woken, `false` on timeout. Composes with
    /// [`crate::egress::AwaitPolicy`]'s deadline: convert
    /// `AwaitPolicy::deadline` (a [`Timestamp`]) to the `std::time::Instant`
    /// your caller already anchors its `Timestamp`s to (see
    /// [`Timestamp::from_instant`]'s inverse — the caller holds the base
    /// `Instant` it built its `Timestamp`s from) and pass that here, so this
    /// call can never park past the caller's own bound.
    ///
    /// Deliberately no unbounded `wait()` is exposed here — only this
    /// deadline-bound form and the `Future` impl below (whose bound is
    /// whatever timeout the caller's own executor wraps it in, exactly the
    /// `hls_runtime::server` "caller-driven wait loop" shape) — matching
    /// this crate's [`crate::egress::AwaitPolicy`] philosophy that a wait on
    /// remote-triggerable input must never be able to park forever.
    pub fn wait_deadline(self, deadline: std::time::Instant) -> bool {
        // Destructuring here (not a method call on `self.listener` while
        // `self` stays intact) is exactly what requires `ProgressListener`
        // itself to carry no `Drop` impl — see `WaiterSlot`'s own doc.
        let ProgressListener { _slot, listener } = self;
        listener.wait_deadline(deadline).is_some()
    }
}

impl Future for ProgressListener {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // `EventListener` is `Unpin` (event_listener's own guarantee), and so
        // is `WaiterSlot` (an `Arc` newtype), so `ProgressListener` as a
        // whole is `Unpin` too — safe to reach the inner listener through a
        // plain `&mut` and poll it directly.
        let this = self.get_mut();
        Pin::new(&mut this.listener).poll(cx)
    }
}

/// The write handle for a [`Trunk`]'s **samples + events** ring group.
/// Obtained via [`Trunk::writer`]. See
/// [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk)
/// for why this group is exactly these two rings, and
/// [`SegmentWriter`] for the sibling handle covering segments + parts.
///
/// `publish` never blocks and never rejects: a full class ring evicts its
/// oldest entry (see the internal per-class log's push logic) rather than waiting for a reader or
/// erroring, so ingest never stalls because some [`SampleCursor`] is slow —
/// the same non-blocking-producer principle as [`crate::byte_tap::ByteTap::record`],
/// for the same reason (a broadcast head-end does not pause live ingest for
/// a lagging analysis tap or a stalled egress peer).
///
/// "Never blocks" describes the absence of any wait-for-a-reader code path,
/// not a claim that the underlying `Mutex` critical section is instant —
/// `publish` briefly contends the same lock [`SampleCursor::poll`] does, a
/// bounded amount of work independent of how far behind any reader is (this
/// is exactly what `spikes/trunk-bench` measured as the O(N)-in-cursor-count
/// cost, not an unbounded wait).
pub struct TrunkWriter {
    trunk: Arc<Trunk>,
}

impl TrunkWriter {
    /// Publish one sample for `track_id` under `retention`.
    pub fn publish(&self, track_id: TrackId, retention: RetentionClass, sample: Sample) {
        let mut state = self.trunk.lock_state();
        match retention {
            RetentionClass::Timed => state.timed.push(track_id, sample),
            RetentionClass::Sparse => state.sparse.push(track_id, sample),
        }
    }

    /// Publish one event. Never blocks and never rejects — a full event log
    /// evicts its oldest entry exactly like the sample/segment logs.
    /// `anchor` is resolved immediately against whatever segment starts /
    /// time anchor this trunk already knows; if it cannot be resolved yet,
    /// the entry is stored exactly as given, and resolves later, in place,
    /// once [`SegmentWriter::note_segment_start`]/[`SegmentWriter::set_time_anchor`]
    /// supplies what was missing. See
    /// [The event log](self#the-event-log-90-khz-absolute-and-the-b1-crux).
    pub fn publish_event(&self, event: TimedEvent, anchor: EventAnchor) {
        let mut state = self.trunk.lock_state();
        state.events.push(event, anchor);
    }

    /// Replace this program's track set wholesale — the write side of
    /// [`Trunk::tracks`]/[`Trunk::track_generation`], and the method
    /// [`crate::ingress::IngestDriver`] calls to seed a freshly-minted
    /// `Trunk` from `SessionEvent::NewProgram`'s `tracks` and to apply a
    /// later `SessionEvent::TracksChanged`.
    ///
    /// `tracks` is taken as the **complete replacement set**, matching
    /// `SessionEvent::TracksChanged`'s own doc: a PMT (or any container's
    /// track-declaration mechanism) carries the whole elementary-stream
    /// list, so this call is idempotent (calling it twice with the same set
    /// leaves the trunk's tracks unchanged in content, only `track_generation`
    /// advances) and immune to delta-ordering bugs — there is no "add
    /// track"/"remove track" pair to apply out of order. A caller that wants
    /// to know *which* track changed diffs the previous [`Trunk::tracks`]
    /// snapshot against this one itself.
    ///
    /// Bumps [`Trunk::track_generation`] by exactly one and wakes any
    /// [`Trunk::listen`] registration, the same
    /// [`event_listener::Event::notify`] fan-out
    /// [`SegmentWriter::publish_segment`]/[`SegmentWriter::publish_part`]
    /// already use — see [`Trunk`]'s `progress` field doc for why a
    /// track-set change is folded into that same broad wake rather than a
    /// new channel.
    pub fn set_tracks(&self, tracks: Vec<TrackSpec>) {
        let mut state = self.trunk.lock_state();
        state.tracks = Arc::from(tracks);
        state.track_generation += 1;
        drop(state);
        self.trunk.progress.notify(usize::MAX);
    }
}

impl Drop for TrunkWriter {
    /// Release this `TrunkWriter`'s slot so [`Trunk::writer`] can issue a
    /// replacement — see that method's doc for why a source reconnect needs
    /// exactly this (a dropped-and-never-reissued writer would otherwise
    /// orphan every already-subscribed [`SampleCursor`] permanently: nothing
    /// would ever publish into their rings again).
    fn drop(&mut self) {
        self.trunk.writer_taken.store(false, Ordering::Release);
    }
}

/// The write handle for a [`Trunk`]'s **segments + parts** ring group.
/// Obtained via [`Trunk::segment_writer`], independently of [`TrunkWriter`]
/// (via [`Trunk::writer`]) — see
/// [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk)
/// for why this split exists, why `note_segment_start`/`set_time_anchor` are
/// grouped here rather than on [`TrunkWriter`], and what is (and is not)
/// guaranteed about ordering relative to the sample/event rings.
///
/// Like [`TrunkWriter`], every method here either never blocks (ordinary
/// eviction, exactly the sample rings' non-blocking-producer principle) or
/// blocks only in the one documented [`ArchiveOverrun::StallIngest`] case —
/// see [`SegmentWriter::publish_segment`].
/// Hard upper bound on how long [`SegmentWriter::publish_segment`] blocks
/// for an [`ArchiveOverrun::StallIngest`] pin before giving up and evicting
/// anyway — falling back to exactly the loss [`ArchiveOverrun::Gap`] always
/// produces (the still-unconsumed pin's cursor reports it as `Gap` the next
/// time it is polled). Without this bound, a caller whose own thread is
/// also responsible for driving the very [`crate::retention::RetentionDriver`]
/// that would release this pin deadlocks itself: nothing else can ever run
/// `drive` to consume the pin, because this call never gives control back to
/// do so. A generous, rare-in-practice wall-clock bound turns that into a
/// bounded delay instead of a permanent hang. See
/// [`SegmentWriter::try_publish_segment`] for the non-blocking alternative
/// that avoids the wait (and this fallback) altogether.
///
/// Much smaller under `#[cfg(test)]`: a test that exercises this bound
/// (rather than the pin advancing first, which every other test does)
/// needs to actually reach it, and this crate's tests must stay fast — the
/// bound's existence is what is under test, not its production-sized value.
/// Comfortably larger than the 200ms "still blocked" confirmation window
/// `archive_overrun_stall_ingest_blocks_writer_until_driver_advances` (in
/// `retention.rs`) uses, so that test's writer unblocks via the pin
/// actually advancing (`RetentionDriver::drive`), not via this bound
/// expiring underneath it.
#[cfg(not(test))]
const STALL_INGEST_MAX_WAIT: Duration = Duration::from_secs(30);
#[cfg(test)]
const STALL_INGEST_MAX_WAIT: Duration = Duration::from_secs(2);

/// [`SegmentWriter::publish_segment`]/[`SegmentWriter::try_publish_segment`]
/// reject a `sequence_number` that is not strictly greater than the last one
/// this `Trunk` accepted — segment numbers are monotonic across every
/// `SegmentWriter` a `Trunk` ever issues, including after a re-issue
/// following a drop (see [`Trunk::segment_writer`]): an HLS media sequence
/// number must never go backwards, and every query keyed on `sequence_number`
/// alone ([`Trunk::part_bytes`]/[`Trunk::parts_in_segment`]/
/// [`Trunk::events_in_segment`]/[`crate::retention::RetentionDriver::locate`])
/// would otherwise resolve ambiguously (or, before those were hardened with
/// `.rev()`/positional lookups, silently stale) the instant a restarted
/// segmenter numbered from `1` again. See [`SegmentWriter::next_sequence_number`]
/// for the number a re-issued writer must resume from instead.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
#[non_exhaustive]
#[error(
    "segment sequence_number {attempted} is not greater than the last \
     published one ({last_published:?}) — sequence numbers are monotonic \
     across writer re-issues; see SegmentWriter::next_sequence_number"
)]
pub struct NonMonotonicSequenceNumber {
    /// The rejected `sequence_number`.
    pub attempted: SegmentSeq,
    /// The last `sequence_number` this `Trunk` accepted, if any.
    pub last_published: Option<SegmentSeq>,
}

/// The failure half of [`SegmentWriter::try_publish_segment`].
#[derive(Debug, Clone, Error)]
#[non_exhaustive]
pub enum TryPublishSegmentError {
    /// Publishing right now would need to block (an
    /// [`ArchiveOverrun::StallIngest`] pin has not caught up) — hands the
    /// entry straight back rather than blocking or losing it.
    #[error("would need to block for an ArchiveOverrun::StallIngest pin to catch up")]
    WouldStall(SegmentEntry),
    /// `sequence_number` was not monotonic — see
    /// [`NonMonotonicSequenceNumber`].
    #[error(transparent)]
    NonMonotonic(#[from] NonMonotonicSequenceNumber),
}

pub struct SegmentWriter {
    trunk: Arc<Trunk>,
}

impl SegmentWriter {
    /// Publish one finished segment, in playlist order.
    ///
    /// Never blocks and never rejects for **every non-pinning**
    /// [`SegmentCursor`] and for every pinning cursor using
    /// [`ArchiveOverrun::Gap`] (the default) or [`ArchiveOverrun::Terminate`]
    /// — a full segment log evicts its oldest entry exactly like
    /// [`TrunkWriter::publish`]'s sample rings. The **one** exception, by
    /// design, is a pinning cursor using [`ArchiveOverrun::StallIngest`]
    /// that has not yet consumed the entry about to be evicted: this call
    /// blocks until that cursor consumes further, is dropped, or
    /// `STALL_INGEST_MAX_WAIT` elapses — see
    /// [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure)
    /// and `STALL_INGEST_MAX_WAIT`'s own doc for why the bound exists. If
    /// the bound elapses with the pin still not caught up, this pin is
    /// **terminated** — exactly [`ArchiveOverrun::Terminate`]'s own
    /// signal ([`SegmentCursorItem::Terminated`] on its cursor's next
    /// `poll`), reused rather than reinvented, because silently falling
    /// back to [`ArchiveOverrun::Gap`] would hide that a pinned (DVR)
    /// consumer's *stronger* guarantee was just broken behind the same
    /// "ordinary, expected" loss report an unpinned cursor gets by design.
    /// The block is a [`parking_lot::Condvar::wait_until`], which releases
    /// the shared `Mutex` while parked, so [`TrunkWriter::publish`] and
    /// every cursor's `poll` on *other* data remain free to proceed even
    /// while this call is stalled. [`Self::try_publish_segment`] is the
    /// non-blocking alternative for a caller that cannot afford to wait at
    /// all.
    ///
    /// Does **not** touch the live-part log — see
    /// [The live-part log](self#the-live-part-log-parts-before-their-segment-closes)
    /// for why a segment closing deliberately leaves that segment's parts
    /// exactly as addressable as they were the instant before. Wakes any
    /// [`Trunk::listen`] registration once this call is about to return
    /// (bare-`_HLS_msn` blocking-reload's condition), even on the
    /// `StallIngest` path — a waiter is woken only after the entry has
    /// actually landed, never merely because a pin released.
    ///
    /// # Errors
    ///
    /// Rejects (before touching anything else) a `sequence_number` that is
    /// not strictly greater than the last one accepted — see
    /// [`NonMonotonicSequenceNumber`] and [`Self::next_sequence_number`].
    pub fn publish_segment(&self, entry: SegmentEntry) -> Result<(), NonMonotonicSequenceNumber> {
        let mut state = self.trunk.lock_state();
        if let Some(last) = state.segments.entries.back().map(|e| e.sequence_number)
            && entry.sequence_number <= last
        {
            return Err(NonMonotonicSequenceNumber {
                attempted: entry.sequence_number,
                last_published: Some(last),
            });
        }
        let deadline = Instant::now() + STALL_INGEST_MAX_WAIT;
        loop {
            if state.segments.entries.len() < state.segments.capacity {
                // Room to push without evicting anything: no pin can be at
                // risk this round.
                break;
            }
            let oldest = state.segments.base;
            let mut must_wait = false;
            for pin in state.segments.pins.values_mut() {
                if pin.terminated || pin.consumed > oldest {
                    // Either already given up on (Terminate already fired),
                    // or this pin has already consumed the entry about to be
                    // evicted — not at risk.
                    continue;
                }
                match pin.policy {
                    // Nothing to do here: eviction proceeds, and the owning
                    // cursor's own `poll` reports the loss as `Gap` the same
                    // way a non-pinning cursor's `poll` reports it as
                    // ordinary `Lagged` — both read `base` vs. their own
                    // progress, after the fact.
                    ArchiveOverrun::Gap => {}
                    ArchiveOverrun::Terminate => pin.terminated = true,
                    ArchiveOverrun::StallIngest => must_wait = true,
                }
            }
            if !must_wait {
                break;
            }
            if Instant::now() >= deadline {
                // The bound elapsed: this `StallIngest` pin's guarantee
                // cannot be honoured any further. Terminate it — the exact
                // same signal `ArchiveOverrun::Terminate` already emits —
                // rather than silently falling through to an ordinary
                // eviction, which would report this as unremarkable `Gap`
                // loss and hide that a pinned consumer's stronger guarantee
                // was just broken. See `STALL_INGEST_MAX_WAIT`'s doc.
                for pin in state.segments.pins.values_mut() {
                    if !pin.terminated
                        && pin.consumed <= oldest
                        && pin.policy == ArchiveOverrun::StallIngest
                    {
                        pin.terminated = true;
                    }
                }
                break;
            }
            #[cfg(test)]
            self.trunk.stalled_publishers.fetch_add(1, Ordering::AcqRel);
            self.trunk
                .segment_pin_released
                .wait_until(&mut state, deadline);
            #[cfg(test)]
            self.trunk.stalled_publishers.fetch_sub(1, Ordering::AcqRel);
            // Loop back around: re-check capacity/oldest/pins after waking —
            // the pin that was blocking may have advanced, been dropped, or
            // (if a *different* pin also needed this entry) still be
            // pending.
        }
        state.segments.push(entry);
        drop(state);
        self.trunk.progress.notify(usize::MAX);
        Ok(())
    }

    /// The `sequence_number` a fresh or re-issued [`SegmentWriter`] must
    /// resume numbering from: the last one this `Trunk` accepted, plus one
    /// — or `1` (the first number, matching [`transmux::Segmenter`]'s own
    /// 1-based numbering) if this `Trunk` has never accepted a segment at
    /// all. A restarted segmenter that numbers from `1` again instead of
    /// calling this is exactly what [`NonMonotonicSequenceNumber`] exists to
    /// reject.
    pub fn next_sequence_number(&self) -> SegmentSeq {
        self.trunk
            .lock_state()
            .segments
            .entries
            .back()
            .map(|e| e.sequence_number.saturating_next())
            .unwrap_or(SegmentSeq::new(1))
    }

    /// Force every [`ArchiveOverrun::StallIngest`] pin currently blocking
    /// eviction of the oldest resident segment to give up — terminated
    /// exactly as [`Self::publish_segment`]'s own bounded wait would once
    /// `STALL_INGEST_MAX_WAIT` elapses (see that constant's doc: the same
    /// [`ArchiveOverrun::Terminate`] signal, reused, not a silent `Gap`
    /// downgrade), but callable proactively and **never blocks**.
    ///
    /// This is for a caller tracking its *own* elapsed wait on a queued
    /// backlog rather than a single blocking call — e.g. a non-blocking
    /// retry loop built on [`Self::try_publish_segment`] that has been
    /// retrying the same entry for its own bounded duration and wants to
    /// force the same give-up decision [`Self::publish_segment`] would have
    /// made by now, before retrying once more. A no-op (returns `0`) if
    /// nothing is currently blocking. Returns the number of pins
    /// terminated.
    pub fn expire_stalled_pins(&self) -> usize {
        let mut state = self.trunk.lock_state();
        if state.segments.entries.len() < state.segments.capacity {
            return 0;
        }
        let oldest = state.segments.base;
        let mut count = 0;
        for pin in state.segments.pins.values_mut() {
            if !pin.terminated
                && pin.consumed <= oldest
                && pin.policy == ArchiveOverrun::StallIngest
            {
                pin.terminated = true;
                count += 1;
            }
        }
        count
    }

    /// Non-blocking alternative to [`Self::publish_segment`] for the one
    /// case that call can stall on: an [`ArchiveOverrun::StallIngest`] pin
    /// that has not yet consumed the entry that would need evicting.
    /// Returns `Err`, handing `entry` straight back rather than blocking or
    /// losing it, if publishing right now would need to wait.
    ///
    /// This is the shape a caller whose own thread also drives the very
    /// [`crate::retention::RetentionDriver`] reading that pin needs: that
    /// caller *is* what would release the pin, so
    /// [`Self::publish_segment`]'s block cannot be serviced by anyone —
    /// see `STALL_INGEST_MAX_WAIT`'s doc. The expected caller shape is a
    /// retry loop that also drives the reader in between attempts:
    ///
    /// ```ignore
    /// let mut pending = Some(entry);
    /// while let Some(entry) = pending.take() {
    ///     match writer.try_publish_segment(entry) {
    ///         Ok(()) => break,
    ///         Err(entry) => {
    ///             retention_driver.drive(now);
    ///             pending = Some(entry);
    ///         }
    ///     }
    /// }
    /// ```
    ///
    /// Never blocks. Every other policy/path behaves identically to
    /// [`Self::publish_segment`] (including the [`ArchiveOverrun::Gap`]/
    /// [`ArchiveOverrun::Terminate`] handling and the wake on success), and
    /// an `Ok(())` result is indistinguishable from what
    /// [`Self::publish_segment`] would have done. Also rejects a
    /// non-monotonic `sequence_number` — see [`NonMonotonicSequenceNumber`].
    ///
    /// A [`ArchiveOverrun::Terminate`] pin is marked terminated only when an
    /// eviction actually happens: this call decides in two passes — first
    /// whether eviction can proceed at all (any [`ArchiveOverrun::StallIngest`]
    /// pin still blocking it), and only once that dry run says yes does it
    /// apply `Terminate`'s side effect and evict. A single-pass version that
    /// mutated `Terminate` pins while *also* discovering a blocking
    /// `StallIngest` pin would mark a pin terminated for an eviction that,
    /// because of that other pin, never actually happened this call.
    pub fn try_publish_segment(&self, entry: SegmentEntry) -> Result<(), TryPublishSegmentError> {
        let mut state = self.trunk.lock_state();
        if let Some(last) = state.segments.entries.back().map(|e| e.sequence_number)
            && entry.sequence_number <= last
        {
            return Err(TryPublishSegmentError::NonMonotonic(
                NonMonotonicSequenceNumber {
                    attempted: entry.sequence_number,
                    last_published: Some(last),
                },
            ));
        }
        if state.segments.entries.len() >= state.segments.capacity {
            let oldest = state.segments.base;
            // Dry run: would eviction be blocked? No pin is mutated yet.
            let must_wait = state.segments.pins.values().any(|pin| {
                !pin.terminated
                    && pin.consumed <= oldest
                    && pin.policy == ArchiveOverrun::StallIngest
            });
            if must_wait {
                return Err(TryPublishSegmentError::WouldStall(entry));
            }
            // Committed to evicting: now apply `Terminate`'s side effect for
            // real (`Gap` needs no action either way — see
            // `Self::publish_segment`'s identical comment).
            for pin in state.segments.pins.values_mut() {
                if !pin.terminated
                    && pin.consumed <= oldest
                    && pin.policy == ArchiveOverrun::Terminate
                {
                    pin.terminated = true;
                }
            }
        }
        state.segments.push(entry);
        drop(state);
        self.trunk.progress.notify(usize::MAX);
        Ok(())
    }

    /// Publish one live part of the segment currently being written — see
    /// [The live-part log](self#the-live-part-log-parts-before-their-segment-closes).
    ///
    /// Never blocks and never rejects: a full part log evicts its oldest
    /// entry exactly like every other ring in this module — the same
    /// non-blocking-producer principle as [`TrunkWriter::publish`]/
    /// [`SegmentWriter::publish_segment`]'s ordinary (non-`StallIngest`) path.
    /// Wakes any [`Trunk::listen`] registration once this part has actually
    /// landed (RFC 8216bis blocking-reload's part-availability condition).
    pub fn publish_part(&self, entry: PartEntry) {
        let mut state = self.trunk.lock_state();
        state.parts.push(entry);
        drop(state);
        self.trunk.progress.notify(usize::MAX);
    }

    /// Report that segment `segment_number` starts at `start` on this
    /// trunk's 90 kHz absolute clock — the boundary an
    /// [`EventAnchor::Segment`] (an `emsg` v0's `presentation_time_delta`)
    /// needs before it can resolve. Called by whoever owns segmentation —
    /// the entity the spec's B1 fix names explicitly: "it cannot be
    /// finalised until the segmenter owns a boundary." Lives here, not on
    /// [`TrunkWriter`], for exactly that reason: only the segmenter can
    /// honestly report it. This does **not** append to the event ring — it
    /// resolves an already-published [`EventAnchor::Segment`] entry in
    /// place, so grouping it with [`SegmentWriter::publish_segment`] does not
    /// create a second appender for [`TrunkWriter::publish_event`]'s ring;
    /// see [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk).
    pub fn note_segment_start(&self, segment_number: SegmentSeq, start: MediaTime) {
        let mut state = self.trunk.lock_state();
        state.events.note_segment_start(segment_number, start);
    }

    /// Give the event log a wall-clock↔media-clock mapping. Resolves every
    /// currently-pending [`EventAnchor::Utc`] entry immediately, and every
    /// future one at publish time, until a later call replaces it. Grouped
    /// with [`SegmentWriter::note_segment_start`] rather than split onto
    /// [`TrunkWriter`] — see
    /// [One writer per ring group](self#one-writer-per-ring-group-not-one-writer-per-trunk)
    /// for why, and the same in-place-resolution reasoning: this is not an
    /// append to the event ring either.
    pub fn set_time_anchor(&self, anchor: TimeAnchor) {
        let mut state = self.trunk.lock_state();
        state.events.set_time_anchor(anchor);
    }
}

impl Drop for SegmentWriter {
    /// Release this `SegmentWriter`'s slot so [`Trunk::segment_writer`] can
    /// issue a replacement — the segments+parts ring group's counterpart to
    /// `TrunkWriter`'s own `Drop`.
    fn drop(&mut self) {
        self.trunk
            .segment_writer_taken
            .store(false, Ordering::Release);
    }
}

/// One item [`SampleCursor::poll`] can hand back: data from either retention
/// class, or a loss report.
///
/// `#[non_exhaustive]`: this is the growth point for anything a cursor might
/// need to surface beyond "sample" or "loss" later, without a breaking
/// change to every match arm in the workspace.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum SampleCursorItem {
    /// A [`RetentionClass::Timed`] sample for `track_id`.
    Timed {
        /// The publishing track.
        track_id: TrackId,
        /// The sample itself. Cloned from the ring's stored copy —
        /// `Sample.data: Bytes` is shared, not copied; see
        /// [Zero-copy fan-out](self#zero-copy-fan-out-honestly).
        sample: Sample,
    },
    /// A [`RetentionClass::Sparse`] sample for `track_id`.
    Sparse {
        /// The publishing track.
        track_id: TrackId,
        /// The sample itself; see the `Timed` variant's doc for the
        /// zero-copy note.
        sample: Sample,
    },
    /// This cursor fell behind the `Timed` ring: `skipped` entries were
    /// evicted before it read them. Ordinary loss — resume from the next
    /// sample; see [`RetentionClass::Timed`].
    Lagged {
        /// Exact count of `Timed` entries evicted since this cursor's last
        /// successful read of that class.
        skipped: u64,
    },
    /// This cursor fell behind the `Sparse` ring: `skipped` entries were
    /// evicted before it read them. **Not** ordinary loss — the consumer's
    /// derived state (e.g. splice-point tracking) is now wrong, not merely
    /// gapped; see [`RetentionClass::Sparse`] for what a consumer is
    /// expected to do about it.
    Degraded {
        /// Exact count of `Sparse` entries evicted since this cursor's last
        /// successful read of that class.
        skipped: u64,
    },
    /// A [`TrunkWriter`] was dropped and a replacement issued (a source
    /// reconnect) — see [`Trunk::writer`]. Reported exactly once, positioned
    /// between the last sample the old writer published and the first the
    /// new one does: a consumer should treat whatever continuity it was
    /// deriving across this point (timestamp cadence, expected sequencing)
    /// as reset, the same "resynchronise before trusting derived state"
    /// contract [`SampleCursorItem::Degraded`] documents, but without any
    /// data having been lost — nothing was evicted, the writer itself was
    /// simply replaced. Never reported for a handover this cursor could not
    /// have observed any continuity across in the first place — see
    /// [`Trunk::subscribe`]'s own doc on that.
    Discontinuity,
    /// This cursor fell more than `WRITER_HANDOVER_LOG_CAPACITY` writer
    /// hand-offs behind: the specific `WriterHandover` marker(s) this
    /// cursor missed are gone (that backlog is bounded, unlike the
    /// `Timed`/`Sparse` rings themselves), so the exact position(s) a
    /// [`SampleCursorItem::Discontinuity`] would have been reported at are
    /// no longer known. **Not** the same signal as `Discontinuity` — this
    /// is a *loss* report (deliberately distinct from `Lagged`/`Degraded`
    /// too, which are about the `Timed`/`Sparse` rings specifically, not
    /// this bookkeeping): `skipped` writer hand-offs happened and cannot be
    /// individually reported, but no actual sample data was lost because of
    /// it.
    HandoversLagged {
        /// Exact count of `WriterHandover` markers evicted since this
        /// cursor's last successful check of that log.
        skipped: u64,
    },
}

/// A subscribed reader of a [`Trunk`]'s sample ring. Obtained via
/// [`Trunk::subscribe`] — **read that method's docs before creating more
/// than a handful of these.**
pub struct SampleCursor {
    trunk: Arc<Trunk>,
    /// How many `Timed` entries this cursor has consumed (returned via
    /// `poll`, or accounted for via a reported `Lagged`) since it
    /// subscribed. Compared against the shared `ClassLog::base` to detect
    /// loss — the same technique as `spikes/trunk-bench`'s `Cursor::read_seq`
    /// vs. `TrunkInner::base_seq`.
    timed_consumed: u64,
    /// The `Sparse`-class equivalent of `timed_consumed`.
    sparse_consumed: u64,
    /// How many `WriterHandover` markers this cursor has already reported
    /// (via [`SampleCursorItem::Discontinuity`]) or otherwise accounted for.
    /// Same `base`/`published`-comparison technique as `timed_consumed`/
    /// `sparse_consumed`, against [`WriterHandoverLog`] instead of a
    /// `ClassLog`.
    handovers_consumed: u64,
    /// This cursor's own `timed_consumed`/`sparse_consumed` at the moment it
    /// subscribed — fixed for the cursor's whole lifetime. A
    /// `WriterHandover` recorded at or before this position represents a
    /// writer replacement this cursor could not have observed any
    /// continuity across (it never read anything from before that point),
    /// so it is consumed silently rather than reported as
    /// [`SampleCursorItem::Discontinuity`] — see [`Trunk::subscribe`]'s doc.
    subscribe_timed_at: u64,
    /// The `Sparse`-class equivalent of `subscribe_timed_at`.
    subscribe_sparse_at: u64,
}

impl SampleCursor {
    /// Pull the next item, if any is ready.
    ///
    /// Loss is always reported before further data, in the same
    /// `Option<SampleCursorItem>` as real samples — following
    /// [`crate::byte_tap::TapItem`]'s precedent: a consumer cannot poll past
    /// a `Lagged`/`Degraded` report to reach the data that follows a gap,
    /// because there is no side channel it could forget to check instead.
    ///
    /// # Merge order across the two retention classes
    ///
    /// A pending `Timed` lag report is checked first, then a pending
    /// `Sparse` lag report, then a pending writer hand-off, then a ready
    /// `Sparse` sample, then a ready `Timed` sample. This gives **no
    /// cross-class chronological interleave guarantee** (see the
    /// [module docs](self) for why that is not something anything
    /// downstream needs) — only that, within each class, entries are
    /// returned in the exact order [`TrunkWriter::publish`] produced them,
    /// with no duplication and no unreported loss.
    ///
    /// # Neither class ever delivers the new writer's data ahead of the
    /// other's still-pending `Discontinuity`
    ///
    /// A pending `WriterHandover` gates **both** classes independently on
    /// its own per-class position: once this cursor's progress on a class
    /// reaches that handover's recorded position for it, that class's data
    /// from the *new* writer is held back — not served — until the
    /// `Discontinuity` itself can fire (i.e. until the *other* class has
    /// also caught up). Without this, a writer replacement that published
    /// only to one class (e.g. a `Sparse` cue with no new `Timed` sample
    /// yet) could let that class's new-writer item overtake the old
    /// writer's still-undelivered `Timed` backlog and the `Discontinuity`
    /// itself, handing a consumer post-handover data before it had even
    /// been told the handover happened.
    pub fn poll(&mut self) -> Option<SampleCursorItem> {
        let state = self.trunk.lock_state();

        if self.timed_consumed < state.timed.base {
            let skipped = state.timed.base - self.timed_consumed;
            self.timed_consumed = state.timed.base;
            return Some(SampleCursorItem::Lagged { skipped });
        }
        if self.sparse_consumed < state.sparse.base {
            let skipped = state.sparse.base - self.sparse_consumed;
            self.sparse_consumed = state.sparse.base;
            return Some(SampleCursorItem::Degraded { skipped });
        }
        if self.handovers_consumed < state.handovers.base {
            // Fell behind the (small, fixed) handover backlog itself: the
            // specific `WriterHandover` marker(s) are gone, so this cursor
            // can no longer report the discontinuity at its exact position
            // — but it must not silently catch up and say nothing, which
            // would hide a real writer hand-off as if it never happened.
            // Its own distinct loss variant, not `Lagged`/`Degraded` (those
            // are about the `Timed`/`Sparse` rings) — see
            // `WRITER_HANDOVER_LOG_CAPACITY`'s doc.
            let skipped = state.handovers.base - self.handovers_consumed;
            self.handovers_consumed = state.handovers.base;
            return Some(SampleCursorItem::HandoversLagged { skipped });
        }

        // The oldest still-unreported handover, if any — used both to
        // decide whether it can fire yet (below) and, further down, to
        // gate new-writer data on EITHER class from overtaking it (see
        // this method's own doc).
        let mut pending_handover = if self.handovers_consumed < state.handovers.published {
            let idx =
                ring_offset(self.handovers_consumed, state.handovers.base).unwrap_or(usize::MAX);
            state.handovers.entries.get(idx).copied()
        } else {
            None
        };
        while let Some(handover) = pending_handover {
            if !(self.timed_consumed >= handover.timed_at
                && self.sparse_consumed >= handover.sparse_at)
            {
                break;
            }
            self.handovers_consumed += 1;
            if handover.timed_at > self.subscribe_timed_at
                || handover.sparse_at > self.subscribe_sparse_at
            {
                return Some(SampleCursorItem::Discontinuity);
            }
            // Suppressed (see `Trunk::subscribe`'s doc): this cursor started
            // at or after this exact hand-off, so it never had continuity
            // to lose across it. Check the next pending one, if any — a
            // back-to-back reconnect with no data published in between can
            // stack more than one such suppressed marker.
            pending_handover = if self.handovers_consumed < state.handovers.published {
                let idx = ring_offset(self.handovers_consumed, state.handovers.base)
                    .unwrap_or(usize::MAX);
                state.handovers.entries.get(idx).copied()
            } else {
                None
            };
        }

        let sparse_blocked = pending_handover.is_some_and(|h| self.sparse_consumed >= h.sparse_at);
        if !sparse_blocked {
            let sparse_idx =
                ring_offset(self.sparse_consumed, state.sparse.base).unwrap_or(usize::MAX);
            if let Some((track_id, sample)) = state.sparse.entries.get(sparse_idx) {
                self.sparse_consumed += 1;
                return Some(SampleCursorItem::Sparse {
                    track_id: *track_id,
                    sample: sample.clone(),
                });
            }
        }

        let timed_blocked = pending_handover.is_some_and(|h| self.timed_consumed >= h.timed_at);
        if !timed_blocked {
            let timed_idx =
                ring_offset(self.timed_consumed, state.timed.base).unwrap_or(usize::MAX);
            if let Some((track_id, sample)) = state.timed.entries.get(timed_idx) {
                self.timed_consumed += 1;
                return Some(SampleCursorItem::Timed {
                    track_id: *track_id,
                    sample: sample.clone(),
                });
            }
        }

        None
    }
}

/// One item [`SegmentCursor::poll`] can hand back: a finished segment, or a
/// loss report.
///
/// `#[non_exhaustive]`: this is the growth point for anything a segment
/// cursor might need to surface beyond "segment" or "loss" later, without a
/// breaking change to every match arm in the workspace.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum SegmentCursorItem {
    /// One finished segment, in playlist order.
    Segment(SegmentEntry),
    /// A **non-pinning** cursor (from [`Trunk::subscribe_segments`]) fell
    /// behind the segment log's ordinary [`TrunkConfig::segment_capacity`]
    /// eviction: `skipped` segments were evicted before it read them.
    /// Ordinary loss, exactly [`SampleCursorItem::Lagged`]'s contract —
    /// resume from the next segment.
    Lagged {
        /// Exact count of segments evicted since this cursor's last
        /// successful read.
        skipped: u64,
    },
    /// A **pinning** cursor's (from [`Trunk::pin_segments`])
    /// [`ArchiveOverrun::Gap`] policy fired: the log evicted `skipped`
    /// segments this cursor had not yet consumed rather than let its pin
    /// grow retention without bound. Unlike `Lagged`, this is the defect a
    /// DVR consumer must record as a hole in the archive — see
    /// [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure).
    Gap {
        /// Exact count of segments evicted out from under this cursor's
        /// pin.
        skipped: u64,
    },
    /// This **pinning** cursor's [`ArchiveOverrun::Terminate`] policy fired:
    /// the log dropped its pin instead of gapping the recording or
    /// stalling ingest. This is the last item this cursor will ever yield —
    /// every `poll` after this one returns `None`.
    Terminated,
}

/// A subscribed reader of a [`Trunk`]'s segment log. Obtained via
/// [`Trunk::subscribe_segments`] (ordinary, lossy-on-overflow) or
/// [`Trunk::pin_segments`] (pinning, [`ArchiveOverrun`]-governed) — **read
/// those methods' docs, and [The DVR contradiction](self#the-dvr-contradiction-losslessness-from-retention-not-back-pressure),
/// before creating more than a handful of these.**
pub struct SegmentCursor {
    trunk: Arc<Trunk>,
    /// Read progress for a **non-pinning** cursor (`pin_id.is_none()`) —
    /// exactly [`SampleCursor`]'s local `*_consumed` fields. Unused (and left
    /// at `0`) for a pinning cursor, whose progress instead lives in the
    /// shared `PinState::consumed` the writer must be able to see; see
    /// [`SegmentLog`].
    consumed: u64,
    /// `Some(id)` for a pinning cursor — the key into
    /// `TrunkState::segments.pins` this cursor's progress and policy are
    /// recorded under. `None` for an ordinary [`Trunk::subscribe_segments`]
    /// cursor.
    pin_id: Option<u64>,
    /// Set once this cursor has reported [`SegmentCursorItem::Terminated`] —
    /// every `poll` after that returns `None` rather than re-reporting it or
    /// resuming as if nothing happened.
    done: bool,
}

impl SegmentCursor {
    /// Pull the next item, if any is ready.
    ///
    /// Loss is always reported before further data, in the same
    /// `Option<SegmentCursorItem>` as real segments — the same
    /// cannot-be-skipped-past precedent as [`SampleCursor::poll`]/
    /// [`crate::byte_tap::TapItem`].
    pub fn poll(&mut self) -> Option<SegmentCursorItem> {
        if self.done {
            return None;
        }

        let Some(pin_id) = self.pin_id else {
            // Non-pinning: local `consumed`, exactly `SampleCursor::poll`'s
            // shape, against the one segment log instead of two class rings.
            let state = self.trunk.lock_state();
            if self.consumed < state.segments.base {
                let skipped = state.segments.base - self.consumed;
                self.consumed = state.segments.base;
                return Some(SegmentCursorItem::Lagged { skipped });
            }
            let idx = ring_offset(self.consumed, state.segments.base).unwrap_or(usize::MAX);
            return if let Some(entry) = state.segments.entries.get(idx) {
                self.consumed += 1;
                Some(SegmentCursorItem::Segment(entry.clone()))
            } else {
                None
            };
        };

        // Pinning: progress lives in the shared `PinState`, because
        // `SegmentWriter::publish_segment` has to consult it before evicting,
        // not merely react to it afterward.
        let mut state = self.trunk.lock_state();
        let Some(pin) = state.segments.pins.get(&pin_id) else {
            // Already removed (defensive: `Drop`/prior `Terminated` report
            // should make this unreachable in practice) — treat as done.
            self.done = true;
            return None;
        };
        if pin.terminated {
            state.segments.pins.remove(&pin_id);
            self.pin_id = None;
            self.done = true;
            return Some(SegmentCursorItem::Terminated);
        }
        let consumed = pin.consumed;
        if consumed < state.segments.base {
            let skipped = state.segments.base - consumed;
            let base = state.segments.base;
            let Some(pin) = state.segments.pins.get_mut(&pin_id) else {
                debug_assert!(false, "pin_id was resolved from this same locked state");
                return None;
            };
            pin.consumed = base;
            drop(state);
            // A pin advancing can free a `StallIngest` writer waiting on
            // exactly this pin.
            self.trunk.segment_pin_released.notify_all();
            return Some(SegmentCursorItem::Gap { skipped });
        }
        let idx = ring_offset(consumed, state.segments.base).unwrap_or(usize::MAX);
        if let Some(entry) = state.segments.entries.get(idx) {
            let item = entry.clone();
            let Some(pin) = state.segments.pins.get_mut(&pin_id) else {
                debug_assert!(false, "pin_id was resolved from this same locked state");
                return None;
            };
            pin.consumed = pin.consumed.saturating_add(1);
            drop(state);
            self.trunk.segment_pin_released.notify_all();
            return Some(SegmentCursorItem::Segment(item));
        }
        None
    }
}

impl Drop for SegmentCursor {
    /// Release this cursor's pin, if it has one, so a dropped/abandoned
    /// pinning cursor cannot hold retention open (or a `StallIngest` writer
    /// blocked) forever — the same "a dead consumer must not grow memory
    /// without limit" guarantee as an actively-`Gap`-ping cursor, for the
    /// case where the consumer disappeared instead of choosing a policy.
    fn drop(&mut self) {
        if let Some(pin_id) = self.pin_id.take() {
            let mut state = self.trunk.lock_state();
            state.segments.pins.remove(&pin_id);
            drop(state);
            self.trunk.segment_pin_released.notify_all();
        }
    }
}

/// One item [`EventCursor::poll`] can hand back: one event-log entry (which
/// may itself still be `Segment`/`Utc`-anchored — a cursor sees an entry
/// the instant it is published, not only once it resolves; see
/// [`EventEntry::anchor`]), or a loss report.
///
/// `#[non_exhaustive]`: the growth point for anything a cursor might need
/// to surface beyond "entry" or "loss" later, without a breaking change to
/// every match arm in the workspace.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum EventCursorItem {
    /// One event-log entry, in publish order.
    Event(EventEntry),
    /// This cursor fell behind the event log's ordinary
    /// [`TrunkConfig::event_capacity`] eviction: `skipped` entries were
    /// evicted before it read them. Exactly [`SampleCursorItem::Lagged`]'s
    /// contract.
    Lagged {
        /// Exact count of entries evicted since this cursor's last
        /// successful read.
        skipped: u64,
    },
}

/// A subscribed reader of a [`Trunk`]'s event log. Obtained via
/// [`Trunk::subscribe_events`] — read that method's docs, and
/// [`Trunk::subscribe`]'s fan-out guidance, before creating more than a
/// handful of these.
pub struct EventCursor {
    trunk: Arc<Trunk>,
    consumed: u64,
}

impl EventCursor {
    /// Pull the next item, if any is ready. Loss is always reported before
    /// further data — the same cannot-be-skipped-past precedent as
    /// [`SampleCursor::poll`]/[`SegmentCursor::poll`].
    pub fn poll(&mut self) -> Option<EventCursorItem> {
        let state = self.trunk.lock_state();
        let log = &state.events;
        if self.consumed < log.base {
            let skipped = log.base - self.consumed;
            self.consumed = log.base;
            return Some(EventCursorItem::Lagged { skipped });
        }
        let idx = ring_offset(self.consumed, log.base).unwrap_or(usize::MAX);
        if let Some(entry) = log.entries.get(idx) {
            self.consumed += 1;
            return Some(EventCursorItem::Event(entry.clone()));
        }
        None
    }
}

/// Test helper (also used by `retention`'s tests): spin, yielding and never
/// sleeping, until `n` publishers are parked in `StallIngest`'s wait. Bounded
/// so a publisher that never parks fails the test instead of hanging it.
#[cfg(test)]
pub(crate) fn wait_until_stalled(trunk: &Trunk, n: usize) {
    let deadline = Instant::now() + Duration::from_secs(60);
    while trunk.stalled_publishers.load(Ordering::Acquire) < n {
        assert!(
            Instant::now() < deadline,
            "publisher never parked in StallIngest"
        );
        std::thread::yield_now();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::mpsc;
    use std::thread;
    use transmux::pipeline::{CodecConfig, DataCarriage};

    /// An opaque `TrackSpec` for track-set tests — mirrors `ingress`'s own
    /// identically-named test helper (same shape, so a track built here and
    /// one built there compare equal field-for-field for any given
    /// `track_id`).
    fn opaque_track(track_id: TrackId) -> TrackSpec {
        TrackSpec::new(
            track_id,
            90_000,
            CodecConfig::Data {
                stream_type: 0x06,
                descriptors: Vec::new(),
                carriage: DataCarriage::Pes,
            },
        )
    }

    /// `NonZeroUsize` from a literal capacity, for readability at the ~30
    /// `TrunkConfig::new` call sites below. Panicking on `0` here is correct
    /// and is *not* the behaviour the deleted `zero_*_capacity_panics` tests
    /// asserted: this is a test helper rejecting a typo in test source, not
    /// the library accepting then rejecting a zero at run time — the library
    /// can no longer be handed one at all.
    fn nz(n: usize) -> NonZeroUsize {
        NonZeroUsize::new(n).expect("test capacity must be non-zero")
    }

    fn sample(byte: u8, len: usize) -> Sample {
        Sample::new(Bytes::from(vec![byte; len]), Some(0), Some(0), None, true)
    }

    fn timed_data(item: &SampleCursorItem) -> Option<(TrackId, &Sample)> {
        match item {
            SampleCursorItem::Timed { track_id, sample } => Some((*track_id, sample)),
            _ => None,
        }
    }

    fn segment_entry(byte: u8, seq: u32) -> SegmentEntry {
        SegmentEntry::new(
            Bytes::from(vec![byte; 16]),
            SegmentSeq::new(seq),
            Duration::from_secs(2),
            Timestamp::from_nanos(u64::from(seq) * 2_000_000_000),
            SegmentMeta {
                discontinuous: false,
            },
        )
    }

    /// #1188 overflow behaviour: at `u32::MAX` the resumption point
    /// saturates (it can never wrap to a number that would re-pass the
    /// monotonic guard), and a further segment is rejected, not accepted
    /// with a wrapped or duplicate number.
    #[test]
    fn segment_sequence_at_u32_max_saturates_and_rejects_the_next() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.segment_writer().unwrap();
        writer.publish_segment(segment_entry(1, u32::MAX)).unwrap();
        let max = SegmentSeq::new(u32::MAX);
        assert_eq!(writer.next_sequence_number(), max);
        assert_eq!(max.checked_next(), None);
        assert_eq!(
            writer
                .publish_segment(segment_entry(2, u32::MAX))
                .unwrap_err(),
            NonMonotonicSequenceNumber {
                attempted: max,
                last_published: Some(max),
            }
        );
        // A wrapped number is also refused.
        assert!(matches!(
            writer.publish_segment(segment_entry(3, 0)),
            Err(NonMonotonicSequenceNumber { .. })
        ));
        assert_eq!(trunk.last_closed_segment(), Some(max));
    }

    fn segment_data(item: &SegmentCursorItem) -> Option<&SegmentEntry> {
        match item {
            SegmentCursorItem::Segment(entry) => Some(entry),
            _ => None,
        }
    }

    /// Drains up to `n` items from `cursor`, stopping early if `poll`
    /// returns `None` — see [`drain`]'s doc for why this is bounded rather
    /// than looping until `None`.
    fn drain_segments(cursor: &mut SegmentCursor, n: usize) -> Vec<SegmentCursorItem> {
        let mut out = Vec::new();
        for _ in 0..n {
            match cursor.poll() {
                Some(item) => out.push(item),
                None => break,
            }
        }
        out
    }

    /// Drains up to `n` items from `cursor`, stopping early if `poll`
    /// returns `None` — a bounded collection loop so a mutation that never
    /// advances `*_consumed` (and would otherwise re-yield the same item
    /// forever) fails the test's length/content assertions instead of
    /// hanging it.
    fn drain(cursor: &mut SampleCursor, n: usize) -> Vec<SampleCursorItem> {
        let mut out = Vec::new();
        for _ in 0..n {
            match cursor.poll() {
                Some(item) => out.push(item),
                None => break,
            }
        }
        out
    }

    // --- 1. multiple cursors, every sample, in order, no dup/no loss -----

    /// MUTATION VERIFIED: removing `self.timed_consumed += 1;` from the
    /// `Timed`-data return arm of `SampleCursor::poll` (so the same ring
    /// index is re-read every call) makes this test fail — `drain` still
    /// returns exactly 5 items (poll never runs out), but they are five
    /// copies of the first published sample (`byte = 0`) instead of the
    /// distinct sequence `0..5`, so the `assert_eq!` on the reconstructed
    /// byte sequence fails with a mismatch at index 1. Recompiled and
    /// re-run to confirm the failure, then reverted.
    #[test]
    fn multiple_cursors_see_every_sample_in_order_with_no_dup_or_loss() {
        let trunk = Trunk::new(TrunkConfig::new(nz(100), nz(10), nz(4), nz(8), nz(8)));
        let mut c1 = trunk.subscribe();
        let mut c2 = trunk.subscribe();
        let mut c3 = trunk.subscribe();
        let writer = trunk.writer().unwrap();

        for i in 0u8..5 {
            writer.publish(TrackId::new(7), RetentionClass::Timed, sample(i, 16));
        }

        for cursor in [&mut c1, &mut c2, &mut c3] {
            let items = drain(cursor, 5);
            assert_eq!(items.len(), 5, "each cursor must see exactly 5 samples");
            let bytes: Vec<u8> = items
                .iter()
                .map(|item| timed_data(item).unwrap().1.data[0])
                .collect();
            assert_eq!(bytes, vec![0, 1, 2, 3, 4], "must be in publish order");
            assert!(cursor.poll().is_none(), "no extra/duplicated items");
        }
    }

    // --- poison recovery (issue #1134): every lock site survives a panic ---

    /// One consumer panicking while it holds the trunk's state lock must not
    /// turn every later writer/reader call — including a cursor's `Drop` —
    /// into a panic. Poisons the mutex for real (a thread panics holding the
    /// guard), then drives every public entry point and asserts literal
    /// results. Bite-check: reverting `lock_state` to
    /// `.lock().expect("poisoned")` makes the first call after the poisoning
    /// panic.
    #[test]
    fn every_entry_point_survives_a_poisoned_state_lock() {
        let trunk = Trunk::new(TrunkConfig::new(nz(8), nz(8), nz(8), nz(8), nz(8)));
        let mut samples = trunk.subscribe();
        let mut segments = trunk.subscribe_segments();
        let mut events = trunk.subscribe_events();
        let writer = trunk.writer().unwrap();
        let segment_writer = trunk.segment_writer().unwrap();

        let t2 = Arc::clone(&trunk);
        let poisoner = thread::spawn(move || {
            let _guard = t2.lock_state();
            panic!("deliberately poisoning the trunk state lock");
        });
        assert!(
            poisoner.join().is_err(),
            "poisoner thread must have panicked"
        );
        assert!(
            trunk.state.try_lock().is_some(),
            "precondition: the lock is free (and, being parking_lot, unpoisoned) after the holder unwound"
        );

        writer.publish(TrackId::new(7), RetentionClass::Timed, sample(5, 4));
        match samples.poll() {
            Some(SampleCursorItem::Timed { track_id, sample }) => {
                assert_eq!(track_id, TrackId::new(7));
                assert_eq!(sample.data.as_ref(), &[5u8; 4]);
            }
            other => panic!("expected the published sample, got {other:?}"),
        }
        segment_writer.publish_segment(segment_entry(9, 1)).unwrap();
        assert_eq!(trunk.last_closed_segment(), Some(SegmentSeq::new(1)));
        match segments.poll() {
            Some(SegmentCursorItem::Segment(e)) => {
                assert_eq!(e.sequence_number, SegmentSeq::new(1))
            }
            other => panic!("expected segment 1, got {other:?}"),
        }
        segment_writer.publish_part(part_entry(3, 2, 0));
        assert_eq!(
            trunk
                .part_bytes(SegmentSeq::new(2), PartIndex::new(0))
                .unwrap()
                .as_ref(),
            &[3u8; 8]
        );
        writer.publish_event(basic_event(4), EventAnchor::Media(MediaTime(10)));
        assert!(matches!(events.poll(), Some(EventCursorItem::Event(_))));
        assert_eq!(trunk.events_between(MediaTime(0), MediaTime(20)).len(), 1);
        // Drop paths take the lock too and must not panic.
        drop(samples);
        drop(segments);
        drop(events);
        drop(writer);
        drop(segment_writer);
        assert!(
            trunk.writer().is_some(),
            "slot released by Drop under poison"
        );
    }

    // --- 2. slow reader lags, writer completes regardless -----------------

    /// MUTATION VERIFIED: changing `ClassLog::push`'s eviction condition from
    /// `self.entries.len() == self.capacity` to `false` (i.e. disabling
    /// eviction, simulating a writer that would instead have to wait/reject
    /// once "full") makes `trunk.timed_len()` grow to 1024 instead of
    /// staying at the configured cap of 4, and the lag report's `skipped`
    /// reads back as `0` (base never advances), not `1020`. Recompiled and
    /// re-run to confirm the failure, then reverted.
    #[test]
    fn slow_reader_lags_but_writer_completes_regardless() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(10), nz(4), nz(8), nz(8)));
        let mut slow = trunk.subscribe();
        let writer = trunk.writer().unwrap();

        // The slow reader never polls while 1024 samples are published —
        // there is no wait-for-reader code path in `publish` for this loop
        // to block on (see `TrunkWriter`'s docs), so this simply completes.
        // A single thread is sufficient to demonstrate this: the absence of
        // a blocking path is a structural property of `publish`, not a race
        // that needs real concurrency to expose (`crate::byte_tap`'s
        // equivalent test uses the same reasoning).
        for i in 0u8..=255u8 {
            for _ in 0..4 {
                writer.publish(TrackId::new(1), RetentionClass::Timed, sample(i, 8));
            }
        }
        // 256 * 4 = 1024 published; ring capacity is 4.
        assert_eq!(
            trunk.timed_len(),
            4,
            "writer unblocked: ring stayed bounded"
        );

        let first = slow.poll().unwrap();
        assert!(
            matches!(first, SampleCursorItem::Lagged { skipped: 1020 }),
            "expected Lagged{{skipped: 1020}}, got {first:?}"
        );
    }

    // --- 3. lag reports an accurate skipped count -------------------------

    /// MUTATION VERIFIED: changing the `skipped` computation in
    /// `SampleCursor::poll`'s `Timed`-lag branch from
    /// `state.timed.base - self.timed_consumed` to
    /// `state.timed.base - self.timed_consumed + 1` makes this test fail:
    /// expected `skipped: 6`, got `skipped: 7`. Recompiled and re-run to
    /// confirm the failure, then reverted.
    #[test]
    fn lag_is_reported_with_an_accurate_skipped_count() {
        let trunk = Trunk::new(TrunkConfig::new(nz(3), nz(10), nz(4), nz(8), nz(8)));
        let mut cursor = trunk.subscribe();
        let writer = trunk.writer().unwrap();

        // Capacity 3, publish 9: 6 evicted before the cursor ever reads.
        for i in 0u8..9 {
            writer.publish(TrackId::new(2), RetentionClass::Timed, sample(i, 4));
        }

        let first = cursor.poll().unwrap();
        assert!(
            matches!(first, SampleCursorItem::Lagged { skipped: 6 }),
            "expected Lagged{{skipped: 6}}, got {first:?}"
        );

        // The remaining 3 (bytes 6,7,8) must still be readable, in order.
        let items = drain(&mut cursor, 3);
        let bytes: Vec<u8> = items
            .iter()
            .map(|item| timed_data(item).unwrap().1.data[0])
            .collect();
        assert_eq!(bytes, vec![6, 7, 8]);
        assert!(cursor.poll().is_none());
    }

    // --- 4. Sparse loss reports Degraded, distinct from Timed's Lagged ----

    /// MUTATION VERIFIED: changing the `Sparse`-lag branch of
    /// `SampleCursor::poll` to also return `SampleCursorItem::Lagged` (i.e.
    /// collapsing the two variants) makes the
    /// `matches!(item, SampleCursorItem::Degraded { .. })` assertion below
    /// fail — the item is a `Lagged` instead. Recompiled and re-run to
    /// confirm the failure, then reverted.
    #[test]
    fn sparse_reader_loses_data_reports_degraded_distinguishable_from_timed_lagged() {
        let trunk = Trunk::new(TrunkConfig::new(nz(2), nz(2), nz(4), nz(8), nz(8)));
        let mut cursor = trunk.subscribe();
        let writer = trunk.writer().unwrap();

        // Overflow the Timed ring (cap 2) with 5 publishes: ordinary loss.
        for i in 0u8..5 {
            writer.publish(TrackId::new(3), RetentionClass::Timed, sample(i, 4));
        }
        // Overflow the Sparse ring (cap 2) with 4 publishes: escalated loss.
        for i in 0u8..4 {
            writer.publish(TrackId::new(9), RetentionClass::Sparse, sample(100 + i, 4));
        }

        let timed_loss = cursor.poll().unwrap();
        assert!(
            matches!(timed_loss, SampleCursorItem::Lagged { skipped: 3 }),
            "expected ordinary Lagged{{skipped: 3}} for the Timed ring, got {timed_loss:?}"
        );

        let sparse_loss = cursor.poll().unwrap();
        assert!(
            matches!(sparse_loss, SampleCursorItem::Degraded { skipped: 2 }),
            "expected escalated Degraded{{skipped: 2}} for the Sparse ring, got {sparse_loss:?}"
        );
        assert_ne!(
            core::mem::discriminant(&timed_loss),
            core::mem::discriminant(&sparse_loss),
            "Lagged and Degraded must be distinct variants, not merely different field values"
        );
    }

    // --- 5. the ring is bounded: flooding cannot grow memory unboundedly --

    /// MUTATION VERIFIED: removing the eviction check in `ClassLog::push`
    /// (replacing `if self.entries.len() == self.capacity { .. }` with a
    /// no-op) makes `trunk.timed_len()`/`trunk.sparse_len()` grow well past
    /// the configured caps (`4`/`3`) instead of staying bounded — the
    /// assertions inside the flood loop below fail on the first
    /// over-capacity iteration. Recompiled and re-run to confirm the
    /// failure, then reverted.
    #[test]
    fn ring_is_bounded_under_flood_on_both_classes() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(3), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();

        for i in 0u32..50_000 {
            writer.publish(
                TrackId::new(5),
                RetentionClass::Timed,
                sample((i % 256) as u8, 2),
            );
            assert!(
                trunk.timed_len() <= 4,
                "Timed ring exceeded its cap mid-flood"
            );
            if i % 7 == 0 {
                writer.publish(
                    TrackId::new(6),
                    RetentionClass::Sparse,
                    sample((i % 256) as u8, 2),
                );
                assert!(
                    trunk.sparse_len() <= 3,
                    "Sparse ring exceeded its cap mid-flood"
                );
            }
        }
        assert_eq!(trunk.timed_len(), 4);
        assert_eq!(trunk.sparse_len(), 3);
    }

    // --- 5b. subscribe_from_backlog: exact replay + Lagged on overwrite ---

    /// [`Trunk::subscribe_from_backlog`]'s core promise: a cursor subscribed
    /// *after* samples have already landed in both rings still sees them,
    /// exactly (in order, no dup), for each retention class independently —
    /// the property issue #808's `ProgramSegmenter` fix depends on.
    ///
    /// MUTATION VERIFIED: changing `subscribe_from_backlog`'s
    /// `timed_consumed: state.timed.base` to
    /// `timed_consumed: state.timed.published` (i.e. accidentally reusing
    /// `subscribe`'s live-tail initialisation) makes this test's
    /// `assert_eq!(timed_bytes, vec![10, 11, 12])` fail: `drain` returns an
    /// empty `Vec` (actual) instead of the expected `[10, 11, 12]`, because
    /// `timed_consumed` now equals `published` — every published entry
    /// already counts as "consumed" the instant the cursor is created, so
    /// `poll` immediately returns `None` instead of replaying the resident
    /// backlog. Recompiled and re-run to confirm this exact failure, then
    /// reverted.
    #[test]
    fn subscribe_from_backlog_replays_exact_resident_entries_both_classes() {
        let trunk = Trunk::new(TrunkConfig::new(nz(100), nz(100), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();

        // Published *before* the cursor exists — subscribe() would never see
        // any of this; subscribe_from_backlog() must replay all of it, since
        // ring capacity (100) is nowhere near exhausted.
        for i in 10u8..13 {
            writer.publish(TrackId::new(1), RetentionClass::Timed, sample(i, 4));
        }
        for i in 20u8..22 {
            writer.publish(TrackId::new(2), RetentionClass::Sparse, sample(i, 4));
        }

        let mut cursor = trunk.subscribe_from_backlog();

        // Merge order (module docs): pending lag reports first (none here),
        // then Sparse data, then Timed data.
        let sparse_items = drain(&mut cursor, 2);
        let sparse_bytes: Vec<u8> = sparse_items
            .iter()
            .map(|item| match item {
                SampleCursorItem::Sparse { sample, .. } => sample.data[0],
                other => panic!("expected Sparse, got {other:?}"),
            })
            .collect();
        assert_eq!(sparse_bytes, vec![20, 21], "exact resident Sparse backlog");

        let timed_items = drain(&mut cursor, 3);
        let timed_bytes: Vec<u8> = timed_items
            .iter()
            .map(|item| timed_data(item).unwrap().1.data[0])
            .collect();
        assert_eq!(
            timed_bytes,
            vec![10, 11, 12],
            "exact resident Timed backlog"
        );

        assert!(
            cursor.poll().is_none(),
            "no extra items beyond the resident backlog"
        );

        // New samples published after subscribing still flow through
        // normally, proving this cursor is a real live cursor afterwards,
        // not a one-shot snapshot.
        writer.publish(TrackId::new(1), RetentionClass::Timed, sample(99, 4));
        let live = cursor.poll().unwrap();
        assert_eq!(timed_data(&live).unwrap().1.data[0], 99);
    }

    /// When the backlog a [`Trunk::subscribe_from_backlog`] cursor would
    /// have replayed has *already* been evicted by ring capacity before the
    /// subscribe call, this cursor must behave exactly like an ordinary
    /// [`Trunk::subscribe`] cursor from that point on: report the loss
    /// in-band as `Lagged`/`Degraded`, never silently skip it.
    ///
    /// MUTATION VERIFIED: changing `subscribe_from_backlog`'s
    /// `timed_consumed: state.timed.base` to `timed_consumed: 0` (simulating
    /// "replay from the beginning of time" rather than "replay what the ring
    /// still holds", anchoring at the wrong reference point) makes this
    /// test's `assert!(matches!(first, SampleCursorItem::Lagged { skipped: 6
    /// }))` fail: actual `Lagged { skipped: 12 }` (`12 - 0`, counting the 6
    /// entries already evicted *before* this cursor even subscribed as if
    /// they were its own loss) instead of the expected `Lagged { skipped: 6
    /// }` (`12 - 6`, only the 6 evictions that happened *after* this cursor
    /// subscribed). Recompiled and re-run to confirm this exact failure,
    /// then reverted.
    #[test]
    fn subscribe_from_backlog_reports_lagged_when_backlog_already_overwritten() {
        let trunk = Trunk::new(TrunkConfig::new(nz(3), nz(10), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();

        // Capacity 3, publish 9: only bytes 6,7,8 remain resident; 0..6 are
        // already gone by the time subscribe_from_backlog is called — this
        // cursor must NOT report those 6 as its own loss (they were never
        // its backlog to miss), which is exactly why it anchors at `base`
        // (6), not `0`.
        for i in 0u8..9 {
            writer.publish(TrackId::new(1), RetentionClass::Timed, sample(i, 4));
        }
        let mut cursor = trunk.subscribe_from_backlog();

        // Publish 6 more before this cursor ever polls — the ring is
        // already full (capacity 3), so each push evicts exactly one
        // resident entry, advancing `base` from 6 to 12. This IS loss this
        // cursor is responsible for (it subscribed to a live cursor at 6,
        // then fell behind by 6 before its first poll).
        for i in 9u8..15 {
            writer.publish(TrackId::new(1), RetentionClass::Timed, sample(i, 4));
        }

        let first = cursor.poll().unwrap();
        assert!(
            matches!(first, SampleCursorItem::Lagged { skipped: 6 }),
            "expected Lagged{{skipped: 6}}, got {first:?}"
        );

        // The remaining resident 3 (bytes 12,13,14) must still be readable.
        let items = drain(&mut cursor, 3);
        let bytes: Vec<u8> = items
            .iter()
            .map(|item| timed_data(item).unwrap().1.data[0])
            .collect();
        assert_eq!(bytes, vec![12, 13, 14]);
        assert!(cursor.poll().is_none());
    }

    // --- 6. payload sharing: Bytes::as_ptr() identity, not equality -------

    /// MUTATION VERIFIED: replacing `sample.clone()` in both of
    /// `SampleCursor::poll`'s data-return arms with a hand-rolled copy
    /// (`Sample::new(Bytes::copy_from_slice(sample.data.as_ref()), ...)`,
    /// preserving every field's *value* so content-equality would still
    /// hold) makes this test's pointer-identity assertion fail — with the
    /// mutation, `p1 == p2` is `false` (two distinct heap allocations with
    /// equal contents), whereas the unmutated `clone()` path yields
    /// `p1 == p2 == p3`. This is precisely the distinction a
    /// content-equality assertion would have missed. Recompiled and re-run
    /// to confirm the failure, then reverted.
    #[test]
    fn payload_is_shared_not_copied_across_cursors() {
        let trunk = Trunk::new(TrunkConfig::new(nz(8), nz(8), nz(4), nz(8), nz(8)));
        let mut c1 = trunk.subscribe();
        let mut c2 = trunk.subscribe();
        let mut c3 = trunk.subscribe();
        let writer = trunk.writer().unwrap();

        writer.publish(TrackId::new(4), RetentionClass::Timed, sample(0xAB, 65536));

        let i1 = c1.poll().unwrap();
        let i2 = c2.poll().unwrap();
        let i3 = c3.poll().unwrap();
        let p1 = timed_data(&i1).unwrap().1.data.as_ptr();
        let p2 = timed_data(&i2).unwrap().1.data.as_ptr();
        let p3 = timed_data(&i3).unwrap().1.data.as_ptr();

        assert_eq!(
            p1, p2,
            "cursor 2's payload must be the SAME allocation as cursor 1's"
        );
        assert_eq!(
            p2, p3,
            "cursor 3's payload must be the SAME allocation as cursor 1's"
        );
        // Not just equal contents (that would also pass for two independent
        // 64KiB copies) — the ptr comparison above is the real assertion;
        // this just confirms the payload wasn't corrupted in the process.
        assert_eq!(timed_data(&i1).unwrap().1.data.len(), 65536);
    }

    // --- Construction invariants -------------------------------------------
    //
    // The five `zero_*_capacity_panics` tests that used to live here are
    // deliberately GONE, not merely disabled: every `TrunkConfig` capacity is
    // now a `NonZeroUsize`, so a zero capacity is unrepresentable rather than
    // rejected at run time, and `Trunk::new` no longer has (or needs) the
    // `assert!`s they pinned. A test asserting a panic that can no longer
    // occur would not compile against the new signature anyway, and keeping a
    // rewritten version would only be asserting that `NonZeroUsize::new(0)`
    // returns `None` — a property of the standard library, not of this crate.

    #[test]
    fn second_writer_is_refused() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let _first = trunk.writer().unwrap();
        assert!(
            trunk.writer().is_none(),
            "a Trunk has exactly one sample/event writer"
        );
    }

    // --- W1. the SegmentWriter half of the split is single-take too, ------
    // --- independently of TrunkWriter ---------------------------------------

    /// MUTATION VERIFIED: replacing `Trunk::segment_writer`'s
    /// `compare_exchange` call with an unconditional `Some(SegmentWriter {
    /// .. })` (i.e. reintroducing "anyone can take it, any number of times")
    /// makes this test's `assert!(trunk.segment_writer().is_none(), ..)`
    /// fail — the second call succeeds instead of being refused. Recompiled
    /// and re-run to confirm the failure, then reverted.
    #[test]
    fn second_segment_writer_is_refused() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let _first = trunk.segment_writer().unwrap();
        assert!(
            trunk.segment_writer().is_none(),
            "a Trunk has exactly one segment/part writer"
        );
    }

    // --- W2. THE GAP THIS STEP CLOSES: a sample/event writer and a --------
    // --- segment/part writer can be held AT THE SAME TIME -------------------

    /// This is the property that was **structurally impossible** before this
    /// step: `Trunk::writer()` and a hypothetical segment-writing capability
    /// shared one `AtomicBool`, so whichever component (the ingest driver)
    /// took the one writer made it impossible for anything else (a
    /// segmenter) to ever publish a segment or a part.
    ///
    /// MUTATION VERIFIED: changing `Trunk::segment_writer` to gate on
    /// `self.writer_taken` instead of its own `self.segment_writer_taken`
    /// (i.e. reintroducing the single-shared-flag bug this step fixes) makes
    /// this test's `let segments = trunk.segment_writer().unwrap();` line
    /// panic — `segment_writer()` returns `None` because the sample/event
    /// writer taken just above already flipped the shared flag. Recompiled
    /// and re-run to confirm the failure, then reverted.
    #[test]
    fn sample_and_segment_writers_can_be_held_simultaneously() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let samples = trunk.writer().unwrap();
        let segments = trunk.segment_writer().expect(
            "the segment/part writer must still be takeable while the sample/event writer is held",
        );

        // Both are simultaneously live and independently usable — not merely
        // both `Some` a moment apart.
        samples.publish(TrackId::new(1), RetentionClass::Timed, sample(1, 4));
        segments.publish_segment(segment_entry(1, 1)).unwrap();
        assert_eq!(trunk.timed_len(), 1);
        assert_eq!(trunk.segment_len(), 1);
    }

    // --- W3. THE SEGMENTER-SHAPED, END-TO-END PROPERTY: a segmenter reads --
    // --- samples through its own cursor and publishes the segment/part it --
    // --- derives from them through the OTHER writer — unreachable before ---
    // --- this step, since there was only one writer for the whole Trunk ----

    /// The load-bearing test for this step: models exactly the component the
    /// gap analysis found could not exist — a segmenter that holds a
    /// [`SampleCursor`] (to read the samples it segments) *and* a
    /// [`SegmentWriter`] (to publish what it produces) at the same time,
    /// distinct from the ingest driver's own [`TrunkWriter`].
    ///
    /// MUTATION VERIFIED: commenting out `state.segments.push(entry);` in
    /// `SegmentWriter::publish_segment` (simulating "the moved method is a
    /// stub that does not actually reach the ring") makes this test's
    /// `let got = seg_cursor.poll().expect(..)` panic — nothing was ever
    /// pushed, so the segment log stays empty and the cursor has nothing to
    /// return. Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn segmenter_holds_sample_cursor_and_segment_writer_at_once() {
        let trunk = Trunk::new(TrunkConfig::new(nz(8), nz(4), nz(4), nz(8), nz(8)));

        // The ingest driver's own handle — a different component, a
        // different ring group.
        let ingest = trunk.writer().unwrap();
        // The segmenter's read side (its own SampleCursor) and write side
        // (the SegmentWriter) — held together, which is exactly what the
        // single-writer-per-Trunk model made impossible.
        let mut samples = trunk.subscribe();
        let segmenter = trunk.segment_writer().expect(
            "a segmenter must be able to take the segment/part writer \
                     while the ingest driver still holds the sample/event writer",
        );
        // `subscribe_segments` only sees entries published *after* this call
        // (exactly `Trunk::subscribe`'s "starts from now" contract) — taken
        // up front so the read-back below has something to see.
        let mut seg_cursor = trunk.subscribe_segments();

        for i in 0u8..3 {
            ingest.publish(TrackId::new(1), RetentionClass::Timed, sample(i, 4));
        }

        // The segmenter consumes exactly the samples it is about to derive
        // a segment from.
        let mut muxed = Vec::new();
        for _ in 0..3 {
            match samples.poll() {
                Some(SampleCursorItem::Timed { sample, .. }) => {
                    muxed.push(sample.data[0]);
                }
                other => panic!("expected a Timed sample, got {other:?}"),
            }
        }
        assert_eq!(muxed, vec![0, 1, 2]);

        // ...then publishes the segment (and one live part of it) derived
        // from exactly those samples — through the OTHER writer.
        segmenter.publish_part(part_entry(0xAB, 1, 0));
        segmenter.publish_segment(segment_entry(0xAA, 1)).unwrap();

        // Read both back through the segment/part log's own query surface —
        // proving the publish actually reached the shared Trunk, not just a
        // private buffer inside the segmenter.
        let got = seg_cursor
            .poll()
            .expect("the segmenter's published segment must be visible");
        assert_eq!(
            segment_data(&got).unwrap().sequence_number,
            SegmentSeq::new(1)
        );
        assert_eq!(
            trunk.part_bytes(SegmentSeq::new(1), PartIndex::new(0)),
            Some(Bytes::from(vec![0xAB; 8])),
            "the segmenter's published part must be individually addressable too"
        );
    }

    #[test]
    fn subscribe_starts_from_now_not_from_history() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        writer.publish(TrackId::new(1), RetentionClass::Timed, sample(1, 4));
        writer.publish(TrackId::new(1), RetentionClass::Timed, sample(2, 4));

        // Subscribing after two publishes must not see either of them.
        let mut cursor = trunk.subscribe();
        assert!(cursor.poll().is_none());

        writer.publish(TrackId::new(1), RetentionClass::Timed, sample(3, 4));
        let item = cursor.poll().unwrap();
        assert_eq!(timed_data(&item).unwrap().1.data[0], 3);
    }

    // ===================== segment log ====================================

    // --- S1. multiple cursors, every segment, in order, no dup/no loss ----

    /// MUTATION VERIFIED: removing the `self.consumed += 1;` from the
    /// non-pinning data-return arm of `SegmentCursor::poll` (so the same
    /// ring index is re-read every call) makes this test fail exactly like
    /// `SampleCursor::poll`'s equivalent mutation: `drain_segments` still
    /// returns 5 items, but all 5 are the first published segment
    /// (`sequence_number == 1`) instead of the distinct sequence `1..=5`, so
    /// the `assert_eq!` on the reconstructed sequence-number list fails at
    /// index 1. Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn multiple_segment_cursors_see_every_segment_in_order_with_no_dup_or_loss() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(100), nz(8), nz(8)));
        let mut c1 = trunk.subscribe_segments();
        let mut c2 = trunk.subscribe_segments();
        let mut c3 = trunk.subscribe_segments();
        let writer = trunk.segment_writer().unwrap();

        for i in 0u32..5 {
            writer
                .publish_segment(segment_entry(i as u8, i + 1))
                .unwrap();
        }

        for cursor in [&mut c1, &mut c2, &mut c3] {
            let items = drain_segments(cursor, 5);
            assert_eq!(items.len(), 5, "each cursor must see exactly 5 segments");
            let seqs: Vec<SegmentSeq> = items
                .iter()
                .map(|item| segment_data(item).unwrap().sequence_number)
                .collect();
            assert_eq!(
                seqs,
                [1, 2, 3, 4, 5].map(SegmentSeq::new),
                "must be in playlist order"
            );
            assert!(cursor.poll().is_none(), "no extra/duplicated items");
        }
    }

    // --- S2. a non-pinning slow reader lags; writer completes regardless --

    /// MUTATION VERIFIED: changing `SegmentLog::push`'s eviction condition
    /// from `self.entries.len() == self.capacity` to `false` (disabling
    /// eviction) makes `trunk.segment_len()` grow to 1024 instead of staying
    /// at the configured cap of 4, and the subsequent `Lagged` assertion
    /// fails because `base` never advanced (`skipped` reads back as `0`, not
    /// `1020`). Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn non_pinning_slow_segment_reader_lags_but_writer_completes_regardless() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let mut slow = trunk.subscribe_segments();
        let writer = trunk.segment_writer().unwrap();

        // The slow (non-pinning) reader never polls while 1024 segments are
        // published — there is no wait-for-reader path for a non-pinning
        // cursor, so this simply completes (same reasoning as
        // `slow_reader_lags_but_writer_completes_regardless`).
        for i in 0u32..1024 {
            writer
                .publish_segment(segment_entry((i % 256) as u8, i + 1))
                .unwrap();
        }
        assert_eq!(
            trunk.segment_len(),
            4,
            "writer unblocked: segment log stayed bounded"
        );

        let first = slow.poll().unwrap();
        assert!(
            matches!(first, SegmentCursorItem::Lagged { skipped: 1020 }),
            "expected Lagged{{skipped: 1020}}, got {first:?}"
        );
    }

    // --- S3. THE DVR PROPERTY: a pinning reader loses nothing while a ------
    // --- non-pinning sibling lags, and StallIngest is what makes it true --

    /// MUTATION VERIFIED: changing the `must_wait` computation in
    /// `SegmentWriter::publish_segment`'s `ArchiveOverrun::StallIngest` arm
    /// from `must_wait = true;` to `{}` (a no-op, i.e. treating
    /// `StallIngest` exactly like `Gap`) makes this test fail: the third
    /// `publish_segment` call no longer blocks, so the background-thread
    /// completion channel's `recv_timeout` at the "still blocked" checkpoint
    /// returns `Ok(())` instead of timing out, and the assertion that it
    /// timed out (`is_err()`) fails. Recompiled and re-run to confirm the
    /// failure, then reverted.
    #[test]
    fn pinning_reader_receives_every_segment_while_non_pinning_reader_lags() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(2), nz(8), nz(8)));
        let mut slow = trunk.subscribe_segments(); // non-pinning: will lag
        let mut archive = trunk.pin_segments(ArchiveOverrun::StallIngest); // pinning: must lose nothing
        let writer = Arc::new(trunk.segment_writer().unwrap());

        // Fill the segment log's capacity (2) without any eviction yet.
        writer.publish_segment(segment_entry(1, 1)).unwrap();
        writer.publish_segment(segment_entry(2, 2)).unwrap();

        // A third publish must evict the oldest (seq 1), which `archive`'s
        // pin has not yet consumed — with `StallIngest`, this call blocks.
        // Run it on a background thread (the same one `SegmentWriter`,
        // shared via `Arc` — this is the segment/part ring group's own
        // single-writer invariant, just called from a different thread) and
        // prove, via a completion channel, that it has NOT returned yet.
        let (done_tx, done_rx) = mpsc::channel();
        let blocked_writer = Arc::clone(&writer);
        let handle = thread::spawn(move || {
            blocked_writer.publish_segment(segment_entry(3, 3)).unwrap();
            done_tx.send(()).unwrap();
        });

        wait_until_stalled(&trunk, 1);
        assert!(
            done_rx.try_recv().is_err(),
            "publish_segment must still be blocked: archive has not consumed seq 1 yet"
        );

        // `archive` catches up: consuming seq 1 releases its pin on it,
        // which must wake and unblock the writer thread.
        let first = archive.poll().unwrap();
        assert_eq!(
            segment_data(&first).unwrap().sequence_number,
            SegmentSeq::new(1)
        );

        // HANG GUARD (issue #807): generous on purpose, same reasoning as
        // `retention.rs`'s `archive_overrun_stall_ingest_blocks_writer_until_driver_advances`
        // (the sibling proof of this same mechanism) -- the claim is "the
        // writer unblocks once the pin is drained", not "within N seconds".
        // The unblock is observed across a thread boundary, so a tight bound
        // measures the machine's scheduler, not this code.
        done_rx
            .recv_timeout(Duration::from_secs(60))
            .expect("publish_segment must unblock once the pin advances");
        handle.join().unwrap();

        // `archive` receives every remaining segment with ZERO loss — no
        // `Gap`, no `Lagged` — proving the DVR property: pinning protected
        // it from the eviction that just happened.
        let second = archive.poll().unwrap();
        assert_eq!(
            segment_data(&second).unwrap().sequence_number,
            SegmentSeq::new(2)
        );
        let third = archive.poll().unwrap();
        assert_eq!(
            segment_data(&third).unwrap().sequence_number,
            SegmentSeq::new(3)
        );
        assert!(archive.poll().is_none());

        // Meanwhile `slow` (non-pinning, never polled) DID lag: exactly one
        // segment (seq 1) was evicted out from under it.
        let lag = slow.poll().unwrap();
        assert!(
            matches!(lag, SegmentCursorItem::Lagged { skipped: 1 }),
            "expected Lagged{{skipped: 1}}, got {lag:?}"
        );
        let remaining: Vec<SegmentSeq> = drain_segments(&mut slow, 2)
            .iter()
            .map(|item| segment_data(item).unwrap().sequence_number)
            .collect();
        assert_eq!(remaining, [2, 3].map(SegmentSeq::new));
    }

    // --- S4. pinning is bounded: an un-acking consumer cannot grow --------
    // --- memory without limit ----------------------------------------------

    /// MUTATION VERIFIED: removing the eviction check in `SegmentLog::push`
    /// (replacing `if self.entries.len() == self.capacity { .. }` with a
    /// no-op, exactly like the sample-ring equivalent mutation) makes
    /// `trunk.segment_len()` grow past the configured cap of `4` instead of
    /// staying bounded — the in-loop assertion below fails on the first
    /// over-capacity iteration. Recompiled and re-run to confirm the
    /// failure, then reverted.
    #[test]
    fn pinning_is_bounded_an_unacking_consumer_cannot_grow_memory_without_limit() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        // Default policy (`Gap`) pinning cursor that never polls at all —
        // the worst case for memory growth: a dead/wedged archive consumer.
        let _archive = trunk.pin_segments(ArchiveOverrun::default());
        let writer = trunk.segment_writer().unwrap();

        for i in 0u32..50_000 {
            writer
                .publish_segment(segment_entry((i % 256) as u8, i + 1))
                .unwrap();
            assert!(
                trunk.segment_len() <= 4,
                "segment log exceeded its cap mid-flood despite an un-acking pinning cursor"
            );
        }
        assert_eq!(trunk.segment_len(), 4);
    }

    // --- S5. ArchiveOverrun::Gap gaps and reports --------------------------

    /// MUTATION VERIFIED: changing the pinning branch of `SegmentCursor::poll`
    /// to report `SegmentCursorItem::Lagged` instead of `SegmentCursorItem::Gap`
    /// (collapsing the two, mirroring the sample path's Timed/Sparse
    /// mutation) makes the `matches!(item, SegmentCursorItem::Gap { .. })`
    /// assertion below fail — the item is a `Lagged` instead. Recompiled and
    /// re-run to confirm the failure, then reverted.
    #[test]
    fn archive_overrun_gap_evicts_and_reports_gap() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(2), nz(8), nz(8)));
        let mut archive = trunk.pin_segments(ArchiveOverrun::Gap);
        let writer = trunk.segment_writer().unwrap();

        // Publish 5 segments into a capacity-2 log without archive ever
        // polling: with `Gap`, eviction proceeds unconditionally, so this
        // never blocks.
        for i in 0u32..5 {
            writer
                .publish_segment(segment_entry(i as u8, i + 1))
                .unwrap();
        }
        assert_eq!(trunk.segment_len(), 2);

        let gap = archive.poll().unwrap();
        assert!(
            matches!(gap, SegmentCursorItem::Gap { skipped: 3 }),
            "expected Gap{{skipped: 3}}, got {gap:?}"
        );
        // The recording has a hole, but the stream survives: archive keeps
        // reading the segments that remain.
        let remaining: Vec<SegmentSeq> = drain_segments(&mut archive, 2)
            .iter()
            .map(|item| segment_data(item).unwrap().sequence_number)
            .collect();
        assert_eq!(remaining, [4, 5].map(SegmentSeq::new));
        assert!(archive.poll().is_none());
    }

    // --- S6. ArchiveOverrun::StallIngest actually applies back-pressure ---

    /// MUTATION VERIFIED: same mutation and same observed failure as
    /// `pinning_reader_receives_every_segment_while_non_pinning_reader_lags`'s
    /// doc comment (removing `must_wait = true;` from the `StallIngest`
    /// arm) — this test is the narrower, single-purpose proof that
    /// `publish_segment` genuinely blocks, isolated from the sibling-lag
    /// scenario. Recompiled and re-run to confirm the failure, then
    /// reverted.
    #[test]
    fn archive_overrun_stall_ingest_actually_blocks_the_writer() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(1), nz(8), nz(8)));
        let mut archive = trunk.pin_segments(ArchiveOverrun::StallIngest);
        let writer = Arc::new(trunk.segment_writer().unwrap());

        writer.publish_segment(segment_entry(1, 1)).unwrap(); // fills capacity-1 log

        let (done_tx, done_rx) = mpsc::channel();
        let blocked_writer = Arc::clone(&writer);
        let handle = thread::spawn(move || {
            blocked_writer.publish_segment(segment_entry(2, 2)).unwrap();
            done_tx.send(()).unwrap();
        });

        wait_until_stalled(&trunk, 1);
        assert!(
            matches!(
                done_rx.try_recv(),
                Err(std::sync::mpsc::TryRecvError::Empty)
            ),
            "publish_segment must block: the pin has not consumed seq 1 yet"
        );

        let first = archive.poll().unwrap();
        assert_eq!(
            segment_data(&first).unwrap().sequence_number,
            SegmentSeq::new(1)
        );

        // HANG GUARD (issue #807): generous on purpose, same reasoning as the
        // sibling test above (`pinning_reader_receives_every_segment_while_non_pinning_reader_lags`)
        // -- claim is "unblocks once drained", not "within N seconds"; the
        // unblock crosses a thread boundary so a tight bound measures the
        // scheduler, not this code.
        done_rx
            .recv_timeout(Duration::from_secs(60))
            .expect("publish_segment must unblock once the pin advances");
        handle.join().unwrap();
    }

    /// The stalled-publisher probe is what replaces the 200 ms negative waits:
    /// it must read 1 exactly while `publish_segment` is parked and 0 after.
    /// It also bounds how long the unblock takes: the pin advance must wake
    /// the parked publisher through `notify_all`, not let it run out
    /// `STALL_INGEST_MAX_WAIT` (the bounded-wait path also returns, so
    /// "it returned" alone cannot tell a lost wake from a delivered one).
    #[test]
    fn stalled_publisher_probe_tracks_a_parked_publish_segment() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(1), nz(8), nz(8)));
        let writer = Arc::new(trunk.segment_writer().unwrap());
        let mut pin = trunk.pin_segments(ArchiveOverrun::StallIngest);
        writer.publish_segment(segment_entry(1, 1)).unwrap();
        assert_eq!(trunk.stalled_publishers.load(Ordering::Acquire), 0);

        let bg = Arc::clone(&writer);
        let handle = thread::spawn(move || bg.publish_segment(segment_entry(2, 2)).unwrap());
        wait_until_stalled(&trunk, 1);
        assert_eq!(trunk.stalled_publishers.load(Ordering::Acquire), 1);

        let released_at = Instant::now();
        assert!(pin.poll().is_some(), "consume seq 1: releases the pin");
        handle.join().unwrap();
        assert!(
            released_at.elapsed() < STALL_INGEST_MAX_WAIT / 2,
            "the publisher must be woken by the pin advance, not time out ({:?})",
            released_at.elapsed()
        );
        assert_eq!(trunk.stalled_publishers.load(Ordering::Acquire), 0);
    }

    /// While a publisher is parked in StallIngest, ordinary sample publish and
    /// cursor polls on the same Trunk must proceed (the Condvar releases the lock).
    #[test]
    fn sample_publish_proceeds_while_a_segment_publisher_is_stalled() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(1), nz(8), nz(8)));
        let seg_writer = Arc::new(trunk.segment_writer().unwrap());
        let writer = trunk.writer().unwrap();
        let mut pin = trunk.pin_segments(ArchiveOverrun::StallIngest);
        seg_writer.publish_segment(segment_entry(1, 1)).unwrap();
        let bg = Arc::clone(&seg_writer);
        let handle = thread::spawn(move || bg.publish_segment(segment_entry(2, 2)).unwrap());
        wait_until_stalled(&trunk, 1);

        writer.publish(TrackId::new(1), RetentionClass::Timed, sample(7, 4)); // must not deadlock
        assert_eq!(trunk.timed_len(), 1);

        assert!(pin.poll().is_some());
        handle.join().unwrap();
    }

    // --- S7. ArchiveOverrun::Terminate drops the cursor --------------------

    /// MUTATION VERIFIED: changing `ArchiveOverrun::Terminate => pin.terminated
    /// = true,` in `SegmentWriter::publish_segment` to `ArchiveOverrun::Terminate
    /// => {}` (a no-op, treating `Terminate` exactly like `Gap`) makes this
    /// test fail: `archive.poll()` returns `Some(Gap { .. })` instead of
    /// `Some(Terminated)`, so the `matches!` assertion on `Terminated` fails.
    /// Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn archive_overrun_terminate_drops_the_cursor() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(2), nz(8), nz(8)));
        let mut archive = trunk.pin_segments(ArchiveOverrun::Terminate);
        let writer = trunk.segment_writer().unwrap();

        // Publish past capacity without archive ever polling: `Terminate`
        // never blocks (like `Gap`), so this completes.
        for i in 0u32..5 {
            writer
                .publish_segment(segment_entry(i as u8, i + 1))
                .unwrap();
        }
        assert_eq!(trunk.segment_len(), 2, "writer unblocked despite Terminate");

        let item = archive.poll().unwrap();
        assert!(
            matches!(item, SegmentCursorItem::Terminated),
            "expected Terminated, got {item:?}"
        );
        // The cursor is done: every poll after `Terminated` returns `None`,
        // never resuming as if nothing happened.
        assert!(archive.poll().is_none());
        assert!(archive.poll().is_none());

        // The log itself is unaffected: publishing continues to work, and a
        // fresh cursor still sees ordinary segment log behaviour.
        writer.publish_segment(segment_entry(9, 6)).unwrap();
        assert_eq!(trunk.segment_len(), 2);
    }

    // --- S8. segment bytes are shared, not copied, across cursors ---------

    /// MUTATION VERIFIED: replacing `entry.clone()` in
    /// `SegmentCursor::poll`'s non-pinning data-return arm with a hand-rolled
    /// copy (`SegmentEntry { bytes: Bytes::copy_from_slice(entry.bytes.as_ref()),
    /// ..entry.clone() }`, preserving every field's *value*) makes this
    /// test's pointer-identity assertion fail — `p1 == p2` becomes `false`
    /// (two distinct heap allocations with equal contents) instead of the
    /// unmutated `clone()` path's `p1 == p2 == p3`. This is exactly the
    /// distinction a content-equality assertion would have missed.
    /// Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn segment_bytes_are_shared_not_copied_across_cursors() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(8), nz(8), nz(8)));
        let mut c1 = trunk.subscribe_segments();
        let mut c2 = trunk.subscribe_segments();
        let mut c3 = trunk.subscribe_segments();
        let writer = trunk.segment_writer().unwrap();

        writer
            .publish_segment(SegmentEntry::new(
                Bytes::from(vec![0xCDu8; 65536]),
                SegmentSeq::new(1),
                Duration::from_secs(2),
                Timestamp::from_nanos(0),
                SegmentMeta {
                    discontinuous: false,
                },
            ))
            .unwrap();

        let i1 = c1.poll().unwrap();
        let i2 = c2.poll().unwrap();
        let i3 = c3.poll().unwrap();
        let p1 = segment_data(&i1).unwrap().bytes.as_ptr();
        let p2 = segment_data(&i2).unwrap().bytes.as_ptr();
        let p3 = segment_data(&i3).unwrap().bytes.as_ptr();

        assert_eq!(
            p1, p2,
            "cursor 2's segment payload must be the SAME allocation as cursor 1's"
        );
        assert_eq!(
            p2, p3,
            "cursor 3's segment payload must be the SAME allocation as cursor 1's"
        );
        assert_eq!(segment_data(&i1).unwrap().bytes.len(), 65536);
    }

    // ===================== event log =======================================

    use timed_metadata::{EventKind, SourcePayload};

    /// A minimal `TimedEvent` for tests that don't care about the SCTE-35
    /// source payload itself — only about how the event *log* addresses and
    /// resolves it. `at`/`duration` are left `None`: this step's
    /// [`EventAnchor`] carries the resolution state, not `TimedEvent::at`.
    fn basic_event(id: u32) -> TimedEvent {
        TimedEvent {
            id: Some(id),
            kind: EventKind::BreakStart,
            at: None,
            duration: None,
            source: SourcePayload::Scte35 { raw: Vec::new() },
        }
    }

    fn event_id(item: &EventCursorItem) -> Option<u32> {
        match item {
            EventCursorItem::Event(e) => e.event.id,
            _ => None,
        }
    }

    /// Build real, valid (Parse/Serialize round-tripping) `splice_insert()`
    /// bytes carrying `pts_time`, via `scte35-splice`'s own builder +
    /// serializer — not hand-rolled/fabricated wire bytes. Used to drive
    /// `timed_metadata::Timeline::push_scte35`'s 33-bit wrap-unroll across a
    /// genuine wrap boundary (see `a_33_bit_pts_wrap_does_not_corrupt_event_log_ordering`).
    fn splice_insert_bytes(event_id: u32, pts_time: u64) -> Vec<u8> {
        use broadcast_common::Serialize;
        use scte35_splice::SpliceInfoSection;
        use scte35_splice::commands::AnyCommand;
        use scte35_splice::commands::splice_insert::SpliceInsert;
        use scte35_splice::time::SpliceTime;

        let si = SpliceInsert {
            splice_event_id: event_id,
            out_of_network_indicator: true,
            splice_time: Some(SpliceTime::with_pts(pts_time)),
            ..SpliceInsert::default()
        };
        let section = SpliceInfoSection::new_clear(AnyCommand::SpliceInsert(si), &[]);
        section.try_to_bytes().unwrap()
    }

    // --- E1. events_between: half-open [from, to), boundaries exact -------

    /// MUTATION VERIFIED: changing the upper-bound comparison in
    /// `Trunk::events_between`'s filter from `t.0 < to.0` to `t.0 <= to.0`
    /// (making the range closed instead of half-open) makes this test fail:
    /// `ids` becomes `[2, 3, 4]` (the boundary event at `to` is wrongly
    /// included) instead of the expected `[2, 3]`. Recompiled and re-run to
    /// confirm the failure, then reverted.
    #[test]
    fn events_between_returns_exactly_the_half_open_range() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();

        for (id, ticks) in [(1u32, 1_000u64), (2, 2_000), (3, 3_000), (4, 4_000)] {
            writer.publish_event(basic_event(id), EventAnchor::Media(MediaTime(ticks)));
        }

        let got = trunk.events_between(MediaTime(2_000), MediaTime(4_000));
        let ids: Vec<u32> = got.iter().map(|e| e.event.id.unwrap()).collect();
        assert_eq!(
            ids,
            vec![2, 3],
            "start (2_000) inclusive, end (4_000) exclusive"
        );
    }

    // --- E2. a Segment-anchored entry resolves at PUBLISH time when the ---
    // --- boundary is already known ------------------------------------------

    /// MUTATION VERIFIED: changing `EventLog::try_resolve`'s `Segment` arm
    /// to always return the anchor unresolved (`_ => anchor` in place of the
    /// `segment_starts` lookup) makes this test fail: `events_in_segment(SegmentSeq::new(3))`
    /// comes back empty instead of containing the published event, because
    /// the entry never leaves `EventAnchor::Segment`. Recompiled and re-run
    /// to confirm the failure, then reverted.
    #[test]
    fn segment_relative_event_resolves_at_publish_time_when_boundary_already_known() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        // Two separate writers, held at once: `note_segment_start` lives on
        // the segmenter's `SegmentWriter`, `publish_event` on the ingest
        // driver's `TrunkWriter` — exactly the split this step introduces.
        let writer = trunk.writer().unwrap();
        let segment_writer = trunk.segment_writer().unwrap();

        segment_writer.note_segment_start(SegmentSeq::new(3), MediaTime(300_000));
        writer.publish_event(
            basic_event(9),
            EventAnchor::Segment {
                segment_number: SegmentSeq::new(3),
                delta: 1_500,
            },
        );

        let got = trunk.events_in_segment(SegmentSeq::new(3));
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].event.id, Some(9));
        assert!(matches!(
            got[0].anchor,
            EventAnchor::Media(MediaTime(t)) if t == 301_500
        ));
    }

    // --- E3. THE B1 SEGMENT CASE: a segment-relative event resolves to ----
    // --- the segment it actually named, not whichever segment is open -----

    /// MUTATION VERIFIED: removing the `if n == segment_number` guard in
    /// `EventLog::note_segment_start` (resolving *every* pending `Segment`
    /// entry against whichever boundary arrives, regardless of which
    /// segment it targets) makes this test fail at the first assertion:
    /// after `note_segment_start(SegmentSeq::new(1), MediaTime(0))` — segment 1, NOT the
    /// event's actual target segment 2 — the entry is wrongly resolved to
    /// `MediaTime(1_000)` (segment 1's start + delta) instead of staying
    /// `EventAnchor::Segment { segment_number: 2, .. }`, so the
    /// `matches!(entry.anchor, EventAnchor::Segment { segment_number: 2, .. })`
    /// assertion fails. Recompiled and re-run to confirm the failure, then
    /// reverted.
    #[test]
    fn segment_relative_event_resolves_to_the_named_segment_not_whichever_is_open() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        let segment_writer = trunk.segment_writer().unwrap();
        let mut cursor = trunk.subscribe_events();

        // The event targets segment 2 specifically, delta 1_000 after ITS
        // start — published before ANY segment boundary is known.
        writer.publish_event(
            basic_event(42),
            EventAnchor::Segment {
                segment_number: SegmentSeq::new(2),
                delta: 1_000,
            },
        );

        // Segment 1 — a DIFFERENT, "currently open" segment — reports its
        // start first. This must NOT resolve the segment-2-targeted event.
        segment_writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));

        let item = cursor.poll().unwrap();
        let entry = match item {
            EventCursorItem::Event(e) => e,
            other => panic!("expected Event, got {other:?}"),
        };
        assert!(
            matches!(
                entry.anchor,
                EventAnchor::Segment {
                    segment_number,
                    delta: 1_000,
                } if segment_number == SegmentSeq::new(2)
            ),
            "must stay pending on segment 2 — segment 1 being open must not \
             resolve it against the wrong boundary: {:?}",
            entry.anchor
        );
        assert!(
            trunk.events_in_segment(SegmentSeq::new(2)).is_empty(),
            "not resolved yet: must not appear under segment 2 either"
        );
        assert!(trunk.events_in_segment(SegmentSeq::new(1)).is_empty());

        // Now segment 2's own start arrives: resolves in place, to the
        // RIGHT segment's start + delta.
        segment_writer.note_segment_start(SegmentSeq::new(2), MediaTime(90_000));

        let in_seg2 = trunk.events_in_segment(SegmentSeq::new(2));
        assert_eq!(in_seg2.len(), 1);
        assert_eq!(in_seg2[0].event.id, Some(42));
        assert!(matches!(
            in_seg2[0].anchor,
            EventAnchor::Media(MediaTime(t)) if t == 91_000
        ));
        assert!(
            trunk.events_in_segment(SegmentSeq::new(1)).is_empty(),
            "must not ALSO appear under segment 1"
        );
    }

    // --- E3b. A reused segment_number resolves against its MOST RECENT ---
    // --- boundary, not a stale earlier one with the same number -----------

    /// MUTATION VERIFIED: reverting `EventLog::try_resolve`'s `Segment` arm
    /// and `Trunk::events_in_segment`'s two lookups from `.iter().rev().find`
    /// back to a plain front-to-back `.iter().find` makes this test's final
    /// assertion fail: the event published against the *second* occurrence
    /// of segment number 1 resolves to `MediaTime(1_500)` (the FIRST
    /// occurrence's start 0 + delta 1_500) instead of `MediaTime(201_500)`
    /// (the second occurrence's start 200_000 + delta 1_500) — a stale
    /// boundary from a segment number that has since been reused.
    /// Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn reused_segment_number_resolves_against_its_most_recent_boundary() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        let segment_writer = trunk.segment_writer().unwrap();

        // Segment number 1 occurs TWICE (a segmenter restart / counter
        // wraparound) with two different starts, far apart in time.
        segment_writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));
        segment_writer.note_segment_start(SegmentSeq::new(2), MediaTime(90_000));
        segment_writer.note_segment_start(SegmentSeq::new(1), MediaTime(200_000)); // reused
        segment_writer.note_segment_start(SegmentSeq::new(2), MediaTime(290_000)); // reused too

        // Published AFTER both occurrences of segment 1 already exist, so
        // it resolves immediately (at `publish_event` time) against
        // whichever boundary `try_resolve` picks.
        writer.publish_event(
            basic_event(7),
            EventAnchor::Segment {
                segment_number: SegmentSeq::new(1),
                delta: 1_500,
            },
        );

        let in_seg1 = trunk.events_in_segment(SegmentSeq::new(1));
        assert_eq!(in_seg1.len(), 1);
        assert!(
            matches!(
                in_seg1[0].anchor,
                EventAnchor::Media(MediaTime(t)) if t == 201_500
            ),
            "must resolve against the MOST RECENT segment-1 boundary \
             (200_000 + 1_500), not the stale first one (0 + 1_500): {:?}",
            in_seg1[0].anchor
        );
    }

    // --- E4. THE B1 CRUX: a UTC-only event stays honestly unanchored ------
    // --- until a TimeAnchor arrives, then resolves correctly ---------------

    /// MUTATION VERIFIED: changing `EventLog::try_resolve`'s `Utc` arm to
    /// fabricate `EventAnchor::Media(MediaTime(0))` whenever no
    /// `time_anchor` is set yet (in place of returning the anchor
    /// unresolved) — i.e. reintroducing the exact B1 bug this design
    /// exists to prevent — makes this test fail at the first assertion:
    /// `entry.anchor` is `EventAnchor::Media(MediaTime(0))` instead of the
    /// expected `EventAnchor::Utc { utc_epoch_ms: 5_000 }`, so the
    /// `matches!` assertion fails. Recompiled and re-run to confirm the
    /// failure, then reverted.
    #[test]
    fn utc_only_event_stays_unanchored_until_a_time_anchor_arrives() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        let segment_writer = trunk.segment_writer().unwrap();
        let mut cursor = trunk.subscribe_events();

        // A GPS/UTC-scheduled event (SCTE-35 splice_schedule.utc_splice_time
        // semantics, §9.7.4) with no media anchor yet.
        writer.publish_event(
            basic_event(7),
            EventAnchor::Utc {
                utc_epoch_ms: 5_000,
            },
        );

        let item = cursor.poll().unwrap();
        let entry = match item {
            EventCursorItem::Event(e) => e,
            other => panic!("expected Event, got {other:?}"),
        };
        assert!(
            matches!(
                entry.anchor,
                EventAnchor::Utc {
                    utc_epoch_ms: 5_000
                }
            ),
            "must stay honestly unanchored — NO fabricated media time: {:?}",
            entry.anchor
        );
        // Nothing to filter a media time against yet: the point-in-time
        // query must not surface it either.
        assert!(
            trunk
                .events_between(MediaTime(0), MediaTime(u64::MAX))
                .is_empty(),
            "an unanchored event must not appear in a media-time query"
        );

        // An anchor arrives: pts 0 == epoch 1_000ms (`TimeAnchor`'s own
        // convention), so epoch 5_000ms is 4_000ms == 360_000 ticks later.
        segment_writer.set_time_anchor(TimeAnchor {
            pts_90k: 0,
            utc_epoch_ms: 1_000,
        });

        let resolved = trunk.events_between(MediaTime(0), MediaTime(u64::MAX));
        assert_eq!(resolved.len(), 1);
        assert_eq!(resolved[0].event.id, Some(7));
        assert!(
            matches!(resolved[0].anchor, EventAnchor::Media(MediaTime(t)) if t == 360_000),
            "expected MediaTime(360_000), got {:?}",
            resolved[0].anchor
        );
    }

    // --- E5. a 33-bit PTS wrap does not corrupt event log ordering --------
    // --- (reuses timed_metadata::Timeline's unroll; does not hand-roll it) -

    /// MUTATION VERIFIED: re-introducing a 33-bit mask on an
    /// already-unrolled `MediaTime` in `EventLog::try_resolve`'s `Media`
    /// arm (`EventAnchor::Media(MediaTime(t)) => EventAnchor::Media(MediaTime(t
    /// & ((1u64 << 33) - 1)))` in place of the pass-through `anchor`) makes
    /// this test fail: `ev2`'s post-wrap absolute tick value
    /// (`(1u64 << 33) + 5`) exceeds 33 bits, so it is stored truncated to
    /// `5` instead of the value `Timeline` actually computed, and
    /// `matches!(got[1].anchor, EventAnchor::Media(t) if t.0 == at2.0)`
    /// fails (stored `5` != `at2.0` ≈ `2^33 + 5`). (The earlier
    /// `at2.0 > at1.0` assertion, which only reads `Timeline`'s local return
    /// value, does NOT catch this mutation — a stored-value mutation only
    /// shows up in what the log hands back, which is exactly why this test
    /// asserts against `got[..].anchor`, not just `at1`/`at2`.) Recompiled
    /// and re-run to confirm the failure, then reverted.
    #[test]
    fn a_33_bit_pts_wrap_does_not_corrupt_event_log_ordering() {
        const PTS_WRAP: u64 = 1u64 << 33;

        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        let mut timeline = timed_metadata::Timeline::new();

        // Event 1: a PTS 10 ticks before the 33-bit wrap point.
        let before_wrap = splice_insert_bytes(1, PTS_WRAP - 10);
        let ev1 = timeline.push_scte35(&before_wrap).unwrap();
        let at1 = ev1.at.unwrap();
        writer.publish_event(ev1, EventAnchor::Media(at1));

        // Event 2: a small RAW PTS after the wrap. `Timeline` must unroll
        // this into a value larger than `at1`, not a small one.
        let after_wrap = splice_insert_bytes(2, 5);
        let ev2 = timeline.push_scte35(&after_wrap).unwrap();
        let at2 = ev2.at.unwrap();
        writer.publish_event(ev2, EventAnchor::Media(at2));

        assert!(
            at2.0 > at1.0,
            "Timeline itself must unroll monotonically: at1={}, at2={}",
            at1.0,
            at2.0
        );

        // The event log must store EXACTLY the MediaTime `Timeline` already
        // unrolled — no re-derivation, re-masking, or truncation of an
        // already-unrolled value anywhere in this module's storage/
        // resolution path. (Publish-order preservation across a wrap is
        // trivial regardless of the anchor's value — `VecDeque` iteration
        // order does not depend on it — so the real assertion here is
        // value-exactness, not position.)
        let got = trunk.events_between(MediaTime(0), MediaTime(u64::MAX));
        assert_eq!(got.len(), 2);
        assert_eq!(got[0].event.id, Some(1));
        assert_eq!(got[1].event.id, Some(2));
        assert!(
            matches!(got[0].anchor, EventAnchor::Media(t) if t.0 == at1.0),
            "event 1's stored anchor must equal Timeline's unrolled value \
             exactly, got {:?}",
            got[0].anchor
        );
        assert!(
            matches!(got[1].anchor, EventAnchor::Media(t) if t.0 == at2.0),
            "event 2's stored (post-wrap) anchor must equal Timeline's \
             unrolled value exactly — not re-masked back into 33 bits, got {:?}",
            got[1].anchor
        );
    }

    // --- E6. the event log is bounded: flooding cannot grow memory --------
    // --- without limit -------------------------------------------------------

    /// MUTATION VERIFIED: removing the eviction check in `EventLog::push`
    /// (replacing `if self.entries.len() == self.capacity { .. }` with a
    /// no-op) makes `trunk.event_len()` grow well past the configured cap
    /// (`3`) instead of staying bounded — the in-loop assertion fails on
    /// the first over-capacity iteration. Recompiled and re-run to confirm
    /// the failure, then reverted.
    #[test]
    fn event_log_is_bounded_under_flood() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(3), nz(8)));
        let writer = trunk.writer().unwrap();

        for i in 0u32..50_000 {
            writer.publish_event(basic_event(i), EventAnchor::Media(MediaTime(u64::from(i))));
            assert!(
                trunk.event_len() <= 3,
                "event log exceeded its cap mid-flood"
            );
        }
        assert_eq!(trunk.event_len(), 3);
    }

    // --- E7. event cursor lag is reported in-band with an accurate --------
    // --- skipped count; the writer never blocks -----------------------------

    /// MUTATION VERIFIED: changing the `skipped` computation in
    /// `EventCursor::poll`'s lag branch from `log.base - self.consumed` to
    /// `log.base - self.consumed + 1` makes this test fail: expected
    /// `skipped: 6`, got `skipped: 7`. Recompiled and re-run to confirm the
    /// failure, then reverted.
    #[test]
    fn event_cursor_lag_is_reported_with_an_accurate_skipped_count_writer_never_blocks() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(3), nz(8)));
        let mut cursor = trunk.subscribe_events();
        let writer = trunk.writer().unwrap();

        // Capacity 3, publish 9: 6 evicted before the cursor ever reads.
        // Never blocks — there is no wait-for-reader path in
        // `EventLog::push`.
        for i in 0u32..9 {
            writer.publish_event(basic_event(i), EventAnchor::Media(MediaTime(u64::from(i))));
        }
        assert_eq!(
            trunk.event_len(),
            3,
            "writer unblocked: event log stayed bounded"
        );

        let first = cursor.poll().unwrap();
        assert!(
            matches!(first, EventCursorItem::Lagged { skipped: 6 }),
            "expected Lagged{{skipped: 6}}, got {first:?}"
        );

        // The remaining 3 (ids 6, 7, 8) must still be readable, in order.
        let mut ids = Vec::new();
        for _ in 0..3 {
            ids.push(event_id(&cursor.poll().unwrap()).unwrap());
        }
        assert_eq!(ids, vec![6, 7, 8]);
        assert!(cursor.poll().is_none());
    }

    // --- E8. `epoch_ms_to_media` really is the inverse of ------------------
    // --- `TimeAnchor::media_to_epoch_ms` -----------------------------------

    /// A wall-clock anchor with room on *both* sides of `pts_90k`, so a sign
    /// error in the (signed) delta shows up rather than being clipped away:
    /// 10 s of media already elapsed, mapped to a realistic epoch instant.
    fn round_trip_anchor() -> TimeAnchor {
        TimeAnchor {
            pts_90k: 900_000,                // 10 s at 90 kHz
            utc_epoch_ms: 1_700_000_000_000, // ~2023-11-14T22:13:20Z
        }
    }

    /// Ticks of the 90 kHz media clock per millisecond of the epoch clock.
    ///
    /// Derived, not asserted: the media clock is [`PTS_HZ`] ticks/second and
    /// `utc_epoch_ms` counts milliseconds, i.e. thousandths of a second, so
    /// one millisecond spans `PTS_HZ / 1000` ticks.
    const TICKS_PER_EPOCH_MS: u64 = PTS_HZ / 1000;

    /// The exact worst-case `media -> epoch_ms -> media` error, in ticks.
    ///
    /// **Derivation (not a tuned constant).**
    /// [`TimeAnchor::media_to_epoch_ms`] computes
    /// `delta_ticks * 1000 / PTS_HZ`, i.e. `delta_ticks / TICKS_PER_EPOCH_MS`,
    /// in integer arithmetic — Rust integer division truncates toward zero,
    /// so it discards a remainder `r` with `|r| <= TICKS_PER_EPOCH_MS - 1`.
    /// `epoch_ms_to_media` then multiplies the surviving whole milliseconds
    /// back up by `TICKS_PER_EPOCH_MS`, reconstructing `delta_ticks - r`
    /// exactly. The round-trip error is therefore *precisely* that discarded
    /// remainder: at most `TICKS_PER_EPOCH_MS - 1` == 89 ticks, i.e. strictly
    /// less than one millisecond. `media_round_trip_is_lossy_by_at_most_one_
    /// millisecond` additionally asserts this bound is **tight** (some input
    /// attains exactly 89), so it cannot silently be loosened into a
    /// tolerance that hides a real error.
    const MEDIA_ROUND_TRIP_MAX_TICKS: u64 = TICKS_PER_EPOCH_MS - 1;

    /// The `epoch_ms -> media -> epoch_ms` direction is **exact** — the media
    /// clock is finer-grained than the millisecond clock (90 ticks per ms),
    /// so no information is lost going to ticks and back. Asserted with
    /// equality, no tolerance.
    ///
    /// MUTATION VERIFIED: flipping the sign of the delta in
    /// `epoch_ms_to_media` (`i128::from(anchor.utc_epoch_ms) -
    /// i128::from(utc_epoch_ms)` in place of the correct
    /// `i128::from(utc_epoch_ms) - i128::from(anchor.utc_epoch_ms)`) makes
    /// this test fail on the first non-zero offset: for `+1` ms the
    /// round-tripped epoch comes back as `1699999999999` instead of
    /// `1700000000001`. **Second mutation, also verified:** changing the
    /// scale conversion from `* PTS_HZ / 1000` to `* PTS_HZ * 1000` fails the
    /// same assertion with `1700001000000` instead of `1700000000001`. So
    /// both the *sign* and the *magnitude* of the inverse are pinned, not
    /// just its shape. Recompiled and re-run to confirm each failure, then
    /// reverted.
    #[test]
    fn epoch_ms_round_trip_through_media_time_is_exact() {
        let anchor = round_trip_anchor();

        // Offsets in ms from the anchor's own epoch instant. Zero, both
        // signs at ±1 ms and ±1 s, a full day forward, a backward offset
        // that lands well clear of the clamp, and one large enough that
        // `delta_ms * PTS_HZ` (2e14 * 9e4 = 1.8e19) exceeds `i64::MAX`
        // (~9.2e18) — the case that exercises the `i128` widening.
        for offset_ms in [
            0i64,
            1,
            -1,
            1_000,
            -1_000,
            86_400_000,
            -9_000,
            200_000_000_000_000,
        ] {
            let epoch_ms = anchor.utc_epoch_ms + offset_ms;
            let media = epoch_ms_to_media(&anchor, epoch_ms).unwrap();
            let back = anchor.media_to_epoch_ms(media);
            assert_eq!(
                back, epoch_ms,
                "epoch_ms -> media -> epoch_ms must be EXACT at offset {offset_ms} ms \
                 (media = {media:?})"
            );
        }
    }

    /// The `media -> epoch_ms -> media` direction is **lossy**, by a bounded
    /// and derived amount: the media clock is 90× finer than the millisecond
    /// clock, so sub-millisecond tick precision cannot survive the trip. See
    /// [`MEDIA_ROUND_TRIP_MAX_TICKS`] for the derivation. This test also
    /// pins the bound as *tight*, so it is a real property and not a loose
    /// tolerance hiding an error.
    ///
    /// MUTATION VERIFIED: flipping the sign of the delta in
    /// `epoch_ms_to_media` (as in
    /// `epoch_ms_round_trip_through_media_time_is_exact`'s note) makes this
    /// test fail at the first offset that is a whole number of milliseconds
    /// away from the anchor: at media offset `+90` ticks the value comes
    /// back as `899_910` instead of `900_090`, a diff of `180` ticks, so the
    /// `diff <= MEDIA_ROUND_TRIP_MAX_TICKS` (89) assertion fails.
    /// **Second mutation, also verified:** the `* PTS_HZ * 1000` scale error
    /// fails the same assertion with a diff of `89_999_910` ticks. Recompiled
    /// and re-run to confirm each failure, then reverted.
    #[test]
    fn media_round_trip_is_lossy_by_at_most_one_millisecond() {
        let anchor = round_trip_anchor();
        let mut worst = 0u64;

        // Tick offsets from the anchor's own `pts_90k`. Both signs, values
        // that are and are not whole multiples of TICKS_PER_EPOCH_MS (so the
        // truncated remainder is genuinely exercised), the exact worst-case
        // remainder on each side (±89), and a large offset well past the
        // i64/i128 boundary region.
        for offset_ticks in [
            0i64,
            1,
            -1,
            89,
            -89,
            90,
            -90,
            91,
            -91,
            18_000_000_000_000_037,
        ] {
            let media = MediaTime((anchor.pts_90k as i64 + offset_ticks) as u64);
            let epoch_ms = anchor.media_to_epoch_ms(media);
            let back = epoch_ms_to_media(&anchor, epoch_ms).unwrap();
            let diff = media.0.abs_diff(back.0);
            assert!(
                diff <= MEDIA_ROUND_TRIP_MAX_TICKS,
                "media -> epoch_ms -> media lost {diff} ticks at offset \
                 {offset_ticks} (bound is {MEDIA_ROUND_TRIP_MAX_TICKS}, i.e. \
                 < 1 ms): {media:?} -> {epoch_ms} -> {back:?}"
            );
            worst = worst.max(diff);
        }

        // The bound is TIGHT: the ±89-tick cases attain it exactly. Without
        // this, `MEDIA_ROUND_TRIP_MAX_TICKS` could be quietly raised to
        // paper over a genuine arithmetic error and the test above would
        // still pass.
        assert_eq!(
            worst, MEDIA_ROUND_TRIP_MAX_TICKS,
            "the derived bound must be attained, not merely respected — \
             otherwise it is a loose tolerance, not a property"
        );
    }

    /// r06-W6: an epoch instant before the anchor's timeline origin implies
    /// a negative media time, which is unrepresentable; `epoch_ms_to_media`
    /// must say so (`None`) instead of clamping to a fabricated `0`.
    /// Mutation check: restoring `media.clamp(0, u64::MAX)` makes the first
    /// assertion fail (`Some(MediaTime(0))`).
    #[test]
    fn epoch_before_the_timeline_origin_is_unresolvable_not_clamped() {
        let anchor = round_trip_anchor();
        // 20 s before the anchor's epoch, but only 10 s of media elapsed
        // at the anchor: the implied media time is -10 s.
        assert_eq!(
            epoch_ms_to_media(&anchor, anchor.utc_epoch_ms - 20_000),
            None
        );
        // Exactly the origin (10 s before the anchor) is representable: 0.
        assert_eq!(
            epoch_ms_to_media(&anchor, anchor.utc_epoch_ms - 10_000),
            Some(MediaTime(0))
        );
        // One millisecond earlier is not.
        assert_eq!(
            epoch_ms_to_media(&anchor, anchor.utc_epoch_ms - 10_001),
            None
        );
        // Beyond u64 ticks is also unresolvable rather than saturated.
        let far = TimeAnchor {
            pts_90k: u64::MAX,
            utc_epoch_ms: 0,
        };
        assert_eq!(epoch_ms_to_media(&far, 1_000), None);
        assert_eq!(epoch_ms_to_media(&far, i64::MIN), None);
    }

    /// r06-W6, end to end: a stale `splice_schedule` UTC instant (before the
    /// timeline origin) stays `Utc` — whether the anchor arrives before or
    /// after the event — and never appears at media time 0.
    #[test]
    fn pre_origin_utc_event_stays_unresolved_through_the_trunk() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        let segment_writer = trunk.segment_writer().unwrap();
        // Event published BEFORE the anchor (resolved in `set_time_anchor`).
        writer.publish_event(
            basic_event(1),
            EventAnchor::Utc {
                utc_epoch_ms: 5_000,
            },
        );
        // Anchor: media 900_000 ticks (10 s) == epoch 100_000 ms, so the
        // timeline origin is epoch 90_000 ms; 5_000 ms is before it.
        segment_writer.set_time_anchor(TimeAnchor {
            pts_90k: 900_000,
            utc_epoch_ms: 100_000,
        });
        // Event published AFTER the anchor (resolved in `try_resolve`); one
        // in-range instant (epoch 100_000 ms == media 900_000) as control.
        writer.publish_event(
            basic_event(2),
            EventAnchor::Utc {
                utc_epoch_ms: 5_000,
            },
        );
        writer.publish_event(
            basic_event(3),
            EventAnchor::Utc {
                utc_epoch_ms: 100_000,
            },
        );
        let got = trunk.events_between(MediaTime(0), MediaTime(u64::MAX));
        assert_eq!(got.len(), 1, "only the in-range event is media-resolvable");
        assert_eq!(got[0].event.id, Some(3));
        assert!(matches!(
            got[0].anchor,
            EventAnchor::Media(MediaTime(900_000))
        ));
        assert!(
            trunk.events_between(MediaTime(0), MediaTime(1)).is_empty(),
            "a pre-origin event must not be reported at the start of the stream"
        );
    }

    // === Step 3b-iv: live-part log + reader-wake primitive =================

    fn part_entry(byte: u8, segment_number: u32, part_index: u32) -> PartEntry {
        PartEntry::new(
            Bytes::from(vec![byte; 8]),
            SegmentSeq::new(segment_number),
            PartIndex::new(part_index),
            Duration::from_millis(200),
            part_index == 0,
        )
    }

    /// r06-O1: part lookups touch only the requested segment's own parts,
    /// never the whole ring. 4_000 resident parts (1_000 segments x 4); the
    /// newest segment's lookups must examine at most its own 4 parts. The
    /// work counter is `PartLog::visits`, incremented at the single point
    /// every lookup dereferences a ring entry. Bite-check: replacing the
    /// two `PartLog` lookups with the old full-ring `iter().filter`/`find`
    /// scan (counting each examined entry) makes this fail at ~4_000.
    #[test]
    fn part_lookups_scan_only_their_own_segment_not_the_ring() {
        use core::sync::atomic::Ordering;
        const SEGMENTS: u32 = 1_000;
        const PARTS_PER_SEGMENT: u32 = 4;
        let ring = usize::try_from(SEGMENTS * PARTS_PER_SEGMENT).unwrap();
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(ring)));
        let writer = trunk.segment_writer().unwrap();
        for seg in 0..SEGMENTS {
            for part in 0..PARTS_PER_SEGMENT {
                writer.publish_part(part_entry(
                    u8::try_from((seg + part) % 251).unwrap(),
                    seg,
                    part,
                ));
            }
        }
        assert_eq!(trunk.part_len(), ring);

        let visits = || {
            trunk
                .lock_state()
                .parts
                .entries
                .visits
                .load(Ordering::Relaxed)
        };
        let before = visits();
        let newest = SEGMENTS - 1;
        let parts = trunk.parts_in_segment(SegmentSeq::new(newest));
        assert_eq!(
            parts
                .iter()
                .map(|p| p.part_index.value())
                .collect::<Vec<_>>(),
            vec![0, 1, 2, 3]
        );
        assert_eq!(visits() - before, 4, "parts_in_segment work units");

        let before = visits();
        let bytes = trunk
            .part_bytes(SegmentSeq::new(newest), PartIndex::new(3))
            .expect("newest part resident");
        assert_eq!(
            bytes.as_ref(),
            &[u8::try_from((newest + 3) % 251).unwrap(); 8]
        );
        assert!(
            visits() - before <= 4,
            "part_bytes must touch at most its own segment's parts, touched {}",
            visits() - before
        );
        // A segment that was never published touches nothing.
        let before = visits();
        assert!(
            trunk
                .part_bytes(SegmentSeq::new(SEGMENTS + 5), PartIndex::new(0))
                .is_none()
        );
        assert!(
            trunk
                .parts_in_segment(SegmentSeq::new(SEGMENTS + 5))
                .is_empty()
        );
        assert_eq!(visits() - before, 0);
        // Index keys are bounded by the resident segments: a 4_000-part
        // ring of 4-part segments holds exactly 1_000 distinct keys.
        assert_eq!(trunk.lock_state().parts.by_segment.len(), 1_000);
        // Publish one more segment: the oldest segment's key is evicted
        // part by part, never left dangling.
        for part in 0..PARTS_PER_SEGMENT {
            writer.publish_part(part_entry(1, SEGMENTS, part));
        }
        let state = trunk.lock_state();
        assert_eq!(state.parts.by_segment.len(), 1_000);
        assert!(!state.parts.by_segment.contains_key(&SegmentSeq::new(0)));
    }

    /// The index must follow eviction exactly: after the ring wraps, evicted
    /// parts are gone, a partly-evicted segment reports only its survivors
    /// (publish order), a reused `segment_number` returns both runs with the
    /// newest winning `part_bytes`, and the index never outgrows the ring.
    #[test]
    fn part_index_tracks_eviction_and_reused_segment_numbers() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(5)));
        let writer = trunk.segment_writer().unwrap();
        // Ring of 5: (7,0) (7,1) (7,2) (8,0) (8,1) ...
        for (seg, part, byte) in [(7, 0, 10), (7, 1, 11), (7, 2, 12), (8, 0, 20), (8, 1, 21)] {
            writer.publish_part(part_entry(byte, seg, part));
        }
        // ... two more evict (7,0) and (7,1); then segment 7 is REUSED.
        writer.publish_part(part_entry(22, 8, 2));
        writer.publish_part(part_entry(99, 7, 2));
        assert_eq!(
            trunk.part_bytes(SegmentSeq::new(7), PartIndex::new(0)),
            None,
            "evicted"
        );
        assert_eq!(
            trunk.part_bytes(SegmentSeq::new(7), PartIndex::new(1)),
            None,
            "evicted"
        );
        let seven: Vec<(PartIndex, u8)> = trunk
            .parts_in_segment(SegmentSeq::new(7))
            .iter()
            .map(|p| (p.part_index, p.bytes[0]))
            .collect();
        assert_eq!(
            seven,
            vec![(PartIndex::new(2), 12), (PartIndex::new(2), 99)],
            "survivor then the reused run"
        );
        assert_eq!(
            trunk
                .part_bytes(SegmentSeq::new(7), PartIndex::new(2))
                .unwrap()
                .as_ref(),
            &[99u8; 8],
            "newest publish of the pair wins"
        );
        assert_eq!(trunk.part_len(), 5);
        let state = trunk.lock_state();
        let indexed: usize = state.parts.by_segment.values().map(|v| v.len()).sum();
        assert_eq!(indexed, 5, "index holds exactly the resident entries");
        assert_eq!(state.parts.entries.iter().count(), 5);
        assert_eq!(state.parts.by_segment.len(), 2, "emptied keys are removed");
    }

    /// The LL-HLS property this whole step exists for: a part must be
    /// addressable and readable **before** its parent segment closes — if
    /// this does not hold, nothing else in this step matters (blocking
    /// reload has nothing to answer with).
    ///
    /// MUTATION VERIFIED: commenting out `state.parts.push(entry);` in
    /// `SegmentWriter::publish_part` (simulating "the part never actually
    /// lands in the log") makes `trunk.part_bytes(SegmentSeq::new(9), PartIndex::new(0))` return `None`
    /// instead of `Some(..)`, failing the `expect` below. Recompiled and
    /// re-run to confirm the failure, then reverted.
    #[test]
    fn part_is_addressable_and_readable_before_its_parent_segment_closes() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.segment_writer().unwrap();

        // Segment 9 has never been closed — no `publish_segment` call for it
        // anywhere in this test.
        assert!(
            trunk.last_closed_segment().is_none(),
            "sanity check: nothing has closed yet"
        );

        writer.publish_part(part_entry(0xAB, 9, 0));

        let bytes = trunk
            .part_bytes(SegmentSeq::new(9), PartIndex::new(0))
            .expect("a published part of an open segment must be addressable now");
        assert_eq!(bytes, Bytes::from(vec![0xAB; 8]));

        // Still true: segment 9 has still never closed.
        assert!(
            trunk.last_closed_segment().is_none(),
            "the part landed without any segment ever closing"
        );
    }

    /// A waiter blocked on a not-yet-existing part wakes once it is
    /// published, and resolves to exactly that part — not merely "wakes",
    /// which a mutation could satisfy vacuously if `part_bytes` mismatched
    /// the wrong entry (this test publishes a decoy part first to make
    /// "the right one" a real assertion).
    ///
    /// MUTATION VERIFIED (two independent mutations, each reverted after
    /// confirming failure):
    /// 1. Removing `self.trunk.progress.notify(usize::MAX);` from
    ///    `SegmentWriter::publish_part` makes `listener.wait_deadline(..)`
    ///    time out (`false`) instead of waking (`true`) within the 2 s bound
    ///    used below — the first assertion fails.
    /// 2. Changing `Trunk::part_bytes`'s filter from
    ///    `p.segment_number == segment_number && p.part_index == part_index`
    ///    to drop the `part_index` half (matching on `segment_number` alone)
    ///    makes the final `assert_eq!` fail: with the decoy part (index 1)
    ///    published first, `part_bytes(SegmentSeq::new(9), PartIndex::new(0))` would resolve to the decoy's
    ///    `0xCC` bytes instead of the awaited part's `0xAB` bytes.
    #[test]
    fn waiter_is_woken_when_the_awaited_part_lands_and_resolves_to_it() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = Arc::new(trunk.segment_writer().unwrap());

        // Register BEFORE re-checking/waiting — the documented no-missed-
        // wakeup ordering.
        let listener = trunk.listen().expect("first registration must succeed");
        assert!(
            trunk
                .part_bytes(SegmentSeq::new(9), PartIndex::new(0))
                .is_none(),
            "not published yet"
        );

        let bg_writer = Arc::clone(&writer);
        let handle = thread::spawn(move || {
            // A decoy: a different part of the same segment, published
            // first. If `part_bytes` ever matched on `segment_number` alone,
            // this would be the value wrongly returned for a request for
            // part 0.
            bg_writer.publish_part(part_entry(0xCC, 9, 1));
            bg_writer.publish_part(part_entry(0xAB, 9, 0));
        });

        // Deliberately generous. The claim under test is "the listener wakes
        // rather than parking forever", NOT "it wakes inside N seconds": the
        // publish happens on another thread, so any tight upper bound is a
        // bound on the *machine's* scheduling, not on this code. A 2s bound
        // here failed once during a loaded full-workspace run and passed on
        // 38 consecutive idle runs -- a false red that trains people to
        // re-run the suite. 10s still fails instantly if the wake channel
        // genuinely never fires, which is the only failure worth reporting.
        let woken = listener.wait_deadline(std::time::Instant::now() + Duration::from_secs(10));
        assert!(
            woken,
            "listener must wake on publish_part, not park forever"
        );
        handle.join().unwrap();

        let bytes = trunk
            .part_bytes(SegmentSeq::new(9), PartIndex::new(0))
            .expect("the awaited part must now be readable");
        assert_eq!(
            bytes,
            Bytes::from(vec![0xAB; 8]),
            "must resolve to the awaited part (index 0), not the decoy (index 1)"
        );
    }

    /// A waiter whose target never arrives is bounded by its own deadline —
    /// it does not park forever. This composes with
    /// `crate::egress::AwaitPolicy`'s deadline exactly the same way: the
    /// caller converts its own bound to a `std::time::Instant` and passes it
    /// here.
    ///
    /// MUTATION VERIFIED: changing `ProgressListener::wait_deadline`'s body
    /// from `listener.wait_deadline(deadline).is_some()` to unconditionally
    /// `true` makes this test's `assert!(!woken, ..)` fail — `woken` is
    /// `true` even though nothing was ever published. Recompiled and re-run
    /// to confirm the failure, then reverted.
    #[test]
    fn waiter_whose_target_never_arrives_is_bounded_not_parked_forever() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let listener = trunk.listen().unwrap();

        let start = std::time::Instant::now();
        let woken = listener.wait_deadline(start + Duration::from_millis(150));
        let elapsed = start.elapsed();

        assert!(!woken, "must report timeout, not a fabricated wake-up");
        // Same reasoning as the generous bound in
        // `awaited_part_wakes_its_listener`: this asserts "returns rather
        // than parking forever", and any tight bound measures the machine.
        // The real assertion is `!woken` above; this one only catches a hang.
        assert!(
            elapsed < Duration::from_secs(60),
            "must actually return at the deadline, not hang: took {elapsed:?}"
        );
    }

    /// The hard invariant this whole file exists to preserve, extended to
    /// the wake channel: `publish_part`/`publish_segment` must still
    /// complete even with a waiter registered and never serviced (no one
    /// ever calls `wait`/`.await`/drops it) — a slow or vanished reader must
    /// never stall the writer. Proven the same way this file's existing
    /// `StallIngest`-blocks proof works, but for the opposite claim: a
    /// background-thread `publish_*` call, and a bounded `recv_timeout`
    /// proving it completed promptly.
    ///
    /// This specific guarantee is structural, not something a local
    /// mutation of this crate's code can plausibly violate: `publish_part`/
    /// `publish_segment` call `Event::notify(usize::MAX)`, which by
    /// `event_listener`'s own documented contract wakes registered listeners
    /// without waiting for any of them to resume — there is no "wait for
    /// the listener to be serviced" code path in this module to remove.
    /// Reaching a blocking wake-up would require swapping the whole
    /// primitive for a different one (the architectural choice already
    /// argued in this module's docs), not a one-line mutation, so no
    /// mutation transcript is claimed for this test — see this crate's
    /// convention that a structural property is reported as such rather
    /// than backed by an invented mutation.
    #[test]
    fn writer_never_blocks_with_a_registered_never_serviced_waiter() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = Arc::new(trunk.segment_writer().unwrap());

        // Registered, kept alive for the whole test, and never waited on or
        // dropped before the assertions below run.
        let _never_serviced = trunk.listen().unwrap();

        let (done_tx, done_rx) = mpsc::channel();
        let bg_writer = Arc::clone(&writer);
        thread::spawn(move || {
            bg_writer.publish_part(part_entry(1, 1, 0));
            bg_writer.publish_segment(segment_entry(2, 1)).unwrap();
            done_tx.send(()).unwrap();
        });

        // HANG GUARD (issue #807): the property under test is "never blocks",
        // i.e. these calls return almost immediately (`Event::notify`
        // doesn't wait for listeners to resume) -- a stuck/blocked writer
        // here is the only thing this should ever catch, so raised for
        // load-tolerance rather than left as a timing claim.
        done_rx.recv_timeout(Duration::from_secs(60)).expect(
            "publish_part/publish_segment must complete promptly even with \
                 a live, never-serviced waiter registered",
        );
    }

    /// The waiter set itself is bounded — a flood of `listen()` calls cannot
    /// grow memory without limit, and reuses `part_capacity` rather than a
    /// sixth, independent knob.
    ///
    /// MUTATION VERIFIED: replacing `Trunk::listen`'s cap check (the
    /// `if current >= self.part_waiter_cap { return None; }` loop) with a
    /// version that always registers (never returns `None`) makes the
    /// `assert!(trunk.listen().is_none(), ..)` below fail — the call
    /// succeeds instead of being refused at the cap. Recompiled and re-run
    /// to confirm the failure, then reverted.
    #[test]
    fn waiter_set_is_bounded_a_flood_of_listen_calls_cannot_grow_without_limit() {
        let cap = nz(4).get();
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(cap)));

        // Fill exactly to the cap, keeping every registration alive.
        let mut held: Vec<ProgressListener> = Vec::new();
        for _ in 0..cap {
            held.push(trunk.listen().expect("must succeed up to the cap"));
        }
        assert_eq!(trunk.waiter_count(), cap);

        // One more must be refused, not silently over-admitted.
        assert!(
            trunk.listen().is_none(),
            "must refuse a registration beyond part_capacity"
        );

        // A flood of register-then-immediately-drop calls (no one keeping
        // them alive) must never push the live count past the cap, however
        // many times it runs.
        for _ in 0..50_000 {
            let l = trunk.listen();
            assert!(
                trunk.waiter_count() <= cap,
                "waiter count exceeded part_capacity mid-flood"
            );
            drop(l);
        }
        assert_eq!(
            trunk.waiter_count(),
            cap,
            "the held registrations are still exactly at the cap"
        );

        // Releasing one held slot frees exactly one registration.
        held.pop();
        assert_eq!(trunk.waiter_count(), cap - 1);
        assert!(
            trunk.listen().is_some(),
            "a released slot must be re-usable"
        );
    }

    /// Releasing a waiter slot wakes ONLY a caller parked on
    /// `waiter_slot_freed`, never an unrelated caller parked on the shared
    /// `progress` event (the part/segment-close wake every `listen()` waiter
    /// is on).
    ///
    /// MUTATION VERIFIED: restoring `self.0.progress.notify(usize::MAX);` in
    /// `WaiterSlot::drop` (the wake-storm bug — C1) makes this test fail: the
    /// 50 ms wait below wakes (`woken == true`) because the drop fires the
    /// shared `progress` event the `b` waiter is registered on, instead of the
    /// dedicated `slot_freed` event. Recompiled and re-run to confirm the
    /// failure, then reverted.
    #[test]
    fn freeing_a_waiter_slot_does_not_wake_an_unrelated_progress_waiter() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));

        let a = trunk.listen().expect("first registration succeeds");
        // `b` is a `progress` waiter (its listener is registered on the shared
        // event at `listen()` time) that must NOT be woken by A's slot release.
        let b = trunk.listen().expect("second registration succeeds");

        drop(a);

        let woken = b.wait_deadline(std::time::Instant::now() + Duration::from_millis(50));
        assert!(
            !woken,
            "releasing a waiter slot must not wake an unrelated `progress` \
             waiter — that is the ping-pong wake-storm (a >2-waiter caller set \
             would 100% CPU with no media progress)"
        );
    }

    /// A caller parked on `waiter_slot_freed` IS woken when a holder releases
    /// its slot — the positive half of the dedicated-event fix.
    #[test]
    fn a_parked_slot_freed_waiter_wakes_when_a_slot_is_freed() {
        let cap = nz(4).get();
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(cap)));

        // Fill to the cap so `waiter_slot_freed` returns a real listener.
        let mut held: Vec<ProgressListener> = Vec::new();
        for _ in 0..cap {
            held.push(trunk.listen().expect("succeeds up to the cap"));
        }
        let parked = trunk
            .waiter_slot_freed()
            .expect("at the cap, a slot-freed future must be returned");

        // A slot freed on another thread must wake the parked waiter.
        let holder = held.pop().expect("a held slot");
        let handle = thread::spawn(move || {
            drop(holder);
        });

        let woken = parked
            .wait_deadline(std::time::Instant::now() + Duration::from_secs(10))
            .is_some();
        assert!(
            woken,
            "a released slot must wake the caller parked on `waiter_slot_freed`"
        );
        handle.join().unwrap();
    }

    /// `waiter_slot_freed` registers BEFORE re-checking the count (the same
    /// no-missed-wakeup ordering `listen` requires), so a slot freed between
    /// the registration and the check cannot be lost. Bites the old
    /// check-then-listen: with the cap saturated, a simultaneous release on
    /// another thread races the check — a check-first shape can observe "full"
    /// after the release and still return a listener that nothing will ever
    /// wake.
    ///
    /// MUTATION VERIFIED: reordering `waiter_slot_freed` to check the count
    /// first and only then `self.slot_freed.listen()` makes this fail
    /// intermittently-to-deterministically (the release lands in the window);
    /// the register-first shape cannot.
    #[test]
    fn waiter_slot_freed_registers_before_rechecking_the_count() {
        let cap = nz(4).get();
        for _ in 0..200 {
            let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(cap)));
            let mut held: Vec<ProgressListener> = Vec::new();
            for _ in 0..cap {
                held.push(trunk.listen().expect("succeeds up to the cap"));
            }

            // Release one slot concurrently with the `waiter_slot_freed` call.
            let holder = held.pop().expect("a held slot");
            let t = Arc::clone(&trunk);
            let handle = thread::spawn(move || {
                drop(holder);
                // Keep the `Trunk` alive until the waiter below has resolved.
                let _ = &t;
            });

            let fut = trunk.waiter_slot_freed();
            match fut {
                // A slot is free now (the release already landed): the caller
                // must fall through to a live `listen`, not park.
                None => {
                    assert!(
                        trunk.listen().is_some(),
                        "a `None` from `waiter_slot_freed` promises a free slot"
                    );
                }
                // The check saw "full": the registered listener must still be
                // woken by the concurrent release (register-before-recheck).
                Some(listener) => {
                    let woken = listener
                        .wait_deadline(std::time::Instant::now() + Duration::from_millis(50))
                        .is_some();
                    assert!(
                        woken,
                        "a slot freed concurrently with `waiter_slot_freed` must \
                         wake the returned listener (check-then-listen misses it)"
                    );
                }
            }
            handle.join().unwrap();
        }
    }

    /// The decided close-behaviour, asserted: a part remains addressable via
    /// `part_bytes` after its parent segment closes (this trunk's `Trunk`
    /// does not evict/transform parts on `publish_segment`), right up until
    /// `part_capacity`'s ordinary eviction reclaims it — at which point a
    /// client requesting that same part gets `None`, indistinguishable from
    /// "never existed", exactly like every other ring's eviction in this
    /// module.
    ///
    /// MUTATION VERIFIED: adding an eviction step to `publish_segment` that
    /// removes every `PartLog` entry whose `segment_number` matches the
    /// just-closed segment (simulating the rejected "evict a segment's
    /// parts the instant it closes" alternative documented in this module's
    /// docs) makes the first `part_bytes(SegmentSeq::new(1), PartIndex::new(0))` assertion below fail
    /// immediately after `publish_segment` — it returns `None` instead of
    /// `Some(..)`. Recompiled and re-run to confirm the failure, then
    /// reverted.
    #[test]
    fn parts_remain_addressable_after_segment_close_until_ordinary_eviction_reclaims_them() {
        let part_cap = nz(4).get();
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(part_cap)));
        let writer = trunk.segment_writer().unwrap();

        writer.publish_part(part_entry(0xAB, 1, 0));
        writer.publish_segment(segment_entry(1, 1)).unwrap();

        // The part a client just watched roll into a closed segment is
        // still `Some` — the same answer as before the close.
        assert_eq!(
            trunk.part_bytes(SegmentSeq::new(1), PartIndex::new(0)),
            Some(Bytes::from(vec![0xAB; 8])),
            "a just-closed segment's part must still be individually fetchable"
        );
        assert_eq!(trunk.last_closed_segment(), Some(SegmentSeq::new(1)));

        // Flood the part ring with `part_cap` more entries for an unrelated
        // segment — enough to evict the original part via ordinary
        // capacity-based eviction, with no further segment closes involved.
        for i in 0..part_cap as u32 {
            writer.publish_part(part_entry(0xFF, 99, i));
        }

        assert!(
            trunk
                .part_bytes(SegmentSeq::new(1), PartIndex::new(0))
                .is_none(),
            "the part is gone once ordinary part_capacity eviction reclaims \
             it — NOT because its segment closed, but because the ring's own \
             bound was exceeded, exactly like every other ring in this module"
        );
    }

    // === Issue #781: track-set snapshot + generation counter ==============

    /// A freshly-minted `Trunk` announces no tracks yet — `set_tracks` has
    /// never been called, so there is nothing to seed `tracks()`/
    /// `track_generation()` from other than the empty defaults `Trunk::new`
    /// establishes.
    ///
    /// MUTATION VERIFIED: changing `Trunk::new`'s `track_generation: 0` to
    /// `track_generation: 1` makes the second `assert_eq!` below fail — it
    /// reads back `1` instead of `0`. Recompiled and re-run to confirm the
    /// failure, then reverted.
    #[test]
    fn fresh_trunk_has_no_tracks_and_generation_zero() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        assert_eq!(trunk.tracks().len(), 0, "nothing has ever set a track set");
        assert_eq!(trunk.track_generation(), 0);
    }

    /// `set_tracks` is a **whole-set replacement**, not a merge/append, and
    /// bumps the generation by exactly one per call.
    ///
    /// MUTATION VERIFIED: changing `TrunkWriter::set_tracks`'s body to merge
    /// the old set with the new one (`let mut merged = state.tracks.to_vec();
    /// merged.extend(tracks); state.tracks = Arc::from(merged);`) instead of
    /// replacing outright makes the second `assert_eq!` below fail —
    /// `trunk.tracks()`'s track ids read back `[1, 7, 9]` (the old track 1
    /// still present) instead of `[7, 9]`. Recompiled and re-run to confirm
    /// the failure, then reverted.
    #[test]
    fn set_tracks_replaces_the_whole_set_and_bumps_generation_by_one_per_call() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();

        writer.set_tracks(vec![opaque_track(TrackId::new(1))]);
        assert_eq!(
            trunk
                .tracks()
                .iter()
                .map(|t| t.track_id)
                .collect::<Vec<_>>(),
            vec![TrackId::new(1)]
        );
        assert_eq!(trunk.track_generation(), 1);

        // A completely different, larger set: if this were a merge/append
        // rather than a replacement, the old track_id 1 would still be
        // present alongside the two new ones.
        writer.set_tracks(vec![
            opaque_track(TrackId::new(7)),
            opaque_track(TrackId::new(9)),
        ]);
        assert_eq!(
            trunk
                .tracks()
                .iter()
                .map(|t| t.track_id)
                .collect::<Vec<_>>(),
            vec![TrackId::new(7), TrackId::new(9)],
            "set_tracks must replace the set wholesale, not append to it"
        );
        assert_eq!(
            trunk.track_generation(),
            2,
            "generation must advance by exactly one per set_tracks call"
        );
    }

    /// `track_generation` is stable across everything that is *not*
    /// `set_tracks` — publishing samples/events/segments/parts must never
    /// bump it, so a consumer polling the generation as a cheap "did the
    /// track set change" check cannot see false positives.
    ///
    /// MUTATION VERIFIED: adding `state.track_generation += 1;` to
    /// `TrunkWriter::publish` (simulating "generation accidentally bumped by
    /// unrelated activity") makes the final `assert_eq!` below fail — the
    /// generation reads back `11` (bumped once per one of the 10 published
    /// samples, on top of the `1` from `set_tracks`) instead of staying at
    /// `1`. Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn generation_is_stable_across_unrelated_activity() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();

        writer.set_tracks(vec![opaque_track(TrackId::new(1))]);
        assert_eq!(trunk.track_generation(), 1);

        for i in 0u8..10 {
            writer.publish(TrackId::new(1), RetentionClass::Timed, sample(i, 4));
        }
        writer.publish_event(basic_event(1), EventAnchor::Media(MediaTime(0)));

        assert_eq!(
            trunk.track_generation(),
            1,
            "publishing samples/events must never bump track_generation"
        );
    }

    /// `set_tracks` wakes a registered [`Trunk::listen`] listener, the same
    /// broad `progress` channel [`SegmentWriter::publish_part`]/
    /// [`SegmentWriter::publish_segment`] already wake — see
    /// [`waiter_is_woken_when_the_awaited_part_lands_and_resolves_to_it`] for
    /// the identical pattern this test mirrors.
    ///
    /// MUTATION VERIFIED: removing `self.trunk.progress.notify(usize::MAX);`
    /// from `TrunkWriter::set_tracks` makes `listener.wait_deadline(..)`
    /// time out (`false`) instead of waking (`true`) within the 10s bound
    /// used below. Recompiled and re-run to confirm the failure, then
    /// reverted.
    #[test]
    fn set_tracks_wakes_a_registered_listener() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = Arc::new(trunk.writer().unwrap());

        // Register BEFORE the change — the documented no-missed-wakeup
        // ordering every other `listen()` test in this module follows.
        let listener = trunk.listen().expect("first registration must succeed");

        let bg_writer = Arc::clone(&writer);
        let handle = thread::spawn(move || {
            bg_writer.set_tracks(vec![opaque_track(TrackId::new(1))]);
        });

        // Same generous, machine-independent bound as this module's other
        // wake tests — see their comments for why 10s asserts only "it woke
        // at all", not "it woke fast".
        let woken = listener.wait_deadline(std::time::Instant::now() + Duration::from_secs(10));
        assert!(woken, "listener must wake on set_tracks, not park forever");
        handle.join().unwrap();

        assert_eq!(trunk.track_generation(), 1);
    }

    /// A panic on another thread while it holds `state` must not break later
    /// calls. `parking_lot::Mutex` has no poisoning, so there is nothing to
    /// recover from — the test pins that no future change reintroduces a
    /// poisoning lock (a `.lock().unwrap()` on the std type would turn this
    /// panic into a permanent one).
    #[test]
    fn subscriber_side_panic_while_holding_the_lock_does_not_poison_later_calls() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();

        let trunk_for_panic = Arc::clone(&trunk);
        let handle = thread::spawn(move || {
            let _guard = trunk_for_panic.state.lock();
            panic!("simulated panic while holding Trunk::state");
        });
        assert!(
            handle.join().is_err(),
            "setup: the thread must have panicked"
        );

        writer.publish(TrackId::new(1), RetentionClass::Timed, sample(1, 4));
        assert_eq!(
            trunk.timed_len(),
            1,
            "publish after the panic must take effect"
        );
        assert!(
            trunk.state.try_lock().is_some(),
            "the lock must be free after the panicking holder unwound"
        );
    }

    /// A segment must never be observable before the samples it was built
    /// from — the guarantee the single lock gave structurally and a split
    /// would keep by program order + rest-then-samples read order.
    #[test]
    fn segment_never_overtakes_its_samples_under_a_split_lock() {
        const ROUNDS: u32 = 2_000;
        let trunk = Trunk::new(TrunkConfig::new(nz(4096), nz(4), nz(4096), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        let seg_writer = trunk.segment_writer().unwrap();
        let mut seg_samples = trunk.subscribe();
        let mut watcher_samples = trunk.subscribe();
        let mut watcher_segments = trunk.subscribe_segments();

        let ingest = thread::spawn(move || {
            for i in 0..ROUNDS {
                writer.publish(
                    TrackId::new(1),
                    RetentionClass::Timed,
                    sample((i % 251) as u8, 4),
                );
            }
        });
        // The "segmenter": publishes segment k only after polling k samples.
        let segmenter = thread::spawn(move || {
            let mut seen = 0u32;
            let mut seq = 1u32;
            while seq <= ROUNDS {
                if let Some(SampleCursorItem::Timed { .. }) = seg_samples.poll() {
                    seen += 1;
                    if seen == seq {
                        seg_writer.publish_segment(segment_entry(0, seq)).unwrap();
                        seq += 1;
                    }
                }
            }
        });
        // The watcher: for each segment it sees (segment state first), the
        // sample cursor (sample state second) must already be able to
        // deliver at least `seq` samples.
        let mut delivered = 0u32;
        let mut segs = 0u32;
        while segs < ROUNDS {
            if let Some(SegmentCursorItem::Segment(entry)) = watcher_segments.poll() {
                segs += 1;
                while delivered < entry.sequence_number.value() {
                    match watcher_samples.poll() {
                        Some(SampleCursorItem::Timed { .. }) => delivered += 1,
                        Some(_) => {}
                        None => panic!(
                            "segment {} visible but only {delivered} samples deliverable",
                            entry.sequence_number
                        ),
                    }
                }
            }
        }
        ingest.join().unwrap();
        segmenter.join().unwrap();
    }

    /// `SegmentWriter::try_publish_segment` lets a single thread that owns
    /// *both* the writer and the pin's own draining loop make progress
    /// without ever blocking itself — exactly the case
    /// `SegmentWriter::publish_segment`'s unconditional block cannot
    /// service: on one thread, nothing else could ever call `poll` to
    /// release the pin.
    ///
    /// MUTATION VERIFIED: changing `try_publish_segment`'s
    /// `if must_wait { return Err(entry); }` to instead call
    /// `self.publish_segment(entry).unwrap(); return Ok(());` (falling back to the
    /// blocking path) was run once on this single-threaded test and, as
    /// expected, never returned — confirmed by wrapping the call in a
    /// `thread::spawn` + bounded `recv_timeout`, which timed out instead of
    /// completing. Reverted immediately (not left in the suite, since a
    /// genuinely-hanging assertion would stall the whole run); this is why
    /// the mutation transcript is reported narratively here rather than as
    /// a recompiled/rerun in-place edit like this module's other entries.
    #[test]
    fn try_publish_segment_lets_a_single_thread_drain_its_own_pin() {
        let trunk = Trunk::new(TrunkConfig::new(nz(10), nz(10), nz(1), nz(8), nz(8)));
        let writer = trunk.segment_writer().unwrap();
        let mut cursor = trunk.pin_segments(ArchiveOverrun::StallIngest);

        // Fills the one-entry segment log.
        writer
            .try_publish_segment(segment_entry(1, 1))
            .expect("room for the first segment");

        // A second publish would need to evict seq 1, which the
        // `StallIngest` pin has not consumed yet: this must be reported
        // immediately as `Err`, not block — there is no other thread on
        // this test that could ever drain it.
        let entry2 = match writer.try_publish_segment(segment_entry(2, 2)) {
            Err(TryPublishSegmentError::WouldStall(entry)) => entry,
            other => panic!(
                "must report WouldStall (Err) instead of silently evicting \
                 the un-consumed seq 1: {other:?}"
            ),
        };
        assert_eq!(
            trunk.segment_len(),
            1,
            "seq 1 must not have been evicted yet"
        );

        // This single thread drains the pin itself — exactly what a
        // caller's own retry loop is expected to do between attempts (see
        // `try_publish_segment`'s doc example).
        assert!(matches!(cursor.poll(), Some(SegmentCursorItem::Segment(_))));

        // The retry now succeeds, without this thread ever having blocked.
        writer
            .try_publish_segment(entry2)
            .expect("must succeed once the pin has advanced past seq 1");
        assert_eq!(trunk.segment_len(), 1);
        assert_eq!(trunk.last_closed_segment(), Some(SegmentSeq::new(2)));
    }

    /// `SegmentWriter::publish_segment`'s block on a stuck
    /// `ArchiveOverrun::StallIngest` pin is bounded, not permanent: past
    /// `STALL_INGEST_MAX_WAIT` it gives up and evicts anyway, exactly like
    /// `ArchiveOverrun::Gap` would, rather than hanging the calling thread
    /// forever. This is what turns the single-thread deadlock case (this
    /// same thread also owns the only cursor that could ever drain the
    /// pin) into a bounded delay instead of a permanent hang.
    #[test]
    fn publish_segment_blocking_path_is_bounded_not_permanent() {
        let trunk = Trunk::new(TrunkConfig::new(nz(10), nz(10), nz(1), nz(8), nz(8)));
        let writer = trunk.segment_writer().unwrap();
        // A pin that is never, ever drained on this single thread.
        let mut cursor = trunk.pin_segments(ArchiveOverrun::StallIngest);

        writer.publish_segment(segment_entry(1, 1)).unwrap();

        // This call would block forever pre-fix (nothing ever drains the
        // pin above); post-fix it returns once the test-sized
        // `STALL_INGEST_MAX_WAIT` elapses. Run on a background thread and
        // observed through a channel with a generous bound, so a
        // regression (an unbounded wait) fails this test instead of
        // hanging the whole suite — the same pattern this module's other
        // cross-thread wake tests already use.
        let (done_tx, done_rx) = mpsc::channel();
        let writer = Arc::new(writer);
        let bg_writer = Arc::clone(&writer);
        let handle = thread::spawn(move || {
            bg_writer.publish_segment(segment_entry(2, 2)).unwrap();
            done_tx.send(()).unwrap();
        });
        done_rx
            .recv_timeout(Duration::from_secs(10))
            .expect("publish_segment must give up and return within its bounded wait");
        handle.join().unwrap();

        assert_eq!(
            trunk.segment_len(),
            1,
            "seq 1 was evicted to make room for seq 2"
        );
        assert_eq!(trunk.last_closed_segment(), Some(SegmentSeq::new(2)));

        // The pin's guarantee was broken by the bound, not silently
        // downgraded to ordinary `Gap` loss: its cursor must report
        // `Terminated`, exactly `ArchiveOverrun::Terminate`'s own signal.
        assert!(
            matches!(cursor.poll(), Some(SegmentCursorItem::Terminated)),
            "an expired StallIngest wait must terminate the pin, not \
             silently gap it"
        );
    }

    // --- Writer re-issue / reconnect (T5, issue #1082) --------------------

    /// The main T5 regression test: an already-subscribed `SampleCursor`,
    /// never resubscribed, keeps reading across a `TrunkWriter` drop +
    /// re-issue (a source reconnect) — A's samples, exactly one
    /// `Discontinuity`, then B's samples, in order.
    ///
    /// Verified to fail pre-fix by temporarily reverting two things in a
    /// scratch copy of this file (backed up first, no `git stash`):
    /// (1) removing `impl Drop for TrunkWriter` makes `trunk.writer()`'s
    /// second call return `None` instead of `Some` — the whole scenario
    /// cannot even be set up, since a dropped writer's slot is never
    /// released; (2) with the `Drop` restored but the `writer_issued_once`/
    /// `WriterHandover` recording in `Trunk::writer` reverted to nothing,
    /// the cursor instead sees A's two samples immediately followed by B's
    /// two samples with **no** `Discontinuity` in between — the middle
    /// `poll()` in this test returns a `Timed` sample instead, failing the
    /// `matches!(.., SampleCursorItem::Discontinuity)` assertion. Both
    /// reverts were compiled, run, confirmed failing exactly as described,
    /// then restored.
    #[test]
    fn writer_handover_keeps_subscribers_reading_across_a_reconnect() {
        let trunk = Trunk::new(TrunkConfig::new(nz(10), nz(10), nz(4), nz(8), nz(8)));
        let mut cursor = trunk.subscribe();

        {
            let writer_a = trunk.writer().expect("first take succeeds");
            writer_a.publish(TrackId::new(1), RetentionClass::Timed, sample(0xAA, 4));
            writer_a.publish(TrackId::new(1), RetentionClass::Timed, sample(0xAB, 4));
        } // A dropped here — its slot must be released.

        let writer_b = trunk
            .writer()
            .expect("re-issue after the first writer was dropped must succeed");
        writer_b.publish(TrackId::new(1), RetentionClass::Timed, sample(0xBA, 4));
        writer_b.publish(TrackId::new(1), RetentionClass::Timed, sample(0xBB, 4));

        for expected in [0xAAu8, 0xAB] {
            match cursor.poll() {
                Some(SampleCursorItem::Timed { sample, .. }) => {
                    assert_eq!(sample.data.as_ref(), &[expected; 4]);
                }
                other => panic!("expected A's sample 0x{expected:02X}, got {other:?}"),
            }
        }

        assert!(
            matches!(cursor.poll(), Some(SampleCursorItem::Discontinuity)),
            "exactly one Discontinuity must appear between A's data and B's"
        );

        for expected in [0xBAu8, 0xBB] {
            match cursor.poll() {
                Some(SampleCursorItem::Timed { sample, .. }) => {
                    assert_eq!(sample.data.as_ref(), &[expected; 4]);
                }
                other => panic!("expected B's sample 0x{expected:02X}, got {other:?}"),
            }
        }

        assert!(
            cursor.poll().is_none(),
            "no extra Discontinuity or data beyond what was published"
        );
    }

    /// Adversarial-review finding: a writer hand-off gates BOTH classes on
    /// its own per-class position, not just whichever class happens to
    /// reach it first — otherwise a `Sparse`-only publish from the new
    /// writer can overtake the old writer's still-undelivered `Timed`
    /// backlog and the `Discontinuity` itself, because the `Sparse` axis
    /// alone was already "caught up" (nothing was published to it before
    /// the hand-off, so its own threshold is trivially satisfied).
    ///
    /// MUTATION VERIFIED: reverting `SampleCursor::poll`'s `sparse_blocked`/
    /// `timed_blocked` gating (serve `Sparse` unconditionally once ready,
    /// the pre-fix shape) makes this test's final assertion fail: `seq` is
    /// `["A-timed", "B-sparse", "DISC"]` instead of the expected
    /// `["A-timed", "DISC", "B-sparse"]` — B's sparse item is delivered
    /// *before* the Discontinuity that should have preceded it. Recompiled
    /// and re-run to confirm the failure, then reverted.
    #[test]
    fn discontinuity_gates_both_classes_not_just_the_one_that_catches_up_first() {
        let trunk = Trunk::new(TrunkConfig::new(nz(10), nz(10), nz(4), nz(8), nz(8)));
        let mut cursor = trunk.subscribe();
        {
            let a = trunk.writer().expect("first take succeeds");
            a.publish(TrackId::new(1), RetentionClass::Timed, sample(0xAA, 4));
        }
        let b = trunk
            .writer()
            .expect("re-issue after the first writer was dropped must succeed");
        b.publish(TrackId::new(2), RetentionClass::Sparse, sample(0xBB, 4));

        let mut seq = Vec::new();
        while let Some(it) = cursor.poll() {
            seq.push(match it {
                SampleCursorItem::Timed { .. } => "A-timed",
                SampleCursorItem::Sparse { .. } => "B-sparse",
                SampleCursorItem::Discontinuity => "DISC",
                other => panic!("unexpected item: {other:?}"),
            });
        }
        assert_eq!(seq, vec!["A-timed", "DISC", "B-sparse"], "got {seq:?}");
    }

    /// The segment+parts ring group's counterpart: a non-pinning
    /// `SegmentCursor`, never resubscribed, keeps reading segments across a
    /// `SegmentWriter` drop + re-issue, in sequence order. (No
    /// `Discontinuity`-equivalent marker exists on this side — only the
    /// sample ring group tracks `WriterHandover`s — so this test's bar is
    /// "does not orphan", the same bar `Drop for SegmentWriter` exists for.)
    ///
    /// Verified to fail pre-fix the same way as the sample-ring test above:
    /// with `impl Drop for SegmentWriter` removed (scratch copy, restored
    /// after), `trunk.segment_writer()`'s second call returns `None`, so
    /// writer B can never be taken at all.
    #[test]
    fn segment_writer_handover_keeps_subscriber_reading_across_a_reconnect() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(10), nz(8), nz(8)));
        let mut cursor = trunk.subscribe_segments();

        {
            let writer_a = trunk.segment_writer().expect("first take succeeds");
            writer_a.publish_segment(segment_entry(0xAA, 1)).unwrap();
            writer_a.publish_segment(segment_entry(0xAB, 2)).unwrap();
        } // A dropped here.

        let writer_b = trunk
            .segment_writer()
            .expect("re-issue after the first segment writer was dropped must succeed");
        writer_b.publish_segment(segment_entry(0xBA, 3)).unwrap();
        writer_b.publish_segment(segment_entry(0xBB, 4)).unwrap();

        for expected_seq in 1..=4u32 {
            match cursor.poll() {
                Some(SegmentCursorItem::Segment(entry)) => {
                    assert_eq!(entry.sequence_number, SegmentSeq::new(expected_seq));
                }
                other => panic!("expected Segment({expected_seq}), got {other:?}"),
            }
        }
        assert!(cursor.poll().is_none());
    }

    /// Taking a second writer while the first is still alive must still be
    /// refused — the re-issue fix must not weaken the single-writer-at-a-time
    /// invariant. Not a "fails pre-fix" test (this behaviour never changed);
    /// a regression guard for the invariant the re-issue fix sits next to.
    #[test]
    fn writer_still_refused_while_the_first_is_alive() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let _writer_a = trunk.writer().expect("first take succeeds");
        assert!(
            trunk.writer().is_none(),
            "a second concurrent TrunkWriter must still be refused"
        );

        let _segment_writer_a = trunk.segment_writer().expect("first take succeeds");
        assert!(
            trunk.segment_writer().is_none(),
            "a second concurrent SegmentWriter must still be refused"
        );
    }

    /// More than `WRITER_HANDOVER_LOG_CAPACITY` hand-offs happen while a
    /// subscribed cursor never polls: the specific `WriterHandover` markers
    /// this cursor missed are gone (the backlog is bounded), but that must
    /// surface as `HandoversLagged`, not vanish silently.
    ///
    /// MUTATION VERIFIED (scratch copy, restored after): reverting
    /// `SampleCursor::poll`'s handover-backlog-overflow arm from
    /// `return Some(SampleCursorItem::HandoversLagged { skipped })` back to
    /// the pre-fix silent catch-up (`self.handovers_consumed =
    /// state.handovers.base;` with no `return`) makes this test's first
    /// assertion fail: `poll()` returns the first `Timed` sample directly,
    /// with no loss report at all, instead of the expected
    /// `HandoversLagged { skipped: 4 }`. Compiled, run, confirmed failing,
    /// then reverted.
    #[test]
    fn handover_backlog_overflow_reports_handovers_lagged_not_silence() {
        // Timed/Sparse rings sized so they never themselves evict anything
        // across this test — the only overflow exercised here is the
        // (much smaller, fixed) handover backlog.
        let trunk = Trunk::new(TrunkConfig::new(nz(1_000), nz(1_000), nz(4), nz(8), nz(8)));
        let mut cursor = trunk.subscribe();

        let total_takes = WRITER_HANDOVER_LOG_CAPACITY + 5;
        for _ in 0..total_takes {
            let w = trunk.writer().expect("reissue must keep succeeding");
            w.publish(TrackId::new(1), RetentionClass::Timed, sample(0x01, 4));
        } // each writer dropped at the end of its own iteration

        // The first take never records a `WriterHandover` (nothing to
        // report yet); every later re-issue does.
        let pushes = total_takes - 1;
        let expected_skipped = (pushes - WRITER_HANDOVER_LOG_CAPACITY) as u64;

        match cursor.poll() {
            Some(SampleCursorItem::HandoversLagged { skipped }) => {
                assert_eq!(skipped, expected_skipped);
            }
            other => {
                panic!("expected HandoversLagged {{ skipped: {expected_skipped} }}, got {other:?}")
            }
        }

        // The remaining, still-resident handovers (the newest
        // `WRITER_HANDOVER_LOG_CAPACITY` of them — only the oldest 4 were
        // evicted and folded into the `Lagged` above) are still
        // individually reportable, so further `Discontinuity` items are
        // expected as the cursor reaches each one's position; what matters
        // is that every published sample still arrives, and nothing panics
        // or loops forever.
        let mut timed_count = 0;
        while let Some(item) = cursor.poll() {
            if matches!(item, SampleCursorItem::Timed { .. }) {
                timed_count += 1;
            }
        }
        assert_eq!(
            timed_count, total_takes,
            "every published sample must still be delivered"
        );
    }

    // --- Adversarial-review round 2 findings (retained, monotonic --------
    // --- sequence numbers, positional boundaries) -------------------------

    /// `Trunk::part_bytes` must resolve a reused `(segment_number,
    /// part_index)` pair to the MOST RECENT entry, not a stale one from an
    /// earlier `SegmentWriter` round — `publish_part` itself places no
    /// monotonicity restriction on an in-progress segment's parts (only
    /// `publish_segment` rejects a non-monotonic `sequence_number`), so this
    /// is a real, currently-reachable case at the `Trunk` API level.
    ///
    /// MUTATION VERIFIED: reverting `part_bytes`'s `.iter().rev().find(..)`
    /// back to a plain front-to-back `.iter().find(..)` makes the final
    /// assertion fail: `part_bytes(SegmentSeq::new(1), PartIndex::new(0))` returns `b"old"` instead of the
    /// expected `b"new"`. Recompiled and re-run to confirm the failure,
    /// then reverted.
    #[test]
    fn part_bytes_reused_segment_number_returns_the_most_recent() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        {
            let w = trunk.segment_writer().unwrap();
            w.publish_part(PartEntry::new(
                Bytes::from_static(b"old"),
                SegmentSeq::new(1),
                PartIndex::new(0),
                Duration::from_millis(200),
                true,
            ));
        }
        let w2 = trunk.segment_writer().unwrap();
        w2.publish_part(PartEntry::new(
            Bytes::from_static(b"new"),
            SegmentSeq::new(1),
            PartIndex::new(0),
            Duration::from_millis(200),
            true,
        ));
        assert_eq!(
            trunk
                .part_bytes(SegmentSeq::new(1), PartIndex::new(0))
                .unwrap()
                .as_ref(),
            b"new"
        );
    }

    /// `Trunk::events_in_segment` for a currently-OPEN reused segment number
    /// must not be hidden by a stale, chronologically-earlier "next
    /// boundary" found by matching on `segment_number + 1` — segment 2's
    /// start from a previous round predates the reused segment 1's start,
    /// so looking it up by number inverts the range into one nothing can
    /// ever fall inside.
    ///
    /// MUTATION VERIFIED: reverting `events_in_segment`'s positional
    /// `segment_starts.get(chosen_idx + 1)` end-boundary lookup back to
    /// searching by number (`.rev().find(|(n, _)| *n == segment_number + 1)`)
    /// makes the assertion fail: `events_in_segment(SegmentSeq::new(1)).len()` is `0` instead
    /// of the expected `1` — the stale segment-2 boundary (start 90_000)
    /// inverts the range against the reused segment-1 start (200_000).
    /// Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn events_in_open_reused_segment_not_hidden_by_a_stale_next_boundary() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let writer = trunk.writer().unwrap();
        let sw = trunk.segment_writer().unwrap();
        sw.note_segment_start(SegmentSeq::new(1), MediaTime(0));
        sw.note_segment_start(SegmentSeq::new(2), MediaTime(90_000));
        sw.note_segment_start(SegmentSeq::new(1), MediaTime(200_000)); // reused, seg 2 not yet re-noted
        writer.publish_event(
            basic_event(7),
            EventAnchor::Segment {
                segment_number: SegmentSeq::new(1),
                delta: 1_500,
            },
        );
        assert_eq!(
            trunk.events_in_segment(SegmentSeq::new(1)).len(),
            1,
            "open reused segment 1 must contain the event"
        );
    }

    /// `SegmentWriter::try_publish_segment` must mark an
    /// `ArchiveOverrun::Terminate` pin terminated only when an eviction
    /// actually happens THIS call — not merely because the pin scan
    /// happened to consider it while a *different* pin (`StallIngest`)
    /// forced the whole call to return `Err` without evicting anything at
    /// all.
    ///
    /// MUTATION VERIFIED: reverting `try_publish_segment` to its pre-fix
    /// single-pass shape (scan pins once, apply `Terminate`'s side effect
    /// unconditionally, THEN check `must_wait` and possibly return `Err`)
    /// makes the final assertion fail: `term.poll()` returns
    /// `Some(SegmentCursorItem::Terminated)` instead of the expected
    /// `Some(SegmentCursorItem::Segment(_))` — the `Terminate` pin was
    /// marked terminated for an eviction of seq 2 that never actually
    /// happened (the call returned `Err` because of the `StallIngest` pin),
    /// so it can no longer read seq 1, which is still genuinely resident.
    /// Recompiled and re-run to confirm the failure, then reverted.
    #[test]
    fn try_publish_segment_does_not_terminate_a_pin_for_an_eviction_that_never_happened() {
        let trunk = Trunk::new(TrunkConfig::new(nz(10), nz(10), nz(1), nz(8), nz(8)));
        let w = trunk.segment_writer().unwrap();
        let _stall = trunk.pin_segments(ArchiveOverrun::StallIngest);
        let mut term = trunk.pin_segments(ArchiveOverrun::Terminate);
        w.try_publish_segment(segment_entry(1, 1))
            .expect("room for the first segment");
        assert!(matches!(
            w.try_publish_segment(segment_entry(2, 2)),
            Err(TryPublishSegmentError::WouldStall(_))
        ));
        // seq 1 is still resident; the Terminate pin must still read it —
        // it must not have been marked terminated for an eviction that,
        // because of the StallIngest pin, never actually happened.
        assert_eq!(trunk.segment_len(), 1);
        let it = term.poll();
        assert!(
            matches!(it, Some(SegmentCursorItem::Segment(_))),
            "got {it:?}"
        );
    }

    /// `SegmentWriter::expire_stalled_pins` lets a caller with its own
    /// non-blocking retry queue force the same give-up decision
    /// `publish_segment`'s bounded wait would eventually make, without ever
    /// blocking — the mechanism `multimux::ProgramSegmenter`'s bounded
    /// pending-segment queue needs (issue #1082's T7/T6 findings).
    #[test]
    fn expire_stalled_pins_terminates_on_demand_without_blocking() {
        let trunk = Trunk::new(TrunkConfig::new(nz(10), nz(10), nz(1), nz(8), nz(8)));
        let w = trunk.segment_writer().unwrap();
        let mut stall = trunk.pin_segments(ArchiveOverrun::StallIngest);

        // A no-op: nothing has even been published yet, so nothing can be
        // blocking eviction.
        assert_eq!(w.expire_stalled_pins(), 0);

        w.publish_segment(segment_entry(1, 1)).unwrap();

        // Fills capacity; a second entry would need to evict seq 1, which
        // the never-drained `stall` pin blocks — `try_publish_segment`
        // reports it non-blockingly rather than actually stalling.
        assert!(w.try_publish_segment(segment_entry(2, 2)).is_err());

        // Force the give-up decision on demand, without blocking:
        assert_eq!(
            w.expire_stalled_pins(),
            1,
            "exactly the one blocking StallIngest pin must be terminated"
        );
        assert!(
            matches!(stall.poll(), Some(SegmentCursorItem::Terminated)),
            "the pin's cursor must see Terminated, exactly ArchiveOverrun::Terminate's own signal"
        );

        // The retry now succeeds, without this thread ever having blocked.
        w.try_publish_segment(segment_entry(2, 2))
            .expect("must succeed once the blocking pin has been terminated");
        assert_eq!(trunk.last_closed_segment(), Some(SegmentSeq::new(2)));
    }

    /// [`Trunk::parts_in_segment`] must find a still-open segment's parts
    /// even when a *later* segment's opening part has already landed in the
    /// ring ahead of it — exactly what a live LL-HLS producer does the
    /// moment it starts buffering segment N+1's first part before segment
    /// N's own close has been queued (`multimux::source::segment`'s own
    /// stream-order publishing still allows this: only the *pending-queue*
    /// order is fixed, and a segment that is not itself stuck is queued
    /// (and drained) immediately, well before this method's caller ever
    /// looks).
    ///
    /// MUTATION VERIFIED: with the from-the-back `take_while` isolation this
    /// method used before (stopping at the first entry whose
    /// `segment_number != segment_number`), this test's
    /// `assert_eq!(trunk.parts_in_segment(SegmentSeq::new(1)).len(), 2, ...)` fails: actual
    /// `0` (the tail entry is segment 2's part, so the `take_while` stops
    /// immediately, before ever reaching either of segment 1's two parts) —
    /// segment 1's still-open parts vanish from the query entirely, which
    /// would silently drop its `#EXT-X-PART` tags from a served LL-HLS
    /// playlist. Recompiled with the `take_while` version restored, re-ran
    /// to confirm this exact failure, then reverted to the plain-filter
    /// fix.
    #[test]
    fn parts_in_segment_finds_open_segments_parts_even_after_a_later_segments_part_arrives() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(8), nz(8)));
        let w = trunk.segment_writer().unwrap();
        w.publish_part(PartEntry::new(
            Bytes::from_static(b"a"),
            SegmentSeq::new(1),
            PartIndex::new(0),
            Duration::from_millis(200),
            true,
        ));
        w.publish_part(PartEntry::new(
            Bytes::from_static(b"b"),
            SegmentSeq::new(1),
            PartIndex::new(1),
            Duration::from_millis(200),
            false,
        ));
        // Segment 2's opening part, already in the ring while segment 1 is
        // still open (not yet closed via `publish_segment`/
        // `try_publish_segment`) — the exact interleaving
        // `ProgramSegmenter::publish_in_stream_order` still permits, and
        // the shape this method must stay correct against.
        w.publish_part(PartEntry::new(
            Bytes::from_static(b"c"),
            SegmentSeq::new(2),
            PartIndex::new(0),
            Duration::from_millis(200),
            true,
        ));

        assert_eq!(
            trunk.parts_in_segment(SegmentSeq::new(1)).len(),
            2,
            "the still-open segment 1's own two parts must not vanish"
        );
        assert_eq!(
            trunk.parts_in_segment(SegmentSeq::new(2)).len(),
            1,
            "segment 2's own opening part must resolve independently"
        );
    }

    #[test]
    fn segment_start_time_returns_reported_start() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(4), nz(4)));
        let writer = trunk.segment_writer().unwrap();
        writer.note_segment_start(SegmentSeq::new(7), MediaTime(12345));
        assert_eq!(
            trunk.segment_start_time(SegmentSeq::new(7)),
            Some(MediaTime(12345))
        );
    }

    #[test]
    fn segment_start_time_unknown_seq_is_none() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(4), nz(4)));
        let writer = trunk.segment_writer().unwrap();
        writer.note_segment_start(SegmentSeq::new(7), MediaTime(12345));
        assert_eq!(trunk.segment_start_time(SegmentSeq::new(8)), None);
    }

    #[test]
    fn segment_start_time_evicted_returns_none() {
        // event_capacity == 2 => segment_starts capacity == 2.
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(2), nz(4)));
        let writer = trunk.segment_writer().unwrap();
        writer.note_segment_start(SegmentSeq::new(1), MediaTime(100));
        writer.note_segment_start(SegmentSeq::new(2), MediaTime(200));
        writer.note_segment_start(SegmentSeq::new(3), MediaTime(300));
        // seq 1 evicted from the bounded table.
        assert_eq!(trunk.segment_start_time(SegmentSeq::new(1)), None);
        assert_eq!(
            trunk.segment_start_time(SegmentSeq::new(2)),
            Some(MediaTime(200))
        );
        assert_eq!(
            trunk.segment_start_time(SegmentSeq::new(3)),
            Some(MediaTime(300))
        );
    }

    #[test]
    fn segment_start_time_two_segments_distinct() {
        let trunk = Trunk::new(TrunkConfig::new(nz(4), nz(4), nz(4), nz(4), nz(4)));
        let writer = trunk.segment_writer().unwrap();
        writer.note_segment_start(SegmentSeq::new(1), MediaTime(1000));
        writer.note_segment_start(SegmentSeq::new(2), MediaTime(2000));
        assert_eq!(
            trunk.segment_start_time(SegmentSeq::new(1)),
            Some(MediaTime(1000))
        );
        assert_eq!(
            trunk.segment_start_time(SegmentSeq::new(2)),
            Some(MediaTime(2000))
        );
    }
}