liminal-server 0.25.4

//! Typed failures at the server's durable connection-incarnation seam, and the
//! durable-write reach that says whether a failed operation may have put bytes
//! on disk.

use liminal::durability::DurabilityError;
use liminal_protocol::{
    lifecycle::{ConnectionIncarnationAllocatorRestoreError, DurableIncarnationReferencesError},
    wire::ConnectionIncarnation,
};

/// Failure at the server's durable connection-incarnation seam.
#[derive(Debug, thiserror::Error)]
pub(in crate::server) enum IncarnationStreamError {
    /// The durable engine rejected a read, append, or flush.
    #[error(transparent)]
    Durability(#[from] DurabilityError),
    /// A replay page did not continue at the requested stream sequence.
    #[error("incarnation event sequence mismatch: expected {expected}, got {actual}")]
    EventSequence {
        /// Requested next stream sequence.
        expected: u64,
        /// Sequence returned by durable storage.
        actual: u64,
    },
    /// A successful append returned a sequence other than its optimistic head.
    #[error("incarnation append sequence mismatch: expected {expected}, got {actual}")]
    AssignedSequence {
        /// Optimistic stream head supplied to append.
        expected: u64,
        /// Sequence reported by durable storage.
        actual: u64,
    },
    /// The append-only stream cannot represent another event sequence.
    #[error("incarnation event stream sequence exhausted at u64::MAX")]
    StreamSequenceExhausted,
    /// The canonical event is shorter than the selected fixed field.
    #[error("incarnation event truncated: required {required} bytes, got {actual}")]
    EventTruncated {
        /// Minimum complete byte count at the failing field boundary.
        required: usize,
        /// Complete bytes supplied.
        actual: usize,
    },
    /// A stored event has the wrong canonical schema magic.
    #[error("incarnation event magic mismatch")]
    EventMagic,
    /// A stored event uses an unsupported schema version.
    #[error("unsupported incarnation event schema version {0}")]
    EventSchemaVersion(u8),
    /// A stored event uses an unassigned kind tag.
    #[error("unknown incarnation event kind {0}")]
    EventKind(u8),
    /// The declared event-body length differs from the complete stored suffix.
    #[error("incarnation event body length mismatch: declared {declared}, actual {actual}")]
    EventBodyLength {
        /// Canonical header declaration.
        declared: u32,
        /// Supplied bytes after the fixed event header.
        actual: usize,
    },
    /// A v1 event used a kind introduced by the v2 grammar.
    #[error("incarnation event kind {kind} requires schema version 2, got {version}")]
    EventKindSchemaVersion {
        /// Stored schema version.
        version: u8,
        /// Stored event-kind tag.
        kind: u8,
    },
    /// An Open's conversation count cannot be represented by this process.
    #[error("connection-fate conversation count {count} exceeds platform usize")]
    ConversationCountPlatformOverflow {
        /// Canonical u64 conversation count.
        count: u64,
    },
    /// An Open's declared conversation bound cannot be represented by this process.
    #[error("connection-fate conversation bound {bound} exceeds platform usize")]
    ConversationBoundPlatformOverflow {
        /// Canonical u64 conversation bound.
        bound: u64,
    },
    /// An Open contains more conversations than its signed declared bound.
    #[error("connection-fate conversation count {actual} exceeds declared bound {maximum}")]
    ConversationCountExceedsBound {
        /// Complete conversation count.
        actual: usize,
        /// Immutable signed bound stored in this Open.
        maximum: usize,
    },
    /// An Open's count does not select its complete fixed-width suffix.
    #[error(
        "connection-fate Open body length mismatch for {count} conversations: expected {expected}, actual {actual}"
    )]
    OpenConnectionFateBodyLength {
        /// Declared complete-conversation count.
        count: u64,
        /// Exact body bytes selected by the count.
        expected: usize,
        /// Supplied body bytes.
        actual: usize,
    },
    /// An Open's conversation ids are not canonical sorted-set bytes.
    #[error("connection-fate Open conversation ids are not strictly increasing at index {index}")]
    ConversationsNotStrictlyIncreasing {
        /// Index of the duplicate or regressing id.
        index: usize,
    },
    /// An Open carries an unassigned typed fate class.
    #[error("unknown connection-fate class {0}")]
    ConnectionFateClass(u8),
    /// A Complete body is not its one exact u64 Open sequence.
    #[error("connection-fate Complete body length mismatch: expected 8, actual {actual}")]
    CompleteConnectionFateBodyLength {
        /// Supplied body bytes.
        actual: usize,
    },
    /// Two Allocate inputs in one Startup generation persisted different bounds.
    #[error(
        "startup generation {startup_sequence} allocation {stored_sequence} changed declared reference bound from {expected} to {actual}"
    )]
    GenerationReferenceBoundConflict {
        startup_sequence: u64,
        expected: usize,
        actual: usize,
        stored_sequence: u64,
    },
    /// An Open names no retained Allocate output in its enclosing generation.
    #[error(
        "connection-fate Open {open_sequence} names unknown allocation {connection_incarnation:?}"
    )]
    UnknownConnectionAllocation {
        open_sequence: u64,
        connection_incarnation: ConnectionIncarnation,
    },
    /// Simultaneous Opens resolved to different Startup generations.
    #[error(
        "connection-fate Open {open_sequence} resolves to startup generation {actual_startup_sequence}, but unmatched Opens belong to {expected_startup_sequence}"
    )]
    CrossGenerationOpen {
        open_sequence: u64,
        expected_startup_sequence: u64,
        actual_startup_sequence: u64,
    },
    /// One connection has two simultaneously unmatched Opens.
    #[error(
        "connection-fate Open {open_sequence} duplicates connection from unmatched Open {existing_open_sequence}"
    )]
    DuplicateConnectionOpen {
        open_sequence: u64,
        existing_open_sequence: u64,
    },
    /// A Complete names no currently unmatched Open.
    #[error("connection-fate Complete {complete_sequence} names absent Open {open_sequence}")]
    CompleteForAbsentOpen {
        complete_sequence: u64,
        open_sequence: u64,
    },
    /// An unmatched set exceeds its selected generation's persisted bound.
    #[error(
        "startup generation {startup_sequence} connection-fate active count {actual} exceeds persisted bound {bound} at Open {open_sequence}"
    )]
    HistoricalConnectionFateBoundExceeded {
        startup_sequence: u64,
        bound: usize,
        actual: usize,
        open_sequence: u64,
    },
    /// Recovery tried to append a Complete for no replay-retained Open.
    #[error("connection-fate recovery names absent Open {open_sequence}")]
    RecoveryCompleteForAbsentOpen { open_sequence: u64 },
    /// Startup recovery cannot finish while durable Opens remain unmatched.
    #[error("connection-fate recovery still has {count} unmatched Opens")]
    RecoveryIncomplete { count: usize },
    /// A startup event carried a body even though its canonical body is empty.
    #[error("incarnation startup event carried {actual} body bytes")]
    StartupBodyLength {
        /// Nonzero body byte count.
        actual: usize,
    },
    /// An allocate event's count cannot be represented by this process.
    #[error("incarnation allocation reference count {count} exceeds platform usize")]
    ReferenceCountPlatformOverflow {
        /// Canonical u64 reference count.
        count: u64,
    },
    /// An allocate event's immutable reference bound cannot be represented by this process.
    #[error("incarnation allocation reference bound {bound} exceeds platform usize")]
    ReferenceBoundPlatformOverflow {
        /// Canonical u64 reference bound.
        bound: u64,
    },
    /// A stored allocation declares more references than its immutable stored bound.
    #[error("incarnation allocation reference count {actual} exceeds stored bound {maximum}")]
    StoredReferenceCountExceedsBound {
        /// Stored complete-reference count.
        actual: usize,
        /// Immutable complete-reference bound encoded in this event.
        maximum: usize,
    },
    /// An allocate event's count does not select its complete fixed-width suffix.
    #[error(
        "incarnation allocation body length mismatch for {count} references: expected {expected}, actual {actual}"
    )]
    AllocateBodyLength {
        /// Declared complete-reference count.
        count: u64,
        /// Exact body bytes selected by the count.
        expected: usize,
        /// Supplied body bytes.
        actual: usize,
    },
    /// Event framing arithmetic or its u32 body-length conversion overflowed.
    #[error("incarnation event encoded length overflow")]
    EventLengthOverflow,
    /// A durable allocation appeared before the stream's first startup.
    #[error("incarnation allocation event {stored_sequence} precedes first startup")]
    AllocateBeforeStartup {
        /// Durable position carrying the illegal event.
        stored_sequence: u64,
    },
    /// A durable startup event asks the protocol to increment an exhausted server counter.
    #[error("incarnation startup event {stored_sequence} follows server exhaustion")]
    StartupAfterServerExhaustion {
        /// Durable position carrying the illegal event.
        stored_sequence: u64,
    },
    /// A durable allocation event asks the protocol to allocate after terminal exhaustion.
    #[error("incarnation allocation event {stored_sequence} follows ordinal exhaustion")]
    AllocateAfterOrdinalExhaustion {
        /// Durable position carrying the illegal event.
        stored_sequence: u64,
    },
    /// Replay was requested as started but its event history had no startup.
    #[error("incarnation event history has no startup")]
    MissingStartup,
    /// A resume replay found the stream under a different server incarnation.
    #[error(
        "incarnation stream advanced from server incarnation {expected} to {actual}: another process owns it"
    )]
    ResumeServerIncarnationMoved {
        /// Server incarnation this process durably started under.
        expected: u64,
        /// Server incarnation the durable stream currently replays to.
        actual: u64,
    },
    /// An epoch stream replayed to an incarnation other than the one its key names.
    ///
    /// The refusal that keeps a regressing scalar header unrepresentable across
    /// the rotation seam: an epoch's resume seed is read off its OWN key and
    /// advanced by the protocol's startup transition, so a replay whose result
    /// disagrees with that key is history the key does not describe — a forged,
    /// stale, or doubled epoch — and is refused rather than adopted.
    #[error(
        "incarnation epoch {epoch_server_incarnation} replayed to server incarnation {replayed_server_incarnation}"
    )]
    EpochHeaderMismatch {
        /// Server incarnation named by the epoch stream key.
        epoch_server_incarnation: u64,
        /// Server incarnation the epoch's canonical inputs actually produce.
        replayed_server_incarnation: u64,
    },
    /// Test-only resume refuses to discard unmatched durable work.
    #[cfg(test)]
    #[error("incarnation event history retains {count} unmatched connection-fate Opens")]
    UnmatchedConnectionFates { count: usize },
    /// The protocol crate rejected a restored allocator genesis or test seed.
    #[error("protocol rejected incarnation allocator header: {0:?}")]
    AllocatorRestore(ConnectionIncarnationAllocatorRestoreError),
    /// The protocol crate rejected an over-bound live complete reference set.
    #[error("protocol rejected durable incarnation references: {0:?}")]
    DurableReferences(DurableIncarnationReferencesError),
}

/// How far a failed incarnation-stream operation got toward durable bytes.
///
/// This is the discriminator the admission authority needs and could not
/// previously see. Every operation on a [`super::StartedIncarnationStream`] is a
/// validate-encode-decide prologue followed by exactly one
/// [`super::append_and_flush`], so "did we reach the store?" is a property of
/// WHERE the operation stopped, not of which error variant it produced — and
/// only the stream itself knows where it stopped. Sniffing the variant at the
/// call site would be a guess: [`IncarnationStreamError::StreamSequenceExhausted`],
/// for one, is raised both by the pre-append `checked_add` in
/// [`super::append_and_flush`] and by replay arithmetic.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(in crate::server) enum DurableWriteReach {
    /// No append was issued. Generation validation, reference bounding, event
    /// encoding, allocator restore, unmatched-set validation, or the stream's
    /// own sequence arithmetic failed first, so the durable stream is
    /// byte-identical to what it was before the call and this handle's header,
    /// sequence, generation and unmatched set are all still exact. The operation
    /// failed; the AUTHORITY did not.
    NotAttempted,
    /// An append or a flush was issued and its durable outcome is UNKNOWN. The
    /// bytes may or may not be on disk, so this handle's next sequence, header
    /// and generation may all be stale and nothing may be appended through it
    /// again until ground truth is re-established by reading the store back.
    Ambiguous,
}

/// A failed incarnation-stream operation carrying its durable-write reach.
///
/// `Display` forwards to the underlying error verbatim so operator-facing
/// messages — which are already in field logs — do not change shape.
#[derive(Debug, thiserror::Error)]
#[error("{error}")]
pub(in crate::server) struct IncarnationOperationError {
    /// Whether durable bytes may have been written before the failure.
    pub(in crate::server) reach: DurableWriteReach,
    /// The typed underlying failure.
    pub(in crate::server) error: IncarnationStreamError,
}

impl IncarnationOperationError {
    /// Marks a failure raised before any append was issued.
    pub(super) const fn not_attempted(error: IncarnationStreamError) -> Self {
        Self {
            reach: DurableWriteReach::NotAttempted,
            error,
        }
    }

    /// Marks a failure raised after an append or flush was issued.
    pub(super) const fn ambiguous(error: IncarnationStreamError) -> Self {
        Self {
            reach: DurableWriteReach::Ambiguous,
            error,
        }
    }
}

impl From<IncarnationStreamError> for IncarnationOperationError {
    /// Every `?` in a prologue lands here, which is why the prologue is the
    /// default: an operation that has not yet called [`super::append_and_flush`]
    /// cannot have written anything, and [`super::append_and_flush`] never
    /// returns a bare [`IncarnationStreamError`] for this conversion to capture.
    fn from(error: IncarnationStreamError) -> Self {
        Self::not_attempted(error)
    }
}

impl From<ConnectionIncarnationAllocatorRestoreError> for IncarnationOperationError {
    fn from(error: ConnectionIncarnationAllocatorRestoreError) -> Self {
        Self::not_attempted(error.into())
    }
}

impl From<DurableIncarnationReferencesError> for IncarnationOperationError {
    fn from(error: DurableIncarnationReferencesError) -> Self {
        Self::not_attempted(error.into())
    }
}

impl From<IncarnationOperationError> for IncarnationStreamError {
    /// Drops the reach for the startup-time callers, which have no authority to
    /// keep alive: a failed startup fails server construction outright.
    fn from(failure: IncarnationOperationError) -> Self {
        failure.error
    }
}

impl From<ConnectionIncarnationAllocatorRestoreError> for IncarnationStreamError {
    fn from(error: ConnectionIncarnationAllocatorRestoreError) -> Self {
        Self::AllocatorRestore(error)
    }
}

impl From<DurableIncarnationReferencesError> for IncarnationStreamError {
    fn from(error: DurableIncarnationReferencesError) -> Self {
        Self::DurableReferences(error)
    }
}