polyc-query 2026.8.3

//! The event-log host as a partition journal, for this crate's own fixtures.
//!
//! # Who still reaches it
//!
//! Nobody a deployment composes. The control plane commits and reads the
//! conversation journal through the State plane (#1565, chunk B6) and seeds a
//! test partition through its own State fixture; this crate's projections take
//! a [`PartitionJournal`] and never build one. What is left is this crate's OWN
//! suite — the authority reads, the dashboard rollup, and the search index each
//! drive a real host in a scratch directory. It retires with this crate's own
//! host, in the chunk that moves the read plane onto a State client (EXC-25,
//! chunk F0).
//!
//! # What is gated, and what is not
//!
//! `over_host` and `classify_host_error` are `cfg(test)`, and the parent's
//! re-exports of them are too. This MODULE is not, so the
//! `impl PartitionJournal for EventLogHost` below compiles into every build,
//! shipped ones included. That is deliberate rather than an oversight: a
//! `cfg(test)` here is active only for this crate's own suite, and the control
//! plane's `QueryEngine::for_test` coerces a host into a journal from its own
//! test build, where this crate's gate is off. Gating the module breaks that
//! compilation outright.
//!
//! So a trait impl, which needs no call site to be reachable, is what actually
//! ships: any crate holding both types can coerce a host into a journal without
//! naming anything here. What bounds that is not this module's visibility. It
//! is that nothing a deployment composes constructs an [`EventLogHost`] to read
//! through, and, for the one plane where that would matter,
//! `scripts/check_journal_authority.py` — which fails on a control-plane source
//! that opens a host at all.
//!
//! # It reports the host's own numbering, not the contract's
//!
//! The host writes one tamper-evidence marker per commit *into the partition*,
//! so its positions are not dense over caller records and its count is inflated
//! by the markers. The module doc's numbering describes what the deployment
//! reads, and this implementation does not restate the host's answers to match
//! it — a fixture that seeds through the host and asserts against the host is
//! self-consistent, and a translation layer here would quietly change what
//! those fixtures mean.

use polyc_eventlog::{BoundedReplay, Event};
use polyc_eventlog_host::{AppendError, EventLogHost};

use super::{JournalError, PartitionJournal};

/// Returns the class a host failure falls into.
///
/// The split is the one the search index already made: bytes that will not read
/// are [`JournalError::Unreadable`], and everything else is worth another
/// attempt. `Log` is the ambiguous one — the journal collapses codec, checksum,
/// and runtime failures into one opaque variant — and it lands on the
/// deterministic side because that is the side that cannot livelock.
pub(crate) fn classify_host_error(error: &AppendError) -> JournalError {
    match error {
        // `PartitionName` and a CORRUPT listing are deterministic: retrying
        // encodes the same name and reads the same directory, so neither can
        // livelock its way to a different answer.
        AppendError::Verify(_)
        | AppendError::Log(_)
        | AppendError::PayloadTooLarge { .. }
        | AppendError::PartitionName(_) => JournalError::Unreadable(error.to_string()),
        AppendError::Listing(listing) => match listing {
            // The volume would not answer. That is the transient half, and it
            // belongs with the other retryable faults rather than with the
            // corruption the same error type also carries.
            polyc_eventlog_host::ListPartitionsError::Storage(_) => {
                JournalError::Unreachable(error.to_string())
            }
            polyc_eventlog_host::ListPartitionsError::Corrupt { .. } => {
                JournalError::Unreadable(error.to_string())
            }
        },
        AppendError::Closed => JournalError::Unreachable(error.to_string()),
    }
}

#[async_trait::async_trait]
impl PartitionJournal for EventLogHost {
    async fn list_partitions(&self) -> Result<Vec<String>, JournalError> {
        Self::list_partitions(self)
            .await
            .map_err(|e| classify_host_error(&e))
    }

    async fn partition_event_count(&self, partition: String) -> Result<u64, JournalError> {
        Self::partition_event_count(self, partition)
            .await
            .map_err(|e| classify_host_error(&e))
    }

    async fn replay_with_positions(
        &self,
        partition: String,
    ) -> Result<Vec<(u64, Event)>, JournalError> {
        Self::replay_with_positions(self, partition)
            .await
            .map_err(|e| classify_host_error(&e))
    }

    async fn replay_with_positions_bounded(
        &self,
        partition: String,
        max_bytes: u64,
    ) -> Result<BoundedReplay, JournalError> {
        Self::replay_with_positions_bounded(self, partition, max_bytes)
            .await
            .map_err(|e| classify_host_error(&e))
    }

    async fn replay_from_with_positions_bounded(
        &self,
        partition: String,
        start: u64,
        max_bytes: u64,
    ) -> Result<BoundedReplay, JournalError> {
        Self::replay_from_with_positions_bounded(self, partition, start, max_bytes)
            .await
            .map_err(|e| classify_host_error(&e))
    }

    async fn replay_range_with_positions_bounded(
        &self,
        partition: String,
        start: u64,
        end: u64,
        max_bytes: u64,
    ) -> Result<BoundedReplay, JournalError> {
        Self::replay_range_with_positions_bounded(self, partition, start, end, max_bytes)
            .await
            .map_err(|e| classify_host_error(&e))
    }

    #[cfg(test)]
    async fn append_batch(
        &self,
        partition: String,
        events: Vec<Event>,
    ) -> Result<(), JournalError> {
        Self::append_batch(self, partition, events)
            .await
            .map(|_| ())
            .map_err(|e| classify_host_error(&e))
    }

    fn is_stopping(&self) -> bool {
        Self::is_shutting_down(self)
    }
}

/// Returns `host` as a partition journal.
///
/// A named call rather than an `as` cast at each fixture, so this crate's own
/// suite has one spelling of the coercion and a reader can find every place
/// that takes it.
#[cfg(test)]
#[must_use]
pub(crate) fn over_host(
    host: std::sync::Arc<EventLogHost>,
) -> std::sync::Arc<dyn PartitionJournal> {
    host
}