liminal-server 0.24.5

//! Checkpoint framing for a bounded live owner and fixed-width history positions.
//!
//! The live owner supplies its canonical codec and joint validation. Historical
//! authority is addressed by the qualified index schema's root, never enumerated
//! by this envelope.

use super::restore_checkpoint_codec;

/// One-way operation schema phase at the checkpoint boundary.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CheckpointSchema {
    /// Legacy rows before the first newer operation.
    V2Prefix,
    /// Rows after the schema-three boundary.
    V3Suffix,
    /// Rows after the schema-four boundary.
    V4Suffix,
}

/// Fixed-width replay positions, independent of the amount of prior history.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct CheckpointPositions {
    /// Permanent identity checked against the durable locator.
    pub conversation_id: u64,
    /// Last operation whose effects the checkpoint includes.
    pub covered_log_sequence: u64,
    /// Next protocol event ordinal.
    pub next_event_ordinal: u64,
    /// Exclusive operation-log head after all covered rows.
    pub next_log_sequence: u64,
    /// Next unallocated delivery sequence.
    pub next_seq: u64,
    /// Next unallocated transaction order.
    pub next_order: u64,
    /// Next unallocated permanent participant identity.
    pub next_participant: u64,
    /// Exclusive extension-log head.
    pub extension_sequence: u64,
    /// Highest delivery sequence retained by the outbox.
    pub highest_delivery_seq: u64,
    /// Durable hard-observer progress.
    pub observer_progress: u64,
    /// Exclusive base-source cursor in the audited merge.
    pub merged_base_sequence: u64,
    /// Exclusive extension-source cursor in the audited merge.
    pub merged_extension_sequence: u64,
    /// Schema phase carried into the audited tail.
    pub schema: CheckpointSchema,
}

impl CheckpointPositions {
    pub(super) fn validate(self) -> Result<(), CheckpointError> {
        let expected_next =
            self.covered_log_sequence
                .checked_add(1)
                .ok_or(CheckpointError::Position {
                    field: "covered log sequence",
                })?;
        if self.next_log_sequence != expected_next {
            return Err(CheckpointError::Position {
                field: "next log sequence",
            });
        }
        if self.highest_delivery_seq >= self.next_seq {
            return Err(CheckpointError::Position {
                field: "next delivery sequence",
            });
        }
        if self.observer_progress > self.highest_delivery_seq {
            return Err(CheckpointError::Position {
                field: "observer progress",
            });
        }
        if self.merged_base_sequence > self.next_log_sequence {
            return Err(CheckpointError::Position {
                field: "merged base sequence",
            });
        }
        if self.merged_extension_sequence > self.extension_sequence {
            return Err(CheckpointError::Position {
                field: "merged extension sequence",
            });
        }
        Ok(())
    }
}

/// The complete bounded live owner's codec and cross-validation boundary.
///
/// Implementations carry protocol and server authority together, including open
/// audit reservations. History-carried fields belong to the authenticated root.
/// Neither a protocol-only capsule nor an unvalidated byte vector is an owner.
pub trait CheckpointLiveState: Sized {
    /// Validated current configuration and durable history lookup authority.
    type RestoreContext;

    /// Encodes all live authority without cloning its owner.
    ///
    /// # Errors
    ///
    /// Returns the owning codec's named failure if capture is not stable or a
    /// field cannot be represented canonically.
    fn encode_live(&self) -> Result<Vec<u8>, CheckpointError>;

    /// Decodes all live authority and validates its coupled protocol values.
    ///
    /// # Errors
    ///
    /// Returns a named owner failure for invalid or noncanonical authority.
    fn decode_live(
        bytes: &[u8],
        positions: &CheckpointPositions,
        history_root: &[u8],
        context: &Self::RestoreContext,
    ) -> Result<Self, CheckpointError>;

    /// Checks the complete owner against its checkpoint positions.
    ///
    /// # Errors
    ///
    /// Returns a named owner failure for identity, cursor, generation, charge,
    /// frontier, debt, reservation or stable-boundary inconsistency.
    fn validate_live(
        &self,
        positions: &CheckpointPositions,
        history_root: &[u8],
    ) -> Result<(), CheckpointError>;
}

/// One checkpoint with a live owner and one authenticated history-index root.
///
/// The root width is supplied by the qualified root schema. This type has no
/// assumed digest width and cannot be cloned through the checkpoint interface.
#[derive(Debug, PartialEq, Eq)]
pub struct RestoreCheckpoint<L, const ROOT_BYTES: usize> {
    positions: CheckpointPositions,
    history_root: [u8; ROOT_BYTES],
    live: L,
}

impl<L: CheckpointLiveState, const ROOT_BYTES: usize> RestoreCheckpoint<L, ROOT_BYTES> {
    /// Takes a complete live owner after validating the stable boundary.
    ///
    /// # Errors
    ///
    /// Refuses an empty or unrepresentable root width, invalid positions or an
    /// owner that fails joint validation. It performs no store publication.
    pub fn new(
        positions: CheckpointPositions,
        history_root: [u8; ROOT_BYTES],
        live: L,
    ) -> Result<Self, CheckpointError> {
        if ROOT_BYTES == 0 || u16::try_from(ROOT_BYTES).is_err() {
            return Err(CheckpointError::RootWidth {
                expected: ROOT_BYTES,
                actual: ROOT_BYTES,
            });
        }
        positions.validate()?;
        live.validate_live(&positions, &history_root)?;
        Ok(Self {
            positions,
            history_root,
            live,
        })
    }

    /// Encodes the ordered positions/root and live sections with an integrity digest.
    ///
    /// # Errors
    ///
    /// Refuses invalid owner state or lengths outside the canonical framing.
    pub fn encode_canonical(&self) -> Result<Vec<u8>, CheckpointError> {
        self.positions.validate()?;
        self.live()
            .validate_live(&self.positions, &self.history_root)?;
        let live = self.live().encode_live()?;
        restore_checkpoint_codec::encode(self.positions, &self.history_root, &live)
    }

    /// Decodes one exact envelope, then delegates joint live-owner validation.
    ///
    /// # Errors
    ///
    /// Refuses malformed framing, unknown versions, wrong identity/root width,
    /// integrity failure, invalid positions or an invalid live owner.
    pub fn decode_canonical(
        input: &[u8],
        expected_conversation_id: u64,
        context: &L::RestoreContext,
    ) -> Result<Self, CheckpointError> {
        Self::decode_canonical_with_association(input, expected_conversation_id, context, |_, _| {
            Ok(())
        })
    }

    /// Decodes framing and proves locator/root association before decoding the owner.
    ///
    /// The callback receives only authenticated positions and the fixed-size
    /// root. Its failure stops decoding before the owning codec can perform a
    /// history lookup. The context must provide the qualified history authority
    /// against which the callback checks that association.
    ///
    /// # Errors
    ///
    /// Propagates framing and association failures before any owner operation,
    /// then applies the same canonical and joint owner validation as decode.
    pub fn decode_canonical_with_association(
        input: &[u8],
        expected_conversation_id: u64,
        context: &L::RestoreContext,
        validate_association: impl FnOnce(
            &CheckpointPositions,
            &[u8; ROOT_BYTES],
        ) -> Result<(), CheckpointError>,
    ) -> Result<Self, CheckpointError> {
        let (positions, history_root, live_bytes) =
            restore_checkpoint_codec::decode(input, expected_conversation_id)?;
        validate_association(&positions, &history_root)?;
        Self::decode_owner(positions, history_root, live_bytes, context)
    }

    fn decode_owner(
        positions: CheckpointPositions,
        history_root: [u8; ROOT_BYTES],
        live_bytes: &[u8],
        context: &L::RestoreContext,
    ) -> Result<Self, CheckpointError> {
        let live = L::decode_live(live_bytes, &positions, &history_root, context)?;
        if live.encode_live()?.as_slice() != live_bytes {
            return Err(CheckpointError::LiveEncoding);
        }
        Self::new(positions, history_root, live)
    }

    /// Borrows the complete live owner.
    #[must_use]
    pub const fn live(&self) -> &L {
        &self.live
    }

    /// Reads the fixed-width replay positions.
    #[must_use]
    pub const fn positions(&self) -> CheckpointPositions {
        self.positions
    }

    /// Borrows the history root without listing its entries.
    #[must_use]
    pub const fn history_root(&self) -> &[u8; ROOT_BYTES] {
        &self.history_root
    }

    /// Moves the validated parts into the sole restore/commit owner.
    #[must_use]
    pub fn into_parts(self) -> (CheckpointPositions, [u8; ROOT_BYTES], L) {
        (self.positions, self.history_root, self.live)
    }
}

/// Named checkpoint envelope or owning-codec failure.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum CheckpointError {
    /// The canonical length cannot be represented.
    #[error("checkpoint length overflow")]
    LengthOverflow,
    /// A required byte range is absent.
    #[error("checkpoint truncated: need {required} bytes, have {available}")]
    Truncated {
        /// Exclusive byte offset needed.
        required: usize,
        /// Actual supplied byte count.
        available: usize,
    },
    /// The envelope magic differs.
    #[error("checkpoint magic differs")]
    InvalidMagic,
    /// The codec version is not supported.
    #[error("checkpoint codec {major}.{minor} is unsupported")]
    UnsupportedCodec {
        /// Declared major version.
        major: u16,
        /// Declared minor version.
        minor: u16,
    },
    /// Ordered sections or exact lengths differ.
    #[error("checkpoint section {tag} is noncanonical")]
    Section {
        /// Expected section tag.
        tag: u16,
    },
    /// The root schema width differs.
    #[error("checkpoint root width differs: expected {expected}, found {actual}")]
    RootWidth {
        /// Width required by the qualified root schema.
        expected: usize,
        /// Width supplied by the envelope.
        actual: usize,
    },
    /// The durable locator and checkpoint name different owners.
    #[error("checkpoint conversation differs: expected {expected}, found {actual}")]
    ForeignConversation {
        /// Conversation named by the durable locator.
        expected: u64,
        /// Conversation named by the checkpoint.
        actual: u64,
    },
    /// Position arithmetic or ordering is inconsistent.
    #[error("checkpoint position is invalid: {field}")]
    Position {
        /// Position that failed validation.
        field: &'static str,
    },
    /// A recovered slot lacks the native commit's validated predecessor source.
    #[error(
        "checkpoint conversation {conversation_id} slot {participant_id} Attached source {attached_source_sequence} requires its original validated FencedAttachCommit"
    )]
    MissingSlotFencedAttachCommit {
        /// Owning conversation.
        conversation_id: u64,
        /// Exact slot identity.
        participant_id: u64,
        /// Attached operation that produced the move-only token.
        attached_source_sequence: u64,
    },
    /// A recovered death intent lacks the native commit's predecessor source.
    #[error(
        "checkpoint conversation {conversation_id} pending fate {participant_id} Attached source {attached_source_sequence} Died source {died_source_sequence} requires its original validated FencedAttachCommit"
    )]
    MissingPendingFencedAttachCommit {
        /// Owning conversation.
        conversation_id: u64,
        /// Participant whose sole token belongs to this death intent.
        participant_id: u64,
        /// Attached operation that produced the token.
        attached_source_sequence: u64,
        /// Durable death operation owning the pending intent.
        died_source_sequence: u64,
    },
    /// Witness coordinates cannot replace the protocol-produced projection.
    #[error(
        "checkpoint conversation {conversation_id} witness participant {participant_id:?} base {base_log_head:?} extension {extension_sequence:?} progress {witness_progress:?} requires its original ObserverProgressProjection source"
    )]
    MissingObserverProgressProjectionSource {
        /// Owning conversation.
        conversation_id: u64,
        /// Participant of the retained witness or occurrence, when still held.
        participant_id: Option<u64>,
        /// Exact merged base coordinate, when still held.
        base_log_head: Option<u64>,
        /// Exact merged extension coordinate, when still held.
        extension_sequence: Option<u64>,
        /// Progress of an extant witness; absent after its pruning.
        witness_progress: Option<u64>,
        /// Actual prune boundary, if the live bookkeeping is available.
        pruned_through: Option<u64>,
        /// Size of a refused encoded section, if supplied for restoration.
        encoded_section_bytes: Option<usize>,
    },
    /// The operation schema phase is unknown.
    #[error("checkpoint operation schema {tag} is unsupported")]
    Schema {
        /// Unrecognised schema phase.
        tag: u8,
    },
    /// Covered envelope bytes do not match their integrity digest.
    #[error("checkpoint integrity digest differs")]
    Integrity,
    /// The authenticated envelope does not match its locator/history authority.
    #[error("checkpoint association differs: {field}")]
    Association {
        /// Associated position or root that differs.
        field: &'static str,
    },
    /// The locator or its content-addressed body refused association.
    #[error("checkpoint locator refused: {source}")]
    Locator {
        /// Exact locator/body validation cause.
        #[from]
        source: super::restore_checkpoint_locator::CheckpointLocatorError,
    },
    /// The live codec did not reproduce the supplied canonical section.
    #[error("checkpoint live section is noncanonical")]
    LiveEncoding,
    /// The complete live owner's codec or joint validation refused its authority.
    #[error("checkpoint live owner refused: {source}")]
    LiveState {
        /// Typed owning-codec or invariant failure, retained as its source.
        source: Box<dyn std::error::Error + Send + Sync>,
    },
}