lodestone-types 0.0.29

//! Recovery-stable facts for one registered document mutation.

use std::error::Error;
use std::fmt::{self, Display, Formatter};

use serde::{Deserialize, Serialize};

use crate::{
    CommittedLogPositionV1, DurabilityBoundary, DurabilityClass, MAX_COMMIT_TIME_UNCERTAINTY_MILLIS, PrimaryTerm,
    SchemaGeneration, SeqNo,
};

pub const SERVER_OBSERVED_COMMIT_TIME_FORMAT_VERSION: u16 = 1;
pub const MUTATION_COMMIT_RECEIPT_FORMAT_VERSION: u16 = 1;

/// UTC sampled by Core immediately before a replicated commit request.
///
/// The uncertainty is explicit because this is an observed wall-clock value,
/// not the ordering authority. The replicated log position owns ordering.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ServerObservedCommitTimeV1 {
    pub format_version: u16,
    pub unix_millis: u64,
    pub uncertainty_millis: u64,
}

impl ServerObservedCommitTimeV1 {
    pub fn validate(self) -> Result<(), MutationCommitReceiptError> {
        if self.format_version != SERVER_OBSERVED_COMMIT_TIME_FORMAT_VERSION
            || self.unix_millis == 0
            || self.uncertainty_millis > MAX_COMMIT_TIME_UNCERTAINTY_MILLIS
        {
            return Err(MutationCommitReceiptError::InvalidCommitTime);
        }
        Ok(())
    }
}

/// Immutable projection of the registered-WAL facts that committed one
/// document mutation. It is evidence of durability, not of search visibility
/// or of a manifest/index cut.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct MutationCommitReceiptV1 {
    pub format_version: u16,
    pub registration_position: CommittedLogPositionV1,
    pub committed_at: ServerObservedCommitTimeV1,
    pub sequence: SeqNo,
    pub primary_term: PrimaryTerm,
    pub mapping_generation: SchemaGeneration,
    pub durability_class: DurabilityClass,
    pub durability_boundary: DurabilityBoundary,
}

impl MutationCommitReceiptV1 {
    pub fn validate(self) -> Result<(), MutationCommitReceiptError> {
        if self.format_version != MUTATION_COMMIT_RECEIPT_FORMAT_VERSION
            || self.registration_position.leader_term == 0
            || self.registration_position.leader_node_id == 0
            || self.registration_position.index == 0
            || self.sequence.get() == 0
            || self.primary_term.get() == 0
            || self.mapping_generation.get() == 0
            || self.durability_boundary != DurabilityBoundary::Registered
            || self.durability_class.require_available().is_err()
        {
            return Err(MutationCommitReceiptError::InvalidReceipt);
        }
        self.committed_at.validate()
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum MutationCommitReceiptError {
    InvalidCommitTime,
    InvalidReceipt,
}

impl Display for MutationCommitReceiptError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidCommitTime => formatter.write_str("mutation commit time is outside the current contract"),
            Self::InvalidReceipt => formatter.write_str("mutation commit receipt is outside the current contract"),
        }
    }
}

impl Error for MutationCommitReceiptError {}

#[cfg(test)]
mod tests {
    use super::*;

    fn fixture() -> Result<MutationCommitReceiptV1, Box<dyn Error>> {
        Ok(MutationCommitReceiptV1 {
            format_version: MUTATION_COMMIT_RECEIPT_FORMAT_VERSION,
            registration_position: CommittedLogPositionV1 {
                leader_term: 7,
                leader_node_id: 9,
                index: 11,
            },
            committed_at: ServerObservedCommitTimeV1 {
                format_version: SERVER_OBSERVED_COMMIT_TIME_FORMAT_VERSION,
                unix_millis: 1_700_000_000_000,
                uncertainty_millis: 5,
            },
            sequence: SeqNo::new(13),
            primary_term: PrimaryTerm::new(15)?,
            mapping_generation: SchemaGeneration::new(17),
            durability_class: DurabilityClass::RegionalObject,
            durability_boundary: DurabilityBoundary::Registered,
        })
    }

    #[test]
    fn registered_mutation_commit_receipt_validates() -> Result<(), Box<dyn Error>> {
        fixture()?.validate()?;
        Ok(())
    }

    #[test]
    fn mutation_commit_receipt_refuses_nonregistration_boundary_and_invalid_position() -> Result<(), Box<dyn Error>> {
        let mut receipt = fixture()?;
        receipt.durability_boundary = DurabilityBoundary::IncarnationMemory;
        assert_eq!(receipt.validate(), Err(MutationCommitReceiptError::InvalidReceipt));

        let mut receipt = fixture()?;
        receipt.registration_position.index = 0;
        assert_eq!(receipt.validate(), Err(MutationCommitReceiptError::InvalidReceipt));
        Ok(())
    }
}