haematite-cli 0.21.0

//! Lossless typed projections retain every admitted outcome and its causes.

use haematite::{BatchOutcome, BatchReceipt, DatabaseError, ShardError};
use serde::{Deserialize, Serialize};

use super::receipt_fragments::{ReceiptCodecError, encode_hex};

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireReceipt {
    pub outcomes: Vec<WireJobOutcome>,
    pub submit_error: Option<WireCause>,
    pub first_error: Option<WireCause>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireJobOutcome {
    pub job_index: u64,
    pub shard: u64,
    pub outcome: WireOutcome,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", deny_unknown_fields)]
pub enum WireOutcome {
    Committed {
        root: [u8; 32],
        callback_failure: Option<WireCause>,
    },
    SequenceRead {
        sequences: Vec<WireSequence>,
    },
    RangeRead {
        sequences: Vec<WireSequence>,
        hit_limit: bool,
        invalid_outside_range: u64,
    },
    Failed {
        cause: WireCause,
    },
    Unknown {
        cause: WireCause,
    },
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireSequence {
    #[serde(deserialize_with = "read_key_hex")]
    pub key_hex: String,
    pub sequence: u64,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireCause {
    pub kind: DatabaseErrorKind,
    pub message: String,
    pub shard: Option<WireShardCause>,
    pub callback: Option<WireCallback>,
    pub batch_receipt: Option<Box<WireReceipt>>,
    pub first_error: Option<Box<Self>>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireShardCause {
    pub kind: ShardErrorKind,
    pub message: String,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum ShardErrorKind {
    QueueFull,
    RequestSizeOverflow,
    ActorUnavailable,
    ReplyDisconnected,
    OutcomeUnknown,
    Wal,
    Tree,
    Store,
    Spawn,
    ReclaimClose,
    ShutdownStorage,
    SequenceConflict,
    ReservedSequenceKey,
    EmptyAppend,
    LeafRecordCorrupt,
    LeafLogCorrupt,
    LeafLogNotAdmitted,
    LeafLogTruncate,
    LeafLogStreamDelete,
    LeafLogTtlWrite,
    LeafLogMergeAdopt,
    PinWouldGoBack,
    PinConflict,
    PinBeyondExtent,
    CasMismatch,
    CasHashMismatch,
    Fenced,
    WriteDuringRootAdvanceEmission,
}

impl ShardErrorKind {
    const fn project(error: &ShardError) -> Self {
        match error {
            ShardError::QueueFull { .. } => Self::QueueFull,
            ShardError::RequestSizeOverflow { .. } => Self::RequestSizeOverflow,
            ShardError::ActorUnavailable { .. } => Self::ActorUnavailable,
            ShardError::ReplyDisconnected { .. } => Self::ReplyDisconnected,
            ShardError::OutcomeUnknown { .. } => Self::OutcomeUnknown,
            ShardError::Wal(_) => Self::Wal,
            ShardError::Tree(_) => Self::Tree,
            ShardError::Store(_) => Self::Store,
            ShardError::Spawn(_) => Self::Spawn,
            ShardError::ReclaimClose { .. } => Self::ReclaimClose,
            ShardError::ShutdownStorage { .. } => Self::ShutdownStorage,
            ShardError::SequenceConflict { .. } => Self::SequenceConflict,
            ShardError::ReservedSequenceKey => Self::ReservedSequenceKey,
            ShardError::EmptyAppend => Self::EmptyAppend,
            ShardError::LeafRecordCorrupt { .. } => Self::LeafRecordCorrupt,
            ShardError::LeafLogCorrupt { .. } => Self::LeafLogCorrupt,
            ShardError::LeafLogNotAdmitted => Self::LeafLogNotAdmitted,
            ShardError::LeafLogTruncate => Self::LeafLogTruncate,
            ShardError::LeafLogStreamDelete => Self::LeafLogStreamDelete,
            ShardError::LeafLogTtlWrite => Self::LeafLogTtlWrite,
            ShardError::LeafLogMergeAdopt => Self::LeafLogMergeAdopt,
            ShardError::PinWouldGoBack { .. } => Self::PinWouldGoBack,
            ShardError::PinConflict { .. } => Self::PinConflict,
            ShardError::PinBeyondExtent { .. } => Self::PinBeyondExtent,
            ShardError::CasMismatch { .. } => Self::CasMismatch,
            ShardError::CasHashMismatch { .. } => Self::CasHashMismatch,
            ShardError::Fenced { .. } => Self::Fenced,
            ShardError::WriteDuringRootAdvanceEmission => Self::WriteDuringRootAdvanceEmission,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireCallback {
    pub job_index: u64,
    pub shard: u64,
    pub pid: u64,
    pub command: String,
    pub subscriber_index: u64,
    pub event: WireAdvance,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireAdvance {
    pub shard_id: u64,
    pub prior_root: [u8; 32],
    pub new_root: [u8; 32],
    pub advance_gen: u64,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum DatabaseErrorKind {
    BatchFailed,
    BatchShardFailed,
    RootCallbackPanicked,
    BatchJobUnknown,
    TailRepairRefused,
    DirectoryCreate,
    ConfigWrite,
    ConfigRead,
    ConfigParse,
    InvalidShardCount,
    InvalidLongWaitInterval,
    InvalidQueueByteLimit,
    ShardQueueFull,
    ShardRequestSizeOverflow,
    ShardSpawn,
    SweepSpawn,
    ShardError,
    OutcomeUnknown,
    SweepError,
    SyncSchedulerSpawn,
    SyncSchedulerError,
    IoError,
    InvalidSweepInterval,
    MissingSyncTopology,
    InvalidSyncInterval,
    FormatVersionTooNew,
    ReservedSequenceKey,
    SequenceConflict,
    CasMismatch,
    CasHashMismatch,
    ConsistencyError,
    Distribution,
    LocalCommitFailed,
    CommittedButOwnerBehind,
    ElectionLost,
    ElectionTimeout,
    Fenced,
    CasConflict,
    ClusterMembers,
    DataDirLocked,
    LockFileIo,
    DataDirAlreadyInitialised,
    LeafLogStore,
    EphemeralStore,
    PathWriterRequired,
    WriteDuringRootAdvanceEmission,
    DuplicateShardId,
    ExecutorThreadsInvalid,
    MissingNodeCacheBudget,
    InvalidNodeCacheBudget,
    RetentionRefused,
    StreamDeleteRefused,
    RootNotRetained,
    ReclaimFailed,
    ExecutorShutdown,
    MigrationInProgress,
    InvalidChunkingStamp,
    MigrationFailed,
    MigrationRefusedFormat,
    MigrationRefusedWalTail,
    MigrationRefusedUncommittedMutations,
}

impl DatabaseErrorKind {
    const fn project(error: &DatabaseError) -> Self {
        match error {
            DatabaseError::BatchFailed(_) => Self::BatchFailed,
            DatabaseError::BatchShardFailed { .. } => Self::BatchShardFailed,
            DatabaseError::RootCallbackPanicked { .. } => Self::RootCallbackPanicked,
            DatabaseError::BatchJobUnknown { .. } => Self::BatchJobUnknown,
            DatabaseError::TailRepairRefused { .. } => Self::TailRepairRefused,
            DatabaseError::DirectoryCreate(_) => Self::DirectoryCreate,
            DatabaseError::ConfigWrite(_) => Self::ConfigWrite,
            DatabaseError::ConfigRead(_) => Self::ConfigRead,
            DatabaseError::ConfigParse(_) => Self::ConfigParse,
            DatabaseError::InvalidShardCount => Self::InvalidShardCount,
            DatabaseError::InvalidLongWaitInterval { .. } => Self::InvalidLongWaitInterval,
            DatabaseError::InvalidQueueByteLimit => Self::InvalidQueueByteLimit,
            DatabaseError::ShardQueueFull { .. } => Self::ShardQueueFull,
            DatabaseError::ShardRequestSizeOverflow { .. } => Self::ShardRequestSizeOverflow,
            DatabaseError::ShardSpawn(_) => Self::ShardSpawn,
            DatabaseError::SweepSpawn(_) => Self::SweepSpawn,
            DatabaseError::ShardError(_) => Self::ShardError,
            DatabaseError::OutcomeUnknown { .. } => Self::OutcomeUnknown,
            DatabaseError::SweepError(_) => Self::SweepError,
            DatabaseError::SyncSchedulerSpawn(_) => Self::SyncSchedulerSpawn,
            DatabaseError::SyncSchedulerError(_) => Self::SyncSchedulerError,
            DatabaseError::IoError(_) => Self::IoError,
            DatabaseError::InvalidSweepInterval => Self::InvalidSweepInterval,
            DatabaseError::MissingSyncTopology => Self::MissingSyncTopology,
            DatabaseError::InvalidSyncInterval => Self::InvalidSyncInterval,
            DatabaseError::FormatVersionTooNew { .. } => Self::FormatVersionTooNew,
            DatabaseError::ReservedSequenceKey => Self::ReservedSequenceKey,
            DatabaseError::SequenceConflict { .. } => Self::SequenceConflict,
            DatabaseError::CasMismatch { .. } => Self::CasMismatch,
            DatabaseError::CasHashMismatch { .. } => Self::CasHashMismatch,
            DatabaseError::ConsistencyError(_) => Self::ConsistencyError,
            DatabaseError::Distribution(_) => Self::Distribution,
            DatabaseError::LocalCommitFailed(_) => Self::LocalCommitFailed,
            DatabaseError::CommittedButOwnerBehind { .. } => Self::CommittedButOwnerBehind,
            DatabaseError::ElectionLost { .. } => Self::ElectionLost,
            DatabaseError::ElectionTimeout { .. } => Self::ElectionTimeout,
            DatabaseError::Fenced { .. } => Self::Fenced,
            DatabaseError::CasConflict { .. } => Self::CasConflict,
            DatabaseError::ClusterMembers(_) => Self::ClusterMembers,
            DatabaseError::DataDirLocked { .. } => Self::DataDirLocked,
            DatabaseError::LockFileIo { .. } => Self::LockFileIo,
            DatabaseError::DataDirAlreadyInitialised { .. } => Self::DataDirAlreadyInitialised,
            DatabaseError::LeafLogStore { .. } => Self::LeafLogStore,
            DatabaseError::EphemeralStore { .. } => Self::EphemeralStore,
            DatabaseError::PathWriterRequired { .. } => Self::PathWriterRequired,
            DatabaseError::WriteDuringRootAdvanceEmission => Self::WriteDuringRootAdvanceEmission,
            DatabaseError::DuplicateShardId { .. } => Self::DuplicateShardId,
            DatabaseError::ExecutorThreadsInvalid(_) => Self::ExecutorThreadsInvalid,
            DatabaseError::MissingNodeCacheBudget => Self::MissingNodeCacheBudget,
            DatabaseError::InvalidNodeCacheBudget(_) => Self::InvalidNodeCacheBudget,
            DatabaseError::RetentionRefused(_) => Self::RetentionRefused,
            DatabaseError::StreamDeleteRefused(_) => Self::StreamDeleteRefused,
            DatabaseError::RootNotRetained { .. } => Self::RootNotRetained,
            DatabaseError::ReclaimFailed { .. } => Self::ReclaimFailed,
            DatabaseError::ExecutorShutdown => Self::ExecutorShutdown,
            DatabaseError::MigrationInProgress { .. } => Self::MigrationInProgress,
            DatabaseError::InvalidChunkingStamp(_) => Self::InvalidChunkingStamp,
            DatabaseError::MigrationFailed(_) => Self::MigrationFailed,
            DatabaseError::MigrationRefusedFormat { .. } => Self::MigrationRefusedFormat,
            DatabaseError::MigrationRefusedWalTail { .. } => Self::MigrationRefusedWalTail,
            DatabaseError::MigrationRefusedUncommittedMutations { .. } => {
                Self::MigrationRefusedUncommittedMutations
            }
        }
    }
}

fn index(value: usize) -> Result<u64, ReceiptCodecError> {
    u64::try_from(value).map_err(|_| ReceiptCodecError::IndexOverflow)
}

fn sequences(values: &[(Vec<u8>, u64)]) -> Result<Vec<WireSequence>, ReceiptCodecError> {
    let mut result = Vec::new();
    result
        .try_reserve_exact(values.len())
        .map_err(ReceiptCodecError::allocation)?;
    for (key, sequence) in values {
        result.push(WireSequence {
            key_hex: encode_hex(key)?,
            sequence: *sequence,
        });
    }
    Ok(result)
}

impl WireReceipt {
    /// Borrow the original receipt while building its complete portable projection.
    ///
    /// # Errors
    ///
    /// Refuses an unrepresentable index or failed allocation; never returns a partial projection.
    pub fn project(receipt: &BatchReceipt) -> Result<Self, ReceiptCodecError> {
        let mut outcomes = Vec::new();
        outcomes
            .try_reserve_exact(receipt.outcomes().len())
            .map_err(ReceiptCodecError::allocation)?;
        for job in receipt.outcomes() {
            let outcome = match &job.outcome {
                BatchOutcome::Committed {
                    root,
                    callback_failure,
                } => WireOutcome::Committed {
                    root: *root.as_bytes(),
                    callback_failure: callback_failure
                        .as_deref()
                        .map(WireCause::project)
                        .transpose()?,
                },
                BatchOutcome::SequenceRead { sequences: values } => WireOutcome::SequenceRead {
                    sequences: sequences(values)?,
                },
                BatchOutcome::RangeRead {
                    sequences: values,
                    hit_limit,
                    invalid_outside_range,
                } => WireOutcome::RangeRead {
                    sequences: sequences(values)?,
                    hit_limit: *hit_limit,
                    invalid_outside_range: index(*invalid_outside_range)?,
                },
                BatchOutcome::Failed { cause } => WireOutcome::Failed {
                    cause: WireCause::project(cause)?,
                },
                BatchOutcome::Unknown { cause } => WireOutcome::Unknown {
                    cause: WireCause::project(cause)?,
                },
            };
            outcomes.push(WireJobOutcome {
                job_index: index(job.job_index)?,
                shard: index(job.shard)?,
                outcome,
            });
        }
        Ok(Self {
            outcomes,
            submit_error: receipt
                .submit_error()
                .map(|cause| WireCause::project(cause))
                .transpose()?,
            first_error: receipt.first_error().map(WireCause::project).transpose()?,
        })
    }
}

impl WireCause {
    /// Retain the exact typed kind and complete diagnostic without accessing a panic payload.
    ///
    /// # Errors
    ///
    /// Refuses unrepresentable metadata or any failed nested receipt projection.
    pub fn project(error: &DatabaseError) -> Result<Self, ReceiptCodecError> {
        let kind = DatabaseErrorKind::project(error);
        let shard = if let DatabaseError::BatchShardFailed { cause } = error {
            Some(WireShardCause {
                kind: ShardErrorKind::project(cause),
                message: cause.to_string(),
            })
        } else {
            None
        };
        let callback = if let DatabaseError::RootCallbackPanicked { cause } = error {
            let identity = cause.identity();
            let event = cause.event();
            Some(WireCallback {
                job_index: index(identity.job_index)?,
                shard: index(identity.shard)?,
                pid: identity.pid,
                command: identity.command.to_owned(),
                subscriber_index: index(cause.subscriber_index())?,
                event: WireAdvance {
                    shard_id: index(event.shard_id)?,
                    prior_root: *event.prior_root.as_bytes(),
                    new_root: *event.new_root.as_bytes(),
                    advance_gen: event.advance_gen,
                },
            })
        } else {
            None
        };
        let (batch_receipt, first_error) = if let DatabaseError::BatchFailed(failure) = error {
            (
                Some(Box::new(WireReceipt::project(failure.receipt())?)),
                Some(Box::new(Self::project(failure.first_error())?)),
            )
        } else {
            (None, None)
        };
        Ok(Self {
            kind,
            message: error.to_string(),
            shard,
            callback,
            batch_receipt,
            first_error,
        })
    }
}

fn read_key_hex<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<String, D::Error> {
    let key = String::deserialize(deserializer)?;
    if !key.len().is_multiple_of(2) || !key.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        return Err(serde::de::Error::custom("receipt_read_key_hex_invalid"));
    }
    Ok(key)
}