polyc-query 2026.9.6

//! Releases exact query results through a consumer-polled bounded stream.

use std::collections::BTreeSet;
use std::error::Error;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
#[cfg(test)]
use datafusion::catalog::TableProvider;
use datafusion::execution::memory_pool::MemoryReservation;
use datafusion::physical_plan::SendableRecordBatchStream;
use futures::{Stream, StreamExt};
use polyc_state::journal::GetJournalSource;
use polyc_state::query_audit::{ErrorClass, QueryOutcome};
use tokio::sync::{OwnedSemaphorePermit, mpsc};
use tokio_util::sync::CancellationToken;

use super::latch::{BatchRelease, DeliveredCounts, TerminalLatch};
use super::{
    CoreExecutionError, CoreMetadataAuthority, CoreOperationContext, CurrentCredentialAuthority,
    EffectiveCoreBounds, PermitGuardian, QueryScope, classify_error, operation_refusal,
    operation_wait,
};
#[cfg(test)]
use crate::core_resolution::CoreTable;

/// Owns a physical plan whose permit, exact pins, and limits cannot change.
pub(crate) struct BoundCoreQuery {
    pub(super) guardian: PermitGuardian,
    pub(super) dataframe: datafusion::dataframe::DataFrame,
    pub(super) manifests: Vec<polyc_state::projection::ProjectionManifest>,
    pub(super) original_scope: QueryScope,
    pub(super) metadata: Arc<dyn CoreMetadataAuthority>,
    pub(super) scope_revalidator: Arc<dyn CurrentCredentialAuthority>,
    pub(super) operation: Arc<CoreOperationContext>,
    pub(super) cancellation: CancellationToken,
    pub(super) bounds: EffectiveCoreBounds,
    pub(super) revalidation_interval: Duration,
    pub(super) _explain_enabled: bool,
    pub(super) execution_admission: OwnedSemaphorePermit,
    pub(super) source_decode_reservation: MemoryReservation,
    /// Deschedules the producer's terminal report by this much.
    ///
    /// Zero in production. A test uses it to hold the producer away from the
    /// guardian after rows were already delivered, which is the exact race a
    /// scheduling grace used to resolve wrongly.
    #[cfg(test)]
    pub(super) report_delay: Duration,
    #[cfg(test)]
    pub(super) providers: std::collections::BTreeMap<CoreTable, Arc<dyn TableProvider>>,
}

impl BoundCoreQuery {
    #[cfg(test)]
    pub(super) const fn delay_report_for_test(&mut self, delay: Duration) {
        self.report_delay = delay;
    }

    /// Restarts the producer's operation budget at this call.
    ///
    /// `prepare` anchors the declared budget before `bind` spends it on
    /// verified reads, so a loaded runner can drain a test's tight deadline
    /// before the scenario under test begins (POLY-373). A test binds under
    /// the generous default and then restarts the budget here: the producer's
    /// own waits — the readiness park, the `operation_wait` around each
    /// upstream poll, revalidation — measure from this call through the real
    /// timer path. Providers keep the context they bound with; their internal
    /// checks keep the original budget while the producer's wait bounds every
    /// poll.
    #[cfg(test)]
    pub(super) fn restart_deadline_for_test(&mut self, budget: Duration) {
        self.operation = Arc::new(CoreOperationContext::for_test(budget));
    }

    #[cfg(test)]
    pub(super) fn provider(&self, table: CoreTable) -> Arc<dyn TableProvider> {
        Arc::clone(
            self.providers
                .get(&table)
                .expect("bound dependency provider"),
        )
    }

    /// Starts the producer task and hands back the consumer-polled stream.
    ///
    /// Nothing is awaited here. The producer waits for the consumer's first
    /// readiness signal. No batch is produced before a poll asks for one.
    pub(crate) fn execute(self) -> CoreResultStream {
        // Captured before the plan is consumed. A caller must be able to
        // announce the result schema before the first batch exists, and the
        // durable audit's own source vector is what the terminal reports.
        let released_schema = self.dataframe.schema().inner().clone();
        let released_source = self.guardian.source().clone();
        let started = tokio::time::Instant::now();
        let Self {
            guardian,
            dataframe,
            manifests,
            original_scope,
            metadata,
            scope_revalidator,
            operation,
            cancellation,
            bounds,
            revalidation_interval,
            _explain_enabled,
            execution_admission,
            source_decode_reservation,
            #[cfg(test)]
            report_delay,
            #[cfg(test)]
                providers: _,
        } = self;
        // The latch is the one place the producer's report, the consumer's
        // withdrawal, and the delivered counts meet. The guardian reads it, so
        // an exact row count never depends on which task ran last.
        let latch = guardian.latch();
        let consumer_latch = Arc::clone(&latch);
        let consumer_cancellation = cancellation.clone();
        let terminal_cancellation = cancellation.clone();
        let (sender, receiver) = mpsc::channel(1);
        let (readiness_signal, readiness_requests) = mpsc::unbounded_channel();
        tokio::spawn(async move {
            let stream = operation_wait(&operation, &cancellation, dataframe.execute_stream())
                .await
                .and_then(|result| result.map_err(CoreExecutionError::from));
            let end = match stream {
                Ok(upstream) => {
                    let state = CoreStreamState {
                        upstream,
                        _execution_admission: execution_admission,
                        _source_decode_reservation: source_decode_reservation,
                        revalidation: CoreRevalidationWitness {
                            manifests,
                            original_scope,
                            metadata,
                            scope_revalidator,
                            operation: Arc::clone(&operation),
                        },
                        operation,
                        cancellation,
                        bounds,
                        revalidation_interval,
                        last_revalidation: None,
                        released_rows: 0,
                        released_bytes: 0,
                        known_extra_row: false,
                        latch: Arc::clone(&latch),
                    };
                    release_from_producer(state, &sender, readiness_requests).await
                }
                Err(error) => ProducerEnd::Failed(error),
            };
            // A consumer that withdrew after the last row still withdrew.
            let end = match end {
                ProducerEnd::Ended
                    if terminal_cancellation.is_cancelled() || sender.is_closed() =>
                {
                    ProducerEnd::Cancelled
                }
                end => end,
            };
            #[cfg(test)]
            if !report_delay.is_zero() {
                tokio::time::sleep(report_delay).await;
            }
            // The outcome only. The counts belong to the terminal, and the
            // terminal is built inside the latch's own critical section, so
            // nothing can be released between reading them and recording it.
            let settlement = guardian.finish(producer_outcome(&end)).await;
            let frame = match (settlement, end) {
                (Err(error), _) | (Ok(()), ProducerEnd::Failed(error)) => {
                    CoreStreamFrame::Failure(error)
                }
                (Ok(()), ProducerEnd::Cancelled) => {
                    CoreStreamFrame::Failure(CoreExecutionError::Cancelled)
                }
                (Ok(()), ProducerEnd::Ended) => CoreStreamFrame::Complete,
            };
            let _ = sender.send(frame).await;
        });
        CoreResultStream {
            receiver,
            readiness_signal,
            poll_outstanding: false,
            terminal_seen: false,
            cancellation: consumer_cancellation,
            latch: consumer_latch,
            schema: released_schema,
            source: released_source,
            started,
        }
    }
}

/// How the producer's own loop ended.
enum ProducerEnd {
    /// The upstream stream ended after every produced row was delivered.
    Ended,
    /// The consumer withdrew.
    Cancelled,
    /// The producer measured this exact failure.
    Failed(CoreExecutionError),
}

struct CoreStreamState {
    upstream: SendableRecordBatchStream,
    _execution_admission: OwnedSemaphorePermit,
    _source_decode_reservation: MemoryReservation,
    revalidation: CoreRevalidationWitness,
    operation: Arc<CoreOperationContext>,
    cancellation: CancellationToken,
    bounds: EffectiveCoreBounds,
    revalidation_interval: Duration,
    last_revalidation: Option<tokio::time::Instant>,
    released_rows: u64,
    released_bytes: u64,
    known_extra_row: bool,
    latch: Arc<TerminalLatch>,
}

async fn release_from_producer(
    mut state: CoreStreamState,
    sender: &mpsc::Sender<CoreStreamFrame>,
    mut readiness: mpsc::UnboundedReceiver<()>,
) -> ProducerEnd {
    loop {
        // The original deadline also bounds the wait for a readiness token. A
        // consumer that stops polling without dropping would otherwise pin the
        // permit, the admission slot, and the memory reservation forever, and
        // leave the durable intent unmatched past its declared deadline.
        let Ok(remaining) = state.operation.remaining() else {
            state.cancellation.cancel();
            return ProducerEnd::Failed(CoreExecutionError::Deadline);
        };
        tokio::select! {
            ready = readiness.recv() => {
                if ready.is_none() {
                    state.cancellation.cancel();
                    return ProducerEnd::Cancelled;
                }
            }
            () = state.cancellation.cancelled() => return ProducerEnd::Cancelled,
            () = tokio::time::sleep(remaining) => {
                state.cancellation.cancel();
                return ProducerEnd::Failed(CoreExecutionError::Deadline);
            }
        }
        match state.next().await {
            Ok(Some((batch, next))) => {
                state = next;
                if sender.send(CoreStreamFrame::Batch(batch)).await.is_err() {
                    state.cancellation.cancel();
                    return ProducerEnd::Cancelled;
                }
            }
            Ok(None) => return ProducerEnd::Ended,
            Err(error) => return ProducerEnd::Failed(error),
        }
    }
}

impl CoreStreamState {
    async fn next(mut self) -> Result<Option<(RecordBatch, Self)>, CoreExecutionError> {
        loop {
            if self.cancellation.is_cancelled() {
                return Err(CoreExecutionError::Cancelled);
            }
            if self.known_extra_row {
                return Ok(None);
            }
            let next =
                match operation_wait(&self.operation, &self.cancellation, self.upstream.next())
                    .await
                {
                    Ok(next) => next,
                    Err(error) => return Err(error),
                };
            let Some(batch) = next else {
                return Ok(None);
            };
            let batch = match batch {
                Ok(batch) => batch,
                Err(error) => return Err(stream_error(error)),
            };
            if batch.num_rows() == 0 {
                continue;
            }
            if self.released_rows >= self.bounds.rows() {
                self.latch.record_truncation();
                return Ok(None);
            }
            let remaining_rows = self.bounds.rows() - self.released_rows;
            let release_rows = usize::try_from(remaining_rows)
                .unwrap_or(usize::MAX)
                .min(batch.num_rows());
            let release = batch.slice(0, release_rows);
            if release_rows < batch.num_rows() {
                self.latch.record_truncation();
                self.known_extra_row = true;
            }
            let bytes = u64::try_from(release.get_array_memory_size()).unwrap_or(u64::MAX);
            let total = match self.released_bytes.checked_add(bytes) {
                Some(total) if total <= self.bounds.result_release_bytes() => total,
                Some(total) => {
                    return Err(CoreExecutionError::ReleaseBound {
                        observed: total,
                        limit: self.bounds.result_release_bytes(),
                    });
                }
                None => {
                    return Err(CoreExecutionError::ReleaseBound {
                        observed: u64::MAX,
                        limit: self.bounds.result_release_bytes(),
                    });
                }
            };
            if self
                .last_revalidation
                .is_none_or(|last| last.elapsed() >= self.revalidation_interval)
            {
                self.revalidation.revalidate(&self.cancellation).await?;
                self.last_revalidation = Some(tokio::time::Instant::now());
            }
            self.released_rows += u64::try_from(release_rows).unwrap_or(u64::MAX);
            self.released_bytes = total;
            return Ok(Some((release, self)));
        }
    }
}

fn stream_error(error: datafusion::error::DataFusionError) -> CoreExecutionError {
    let mut source: Option<&(dyn Error + 'static)> = Some(&error);
    while let Some(current) = source {
        if matches!(
            current.downcast_ref::<CoreExecutionError>(),
            Some(CoreExecutionError::Deadline)
        ) {
            return CoreExecutionError::Deadline;
        }
        if matches!(
            current.downcast_ref::<CoreExecutionError>(),
            Some(CoreExecutionError::Cancelled)
        ) {
            return CoreExecutionError::Cancelled;
        }
        source = current.source();
    }
    CoreExecutionError::DataFusion(error)
}

struct CoreRevalidationWitness {
    manifests: Vec<polyc_state::projection::ProjectionManifest>,
    original_scope: QueryScope,
    metadata: Arc<dyn CoreMetadataAuthority>,
    scope_revalidator: Arc<dyn CurrentCredentialAuthority>,
    operation: Arc<CoreOperationContext>,
}

impl CoreRevalidationWitness {
    async fn revalidate(&self, cancellation: &CancellationToken) -> Result<(), CoreExecutionError> {
        operation_wait(&self.operation, cancellation, self.revalidate_inner()).await?
    }

    async fn revalidate_inner(&self) -> Result<(), CoreExecutionError> {
        let current = self
            .scope_revalidator
            .current_scope(&self.operation)
            .await?;
        if !scope_contains(&current, &self.original_scope) {
            return Err(CoreExecutionError::AuthorityNarrowed);
        }
        for manifest in &self.manifests {
            self.operation.check().map_err(operation_refusal)?;
            // The liveness re-read asks the authority the manifest's FAMILY
            // declares. A journal family asks a partition head; a Versioned
            // family asks its aggregate's lineage and head. Asking one about
            // the other would compare two position spaces.
            let family = crate::core_execution::family_for_manifest(manifest)?;
            match (family.source().evidence_variant(), manifest.evidence()) {
                (
                    polyc_projection::family::EvidenceVariant::Journal,
                    polyc_state::feed::SourceEvidence::Journal(checkpoint),
                ) => {
                    let expected = checkpoint.source();
                    let observed = self
                        .metadata
                        .source_head(
                            &self.operation,
                            GetJournalSource::new(expected.partition().clone()),
                        )
                        .await?
                        .ok_or_else(|| {
                            CoreExecutionError::SourceChanged(expected.partition().clone())
                        })?;
                    if observed.source() != expected {
                        return Err(CoreExecutionError::SourceChanged(
                            expected.partition().clone(),
                        ));
                    }
                }
                (
                    polyc_projection::family::EvidenceVariant::Versioned,
                    polyc_state::feed::SourceEvidence::Versioned(checkpoint),
                ) => {
                    let observed = self
                        .metadata
                        .versioned_source_head(&self.operation, checkpoint.source().scope())
                        .await?;
                    versioned_liveness(checkpoint, &observed)?;
                }
                (
                    polyc_projection::family::EvidenceVariant::PersonaMemory,
                    polyc_state::feed::SourceEvidence::PersonaMemory(checkpoint),
                ) => {
                    let observed = self
                        .metadata
                        .persona_memory_source_head(
                            &self.operation,
                            checkpoint.source().partition(),
                        )
                        .await?;
                    persona_memory_liveness(checkpoint, &observed)?;
                }
                (
                    polyc_projection::family::EvidenceVariant::QueryAudit,
                    polyc_state::feed::SourceEvidence::QueryAudit(checkpoint),
                ) => {
                    // This plane holds no live read of the query-audit
                    // authority — the same reason `resolve_query_audit_family`
                    // resolves through `resolve_manifest` rather than a source
                    // head. Revalidation asks the identical question the same
                    // way: `resolve_manifest` under the PINNED source, never a
                    // guess, so a reply of anything but `Current` means the
                    // pin's own lineage no longer names the current one.
                    let partition = polyc_state::id::PartitionId::new(
                        polyc_projection::family::QUERY_AUDIT_SOURCE,
                    );
                    let key = polyc_state::projection::ProjectionKey::new(
                        polyc_state::projection::FamilyId::new(family.family_str()),
                        partition,
                    );
                    let source = polyc_state::feed::ProjectionSource::QueryAudit(
                        checkpoint.source().clone(),
                    );
                    let owner = manifest.object_descriptor().owner().clone();
                    let resolution = self
                        .metadata
                        .resolve_manifest(
                            &self.operation,
                            polyc_state::projection::ResolveManifest::new(key, source, owner),
                        )
                        .await?;
                    query_audit_liveness(checkpoint, &resolution)?;
                }
                (
                    polyc_projection::family::EvidenceVariant::Observed,
                    polyc_state::feed::SourceEvidence::Observed(checkpoint),
                ) => {
                    let observed = self
                        .metadata
                        .observed_head(&self.operation, checkpoint.source().collection())
                        .await?
                        .ok_or_else(|| {
                            CoreExecutionError::SourceChanged(
                                checkpoint.source().projection_partition().clone(),
                            )
                        })?;
                    observed_liveness(checkpoint, &observed)?;
                }
                // Any other pair is a manifest whose evidence is not the
                // variant its family issues. Catalog admission refuses that,
                // so reaching it here means a generation was admitted under a
                // different rule.
                _ => return Err(CoreExecutionError::AuthorityNarrowed),
            }
        }
        Ok(())
    }
}

/// Requires the generation to describe the authority's latest full snapshot.
///
/// Unlike an append-only history, an observation ordinal replaces the whole
/// collection. A later ordinal therefore makes an older projection stale,
/// even when both belong to the same lineage.
fn observed_liveness(
    pin: &polyc_state::feed::ObservedCheckpoint,
    observed: &polyc_state::observation::ObservationHead,
) -> Result<(), CoreExecutionError> {
    if observed.source() != pin.source() {
        return Err(CoreExecutionError::SourceChanged(
            pin.source().projection_partition().clone(),
        ));
    }
    if observed.ordinal() != pin.ordinal() {
        return Err(CoreExecutionError::AuthorityNarrowed);
    }
    if observed.payload_digest() != *pin.payload_digest() {
        return Err(CoreExecutionError::AuthorityNarrowed);
    }
    Ok(())
}

/// Decides whether a pinned Versioned generation is still live.
///
/// The LINEAGE decides first. A pin from a replaced log is `SourceChanged`
/// for that reason, not for how its position compares against a log that
/// never issued it: a wipe that has refilled only partway would otherwise
/// report the pin as merely ahead of the head, which is true of the new log
/// and says nothing about the real fault.
///
/// Inside one lineage, a head below the pin means the authority no longer
/// contains what the generation folded, which is `AuthorityNarrowed`.
///
/// # Errors
///
/// Returns the refusal the caller reports for a pin that is no longer live.
fn versioned_liveness(
    pin: &polyc_state::feed::VersionedCheckpoint,
    observed: &polyc_state::versioned::VersionedSourceHead,
) -> Result<(), CoreExecutionError> {
    let expected = pin.source();
    if observed.incarnation() != expected.incarnation() {
        return Err(CoreExecutionError::SourceChanged(
            expected.scope().partition().clone(),
        ));
    }
    if observed.head().position() < pin.position() {
        return Err(CoreExecutionError::AuthorityNarrowed);
    }
    Ok(())
}

/// Decides whether a pinned persona-memory generation is still live.
///
/// The lineage decides first, exactly as it does for a Versioned pin: a
/// `rewrite`, `migrate`, or `destroy` since the pin was taken changes the
/// derived incarnation, and that is `SourceChanged` whatever the position
/// says. Inside one lineage, a head below the pin means the authority no
/// longer contains what the generation folded, which is `AuthorityNarrowed`.
///
/// # Errors
///
/// Returns the refusal the caller reports for a pin that is no longer live.
fn persona_memory_liveness(
    pin: &polyc_state::feed::PersonaMemoryHistoryCheckpoint,
    observed: &polyc_state::persona_memory::journal::PersonaMemorySourceHead,
) -> Result<(), CoreExecutionError> {
    let expected = pin.source();
    if observed.incarnation() != expected.incarnation() {
        return Err(CoreExecutionError::SourceChanged(
            expected.projection_partition().clone(),
        ));
    }
    if observed.head().position() < pin.position() {
        return Err(CoreExecutionError::AuthorityNarrowed);
    }
    Ok(())
}

/// Decides whether a pinned query-audit generation is still live.
///
/// There is no live head to read this against — the same reason
/// `resolve_query_audit_family` never reads one. Revalidation instead
/// resolves under the PINNED source itself: `Current` means the catalog
/// still records that exact lineage for the family, and the returned
/// generation's own ordinal — which only ever advances — decides whether the
/// authority still contains what this pin folded. Anything else (a reveal of
/// a different source, or no generation at all) means the pinned lineage is
/// no longer the recorded one.
///
/// # Errors
///
/// Returns the refusal the caller reports for a pin that is no longer live.
fn query_audit_liveness(
    pin: &polyc_state::feed::AuditSourceCheckpoint,
    resolution: &polyc_state::projection::ProjectionResolution,
) -> Result<(), CoreExecutionError> {
    let partition =
        || polyc_state::id::PartitionId::new(polyc_projection::family::QUERY_AUDIT_SOURCE);
    let Some(manifest) = resolution.current() else {
        return Err(CoreExecutionError::SourceChanged(partition()));
    };
    let polyc_state::feed::SourceEvidence::QueryAudit(observed) = manifest.evidence() else {
        return Err(CoreExecutionError::SourceChanged(partition()));
    };
    if observed.source().incarnation() != pin.source().incarnation() {
        return Err(CoreExecutionError::SourceChanged(partition()));
    }
    if observed.ordinal() < pin.ordinal() {
        return Err(CoreExecutionError::AuthorityNarrowed);
    }
    Ok(())
}

fn scope_contains(current: &QueryScope, original: &QueryScope) -> bool {
    match (current, original) {
        (QueryScope::Fleet, _) => true,
        (QueryScope::Conversations { .. }, QueryScope::Fleet) => false,
        (
            QueryScope::Conversations {
                conversations: current,
                memory: current_memory,
            },
            QueryScope::Conversations {
                conversations: original,
                memory: original_memory,
            },
        ) => {
            let current_conversations = current.iter().collect::<BTreeSet<_>>();
            let conversations_contained = original
                .iter()
                .all(|conversation| current_conversations.contains(conversation));
            // A narrowed owner (persona de-activated mid-stream) or a shrunk
            // participant set is `AuthorityNarrowed` (N-8): every persona the
            // ORIGINAL scope admitted must still be admitted by the CURRENT
            // one. A widened memory scope (N-7) is accepted here — the
            // plan's own file registration stays fixed at bind, so no new
            // partition's rows enter a running stream regardless.
            let current_memory_set = current_memory
                .partitions()
                .into_iter()
                .collect::<BTreeSet<_>>();
            let memory_contained = original_memory
                .partitions()
                .into_iter()
                .all(|persona| current_memory_set.contains(&persona));
            conversations_contained && memory_contained
        }
    }
}

/// Returns the outcome the producer measured.
///
/// The outcome only. The rows and truncation that go with it belong to the
/// latch, which builds the completion from the counts it holds, in the same
/// critical section that records the terminal.
fn producer_outcome(end: &ProducerEnd) -> QueryOutcome {
    match end {
        ProducerEnd::Ended => QueryOutcome::Succeeded,
        ProducerEnd::Cancelled => QueryOutcome::Failed(ErrorClass::Cancelled),
        ProducerEnd::Failed(error) => QueryOutcome::Failed(classify_error(error)),
    }
}

/// Carries the private bounded Arrow stream.
///
/// End-of-stream becomes visible only after the producer durably settles its
/// exact completion. Dropping this value signals cancellation and nothing
/// more. The producer still owns the one-shot permit. It settles under its own
/// terminal budget.
pub(crate) struct CoreResultStream {
    receiver: mpsc::Receiver<CoreStreamFrame>,
    readiness_signal: mpsc::UnboundedSender<()>,
    poll_outstanding: bool,
    terminal_seen: bool,
    cancellation: CancellationToken,
    latch: Arc<TerminalLatch>,
    /// The plan's own output schema, captured before the plan was consumed.
    schema: SchemaRef,
    /// The durable audit's source vector for this execution.
    source: polyc_state::query_audit::SourceSnapshot,
    started: tokio::time::Instant,
}

impl CoreResultStream {
    /// Returns the result schema, available before the first batch.
    pub(crate) const fn schema(&self) -> &SchemaRef {
        &self.schema
    }

    /// Returns the exact source vector this execution read.
    pub(crate) const fn source(&self) -> &polyc_state::query_audit::SourceSnapshot {
        &self.source
    }

    /// Returns what the consumer has actually received so far.
    ///
    /// Read from the latch, not from a producer-side tally. What "received"
    /// means follows the live accounting: batches that left this stream, or —
    /// once a consumer re-frames them — the frames that left that consumer.
    pub(crate) fn delivered(&self) -> DeliveredCounts {
        self.latch.delivered()
    }

    /// Returns how long this execution has run.
    pub(crate) fn elapsed(&self) -> Duration {
        self.started.elapsed()
    }

    /// Hands the delivery count to a consumer that re-frames each batch.
    ///
    /// After this, a batch leaving the stream counts nothing. The consumer
    /// admits what it releases, through [`Self::admit_release`], so the
    /// durable record counts rows a caller actually received rather than rows
    /// this stream handed to a framer.
    pub(crate) fn account_at_consumer(&self) {
        self.latch.account_at_consumer();
    }

    /// Admits one release into the record this query will settle.
    ///
    /// Returns whether it was admitted. A refusal means a terminal has
    /// already been chosen, so the frame must not reach the caller: the
    /// record it would belong to is already fixed without it.
    pub(crate) fn admit_release(&self, rows: u64, bytes: u64) -> bool {
        self.latch.admit_release(rows, bytes)
    }

    /// Settles the terminal the consumer measured at its own admitted bound.
    ///
    /// The consumer stopped because it reached a ceiling the caller asked
    /// for. It then drops this stream to stop the producer, and that drop
    /// must not turn a bounded success into a withdrawal. The stop is always
    /// a truncation: release ended because the next frame did not fit.
    ///
    /// Returns the counts the record carries, or `None` when another party
    /// had already ended the query.
    pub(crate) fn settle_consumer_bound(&self) -> Option<DeliveredCounts> {
        self.latch
            .settle_consumer_bound(&self.source, self.started.elapsed())
    }

    /// Records a failure the consumer measured after the batch left this
    /// stream.
    ///
    /// A consumer that cannot use a delivered batch ends the query, and
    /// dropping this stream would then record a withdrawal. That is the wrong
    /// word: the caller withdrew nothing. The latch keeps a measured failure
    /// over a later cancellation, so reporting here is what makes the durable
    /// terminal say what actually happened.
    /// The record's counts are the admitted releases, which is what the
    /// caller received: a batch that left this stream but could not be framed
    /// reached nobody, and was never admitted.
    ///
    /// Returns whether this failure became the selected terminal. A caller
    /// must not publish its local terminal when another terminal won first.
    pub(crate) fn report_failure(&self, class: ErrorClass) -> bool {
        self.latch.report(
            QueryOutcome::Failed(class),
            self.started.elapsed(),
            &self.source,
        )
    }

    /// Requests one producer batch without dequeuing it.
    ///
    /// A case uses this to hold a batch in front of a producer terminal. It
    /// can then prove that a later framing failure does not replace the
    /// terminal that already became the durable record.
    #[cfg(test)]
    pub(crate) fn request_buffered_batch(&mut self) {
        if !self.poll_outstanding {
            let _ = self.readiness_signal.send(());
            self.poll_outstanding = true;
        }
    }

    /// Returns how many producer frames wait in front of this consumer.
    #[cfg(test)]
    pub(crate) fn buffered_frames(&self) -> usize {
        self.receiver.len()
    }

    /// Returns whether a terminal has won the latch.
    #[cfg(test)]
    pub(crate) fn terminal_selected(&self) -> bool {
        self.latch.is_settled()
    }
}

enum CoreStreamFrame {
    Batch(RecordBatch),
    Failure(CoreExecutionError),
    Complete,
}

impl Stream for CoreResultStream {
    type Item = Result<RecordBatch, CoreExecutionError>;

    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        loop {
            if !self.poll_outstanding {
                // The producer may close readiness while it is durably
                // settling a terminal result. The result channel, not this
                // hint, owns EOF.
                let _ = self.readiness_signal.send(());
                self.poll_outstanding = true;
            }
            let frame = self.receiver.poll_recv(context);
            return match frame {
                Poll::Ready(Some(CoreStreamFrame::Batch(batch))) => {
                    self.poll_outstanding = false;
                    // Offered at the exact point the batch leaves this stream,
                    // through the same admission a frame goes through.
                    let rows = u64::try_from(batch.num_rows()).unwrap_or(u64::MAX);
                    let bytes = u64::try_from(batch.get_array_memory_size()).unwrap_or(u64::MAX);
                    match self.latch.record_batch_delivery(rows, bytes) {
                        // A terminal was already chosen and this batch is not
                        // in the record. Handing it over would give the caller
                        // rows the record does not carry, so it is dropped and
                        // the producer's own terminal is taken instead. The
                        // producer sends nothing after its terminal, so this
                        // skips at most what the channel already buffered.
                        BatchRelease::Refused => continue,
                        BatchRelease::Admitted | BatchRelease::CountedByConsumer => {
                            Poll::Ready(Some(Ok(batch)))
                        }
                    }
                }
                Poll::Ready(Some(CoreStreamFrame::Failure(error))) => {
                    self.poll_outstanding = false;
                    self.terminal_seen = true;
                    Poll::Ready(Some(Err(error)))
                }
                Poll::Ready(Some(CoreStreamFrame::Complete)) => {
                    self.poll_outstanding = false;
                    self.terminal_seen = true;
                    Poll::Ready(None)
                }
                Poll::Ready(None) if self.terminal_seen => Poll::Ready(None),
                Poll::Ready(None) => {
                    self.poll_outstanding = false;
                    self.terminal_seen = true;
                    Poll::Ready(Some(Err(CoreExecutionError::AuditCompletionUnavailable)))
                }
                Poll::Pending => Poll::Pending,
            };
        }
    }
}

impl Drop for CoreResultStream {
    fn drop(&mut self) {
        // Withdraw through the latch first, so the transition is recorded
        // before the producer can observe the cancellation token and report a
        // success the consumer will never see.
        self.latch.cancel();
        self.cancellation.cancel();
    }
}

#[cfg(test)]
mod versioned_liveness_tests {
    use super::{CoreExecutionError, versioned_liveness};
    use polyc_state::{
        command::CommandScope,
        digest::ContentDigest,
        feed::{VersionedCheckpoint, VersionedSource},
        id::{AggregateId, NamespaceId, PartitionId},
        revision::{JournalHead, JournalPosition, PartitionIncarnation},
        versioned::VersionedSourceHead,
    };

    fn scope() -> CommandScope {
        CommandScope::new(
            AggregateId::new("credentials"),
            PartitionId::new("credentials"),
            NamespaceId::new("polychrome"),
        )
    }

    fn pin(lineage: u8, position: u64) -> VersionedCheckpoint {
        VersionedCheckpoint::try_new(
            VersionedSource::new(
                scope(),
                PartitionIncarnation::from_bytes([lineage; PartitionIncarnation::LEN]),
            ),
            JournalPosition::new(position),
            ContentDigest::from_bytes([7; 32]),
        )
        .expect("a non-origin position is a checkpoint")
    }

    fn observed(lineage: u8, head: u64) -> VersionedSourceHead {
        VersionedSourceHead::new(
            PartitionIncarnation::from_bytes([lineage; PartitionIncarnation::LEN]),
            JournalHead::new(JournalPosition::new(head), None),
        )
    }

    /// A pin inside its own lineage, at or below the head, is live.
    #[test]
    fn a_pin_within_its_lineage_is_live() {
        versioned_liveness(&pin(1, 4), &observed(1, 4)).expect("a pin at the head is live");
        versioned_liveness(&pin(1, 4), &observed(1, 9)).expect("a pin behind the head is live");
    }

    /// A replaced log is reported as a changed source, whatever its head.
    ///
    /// The lineage decides first. A wipe that refilled past the pin would
    /// otherwise look live, and one that refilled only partway would report a
    /// narrowed authority. Both would hide that the log is not the one the
    /// generation folded.
    #[test]
    fn a_replaced_lineage_is_a_changed_source_at_any_head() {
        for head in [0, 4, 99] {
            let refusal = versioned_liveness(&pin(1, 4), &observed(2, head))
                .expect_err("a replaced lineage must refuse");
            assert!(
                matches!(refusal, CoreExecutionError::SourceChanged(_)),
                "head {head} reported {refusal:?}, not a changed source"
            );
        }
    }

    /// Inside one lineage, a head below the pin is a narrowed authority.
    #[test]
    fn a_head_below_the_pin_is_a_narrowed_authority() {
        let refusal = versioned_liveness(&pin(1, 8), &observed(1, 7))
            .expect_err("a head below the pin must refuse");
        assert!(matches!(refusal, CoreExecutionError::AuthorityNarrowed));
    }
}

#[cfg(test)]
mod persona_memory_liveness_tests {
    use super::{CoreExecutionError, persona_memory_liveness};
    use polyc_state::{
        digest::ContentDigest,
        feed::PersonaMemoryHistoryCheckpoint,
        persona_memory::journal::{
            MemoryJournalPartition, PersonaMemorySource, PersonaMemorySourceHead,
        },
        revision::{JournalHead, JournalPosition, PartitionIncarnation},
    };

    fn partition() -> MemoryJournalPartition {
        MemoryJournalPartition::parse("persona-1-mem").unwrap()
    }

    fn pin(lineage: u8, position: u64) -> PersonaMemoryHistoryCheckpoint {
        PersonaMemoryHistoryCheckpoint::try_new(
            PersonaMemorySource::new(
                partition(),
                PartitionIncarnation::from_bytes([lineage; PartitionIncarnation::LEN]),
            ),
            JournalPosition::new(position),
            ContentDigest::from_bytes([7; 32]),
        )
        .expect("a non-origin position is a checkpoint")
    }

    fn observed(lineage: u8, head: u64) -> PersonaMemorySourceHead {
        PersonaMemorySourceHead::new(
            PartitionIncarnation::from_bytes([lineage; PartitionIncarnation::LEN]),
            JournalHead::new(JournalPosition::new(head), None),
        )
    }

    /// A pin inside its own lineage, at or below the head, is live.
    #[test]
    fn a_pin_within_its_lineage_is_live() {
        persona_memory_liveness(&pin(1, 4), &observed(1, 4)).expect("a pin at the head is live");
        persona_memory_liveness(&pin(1, 4), &observed(1, 9))
            .expect("a pin behind the head is live");
    }

    /// A rewrite, migrate, or destroy that advanced the incarnation is
    /// reported as a changed source, whatever its head — the same rule
    /// `versioned_liveness` follows, so a rewrite that refilled past the pin
    /// does not look live and one that refilled only partway does not look
    /// merely narrowed.
    #[test]
    fn a_replaced_lineage_is_a_changed_source_at_any_head() {
        for head in [0, 4, 99] {
            let refusal = persona_memory_liveness(&pin(1, 4), &observed(2, head))
                .expect_err("a replaced lineage must refuse");
            assert!(
                matches!(refusal, CoreExecutionError::SourceChanged(_)),
                "head {head} reported {refusal:?}, not a changed source"
            );
        }
    }

    /// Inside one lineage, a head below the pin is a narrowed authority.
    #[test]
    fn a_head_below_the_pin_is_a_narrowed_authority() {
        let refusal = persona_memory_liveness(&pin(1, 8), &observed(1, 7))
            .expect_err("a head below the pin must refuse");
        assert!(matches!(refusal, CoreExecutionError::AuthorityNarrowed));
    }
}

#[cfg(test)]
mod observed_liveness_tests {
    use super::{CoreExecutionError, observed_liveness};
    use polyc_state::{
        deadline::MonotonicInstant,
        digest::ContentDigest,
        feed::ObservedCheckpoint,
        observation::{
            CollectionId, CollectionKind, ObservationHead, ObservationOrdinal, ObservationSource,
            ResourceVersion,
        },
        revision::{JournalPosition, PartitionIncarnation},
    };

    fn source(lineage: u8) -> ObservationSource {
        ObservationSource::new(
            CollectionId::try_new(CollectionKind::Routines, "namespace-a").unwrap(),
            PartitionIncarnation::from_bytes([lineage; PartitionIncarnation::LEN]),
        )
    }

    fn pin(lineage: u8, ordinal: u64, digest: u8) -> ObservedCheckpoint {
        ObservedCheckpoint::try_new(
            source(lineage),
            ObservationOrdinal::new(JournalPosition::new(ordinal)),
            ContentDigest::from_bytes([digest; ContentDigest::LEN]),
        )
        .unwrap()
    }

    fn head(lineage: u8, ordinal: u64, digest: u8) -> ObservationHead {
        ObservationHead::from_parts(
            source(lineage),
            ObservationOrdinal::new(JournalPosition::new(ordinal)),
            ContentDigest::from_bytes([digest; ContentDigest::LEN]),
            ResourceVersion::try_new("resource-1").unwrap(),
            MonotonicInstant::from_nanos(1),
            MonotonicInstant::from_nanos(2),
        )
    }

    #[test]
    fn routine_pin_stays_live_only_in_its_observation_lineage() {
        observed_liveness(&pin(1, 4, 7), &head(1, 4, 7)).unwrap();
        assert!(matches!(
            observed_liveness(&pin(1, 4, 7), &head(1, 5, 8)),
            Err(CoreExecutionError::AuthorityNarrowed)
        ));
        assert!(matches!(
            observed_liveness(&pin(1, 4, 7), &head(2, 5, 8)),
            Err(CoreExecutionError::SourceChanged(_))
        ));
        assert!(matches!(
            observed_liveness(&pin(1, 4, 7), &head(1, 3, 7)),
            Err(CoreExecutionError::AuthorityNarrowed)
        ));
        assert!(matches!(
            observed_liveness(&pin(1, 4, 7), &head(1, 4, 8)),
            Err(CoreExecutionError::AuthorityNarrowed)
        ));
    }
}