polyc-query 2026.9.0

//! The partition journal this crate reads, named as a capability (#1565,
//! chunk B6).
//!
//! Every read the query plane makes — the sealed funnel's scoped replays, the
//! fleet dashboard's rebuild, the search index's reconcile and its coverage
//! checks — must not name [`polyc_eventlog_host::EventLogHost`] directly: the
//! journal's single writer lives in the State plane, and a read plane
//! pointed at a store nothing writes reads an empty deployment forever.
//! So this module names what those reads actually need, and the Container
//! decides what stands behind it.
//!
//! # What a position means here
//!
//! A position is a **zero-based index into the partition's own caller
//! records**: a partition holding `n` records answers
//! [`PartitionJournal::partition_event_count`] with `n` and reports positions
//! `0..n`. Every range in this module is half-open — `[start, end)` — and every
//! watermark this crate stores is "one past the last record read", which under
//! that numbering equals the count of records covered.
//!
//! That is stated here rather than left to whichever implementation is mounted
//! because the two differ underneath. The event-log host numbers its own
//! tamper-evidence markers into the same sequence; State numbers caller records
//! only, one-based, and filters its markers out of every read. An
//! implementation reconciles its own numbering to this one — it does not leak
//! it — so nothing in this crate's arithmetic depends on which is mounted.
//!
//! [`port_position`] and [`state_position`] are that reconciliation, stated
//! once. Every place a State position crosses into this crate's coordinate —
//! the read adapter the Container mounts, and [`crate::feed::commit_events`] on
//! the way in from the durable commit feed — converts through these two rather
//! than through arithmetic of its own. See #2146 for why: the feed spelled the
//! conversion by omission and the read path spelled it by subtraction, the two
//! sides disagreed by one, and every watermark the search index derived from
//! the feed named a position its own replay could not reach.
//!
//! # Why the errors are two, and only two
//!
//! Because two is what the callers act on. A read that did not complete for a
//! reason unrelated to this partition's bytes ([`JournalError::Unreachable`])
//! is worth another attempt, and one where the bytes themselves are the problem
//! ([`JournalError::Unreadable`]) is not — replaying them again reads the same
//! bytes. The search index's own
//! `UnavailableReason` split is exactly that distinction, and the dashboard and
//! the query funnel both treat any failure as a skip-and-log, so a finer
//! taxonomy here would carry information nothing downstream could use.

// NOT `cfg(test)`, and it cannot be. The two re-exports below are, but the
// module is not, so the `impl PartitionJournal for EventLogHost` inside it
// compiles into every build. 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.
// See `host.rs` for what that leaves reachable and what actually bounds it.
mod host;

#[cfg(test)]
pub(crate) use host::classify_host_error;
#[cfg(test)]
pub(crate) use host::over_host;

use polyc_eventlog::{BoundedReplay, Event};
use polyc_state::revision::{JournalPosition, PartitionIncarnation};

/// Returns the port position one State position occupies.
///
/// State's first durable record is position one — a commit assigns
/// `position.next()` from [`JournalPosition::ORIGIN`] — and this module's
/// positions are zero-based, so the two differ by exactly one.
///
/// Stated here, once, and reached for by name on both sides of the boundary.
/// A caller that instead reads [`JournalPosition::get`] straight through has
/// not converted at all, which is a skew rather than a compile error; naming
/// the conversion is what makes the omission visible at the call site.
#[must_use]
pub const fn port_position(position: JournalPosition) -> u64 {
    position.get().saturating_sub(1)
}

/// Returns the State position one port position names.
///
/// The inverse of [`port_position`] over every position a record can occupy.
#[must_use]
pub const fn state_position(position: u64) -> JournalPosition {
    JournalPosition::new(position.saturating_add(1))
}

/// Why a partition read did not answer.
#[derive(Debug, thiserror::Error)]
pub enum JournalError {
    /// The journal did not answer, and nothing about this partition's bytes
    /// caused it — the authority is unreachable, its answer was lost, or this
    /// process is stopping.
    ///
    /// Worth another attempt: the same read against a healthy journal answers.
    #[error("the partition journal did not answer: {0}")]
    Unreachable(String),
    /// The journal answered, refusing to read what this partition holds — the
    /// records do not decode, their tamper evidence does not hold, or the
    /// storage under them failed.
    ///
    /// Not worth an immediate retry: a second replay reads the same bytes.
    #[error("the partition journal could not read the partition: {0}")]
    Unreadable(String),
}

/// The partition journal the query plane reads.
///
/// One trait rather than one per consumer, because the three consumers overlap
/// almost entirely and a split would make the Container hand out three views of
/// one client for no boundary anybody enforces.
///
/// # Cancellation safety
///
/// Every method a deployment can reach is a read: dropping one mid-await loses
/// an answer and nothing more. The one append this trait declares exists only
/// in this crate's own test builds, and dropping that abandons the call rather
/// than undoing it.
#[async_trait::async_trait]
pub trait PartitionJournal: Send + Sync {
    /// Returns the durable, non-repeating lineage of `partition`.
    ///
    /// `None` means no canonical physical source exists. Missing, malformed, or
    /// duplicate source evidence must never be synthesized into an identity.
    ///
    /// Position is not part of the rule. A partition the State plane adopted —
    /// one written before the lineage marker existed — carries its marker at
    /// the tail, because an append is the only repair available to a sweep that
    /// may not rewrite. Both planes name the lineage the same way (POLY-264).
    ///
    /// # Errors
    ///
    /// Returns the class the exact-source read failed with.
    async fn partition_incarnation(
        &self,
        partition: String,
    ) -> Result<Option<PartitionIncarnation>, JournalError>;

    /// Names every partition the journal holds.
    ///
    /// # Errors
    ///
    /// Returns the class the enumeration failed with.
    async fn list_partitions(&self) -> Result<Vec<String>, JournalError>;

    /// How many caller records `partition` holds.
    ///
    /// A partition that does not exist is `0`, never an error: the query plane
    /// asks this about conversations a caller named, and "never written to" is
    /// an answer rather than a failure.
    ///
    /// # Errors
    ///
    /// Returns the class the read failed with.
    async fn partition_event_count(&self, partition: String) -> Result<u64, JournalError>;

    /// Replays every record in `partition`, each with its position.
    ///
    /// Unbounded, and the one read here that is: the fleet dashboard's boot
    /// rebuild folds a whole conversation and has no partial answer to give.
    ///
    /// # Errors
    ///
    /// Returns the class the replay failed with.
    async fn replay_with_positions(
        &self,
        partition: String,
    ) -> Result<Vec<(u64, Event)>, JournalError>;

    /// Replays `partition` from its start, stopping once the cumulative
    /// payload bytes read exceed `max_bytes`.
    ///
    /// The record that trips the budget is included, and
    /// [`BoundedReplay::budget_exceeded`] says the read stopped early — a
    /// truncated replay must never be mistaken for a complete one.
    ///
    /// # Errors
    ///
    /// Returns the class the replay failed with.
    async fn replay_with_positions_bounded(
        &self,
        partition: String,
        max_bytes: u64,
    ) -> Result<BoundedReplay, JournalError>;

    /// Replays `partition` from `start` under the same byte budget.
    ///
    /// A `start` at or past the partition's end is an empty,
    /// `budget_exceeded: false` reply rather than an error.
    ///
    /// # Errors
    ///
    /// Returns the class the replay failed with.
    async fn replay_from_with_positions_bounded(
        &self,
        partition: String,
        start: u64,
        max_bytes: u64,
    ) -> Result<BoundedReplay, JournalError>;

    /// Replays `[start, end)` of `partition` under the same byte budget.
    ///
    /// An empty or inverted range is an empty, `budget_exceeded: false` reply
    /// rather than an error.
    ///
    /// # Errors
    ///
    /// Returns the class the replay failed with.
    async fn replay_range_with_positions_bounded(
        &self,
        partition: String,
        start: u64,
        end: u64,
        max_bytes: u64,
    ) -> Result<BoundedReplay, JournalError>;

    /// Commits `events` to `partition` as one atomic batch.
    ///
    /// Declared only in this crate's own test builds, and that is the whole
    /// point: the query plane reads. The one append it makes is a test
    /// interleaving a write with a read it is racing, and every production
    /// write to a conversation partition is issued by the Container through the
    /// State journal contract. A port that offered the write unconditionally
    /// would let a read surface grow a second way to commit — which is exactly
    /// the "one writer per data family" property chunk B6 exists to establish.
    ///
    /// # Errors
    ///
    /// Returns the class the commit failed with.
    #[cfg(test)]
    async fn append_batch(&self, partition: String, events: Vec<Event>)
    -> Result<(), JournalError>;

    /// Whether this process is stopping.
    ///
    /// The search index asks before it turns a failed read into a published
    /// `available: false`: a read that failed because everything is shutting
    /// down says nothing about the conversation it was reading, and marking it
    /// unsearchable on the way out would take participation-wide search down
    /// for that conversation until something rebuilt it — with the mark that
    /// would have scheduled the rebuild dying with the process.
    fn is_stopping(&self) -> bool;
}