polyc-query 2026.9.0

//! 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 polyc_state::{journal::partition_incarnation_from_record, revision::PartitionIncarnation};

use super::{JournalError, PartitionJournal};

/// Returns the one lineage a physical journal names, refusing any other shape.
///
/// Exactly one marker, at any position. Position is deliberately not part of
/// the rule: the State plane's startup sweep gives a partition written before
/// the marker existed its lineage by appending one at the TAIL, because the
/// only writer that can place a marker at position zero is the append-time
/// insert, and that insert needs a partition that already loads. A rule keyed
/// on position zero would read every such partition as unreadable while the
/// State plane serves it perfectly well, which is two planes disagreeing about
/// one durable fact.
///
/// What the rule protects is unchanged and still enforced below: a missing
/// marker, a duplicate, and a malformed frame are each refused rather than
/// synthesized into an identity.
fn canonical_incarnation(
    events: &[(u64, Event)],
) -> Result<Option<PartitionIncarnation>, JournalError> {
    if events.is_empty() {
        return Ok(None);
    }
    let mut observed = None;
    for (_, event) in events {
        let marker = partition_incarnation_from_record(&event.kind, &event.payload)
            .map_err(|error| JournalError::Unreadable(error.to_string()))?;
        if let Some(marker) = marker
            && observed.replace(marker).is_some()
        {
            return Err(JournalError::Unreadable(
                "a physical journal carries exactly one incarnation marker".to_owned(),
            ));
        }
    }
    observed.map_or_else(
        || {
            Err(JournalError::Unreadable(
                "a nonempty physical journal carries an incarnation marker".to_owned(),
            ))
        },
        |incarnation| Ok(Some(incarnation)),
    )
}

/// 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 { .. }
            | polyc_eventlog_host::ListPartitionsError::EntriesExceeded { .. }
            | polyc_eventlog_host::ListPartitionsError::NameBytesExceeded { .. } => {
                JournalError::Unreadable(error.to_string())
            }
        },
        AppendError::Closed
        | AppendError::AppendOutcomeUnknown(_)
        | AppendError::AttestationOutcomeUnknown(_) => JournalError::Unreachable(error.to_string()),
    }
}

#[async_trait::async_trait]
impl PartitionJournal for EventLogHost {
    async fn partition_incarnation(
        &self,
        partition: String,
    ) -> Result<Option<PartitionIncarnation>, JournalError> {
        let events = Self::replay_with_positions(self, partition)
            .await
            .map_err(|error| classify_host_error(&error))?;
        canonical_incarnation(&events)
    }

    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
}

#[cfg(test)]
mod tests {
    use super::*;
    use polyc_state::journal::{INCARNATION_MARKER_KIND, incarnation_marker_payload};

    fn marker(byte: u8) -> Event {
        Event::new(
            INCARNATION_MARKER_KIND,
            incarnation_marker_payload(PartitionIncarnation::from_bytes([byte; 32])),
        )
    }

    #[test]
    fn exact_source_read_refuses_missing_duplicate_or_malformed_markers() {
        assert!(canonical_incarnation(&[(0, Event::new("record", vec![]))]).is_err());
        assert!(canonical_incarnation(&[(0, marker(1)), (1, marker(2))]).is_err());
        assert!(
            canonical_incarnation(&[(0, Event::new(INCARNATION_MARKER_KIND, vec![1; 3]))]).is_err()
        );
        assert_eq!(canonical_incarnation(&[]).unwrap(), None);
        assert_eq!(
            canonical_incarnation(&[(0, marker(7)), (1, Event::new("record", vec![]))]).unwrap(),
            Some(PartitionIncarnation::from_bytes([7; 32]))
        );
    }

    /// A partition the State plane adopted carries its marker at the tail, and
    /// both planes have to name the same lineage for it.
    ///
    /// This case asserted `is_err` until POLY-264. The sweep that repairs a
    /// partition written before the marker existed can only append, so the
    /// marker it adds lands last — and reading that as unreadable here would
    /// have this plane refuse a partition State serves.
    #[test]
    fn a_marker_the_state_plane_appended_at_the_tail_names_the_same_lineage() {
        assert_eq!(
            canonical_incarnation(&[(0, Event::new("record", vec![])), (1, marker(1))]).unwrap(),
            Some(PartitionIncarnation::from_bytes([1; 32])),
            "a tail marker is the shape the State plane's adoption leaves"
        );
        assert!(
            canonical_incarnation(&[
                (0, Event::new("record", vec![])),
                (1, marker(1)),
                (2, marker(2)),
            ])
            .is_err(),
            "two markers stay a refusal wherever they sit"
        );
    }
}