polyc-query 2026.9.0

//! The one linearization point between a producer report and a withdrawal.
//!
//! Two independent events end a projected query. The producer measures a
//! terminal. The consumer withdraws by dropping its stream. Both must resolve
//! into exactly one durable completion, and the choice must not depend on
//! which task the scheduler runs first.
//!
//! A third event ends some queries: the consumer stops at a bound the caller
//! asked for, having received exactly what it reports. That is neither a
//! producer report nor a withdrawal, so it has its own transition.
//!
//! Every transition happens under this latch's lock. `Pending` moves to
//! `Reported`, `ConsumerBound`, or `Cancelled`. From there the only moves are
//! narrow ones: a consumer bound may replace a `Reported` success, since the
//! caller received what the bound reports and not what the producer made; and
//! a measured failure may replace a withdrawal or a consumer bound, since a
//! query that failed did not succeed for anyone. Exactly one later move to
//! `Dispatched` wins, and after it the permit has become a completion command,
//! so nothing can change its content.
//!
//! Those rules read the same in both arrival orders, so a race between the
//! producer's terminal and the consumer's settles the same way whichever task
//! the scheduler ran first. Two pairs are deliberately not symmetric, and
//! neither is a race: a producer terminal never replaces an earlier producer
//! terminal, because the first one measured the query; and a `Cancelled`
//! report never replaces a consumer bound, because that report is the echo of
//! the drop the bound itself performed.
//!
//! Delivered counts live under the same lock, and that is the point. A frame
//! reaching the caller and a terminal being chosen are the two events that
//! must not interleave: a count taken outside the lock can be added after a
//! terminal was built from it, and the caller then holds a row the durable
//! record does not know about. So releasing is itself a transition —
//! `admit_release` takes the lock, and a release is admitted only while the
//! query is still `Pending`. Once any terminal has been chosen, a release is
//! refused and the frame never reaches the caller.
//!
//! Every terminal is built from the counts under that same lock, in the same
//! critical section that records it. Nothing snapshots the counts and reports
//! a terminal built from them later. Which event feeds the counts still
//! depends on who is counting — the result stream as each batch leaves it, or
//! a consumer that re-frames those batches as each frame leaves, see
//! `account_at_consumer` — and exactly one of the two is ever live.
//!
//! A panic under the lock leaves the transition untrustworthy, and the sticky
//! `poisoned` flag makes `dispatch` fail closed on it. The counts are read
//! back from the recovered guard: each update computes its new value before
//! assigning it, so no update can be half applied, and what already left the
//! stream is still reported exactly.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, MutexGuard, PoisonError};
use std::time::Duration;

use polyc_state::query_audit::{
    ErrorClass, QueryCompletion, QueryOutcome, RowCount, SourceSnapshot, Truncation,
};

/// What the consumer actually received.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct DeliveredCounts {
    rows: u64,
    result_bytes: u64,
    truncated: bool,
}

impl DeliveredCounts {
    /// Returns the rows the consumer received.
    pub(crate) const fn rows(self) -> u64 {
        self.rows
    }

    /// Returns the result bytes the consumer received.
    ///
    /// The unit follows whichever accounting is live: decoded batch memory
    /// while the result stream counts, encoded frame bytes once a consumer
    /// re-frames. The terminal frame counts the bytes it released itself, so
    /// this is the latch's own view and only its cases read it.
    #[cfg(test)]
    pub(crate) const fn result_bytes(self) -> u64 {
        self.result_bytes
    }

    /// Returns whether release stopped at a ceiling with rows still available.
    pub(crate) const fn truncated(self) -> bool {
        self.truncated
    }

    pub(crate) const fn truncation(self) -> Truncation {
        if self.truncated {
            Truncation::TruncatedAt(self.rows)
        } else {
            Truncation::Complete
        }
    }
}

#[derive(Default)]
enum LatchState {
    /// Neither side has ended the query.
    #[default]
    Pending,
    /// The producer measured this exact terminal.
    Reported(Box<QueryCompletion>),
    /// The consumer stopped at a bound it asked for, and measured this exact
    /// terminal itself.
    ///
    /// Distinct from both neighbours. It is not a producer report, because the
    /// party that ended the query is the one describing it. It is not a
    /// withdrawal, because nothing was abandoned: the caller received exactly
    /// what this completion says, and asked for no more. The withdrawal that
    /// follows — the consumer drops the stream to stop the producer — must
    /// therefore not relabel it.
    ConsumerBound(Box<QueryCompletion>),
    /// The consumer withdrew before a terminal was dispatched.
    Cancelled,
    /// The permit became a completion command. The content is now immutable.
    Dispatched,
}

/// Serializes the producer report, the consumer withdrawal, and the dispatch.
///
/// Every method is synchronous and releases the lock before returning, so no
/// caller can await while holding it.
pub(crate) struct TerminalLatch {
    /// The transition and the counts every terminal is built from.
    ///
    /// One lock, because they are one decision. A release admitted here is in
    /// the record; a release refused here never reaches the caller.
    settlement: Mutex<Settlement>,
    /// Sticky: a panic under the transition lock may have torn the state.
    poisoned: AtomicBool,
}

/// What became of a whole batch offered to the record.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum BatchRelease {
    /// Counted, and part of every terminal built afterwards.
    Admitted,
    /// Not counted here: a consumer re-frames these batches and admits the
    /// frames it releases instead.
    CountedByConsumer,
    /// Refused. A terminal has been chosen and this batch is not in it, so it
    /// must not reach the caller either.
    Refused,
}

/// Admits one release into counts the caller already holds the lock on.
///
/// Each field takes a value computed before the assignment, so a panic cannot
/// leave one updated and the other not.
const fn admit_locked(settlement: &mut Settlement, rows: u64, result_bytes: u64) -> bool {
    if !matches!(settlement.state, LatchState::Pending) {
        return false;
    }
    let counted = DeliveredCounts {
        rows: settlement.delivered.rows.saturating_add(rows),
        result_bytes: settlement
            .delivered
            .result_bytes
            .saturating_add(result_bytes),
        truncated: settlement.delivered.truncated,
    };
    settlement.delivered = counted;
    true
}

/// The state a terminal is chosen from, and the counts it reports.
#[derive(Default)]
struct Settlement {
    state: LatchState,
    delivered: DeliveredCounts,
    /// Whether a consumer that re-frames each batch owns the delivery count.
    ///
    /// A batch leaving the result stream is not the same event as a caller
    /// receiving its rows: a consumer that splits one batch into protocol
    /// frames may release only some of them. Exactly one accounting may be
    /// live, and this is where that is decided, because this is what both
    /// would write to.
    consumer_accounted: bool,
}

impl Default for TerminalLatch {
    fn default() -> Self {
        Self {
            settlement: Mutex::new(Settlement::default()),
            poisoned: AtomicBool::new(false),
        }
    }
}

impl TerminalLatch {
    /// Takes the transition lock, remembering a poisoning rather than
    /// trusting the state it guards.
    ///
    /// A poisoned lock means a thread panicked mid-transition, so the chosen
    /// outcome may be torn and [`Self::dispatch`] fails closed on it. The
    /// counts are recovered from the guard: every update assigns a value it
    /// computed first, so none can be half applied.
    fn lock(&self) -> MutexGuard<'_, Settlement> {
        self.settlement.lock().unwrap_or_else(|poisoned| {
            self.poisoned.store(true, Ordering::SeqCst);
            PoisonError::into_inner(poisoned)
        })
    }

    /// Reports whether this latch can still vouch for its own transition.
    ///
    /// Consults the mutex directly as well as the sticky flag, so a panic is
    /// visible immediately rather than only after the next lock.
    pub(crate) fn poisoned(&self) -> bool {
        self.poisoned.load(Ordering::SeqCst) || self.settlement.is_poisoned()
    }

    /// Admits one release into the record, or refuses it.
    ///
    /// This is the boundary. A release is admitted only while no terminal has
    /// been chosen, and it is counted in the same critical section that admits
    /// it, so every terminal built afterwards reports it. A refusal means some
    /// other party has already ended the query: the caller must discard what
    /// it was about to release rather than hand out a row the record will not
    /// carry.
    ///
    /// Returns whether the release was admitted.
    pub(crate) fn admit_release(&self, rows: u64, result_bytes: u64) -> bool {
        let mut settlement = self.lock();
        admit_locked(&mut settlement, rows, result_bytes)
    }

    /// Hands the delivery count to a consumer that re-frames each batch.
    ///
    /// After this, [`Self::record_batch_delivery`] records nothing, and the
    /// consumer admits what it actually releases through
    /// [`Self::admit_release`].
    pub(crate) fn account_at_consumer(&self) {
        self.lock().consumer_accounted = true;
    }

    /// Admits a whole batch as delivered, unless a consumer owns the count.
    ///
    /// One acquisition decides both questions. Reading the flag under one lock
    /// and counting under another would let a consumer take ownership between
    /// them, and the batch would then be counted twice — once here and again
    /// as the frames it becomes.
    pub(crate) fn record_batch_delivery(&self, rows: u64, result_bytes: u64) -> BatchRelease {
        let mut settlement = self.lock();
        let release = if settlement.consumer_accounted {
            BatchRelease::CountedByConsumer
        } else if admit_locked(&mut settlement, rows, result_bytes) {
            BatchRelease::Admitted
        } else {
            BatchRelease::Refused
        };
        drop(settlement);
        release
    }

    /// Records that release stopped at a ceiling with rows still available.
    ///
    /// Ignored once a terminal has been chosen: that terminal was built from
    /// the counts as they stood, and this would describe a result it does not.
    pub(crate) fn record_truncation(&self) {
        let mut settlement = self.lock();
        if matches!(settlement.state, LatchState::Pending) {
            settlement.delivered.truncated = true;
        }
    }

    /// Returns what the consumer has received so far.
    pub(crate) fn delivered(&self) -> DeliveredCounts {
        self.lock().delivered
    }

    /// Offers the producer's measured terminal.
    ///
    /// A measured failure is the exact truth for this intent, so it is kept
    /// even after a withdrawal. A measured success is not: a consumer that
    /// withdrew must never see its query recorded as a success.
    ///
    /// Returns whether this report became the selected terminal.
    pub(crate) fn report(
        &self,
        outcome: QueryOutcome,
        duration: Duration,
        source: &SourceSnapshot,
    ) -> bool {
        let mut settlement = self.lock();
        let completion = QueryCompletion::new(
            outcome,
            duration,
            RowCount::new(settlement.delivered.rows),
            settlement.delivered.truncation(),
            source.clone(),
        );
        let keep = match &settlement.state {
            LatchState::Pending => true,
            LatchState::Cancelled => completion.outcome() != QueryOutcome::Succeeded,
            // The consumer stopped at a bound it asked for and released
            // exactly what it recorded. A later producer SUCCESS counts rows
            // that reached no caller, so it cannot replace that. A measured
            // failure can, in either arrival order, because a query that
            // failed did not succeed for anyone.
            //
            // A cancellation is neither. Stopping at the bound means dropping
            // the stream to stop the producer, and the producer reports that
            // drop back as `Cancelled` — it is this transition's own echo,
            // not a measurement of anything.
            LatchState::ConsumerBound(_) => !matches!(
                completion.outcome(),
                QueryOutcome::Succeeded | QueryOutcome::Failed(ErrorClass::Cancelled)
            ),
            LatchState::Reported(_) | LatchState::Dispatched => false,
        };
        if keep {
            settlement.state = LatchState::Reported(Box::new(completion));
        }
        keep
    }

    /// Records the terminal the consumer measured at its own admitted bound.
    ///
    /// The consumer released exactly what this records and asked for nothing
    /// further. That outranks a producer success, which counts rows this
    /// caller never received, and it does not outrank a measured producer
    /// failure: a query that failed did not succeed for anyone.
    ///
    /// The counts come from this same critical section, so the terminal the
    /// caller reads and the record settled here cannot describe different
    /// rows. Returns those counts when the bound is taken, and `None` when
    /// another party had already ended the query — in which case the caller
    /// must release nothing further and take that party's terminal.
    pub(crate) fn settle_consumer_bound(
        &self,
        source: &SourceSnapshot,
        duration: Duration,
    ) -> Option<DeliveredCounts> {
        let mut settlement = self.lock();
        let keep = match &settlement.state {
            LatchState::Pending => true,
            LatchState::Reported(reported) => reported.outcome() == QueryOutcome::Succeeded,
            LatchState::ConsumerBound(_) | LatchState::Cancelled | LatchState::Dispatched => false,
        };
        let settled = keep.then(|| {
            // Release stopped with rows still available, by definition: the
            // bound was reached because the next frame did not fit.
            let delivered = DeliveredCounts {
                rows: settlement.delivered.rows,
                result_bytes: settlement.delivered.result_bytes,
                truncated: true,
            };
            // Built before anything is mutated. The allocation below is the
            // one step here that can unwind, and it must not leave the counts
            // describing a truncation no terminal recorded.
            let completion = Box::new(QueryCompletion::new(
                QueryOutcome::Succeeded,
                duration,
                RowCount::new(delivered.rows),
                delivered.truncation(),
                source.clone(),
            ));
            settlement.delivered = delivered;
            settlement.state = LatchState::ConsumerBound(completion);
            delivered
        });
        drop(settlement);
        settled
    }

    /// Withdraws the query.
    ///
    /// This is a no-op once the permit has become a command. It is also a
    /// no-op against a measured failure, so a deadline or a bound refusal is
    /// never relabelled as a cancellation.
    pub(crate) fn cancel(&self) {
        let mut settlement = self.lock();
        let withdraw = match &settlement.state {
            LatchState::Pending => true,
            LatchState::Reported(completion) => completion.outcome() == QueryOutcome::Succeeded,
            // A consumer that stopped at its own bound then drops the stream
            // to stop the producer. That drop is how the bound is enforced,
            // not a withdrawal from it.
            LatchState::ConsumerBound(_) | LatchState::Cancelled | LatchState::Dispatched => false,
        };
        if withdraw {
            settlement.state = LatchState::Cancelled;
        }
    }

    /// Reports whether either side has already ended the query.
    pub(crate) fn is_settled(&self) -> bool {
        !matches!(self.lock().state, LatchState::Pending)
    }

    /// Freezes the terminal and returns the completion to settle.
    ///
    /// This is the linearization point. Exactly one call moves the latch to
    /// `Dispatched`; every later withdrawal is ignored, because the permit is
    /// already a command whose content cannot change.
    pub(crate) fn dispatch(&self, source: &SourceSnapshot, duration: Duration) -> QueryCompletion {
        // The counts are read inside the lock that freezes the transition, so
        // no release can be admitted between reading them and freezing it.
        let mut guard = self.lock();
        let delivered = guard.delivered;
        let state = std::mem::replace(&mut guard.state, LatchState::Dispatched);
        drop(guard);
        if self.poisoned() {
            // The transition is no longer trustworthy, so the outcome fails
            // closed. The counts survive it: each update assigned a value it
            // computed first, so what already left the stream is still
            // reported exactly.
            return QueryCompletion::new(
                QueryOutcome::Failed(ErrorClass::Internal),
                duration,
                RowCount::new(delivered.rows()),
                delivered.truncation(),
                source.clone(),
            );
        }
        match state {
            LatchState::Reported(completion) | LatchState::ConsumerBound(completion) => *completion,
            LatchState::Pending | LatchState::Cancelled | LatchState::Dispatched => {
                cancelled(source, duration, delivered)
            }
        }
    }
}

/// Builds a cancelled terminal that reports what the consumer received.
fn cancelled(
    source: &SourceSnapshot,
    duration: Duration,
    delivered: DeliveredCounts,
) -> QueryCompletion {
    QueryCompletion::new(
        QueryOutcome::Failed(ErrorClass::Cancelled),
        duration,
        RowCount::new(delivered.rows()),
        delivered.truncation(),
        source.clone(),
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    fn source() -> SourceSnapshot {
        SourceSnapshot::from_canonical(vec![]).expect("an empty source vector is canonical")
    }

    const TICK: Duration = Duration::from_millis(1);

    /// Records a producer terminal the way the guardian does.
    fn report(latch: &TerminalLatch, outcome: QueryOutcome) {
        let _ = latch.report(outcome, TICK, &source());
    }

    /// Settles a consumer bound the way the result stream does.
    fn settle_bound(latch: &TerminalLatch) -> Option<DeliveredCounts> {
        latch.settle_consumer_bound(&source(), TICK)
    }

    #[test]
    fn a_withdrawal_without_any_report_still_reports_delivered_rows() {
        let latch = TerminalLatch::default();
        assert!(latch.admit_release(600, 4800));
        latch.cancel();

        let settled = latch.dispatch(&source(), Duration::from_millis(4));

        assert_eq!(
            settled.outcome(),
            QueryOutcome::Failed(ErrorClass::Cancelled)
        );
        assert_eq!(
            settled.rows(),
            RowCount::new(600),
            "a query that released rows must never settle as zero rows"
        );
        assert_eq!(
            latch.delivered().result_bytes(),
            4800,
            "the released byte count survives a withdrawal with no report"
        );
    }

    /// A consumer that stopped at its own bound is not a consumer that
    /// withdrew, and the drop it uses to stop the producer must not say so.
    #[test]
    fn a_consumer_bound_survives_the_withdrawal_that_stops_the_producer() {
        let latch = TerminalLatch::default();
        assert!(latch.admit_release(4, 32));
        settle_bound(&latch);
        latch.cancel();
        let settled = latch.dispatch(&source(), Duration::from_millis(1));
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Succeeded,
            "a bounded read is a success, and the drop that ends it is not a withdrawal"
        );
        assert_eq!(settled.rows(), RowCount::new(4));
    }

    /// The producer counts every row it produced. The consumer counts what it
    /// released. When they disagree, the caller's view is the record.
    #[test]
    fn a_consumer_bound_outranks_a_producer_success_and_yields_to_a_failure() {
        let over_success = TerminalLatch::default();
        assert!(over_success.admit_release(4, 32));
        report(&over_success, QueryOutcome::Succeeded);
        settle_bound(&over_success);
        assert_eq!(
            over_success
                .dispatch(&source(), Duration::from_millis(1))
                .rows(),
            RowCount::new(4),
            "the rows the caller received, not the rows the producer made"
        );

        // Both arrival orders, because the outcome must not depend on which
        // task the scheduler ran first. The deadline wins either way.
        let failure_first = TerminalLatch::default();
        report(&failure_first, QueryOutcome::Failed(ErrorClass::Deadline));
        settle_bound(&failure_first);
        let bound_first = TerminalLatch::default();
        settle_bound(&bound_first);
        report(&bound_first, QueryOutcome::Failed(ErrorClass::Deadline));
        for (latch, order) in [
            (failure_first, "failure first"),
            (bound_first, "bound first"),
        ] {
            assert_eq!(
                latch
                    .dispatch(&source(), Duration::from_millis(1))
                    .outcome(),
                QueryOutcome::Failed(ErrorClass::Deadline),
                "a query that failed did not succeed for anyone, {order}"
            );
        }
    }

    /// The producer's report of the drop that stopped it is that drop's own
    /// echo, and must not overwrite the bound the drop was enforcing.
    #[test]
    fn a_producer_cancellation_after_a_consumer_bound_is_refused() {
        let latch = TerminalLatch::default();
        assert!(latch.admit_release(4, 32));
        settle_bound(&latch);
        latch.cancel();
        report(&latch, QueryOutcome::Failed(ErrorClass::Cancelled));
        let settled = latch.dispatch(&source(), Duration::from_millis(1));
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Succeeded,
            "the producer noticing the stop is not a measurement of failure"
        );
        assert_eq!(settled.rows(), RowCount::new(4));
    }

    /// A withdrawal that already happened is not undone by a later bound.
    #[test]
    fn a_consumer_bound_after_a_withdrawal_is_refused() {
        let latch = TerminalLatch::default();
        latch.cancel();
        settle_bound(&latch);
        assert_eq!(
            latch
                .dispatch(&source(), Duration::from_millis(1))
                .outcome(),
            QueryOutcome::Failed(ErrorClass::Cancelled)
        );
    }

    /// Batch-level delivery stops the moment a consumer owns the count.
    #[test]
    fn consumer_accounting_replaces_batch_accounting_rather_than_adding_to_it() {
        let latch = TerminalLatch::default();
        latch.record_batch_delivery(13, 104);
        assert_eq!(
            latch.delivered().rows(),
            13,
            "the batch count is live first"
        );

        let framed = TerminalLatch::default();
        framed.account_at_consumer();
        framed.record_batch_delivery(13, 104);
        let _ = framed.admit_release(4, 32);
        assert_eq!(
            framed.delivered().rows(),
            4,
            "only what the consumer released is counted"
        );
    }

    #[test]
    fn a_withdrawal_keeps_a_measured_failure_and_drops_a_measured_success() {
        let failed = TerminalLatch::default();
        report(&failed, QueryOutcome::Failed(ErrorClass::Deadline));
        failed.cancel();
        assert_eq!(
            failed
                .dispatch(&source(), Duration::from_millis(1))
                .outcome(),
            QueryOutcome::Failed(ErrorClass::Deadline),
            "a measured deadline is the exact truth and is not relabelled"
        );

        let succeeded = TerminalLatch::default();
        let _ = succeeded.admit_release(3, 24);
        report(&succeeded, QueryOutcome::Succeeded);
        succeeded.cancel();
        let settled = succeeded.dispatch(&source(), Duration::from_millis(1));
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Failed(ErrorClass::Cancelled)
        );
        assert_eq!(settled.rows(), RowCount::new(3));
    }

    #[test]
    fn a_late_report_after_a_withdrawal_keeps_its_measured_failure() {
        let latch = TerminalLatch::default();
        let _ = latch.admit_release(9, 72);
        latch.cancel();
        report(&latch, QueryOutcome::Failed(ErrorClass::Bounds));

        assert_eq!(
            latch
                .dispatch(&source(), Duration::from_millis(1))
                .outcome(),
            QueryOutcome::Failed(ErrorClass::Bounds),
        );
    }

    #[test]
    fn a_late_success_after_a_withdrawal_is_refused() {
        let latch = TerminalLatch::default();
        latch.cancel();
        report(&latch, QueryOutcome::Succeeded);

        assert_eq!(
            latch
                .dispatch(&source(), Duration::from_millis(1))
                .outcome(),
            QueryOutcome::Failed(ErrorClass::Cancelled),
        );
    }

    #[test]
    fn dispatch_and_cancellation_are_mutually_exclusive() {
        let latch = TerminalLatch::default();
        assert!(latch.admit_release(4, 32));
        report(&latch, QueryOutcome::Succeeded);

        let settled = latch.dispatch(&source(), Duration::from_millis(1));
        assert_eq!(settled.outcome(), QueryOutcome::Succeeded);

        // The permit is a command now. A withdrawal cannot rewrite it.
        latch.cancel();
        let again = latch.dispatch(&source(), Duration::from_millis(1));
        assert_eq!(
            again.outcome(),
            QueryOutcome::Failed(ErrorClass::Cancelled),
            "a second dispatch never re-issues the first command's content"
        );
    }

    #[test]
    fn truncation_reports_the_delivered_row_count() {
        let latch = TerminalLatch::default();
        let _ = latch.admit_release(2, 16);
        latch.record_truncation();
        latch.cancel();

        assert_eq!(
            latch
                .dispatch(&source(), Duration::from_millis(1))
                .truncation(),
            Truncation::TruncatedAt(2),
        );
    }

    #[test]
    fn no_transition_can_overwrite_or_discard_a_delivered_count() {
        let latch = TerminalLatch::default();
        assert!(latch.admit_release(4, 32));

        // Every transition, in every order a real stream can produce them.
        report(&latch, QueryOutcome::Succeeded);
        latch.cancel();
        report(&latch, QueryOutcome::Failed(ErrorClass::Deadline));

        // A release attempted after a terminal was chosen is refused, and
        // counts nothing. That is the other half of the boundary: the record
        // cannot gain a row the caller was never handed, and the caller
        // cannot be handed a row the record does not carry.
        assert!(
            !latch.admit_release(2, 16),
            "a settled query admits no further release"
        );
        latch.cancel();

        assert_eq!(latch.delivered().rows(), 4);
        assert_eq!(latch.delivered().result_bytes(), 32);

        let settled = latch.dispatch(&source(), Duration::from_millis(1));
        // A measured failure arrived after the withdrawal, so it wins the
        // outcome, and reports the counts admitted before it.
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Failed(ErrorClass::Deadline)
        );
        assert_eq!(settled.rows(), RowCount::new(4));
        // The delivered counts are still intact behind it.
        assert_eq!(latch.delivered().rows(), 4);
        assert_eq!(latch.delivered().result_bytes(), 32);
    }

    /// Release and terminal selection are one decision, in both orders.
    ///
    /// These are the two schedules the boundary exists for. Neither may leave
    /// the caller holding a row the record omits, nor the record carrying a
    /// row the caller never received.
    #[test]
    fn a_release_and_a_terminal_cannot_interleave() {
        // Release wins. The frame is in the record the failure then reports.
        let release_first = TerminalLatch::default();
        assert!(
            release_first.admit_release(4, 32),
            "nothing has ended the query yet"
        );
        report(&release_first, QueryOutcome::Failed(ErrorClass::Deadline));
        let settled = release_first.dispatch(&source(), TICK);
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Failed(ErrorClass::Deadline)
        );
        assert_eq!(
            settled.rows(),
            RowCount::new(4),
            "a release admitted before the terminal is part of it"
        );

        // The terminal wins. The release is refused, so the caller is never
        // handed the frame and the record does not count it.
        let terminal_first = TerminalLatch::default();
        report(&terminal_first, QueryOutcome::Failed(ErrorClass::Deadline));
        assert!(
            !terminal_first.admit_release(4, 32),
            "a query that has already ended admits no release"
        );
        let settled = terminal_first.dispatch(&source(), TICK);
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Failed(ErrorClass::Deadline)
        );
        assert_eq!(
            settled.rows(),
            RowCount::new(0),
            "a refused release is in neither the record nor the caller's hands"
        );
    }

    /// A consumer can publish its own failure terminal only when that failure
    /// became the record. If another terminal won first, the consumer must
    /// take that terminal instead of describing a different outcome.
    #[test]
    fn a_report_says_whether_its_terminal_became_the_record() {
        let consumer_first = TerminalLatch::default();
        assert!(consumer_first.report(QueryOutcome::Failed(ErrorClass::Bounds), TICK, &source(),));
        assert!(!consumer_first.report(
            QueryOutcome::Failed(ErrorClass::Deadline),
            TICK,
            &source(),
        ));
        assert_eq!(
            consumer_first.dispatch(&source(), TICK).outcome(),
            QueryOutcome::Failed(ErrorClass::Bounds),
        );

        let producer_first = TerminalLatch::default();
        assert!(
            producer_first.report(QueryOutcome::Failed(ErrorClass::Deadline), TICK, &source(),)
        );
        assert!(!producer_first.report(QueryOutcome::Failed(ErrorClass::Bounds), TICK, &source(),));
        assert_eq!(
            producer_first.dispatch(&source(), TICK).outcome(),
            QueryOutcome::Failed(ErrorClass::Deadline),
        );
    }

    /// A batch offered after a terminal is refused, exactly as a frame is.
    ///
    /// The batch path is what a consumer that does not re-frame uses. It must
    /// not become a second way into the record, so it goes through the same
    /// admission and answers with which of the three happened rather than a
    /// bool that reads the same for "already counted" and "refused".
    #[test]
    fn a_batch_offered_after_a_terminal_is_refused() {
        let latch = TerminalLatch::default();
        assert_eq!(
            latch.record_batch_delivery(4, 32),
            BatchRelease::Admitted,
            "nothing has ended the query yet"
        );
        report(&latch, QueryOutcome::Failed(ErrorClass::Deadline));
        assert_eq!(
            latch.record_batch_delivery(9, 72),
            BatchRelease::Refused,
            "the record is chosen; this batch is not in it"
        );
        assert_eq!(latch.dispatch(&source(), TICK).rows(), RowCount::new(4));
    }

    /// A consumer that re-frames owns the count, and the batch path says so
    /// rather than reporting a refusal it did not make.
    #[test]
    fn a_batch_is_not_counted_twice_when_a_consumer_owns_the_count() {
        let latch = TerminalLatch::default();
        latch.account_at_consumer();
        assert_eq!(
            latch.record_batch_delivery(9, 72),
            BatchRelease::CountedByConsumer,
            "the frames this batch becomes are what get admitted"
        );
        assert!(latch.admit_release(4, 32));
        assert_eq!(
            latch.delivered().rows(),
            4,
            "only the released frames are counted"
        );
    }

    /// Dispatch freezes the record, so nothing can be added behind it.
    #[test]
    fn a_release_after_dispatch_is_refused() {
        let latch = TerminalLatch::default();
        assert!(latch.admit_release(4, 32));
        let settled = latch.dispatch(&source(), TICK);
        assert_eq!(settled.rows(), RowCount::new(4));
        assert!(
            !latch.admit_release(1, 8),
            "the permit is already a command; its content cannot change"
        );
        assert_eq!(
            latch.delivered().rows(),
            4,
            "and the counts behind it are untouched"
        );
    }

    /// Every terminal is built from the counts under the lock that records it.
    ///
    /// A producer that snapshotted the counts and reported a completion built
    /// from them later would report a number a release could still change.
    /// The latch takes an outcome, never a completion.
    #[test]
    fn a_terminal_reports_the_counts_admitted_when_it_was_recorded() {
        let latch = TerminalLatch::default();
        assert!(latch.admit_release(3, 24));
        report(&latch, QueryOutcome::Succeeded);
        assert!(
            !latch.admit_release(5, 40),
            "the terminal has been chosen; nothing more is admitted"
        );
        let settled = latch.dispatch(&source(), TICK);
        assert_eq!(settled.rows(), RowCount::new(3));
        assert_eq!(settled.truncation(), Truncation::Complete);
    }

    /// A latch that cannot vouch for its own state must not report success.
    #[test]
    fn a_poisoned_latch_fails_closed_and_keeps_its_counts() {
        let latch = std::sync::Arc::new(TerminalLatch::default());
        let _ = latch.admit_release(5, 40);
        report(&latch, QueryOutcome::Succeeded);

        let poisoner = std::sync::Arc::clone(&latch);
        let _ = std::thread::spawn(move || {
            let _guard = poisoner
                .settlement
                .lock()
                .expect("the lock is not yet poisoned");
            panic!("tear the lifecycle state");
        })
        .join();

        assert!(latch.poisoned(), "the panic marks the latch untrustworthy");
        let settled = latch.dispatch(&source(), Duration::from_millis(1));
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Failed(ErrorClass::Internal),
            "a torn latch never reports the success it can no longer vouch for"
        );
        assert_eq!(
            settled.rows(),
            RowCount::new(5),
            "a torn transition never makes released rows inexact"
        );
        assert_eq!(
            settled.truncation(),
            Truncation::Complete,
            "truncation survives the poisoning with the counts it belongs to"
        );
        assert_eq!(
            latch.delivered().result_bytes(),
            40,
            "a torn transition never makes released bytes inexact"
        );
    }

    /// Poisoning must not cost a single released row or byte.
    ///
    /// The counts live under the poisoned lock, and every read of them
    /// recovers it: `lock` takes the guard out of the `PoisonError` rather
    /// than refusing. What the poisoning costs is trust in the transition,
    /// which `dispatch` answers by failing closed — never the counts.
    #[test]
    fn a_poisoned_latch_reports_the_exact_counts_it_already_released() {
        let latch = std::sync::Arc::new(TerminalLatch::default());
        let _ = latch.admit_release(600, 4096);
        latch.record_truncation();

        let poisoner = std::sync::Arc::clone(&latch);
        let _ = std::thread::spawn(move || {
            let _guard = poisoner
                .settlement
                .lock()
                .expect("the lock is not yet poisoned");
            panic!("tear the transition");
        })
        .join();

        // More rows leave the stream after the tear. They still count.
        let _ = latch.admit_release(7, 64);

        assert!(latch.poisoned());
        let settled = latch.dispatch(&source(), Duration::from_millis(2));
        assert_eq!(
            settled.outcome(),
            QueryOutcome::Failed(ErrorClass::Internal)
        );
        assert_eq!(settled.rows(), RowCount::new(607));
        assert_eq!(settled.truncation(), Truncation::TruncatedAt(607));
        assert_eq!(latch.delivered().result_bytes(), 4160);
    }
}