liminal-server 0.24.9

//! Typed semantic refusals produced while decoding participant log rows, and
//! [`OperationLogError`], moved here from `log.rs` by a pure move under the
//! 500-line lens before DUR-027 R4 adds to it.

use liminal::durability::DurabilityError;

use super::facts::FactsError;
use super::routed_stream::{RoutedStreamError, RoutedStreamRefusal};

/// Why the registry's live and sweeping sets refuse a row (DUR-027 R4).
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub(super) enum RegistryRowRefusal {
    #[error("a retire of an id that is not live")]
    RetireNotLive,
    #[error("a swept of an id that is not sweeping")]
    SweptNotSweeping,
    #[error("a registration of an id that is sweeping")]
    RegisterSweeping,
}

/// Exact typed reason a closed Attached-v3 mode or fenced proof is refused.
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub(super) enum FencedAttachProofRefusal {
    #[error("Ordinary mode carries a request marker")]
    OrdinaryRequestMarker,
    #[error("Superseding mode carries a request marker")]
    SupersedingRequestMarker,
    #[error("Superseding terminal order differs from the Attached order")]
    SupersedingTerminalOrder,
    #[error("request marker differs from the fenced mode marker")]
    RequestMarkerMismatch,
    #[error("DetachedCredentialRecovery bytes are malformed")]
    DetachedCredentialRecoveryMalformed,
    #[error("DetachedCredentialRecovery bytes are not canonical")]
    DetachedCredentialRecoveryNonCanonical,
    #[error("predecessor debt bytes are malformed")]
    PredecessorDebtMalformed,
    #[error("predecessor debt bytes are not canonical")]
    PredecessorDebtNonCanonical,
    #[error("predecessor debt is zero")]
    PredecessorDebtZero,
    #[error("restricted successor bytes are malformed")]
    SuccessorMalformed,
    #[error("restricted successor bytes are not canonical")]
    SuccessorNonCanonical,
    #[error("restricted successor debt is zero")]
    SuccessorDebtZero,
    #[error("physical-compaction successor range is inverted")]
    SuccessorCompactionRange,
    #[error("recovery conversation differs from the Attached request")]
    RecoveryConversationMismatch,
    #[error("recovery participant differs from the Attached request")]
    RecoveryParticipantMismatch,
    #[error("recovery marker differs from the fenced mode")]
    RecoveryMarkerMismatch,
    #[error("recovery prior epoch differs from the fenced mode")]
    RecoveryPriorEpochMismatch,
    #[error("new binding generation is not immediately after the prior epoch")]
    NewBindingGenerationMismatch,
    #[error("cursor-progress conversation differs from recovery")]
    ProgressConversationMismatch,
    #[error("cursor-progress participant differs from recovery")]
    ProgressParticipantMismatch,
    #[error("cursor-progress epoch differs from recovery")]
    ProgressEpochMismatch,
    #[error("cursor-progress marker differs from recovery")]
    ProgressMarkerMismatch,
    #[error("cursor-progress through boundary differs from its marker")]
    ProgressThroughMismatch,
    #[error("marker-delivery participant differs from cursor progress")]
    DeliveryParticipantMismatch,
    #[error("marker-delivery epoch differs from cursor progress")]
    DeliveryEpochMismatch,
    #[error("marker-delivery sequence differs from cursor progress")]
    DeliveryMarkerMismatch,
    #[error("terminal conversation differs from recovery")]
    TerminalConversationMismatch,
    #[error("terminal participant differs from recovery")]
    TerminalParticipantMismatch,
    #[error("terminal epoch differs from recovery")]
    TerminalEpochMismatch,
    #[error("composed terminal kind and cause class disagree")]
    ComposedTerminalKindCause,
    #[error("composed terminal order differs from the enclosing Attached order")]
    ComposedTerminalOrder,
    #[error("composed terminal pending source does not precede Attached")]
    ComposedPendingSourceOrder,
    #[error("recovered reservation source does not precede Attached")]
    ComposedRecoveredSourceOrder,
    #[error("only Pending Died may consume a recovered finalizer reservation")]
    ComposedRecoveredReservationKind,
    #[error("composed terminal pending source row disagrees with its audit")]
    ComposedPendingSourceMismatch,
    #[error("composed terminal recovered reservation row disagrees with its audit")]
    ComposedRecoveredReservationMismatch,
    #[error("open recovered finalizer reservations exceed the signed identity-slot bound")]
    ComposedRecoveredReservationCapacity,
    #[error("composed terminal disagrees with replay binding prestate")]
    ComposedReplayStateMismatch,
}

/// Failure to encode, decode, append, or read one durable log entry.
#[derive(Debug, thiserror::Error)]
pub enum OperationLogError {
    /// The underlying durable store rejected an operation.
    #[error(transparent)]
    Durability(#[from] DurabilityError),
    /// Routed stream structure or framing was refused.
    #[error("routed stream refused: {0}")]
    RoutedStream(#[source] RoutedStreamRefusal),
    /// The bounded synchronous durability bridge rejected a suspending backend.
    #[error(transparent)]
    Bridge(#[from] liminal::durability::bridge::BridgeError),
    /// A durable entry could not be serialized or deserialized.
    #[error("participant production log serialization failed: {0}")]
    Serialization(#[from] serde_json::Error),
    /// A durable entry uses an unsupported schema version.
    #[error("unsupported participant production log schema version {0}")]
    SchemaVersion(u8),
    /// A durable row's retained-charge audit does not have the form its
    /// schema version requires.
    #[error(
        "participant production log row at sequence {sequence} carries the wrong retained-charge audit form for schema version {schema_version}"
    )]
    RetainedChargeForm { sequence: u64, schema_version: u8 },
    /// A durable stream regressed from a later schema back to an earlier one.
    #[error(
        "participant production log schema regressed at sequence {sequence}: previous {previous}, actual {actual}"
    )]
    SchemaVersionTransition {
        sequence: u64,
        previous: u8,
        actual: u8,
    },
    /// The durable stream was not contiguous at the expected sequence.
    #[error("participant production log expected sequence {expected}, found {actual}")]
    Sequence {
        /// Next sequence required by replay.
        expected: u64,
        /// Sequence read from durable storage.
        actual: u64,
    },
    /// The store assigned a different sequence than the optimistic head.
    #[error("participant production log append expected {expected}, got {actual}")]
    AssignedSequence {
        /// Optimistic-concurrency sequence supplied to the store.
        expected: u64,
        /// Assigned sequence returned by the store.
        actual: u64,
    },
    /// A stored numeric field violated a protocol domain (zero generation).
    #[error("participant production log entry carries zero generation")]
    ZeroGeneration,
    /// A durable row the protocol aggregate itself refuses to replay.
    #[error("durable participant row at sequence {sequence} was refused during restore")]
    CorruptRow {
        /// Durable sequence of the refused row.
        sequence: u64,
    },
    /// A prune receipt the checkpoint would stage does not prove its own cut
    /// (DUR-018): the plan is a defect, named before any row goes.
    #[error("prune receipt {key} refused its proof: {source}")]
    PruneReceipt {
        /// The receipt's ledger key inside the conversation's route.
        key: String,
        #[source]
        source: super::prune::PruneProofError,
    },
    /// Frozen v2 marker-bearing Attached bytes cannot supply a fenced proof.
    #[error("v2 Attached row at sequence {sequence} has no durable fenced proof")]
    V2AttachedFencedProofUnavailable { sequence: u64 },
    /// Frozen v2 Attached option fields disagree with replay prestate.
    #[error("v2 Attached row at sequence {sequence} has contradictory mode evidence")]
    V2AttachedModeMismatch { sequence: u64 },
    /// A schema-v3 Attached row carries non-canonical or contradictory fenced proof facts.
    #[error("v3 fenced Attached proof at sequence {sequence} was refused: {reason}")]
    FencedAttachProof {
        sequence: u64,
        reason: FencedAttachProofRefusal,
    },
    /// A retained-charge list is longer than the stored count domain.
    #[error("participant production retained-charge count exceeds u64")]
    RetainedChargeCount,
    /// Reconstructed fate operation disagreed with the complete durable row.
    #[error("reconstructed fate operation drifted from durable sequence {sequence}")]
    FateReplayDrift { sequence: u64 },
    /// The clock could not stamp a row written on its own.
    #[error("participant production log row timestamp: {0}")]
    Clock(#[from] FactsError),
    /// The conversation registry's snapshot was refused by name (DUR-019);
    /// a refused snapshot stops the load and is never walked around.
    #[error("participant conversation registry snapshot refused: {0}")]
    RegistrySnapshot(#[source] super::registry_snapshot::RegistrySnapshotError),
    /// A registry row the live and sweeping sets refuse (DUR-027 R4): on
    /// restore a corrupt registry, on a write a retirement out of order.
    #[error(
        "participant conversation registry row {sequence} for conversation {conversation_id} refused: {refusal}"
    )]
    RegistryRow {
        /// The row's sequence (the head, for a refused write).
        sequence: u64,
        /// The conversation the row names.
        conversation_id: u64,
        /// Why the sets refused it.
        refusal: RegistryRowRefusal,
    },
    /// The conversation's generation key holds the retirement tombstone: it
    /// is retired and is never reopened (DUR-027 R4).
    #[error("participant conversation {conversation_id} is retired")]
    ConversationRetired {
        /// The retired conversation.
        conversation_id: u64,
    },
}

impl From<RoutedStreamError> for OperationLogError {
    fn from(error: RoutedStreamError) -> Self {
        match error {
            RoutedStreamError::Store(original) => Self::Durability(original),
            RoutedStreamError::Refusal(refusal) => Self::RoutedStream(refusal),
        }
    }
}

impl From<super::operation_batch::CheckpointStageError> for OperationLogError {
    fn from(error: super::operation_batch::CheckpointStageError) -> Self {
        use super::operation_batch::CheckpointStageError;
        match error {
            CheckpointStageError::Store(original) => Self::Durability(original),
            CheckpointStageError::Checkpoint(original) => {
                Self::RoutedStream(RoutedStreamRefusal::Checkpoint(Box::new(original)))
            }
            CheckpointStageError::Locator(original) => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointLocator(original))
            }
            CheckpointStageError::BatchRefused => {
                Self::RoutedStream(RoutedStreamRefusal::BatchRefused)
            }
            CheckpointStageError::RouteMismatch => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointRouteMismatch)
            }
            CheckpointStageError::PositionRegression => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointPositionRegression)
            }
            CheckpointStageError::BodyMismatch => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointBodyMismatch)
            }
        }
    }
}

/// Why a log could not name the stream of its generation: the two ways
/// [`OperationLog::stream_key`](super::log::OperationLog::stream_key) fails.
#[derive(Debug, thiserror::Error)]
pub(super) enum StreamKeyRefusal {
    /// The store refused the generation read.
    #[error(transparent)]
    Store(#[from] DurabilityError),
    /// The generation key holds the retirement tombstone (DUR-027 R4).
    #[error("participant conversation {conversation_id} is retired")]
    Retired {
        /// The retired conversation.
        conversation_id: u64,
    },
}

impl From<StreamKeyRefusal> for OperationLogError {
    fn from(refusal: StreamKeyRefusal) -> Self {
        match refusal {
            StreamKeyRefusal::Store(store) => Self::Durability(store),
            StreamKeyRefusal::Retired { conversation_id } => {
                Self::ConversationRetired { conversation_id }
            }
        }
    }
}