polyc-state-connect 2026.10.2

//! The shared tail loop that replaces one thread per caught-up subscription.
//!
//! Before this module existed, a caught-up subscription's pump slept and
//! re-asked [`JournalFeed::commits`] on its own dedicated thread for as long
//! as the stream ran — cheap for one subscription, and the shape of cost that
//! grew straight into the thread exhaustion [`crate::feed::executor`]
//! isolated on 2026-09-11 (POLY-324). Isolating that cost onto its own
//! executor stopped it from starving the other RPC families sharing this
//! listener's blocking pool; it did not stop the cost itself from growing
//! with the subscription count.
//!
//! This module replaces "one thread, permanently held, per caught-up
//! subscription" with one shared thread that polls every parked
//! subscription's source in a single batched [`JournalFeed::heads`] call per
//! tick, and hands a subscription back to [`FeedExecutor`] only when there is
//! a real chunk to build — the same bounded worker pool [`FeedExecutor`]
//! already was, used for a pump's actual work instead of for waiting on it.
//!
//! [`JournalFeed::commits`]: polyc_state::feed::JournalFeed::commits
//! [`JournalFeed::heads`]: polyc_state::feed::JournalFeed::heads

use std::{
    sync::{
        Arc, Mutex, PoisonError,
        atomic::{AtomicUsize, Ordering},
        mpsc,
    },
    thread::JoinHandle,
    time::Duration,
};

use polyc_state::{feed::JournalFeed, revision::JournalPosition};

use crate::feed::{
    executor::FeedExecutor,
    service::{ChunkSender, MAX_CONCURRENT_SUBSCRIPTIONS, Pump, StepOutcome, TAIL_POLL},
};

/// One parked subscription, waiting for its source's head to move.
struct Parked {
    pump: Pump,
    sender: ChunkSender,
    /// The feed position [`Pump::head_hint`] reported when this entry was
    /// last parked or ticked — not the durable head itself, just what this
    /// one pump has already confirmed. A tick redispatches the pump once
    /// [`JournalFeed::heads`] reports something past it.
    known: JournalPosition,
    /// How long this entry has sat unchanged across ticks.
    idle: Duration,
}

/// The parked-subscription list the tail loop and [`FeedExecutor`] workers
/// both touch.
///
/// Cloning shares the same underlying list: every clone parks into, and the
/// one tail loop thread drains from, the same set. [`Tailer`] hands one out
/// through [`Tailer::registry`] so [`drive`] can park into it from a worker
/// thread without holding a reference to the `Tailer` itself.
#[derive(Clone)]
pub(crate) struct Registry(Arc<Mutex<Vec<Parked>>>);

impl Registry {
    fn new() -> Self {
        Self(Arc::new(Mutex::new(Vec::new())))
    }

    /// Parks `pump`, caught up as of its own last confirmed position.
    ///
    /// Takes `self` by value: [`drive`] holds the one handle this call needs
    /// and has nothing left to do with it once the pump is parked.
    ///
    /// Called by [`drive`] once [`Pump::step`] reports
    /// [`StepOutcome::CaughtUp`] — never by the tail loop itself, which only
    /// ever removes an entry from this list or hands it to
    /// [`FeedExecutor::spawn_pump`].
    pub(crate) fn park(self, pump: Pump, sender: ChunkSender) {
        let known = pump.head_hint();
        self.0
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .push(Parked {
                pump,
                sender,
                known,
                idle: Duration::ZERO,
            });
    }

    /// Removes and returns every currently parked entry.
    fn drain(&self) -> Vec<Parked> {
        std::mem::take(&mut *self.0.lock().unwrap_or_else(PoisonError::into_inner))
    }

    /// Adds `entries` back, keeping whatever a concurrent [`Registry::park`]
    /// added since the matching [`Registry::drain`].
    fn restore(&self, entries: Vec<Parked>) {
        if entries.is_empty() {
            return;
        }
        self.0
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .extend(entries);
    }
}

/// Drives `pump` until it ends or catches up.
///
/// Runs on [`FeedExecutor`]'s blocking pool, dispatched both for a
/// subscription's first chunk (from
/// [`FeedSvc::subscribe_commits`](crate::feed::service::FeedSvc::subscribe_commits))
/// and for a parked subscription the tail loop found new work for. A
/// caught-up pump parks itself back into `registry` and returns, rather than
/// looping in place — the entire reason this occupies a worker thread only
/// while there is a real chunk to build and send.
pub(crate) fn drive(mut pump: Pump, mut sender: ChunkSender, registry: Registry) {
    loop {
        match pump.step(&mut sender) {
            StepOutcome::Sent => {}
            StepOutcome::CaughtUp => {
                registry.park(pump, sender);
                return;
            }
            StepOutcome::Ended => return,
        }
    }
}

/// The shared tail loop's dedicated thread, and the registry it drains.
///
/// One `Tailer` serves every `FeedSvc` mounted over the same feed module — a
/// feed module is commonly served on more than one router in production (see
/// [`FeedExecutor`]'s own doc for why), and parking a subscription with one
/// `Tailer` while a second, disjoint `Tailer` polled the same partition would
/// just be two threads doing the same durable read instead of one.
pub struct Tailer {
    registry: Registry,
    admitted: Arc<AtomicUsize>,
    budget: usize,
    stop: Option<mpsc::Sender<()>>,
    thread: Option<JoinHandle<()>>,
}

impl std::fmt::Debug for Tailer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Tailer").finish_non_exhaustive()
    }
}

impl Tailer {
    /// Starts the tail loop sized for [`MAX_CONCURRENT_SUBSCRIPTIONS`] —
    /// what every call site outside a proof of the budget itself should
    /// call.
    ///
    /// # Errors
    ///
    /// Returns an error if the tail loop's thread cannot be spawned.
    pub fn start_production(
        feed: Arc<dyn JournalFeed>,
        executor: Arc<FeedExecutor>,
    ) -> std::io::Result<Self> {
        Self::start(feed, executor, MAX_CONCURRENT_SUBSCRIPTIONS)
    }

    /// Starts the tail loop over `feed`, admitting at most `budget`
    /// concurrent subscriptions and dispatching a redispatched pump onto
    /// `executor`.
    ///
    /// Takes the budget explicitly, rather than always reading
    /// [`MAX_CONCURRENT_SUBSCRIPTIONS`], so a proof of the admission gate
    /// itself exercises a refusal after admitting a handful of subscriptions
    /// instead of the full production budget.
    ///
    /// # Errors
    ///
    /// Returns an error if the tail loop's thread cannot be spawned.
    pub fn start(
        feed: Arc<dyn JournalFeed>,
        executor: Arc<FeedExecutor>,
        budget: usize,
    ) -> std::io::Result<Self> {
        let registry = Registry::new();
        let (stop, stopped) = mpsc::channel();
        let loop_registry = registry.clone();
        let thread = std::thread::Builder::new()
            .name("polychrome-feed-tailer".to_owned())
            .spawn(move || run(&feed, &loop_registry, &executor, &stopped))?;
        Ok(Self {
            registry,
            admitted: Arc::new(AtomicUsize::new(0)),
            budget,
            stop: Some(stop),
            thread: Some(thread),
        })
    }

    /// Returns the registry a worker thread parks a caught-up pump into.
    ///
    /// [`FeedSvc::subscribe_commits`](crate::feed::service::FeedSvc::subscribe_commits)
    /// clones this once per subscription and hands it to [`drive`]; nothing
    /// else needs a handle on the registry directly.
    pub(crate) fn registry(&self) -> Registry {
        self.registry.clone()
    }

    /// Tries to claim one of this tail loop's admission slots.
    /// [`None`] means the listener is already at its subscription budget —
    /// the caller must refuse the call rather than open its stream.
    ///
    /// The returned permit holds the slot until dropped; a [`Pump`] holds one
    /// for as long as its subscription is admitted, so the slot frees itself
    /// the moment [`drive`] or a tail-loop tick finally drops that `Pump` —
    /// whichever path ends it.
    #[must_use]
    pub(crate) fn try_admit(&self) -> Option<SubscriptionPermit> {
        // A compare-and-swap loop rather than a plain `fetch_add` followed by
        // a check: the latter can transiently over-admit, incrementing the
        // counter past `budget` before the offending caller notices and
        // backs it out, which is exactly the race a budget exists to close.
        self.admitted
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| {
                (count < self.budget).then_some(count + 1)
            })
            .ok()
            .map(|_| SubscriptionPermit(Arc::clone(&self.admitted)))
    }
}

/// One admitted subscription's claim on [`Tailer::try_admit`]'s budget.
///
/// Released automatically when dropped — whichever path ends the
/// subscription (the stream drains, pauses, is shed, or its consumer simply
/// disconnects), the slot it held returns to the budget the moment the
/// [`Pump`] holding this permit is dropped. A manual decrement at every
/// removal site is exactly the kind of bookkeeping a missed code path
/// silently leaks; this closes over the counter instead so `Drop` cannot
/// forget.
pub(crate) struct SubscriptionPermit(Arc<AtomicUsize>);

impl Drop for SubscriptionPermit {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::AcqRel);
    }
}

#[cfg(test)]
impl SubscriptionPermit {
    /// A permit for a case that drives a [`Pump`] directly, off the
    /// admission path [`Tailer::try_admit`] covers.
    pub(crate) fn for_test() -> Self {
        Self(Arc::new(AtomicUsize::new(1)))
    }
}

impl Drop for Tailer {
    fn drop(&mut self) {
        // Dropping the stop sender alone would end the loop on its next
        // `recv_timeout` wake; sending first wakes it immediately instead of
        // waiting out the last `TAIL_POLL`. Same shape as
        // `crate::journal::bridge::Bridge`'s own drop, adapted for a
        // polling loop instead of a blocking `recv`.
        if let Some(stop) = self.stop.take() {
            let _ = stop.send(());
        }
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }
    }
}

/// How many sources one [`JournalFeed::heads`] call names at once.
///
/// A tick with more parked sources than this asks in more than one batch,
/// each its own module lock acquisition. Never one call for every parked
/// source: `heads` resolves a source by loading its partition if it is not
/// already resident, and a batch wider than the durable journal's own
/// resident-partition cache can evict and reload partitions it will need
/// again later in the same batch — holding the module's mutation lock, the
/// same one a real commit needs, for as long as that churn takes. Bounding
/// the batch bounds that hold to a fixed number of loads, whatever the
/// parked count grows to (POLY-331).
const HEADS_BATCH: usize = 128;

/// Runs one tick: drops closed and draining entries without a durable read,
/// then asks [`JournalFeed::heads`] for every source still parked, in
/// batches of at most [`HEADS_BATCH`].
///
/// `elapsed` is the real time since the previous tick started — not the
/// nominal [`TAIL_POLL`] — so a tick slowed by its own batched reads still
/// accounts idle time honestly instead of undercounting it.
fn tick(
    feed: &Arc<dyn JournalFeed>,
    registry: &Registry,
    executor: &Arc<FeedExecutor>,
    elapsed: Duration,
) {
    let parked = registry.drain();
    if parked.is_empty() {
        return;
    }

    // Closed and draining entries cost nothing durable to resolve, so they
    // are removed before the batched reads below rather than folded into
    // them.
    let mut live = Vec::with_capacity(parked.len());
    for mut entry in parked {
        if entry.sender.is_closed() {
            continue;
        }
        if entry.pump.is_draining() {
            entry.pump.drain_now(&mut entry.sender);
            continue;
        }
        live.push(entry);
    }
    if live.is_empty() {
        return;
    }

    let mut still_parked = Vec::with_capacity(live.len());
    while !live.is_empty() {
        let batch_len = live.len().min(HEADS_BATCH);
        let batch: Vec<Parked> = live.drain(..batch_len).collect();

        let sources: Vec<_> = batch
            .iter()
            .map(|entry| entry.pump.source().clone())
            .collect();
        let heads = feed.heads(&sources);
        debug_assert_eq!(
            heads.len(),
            batch.len(),
            "JournalFeed::heads must answer one entry per source asked about, in the same order"
        );

        for (entry, (_, head)) in batch.into_iter().zip(heads) {
            let Parked {
                pump,
                mut sender,
                known,
                idle,
            } = entry;
            // A source that failed to resolve (a compacted or
            // destroyed-and-recreated incarnation, for example) is not this
            // loop's call to make: dispatch the pump so `Pump::step` reports
            // the same typed refusal on the stream it would have without
            // this fast path.
            let moved = head != Ok(known);
            if moved {
                let registry = registry.clone();
                executor.spawn_pump(move || drive(pump, sender, registry));
                continue;
            }
            let idle = idle.saturating_add(elapsed);
            if idle >= pump.max_tail_idle() {
                pump.pause_now(&mut sender);
                continue;
            }
            still_parked.push(Parked {
                pump,
                sender,
                known,
                idle,
            });
        }
    }
    registry.restore(still_parked);
}

/// The tail loop's body: wait one [`TAIL_POLL`] interval, tick, repeat, until
/// told to stop.
fn run(
    feed: &Arc<dyn JournalFeed>,
    registry: &Registry,
    executor: &Arc<FeedExecutor>,
    stopped: &mpsc::Receiver<()>,
) {
    let mut previous = std::time::Instant::now();
    loop {
        match stopped.recv_timeout(TAIL_POLL) {
            Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => return,
            Err(mpsc::RecvTimeoutError::Timeout) => {}
        }
        let now = std::time::Instant::now();
        let elapsed = now.duration_since(previous);
        previous = now;
        tick(feed, registry, executor, elapsed);
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use polyc_state::memory::MemoryFeed;

    use super::{FeedExecutor, Tailer};

    fn tailer_over(budget: usize) -> Tailer {
        let feed = Arc::new(MemoryFeed::new());
        let executor = Arc::new(FeedExecutor::start(2).expect("start the executor"));
        Tailer::start(feed, executor, budget).expect("start the tailer")
    }

    #[test]
    fn admits_up_to_the_budget_then_refuses() {
        let tailer = tailer_over(2);
        let first = tailer.try_admit();
        assert!(first.is_some());
        let second = tailer.try_admit();
        assert!(second.is_some());
        assert!(
            tailer.try_admit().is_none(),
            "a third subscription past a budget of two must be refused"
        );
        drop(first);
        drop(second);
    }

    #[test]
    fn dropping_a_permit_frees_its_slot() {
        let tailer = tailer_over(1);
        let permit = tailer.try_admit();
        assert!(permit.is_some());
        assert!(
            tailer.try_admit().is_none(),
            "already at the one-subscription budget"
        );
        drop(permit);
        assert!(
            tailer.try_admit().is_some(),
            "the freed slot must be admittable again"
        );
    }
}

/// Proofs of the tail loop's batching behavior itself — every case here
/// drives [`tick`] and [`drive`] directly against a synthetic feed, never
/// through [`crate::feed::service::FeedSvc::subscribe_commits`] and a real
/// listener, because what each proves is a property of the pump and the
/// registry it parks in, not of the transport in front of them.
#[cfg(test)]
mod proofs {
    use std::{
        sync::{
            Arc, Mutex as StdMutex,
            atomic::{AtomicBool, AtomicUsize, Ordering},
        },
        time::Duration,
    };

    use polyc_state::{
        context::CallContext,
        error::StateError,
        feed::{
            AcknowledgeProjectorCursor, CompactFeedPrefix, CreateSnapshot, FeedChunk,
            FeedCompaction, FeedCursor, FeedRetention, FeedSnapshot, GetFeedRetention,
            GetProjectorStatus, JournalFeed, ListProjectors, ProjectorListing, ProjectorStatus,
            RegisterProjector, SubscribeCommits,
        },
        id::PartitionId,
        receipt::Receipt,
        revision::{JournalPosition, JournalSource, PartitionIncarnation},
        stream::StreamEnd,
    };

    use super::{FeedExecutor, Registry, TAIL_POLL, tick};
    use crate::feed::service::{FeedStreamTuning, Pump, PumpedChunk};

    /// A source distinct from every other one a proof asks for.
    fn source(n: usize) -> JournalSource {
        JournalSource::new(
            PartitionId::new(format!("tailer-proof-{n}")),
            PartitionIncarnation::from_bytes([1; PartitionIncarnation::LEN]),
        )
    }

    /// A `JournalFeed` that answers only `heads` and `commits`, counting
    /// every call to each. Every other method panics: nothing the tail loop
    /// drives ever reaches them.
    struct FakeFeed {
        heads_calls: AtomicUsize,
        commits_calls: AtomicUsize,
        /// The position `heads` reports for source `n`, by index.
        positions: StdMutex<Vec<JournalPosition>>,
        /// How long `commits` sleeps before answering source `n`, by index —
        /// [`Duration::ZERO`] for every source a proof does not name.
        commit_delay: StdMutex<Vec<Duration>>,
        /// Whether `commits` has finished answering source `n`, by index —
        /// what a proof polls instead of a chunk's content, since a chunk
        /// with no records never reaches a pump's channel at all.
        completed: StdMutex<Vec<bool>>,
        /// Every source's completion state, captured atomically — in the
        /// same critical section that marks it complete, before its own
        /// `commits` call even returns — the instant the deliberately
        /// delayed source (the one [`Self::delay`] named) finishes.
        ///
        /// A proof reading this instead of polling [`Self::completed`]
        /// after the fact closes a real race: polling for the delayed
        /// source's own flag and then separately reading the others has a
        /// window, however small, in which sources dispatched AFTER the
        /// delayed one — under the very mutation this proof exists to
        /// catch, which runs them serialized on the same thread right
        /// behind it — can complete before the read happens, especially
        /// once a retry gives them a second attempt. This snapshot has no
        /// such window: nothing dispatched behind the delayed source can
        /// possibly be recorded here, concurrent or not, because it is
        /// taken from inside the delayed source's own `commits` call,
        /// before that call — and therefore anything serialized behind it —
        /// has returned.
        snapshot_at_delay: StdMutex<Option<Vec<bool>>>,
    }

    impl FakeFeed {
        fn new(sources: usize) -> Self {
            Self {
                heads_calls: AtomicUsize::new(0),
                commits_calls: AtomicUsize::new(0),
                positions: StdMutex::new(vec![JournalPosition::ORIGIN; sources]),
                commit_delay: StdMutex::new(vec![Duration::ZERO; sources]),
                completed: StdMutex::new(vec![false; sources]),
                snapshot_at_delay: StdMutex::new(None),
            }
        }

        /// Waits up to `timeout` for [`Self::delay`]'s source to finish, and
        /// returns the snapshot its own `commits` call captured at that
        /// instant.
        ///
        /// Once written, the snapshot never changes — polling for it to
        /// exist carries none of the race polling for a live flag would,
        /// because what is read back is frozen the moment it was taken.
        fn snapshot_at_delay_within(&self, timeout: Duration) -> Vec<bool> {
            let deadline = std::time::Instant::now() + timeout;
            loop {
                let snapshot = self.snapshot_at_delay.lock().unwrap().clone();
                if let Some(snapshot) = snapshot {
                    return snapshot;
                }
                assert!(
                    std::time::Instant::now() < deadline,
                    "the deliberately delayed source never completed within the hang-guard"
                );
                std::thread::sleep(Duration::from_millis(2));
            }
        }

        /// Reports `position` as source `n`'s head from the next `heads`
        /// call onward.
        fn advance(&self, n: usize, position: JournalPosition) {
            self.positions.lock().unwrap()[n] = position;
        }

        /// Makes `commits` sleep `delay` before answering source `n`, and
        /// marks `n` as the source [`Self::snapshot_at_delay_within`]
        /// triggers on.
        fn delay(&self, n: usize, delay: Duration) {
            self.commit_delay.lock().unwrap()[n] = delay;
        }

        fn index_of(source: &JournalSource) -> usize {
            source
                .partition()
                .as_str()
                .strip_prefix("tailer-proof-")
                .expect("every source this fake serves is one it minted")
                .parse()
                .expect("the index is the whole suffix")
        }
    }

    impl JournalFeed for FakeFeed {
        fn create_snapshot(
            &self,
            _command: CreateSnapshot,
            _context: &CallContext,
        ) -> Result<FeedSnapshot, StateError> {
            unimplemented!("not exercised by a tail-loop proof")
        }

        fn commits(
            &self,
            request: SubscribeCommits,
            _context: &CallContext,
        ) -> Result<FeedChunk, StateError> {
            self.commits_calls.fetch_add(1, Ordering::SeqCst);
            let index = Self::index_of(request.source());
            let delay = self.commit_delay.lock().unwrap()[index];
            if delay > Duration::ZERO {
                std::thread::sleep(delay);
            }
            let mut completed = self.completed.lock().unwrap();
            completed[index] = true;
            if delay > Duration::ZERO {
                // Still holding `completed`'s lock: nothing dispatched
                // behind this call — concurrently or, under the mutation
                // this proof exists to catch, serialized right behind it on
                // the same thread — can be recorded complete before this
                // clone is taken.
                *self.snapshot_at_delay.lock().unwrap() = Some(completed.clone());
            }
            drop(completed);
            Ok(FeedChunk::new(
                Vec::new(),
                FeedCursor::origin(request.source().clone()),
                StreamEnd::More,
            ))
        }

        fn heads(
            &self,
            sources: &[JournalSource],
        ) -> Vec<(JournalSource, Result<JournalPosition, StateError>)> {
            self.heads_calls.fetch_add(1, Ordering::SeqCst);
            let positions = self.positions.lock().unwrap();
            sources
                .iter()
                .map(|source| {
                    let index = Self::index_of(source);
                    (source.clone(), Ok(positions[index]))
                })
                .collect()
        }

        fn register_projector(
            &self,
            _command: RegisterProjector,
            _context: &CallContext,
        ) -> Result<Receipt, StateError> {
            unimplemented!("not exercised by a tail-loop proof")
        }

        fn acknowledge(
            &self,
            _command: AcknowledgeProjectorCursor,
            _context: &CallContext,
        ) -> Result<Receipt, StateError> {
            unimplemented!("not exercised by a tail-loop proof")
        }

        fn projector_status(
            &self,
            _request: GetProjectorStatus,
            _context: &CallContext,
        ) -> Result<Option<ProjectorStatus>, StateError> {
            unimplemented!("not exercised by a tail-loop proof")
        }

        fn projectors(
            &self,
            _request: ListProjectors,
            _context: &CallContext,
        ) -> Result<ProjectorListing, StateError> {
            unimplemented!("not exercised by a tail-loop proof")
        }

        fn retention(
            &self,
            _request: GetFeedRetention,
            _context: &CallContext,
        ) -> Result<FeedRetention, StateError> {
            unimplemented!("not exercised by a tail-loop proof")
        }

        fn compact(
            &self,
            _command: CompactFeedPrefix,
            _context: &CallContext,
        ) -> Result<FeedCompaction, StateError> {
            unimplemented!("not exercised by a tail-loop proof")
        }
    }

    /// Parks a fresh pump over `feed` for `source`, sharing `draining`.
    fn park(
        registry: &Registry,
        feed: &Arc<dyn JournalFeed>,
        draining: &Arc<AtomicBool>,
        n: usize,
    ) -> futures::channel::mpsc::Receiver<PumpedChunk> {
        let (sender, receiver) = futures::channel::mpsc::channel(8);
        let pump = Pump::for_proof(
            Arc::clone(feed),
            Arc::clone(draining),
            source(n),
            FeedStreamTuning::new(),
        );
        registry.clone().park(pump, sender);
        receiver
    }

    /// Polls `receiver` for up to `timeout`, without an executor: these
    /// proofs run on plain threads, not inside a Tokio runtime.
    fn recv_within(
        receiver: &mut futures::channel::mpsc::Receiver<PumpedChunk>,
        timeout: Duration,
    ) -> bool {
        let deadline = std::time::Instant::now() + timeout;
        loop {
            if receiver.try_recv().is_ok() {
                return true;
            }
            if std::time::Instant::now() >= deadline {
                return false;
            }
            std::thread::sleep(Duration::from_millis(2));
        }
    }

    /// Proof 3 (FS-1): a caught-up partition costs nothing durable. With
    /// 1,000 subscriptions parked on sources whose head never moves,
    /// `commits` is never called, and `heads` is called a fixed, bounded
    /// number of times per tick — [`HEADS_BATCH`]-sized batches, never once
    /// per parked subscription per tick.
    ///
    /// Mutation: call `commits` for every parked subscription instead of
    /// batching through `heads` — caught, because `commits_calls` would
    /// then grow instead of staying at zero.
    #[test]
    fn a_caught_up_partition_costs_no_durable_read() {
        const PARKED: usize = 1_000;
        const TICKS: usize = 20;

        let feed = Arc::new(FakeFeed::new(PARKED));
        let dyn_feed: Arc<dyn JournalFeed> = feed.clone();
        let executor = Arc::new(FeedExecutor::start(4).expect("start the executor"));
        let registry = Registry::new();
        let draining = Arc::new(AtomicBool::new(false));
        let _receivers: Vec<_> = (0..PARKED)
            .map(|n| park(&registry, &dyn_feed, &draining, n))
            .collect();

        for _ in 0..TICKS {
            tick(&dyn_feed, &registry, &executor, TAIL_POLL);
        }

        assert_eq!(
            feed.commits_calls.load(Ordering::SeqCst),
            0,
            "nothing moved; commits must never run"
        );
        assert_eq!(
            feed.heads_calls.load(Ordering::SeqCst),
            TICKS * PARKED.div_ceil(super::HEADS_BATCH),
            "heads() batches by HEADS_BATCH, not once per parked subscription and not once for \
             all 1,000 at once"
        );
    }

    /// Proof 4 (FS-1): one commit among 1,000 parked reaches only that
    /// source's pump, on the tick after `heads` reports it moved, and
    /// touches no other parked subscription.
    ///
    /// Mutation: build a chunk for every parked source instead of only the
    /// one whose head moved — caught, because every other source's
    /// `commits_calls` share would then be nonzero too.
    #[test]
    fn only_the_moved_source_is_dispatched() {
        const PARKED: usize = 1_000;
        const MOVED: usize = 507;

        let feed = Arc::new(FakeFeed::new(PARKED));
        let dyn_feed: Arc<dyn JournalFeed> = feed.clone();
        let executor = Arc::new(FeedExecutor::start(4).expect("start the executor"));
        let registry = Registry::new();
        let draining = Arc::new(AtomicBool::new(false));
        let _receivers: Vec<_> = (0..PARKED)
            .map(|n| park(&registry, &dyn_feed, &draining, n))
            .collect();

        tick(&dyn_feed, &registry, &executor, TAIL_POLL);
        assert_eq!(
            feed.commits_calls.load(Ordering::SeqCst),
            0,
            "nothing moved on the first tick"
        );

        feed.advance(MOVED, JournalPosition::new(1));
        tick(&dyn_feed, &registry, &executor, TAIL_POLL);

        let deadline = std::time::Instant::now() + Duration::from_secs(2);
        while feed.commits_calls.load(Ordering::SeqCst) == 0 && std::time::Instant::now() < deadline
        {
            std::thread::sleep(Duration::from_millis(2));
        }

        assert_eq!(
            feed.commits_calls.load(Ordering::SeqCst),
            1,
            "only the one moved source's pump is dispatched"
        );
    }

    /// Proof 6 (FS-1): a drain ends every parked subscription's stream
    /// within one tick, without a durable read.
    ///
    /// Mutation: skip parked entries on drain — caught, because the
    /// registry would still hold every entry after the tick instead of
    /// being empty, and no receiver would see a marker.
    #[test]
    fn a_drain_ends_every_parked_stream_in_one_tick() {
        const PARKED: usize = 1_000;

        let feed = Arc::new(FakeFeed::new(PARKED));
        let dyn_feed: Arc<dyn JournalFeed> = feed.clone();
        let executor = Arc::new(FeedExecutor::start(4).expect("start the executor"));
        let registry = Registry::new();
        let draining = Arc::new(AtomicBool::new(false));
        let receivers: Vec<_> = (0..PARKED)
            .map(|n| park(&registry, &dyn_feed, &draining, n))
            .collect();

        draining.store(true, Ordering::SeqCst);
        tick(&dyn_feed, &registry, &executor, TAIL_POLL);

        assert_eq!(
            registry.drain().len(),
            0,
            "every parked entry left the registry in one tick"
        );
        assert_eq!(
            feed.commits_calls.load(Ordering::SeqCst),
            0,
            "a drain costs no durable read"
        );
        for mut receiver in receivers {
            assert!(
                recv_within(&mut receiver, Duration::from_secs(1)),
                "every parked subscription receives its drain marker"
            );
        }
    }

    /// Proof 7 (FS-1): one source's slow `commits` call delays no other
    /// source's chunk.
    ///
    /// Mutation: serve chunks inline on the tail loop's own thread instead
    /// of dispatching to the executor pool — caught, because the fast
    /// sources would then wait behind the slow one's real sleep instead of
    /// running concurrently on their own worker threads.
    ///
    /// `SLOW_DELAY`/`SAFETY_TIMEOUT` (POLY-343, following POLY-342): this
    /// proof used to assert an ABSOLUTE bound — every fast source completes
    /// within some fixed low-millisecond window while the slow one sleeps.
    /// POLY-342 already widened that bound once ("two orders of magnitude
    /// of scheduling slack") and it still flaked again (POLY-343).
    ///
    /// Root-caused further: the residual flake reproduces even in isolation
    /// on an otherwise-idle machine, at any margin tried (500ms through
    /// 4,000ms) and at any executor pool size tried (12 through 64
    /// threads), so it is not primarily CI co-scheduling or a pool sized
    /// too small. It is real OS thread-scheduling latency: dispatching ten
    /// blocking closures to [`FeedExecutor`]'s pool in a tight loop
    /// occasionally leaves one of them waiting an unpredictable, unbounded
    /// tail before an idle worker thread actually picks it up and runs
    /// it — the same class of scheduling-latency tail this crate's own
    /// `.config/nextest.toml` already documents for suites that "race the
    /// OS scheduler". No fixed real-time bound can be unconditionally
    /// reliable against a tail with no proven upper limit.
    ///
    /// A first attempt replaced the absolute bound with a RELATIVE one —
    /// poll for the slow source's own flag, then read the fast sources'
    /// flags without waiting — reasoning that a source dispatched after the
    /// slow one cannot even start until the slow one's blocking call
    /// returns. That reasoning holds, but the polling itself reopened the
    /// same shape of race one level down: between the poll noticing the
    /// slow flag and this test reading the others, a few sub-millisecond
    /// sequential calls — exactly what the mutation this proof exists to
    /// catch produces, immediately after the slow one finally unblocks —
    /// can complete and be misread as concurrent. Caught only by testing
    /// the mutation directly: it passed on a retry roughly one time in
    /// three, which would have made `.config/nextest.toml`'s retry (below)
    /// actively mask the very regression this proof exists to catch.
    ///
    /// This test's `FakeFeed` closes that window instead of narrowing it:
    /// the slow source's own `commits` call, while still
    /// holding the lock that marks it complete, atomically clones every
    /// source's completion state into a snapshot — before that call, and
    /// therefore anything a serializing mutation runs behind it, has even
    /// returned. There is no read anywhere that can observe a fast source
    /// as complete before the slow one if it was not already complete at
    /// that instant. Verified directly: with the mutation applied (drive
    /// each pump inline instead of dispatching it — see the top of this
    /// doc) and the retry override below active, 20 consecutive
    /// `cargo nextest run` invocations each exhausted every attempt the
    /// retry allows and still reported the test failed overall — zero
    /// exceptions, and zero passing attempts within any of them.
    ///
    /// `.config/nextest.toml` still retries this exact test, the same
    /// remedy that file already applies to every other suite that races the
    /// OS scheduler: the snapshot removes the assertion's OWN race, but the
    /// dispatch latency it measures against is still real OS scheduling
    /// time with no proven upper bound, and a hang-guard timeout under
    /// truly pathological CI contention is a scheduling race, not a
    /// correctness bug.
    #[test]
    fn a_slow_source_does_not_delay_the_others() {
        const PARKED: usize = 10;
        const SLOW: usize = 3;
        const SLOW_DELAY: Duration = Duration::from_millis(2_000);
        /// Bounds only how long this test itself waits for the slow source
        /// to finish its own deliberate delay — a hang-guard for a broken
        /// build, not a claim about how fast anything must run.
        const SAFETY_TIMEOUT: Duration = Duration::from_secs(10);

        let feed = Arc::new(FakeFeed::new(PARKED));
        let dyn_feed: Arc<dyn JournalFeed> = feed.clone();
        let executor = Arc::new(FeedExecutor::start(PARKED + 2).expect("start the executor"));
        let registry = Registry::new();
        let draining = Arc::new(AtomicBool::new(false));
        let _receivers: Vec<_> = (0..PARKED)
            .map(|n| {
                feed.advance(n, JournalPosition::new(1));
                park(&registry, &dyn_feed, &draining, n)
            })
            .collect();
        feed.delay(SLOW, SLOW_DELAY);

        // Off its own thread, exactly as the real tail loop calls it from
        // `run` — a proof that checked completion only after this call
        // returned would still pass under a mutation that ran every
        // dispatched pump inline on the calling thread, because by the time
        // a blocking call returns every pump it drove has already finished
        // too. Checking while this may still be running is what separates
        // "dispatched to run concurrently" from "run one after another
        // before this returns."
        std::thread::spawn({
            let dyn_feed = Arc::clone(&dyn_feed);
            move || tick(&dyn_feed, &registry, &executor, TAIL_POLL)
        });

        let snapshot = feed.snapshot_at_delay_within(SAFETY_TIMEOUT);
        for (n, completed) in snapshot.iter().enumerate().take(PARKED) {
            if n == SLOW {
                continue;
            }
            assert!(
                *completed,
                "source {n} had not completed by the instant the slow source's own \
                 {SLOW_DELAY:?} delay did, so it must have been queued behind it \
                 instead of running concurrently"
            );
        }
    }
}