lodestone_types/
mutation_commit.rs

1//! Recovery-stable facts for one registered document mutation.
2
3use std::error::Error;
4use std::fmt::{self, Display, Formatter};
5
6use serde::{Deserialize, Serialize};
7
8use crate::{
9    CommittedLogPositionV1, DurabilityBoundary, DurabilityClass, MAX_COMMIT_TIME_UNCERTAINTY_MILLIS, PrimaryTerm,
10    SchemaGeneration, SeqNo,
11};
12
13pub const SERVER_OBSERVED_COMMIT_TIME_FORMAT_VERSION: u16 = 1;
14pub const MUTATION_COMMIT_RECEIPT_FORMAT_VERSION: u16 = 1;
15
16/// UTC sampled by Core immediately before a replicated commit request.
17///
18/// The uncertainty is explicit because this is an observed wall-clock value,
19/// not the ordering authority. The replicated log position owns ordering.
20#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
21#[serde(deny_unknown_fields)]
22pub struct ServerObservedCommitTimeV1 {
23    pub format_version: u16,
24    pub unix_millis: u64,
25    pub uncertainty_millis: u64,
26}
27
28impl ServerObservedCommitTimeV1 {
29    pub fn validate(self) -> Result<(), MutationCommitReceiptError> {
30        if self.format_version != SERVER_OBSERVED_COMMIT_TIME_FORMAT_VERSION
31            || self.unix_millis == 0
32            || self.uncertainty_millis > MAX_COMMIT_TIME_UNCERTAINTY_MILLIS
33        {
34            return Err(MutationCommitReceiptError::InvalidCommitTime);
35        }
36        Ok(())
37    }
38}
39
40/// Immutable projection of the registered-WAL facts that committed one
41/// document mutation. It is evidence of durability, not of search visibility
42/// or of a manifest/index cut.
43#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
44#[serde(deny_unknown_fields)]
45pub struct MutationCommitReceiptV1 {
46    pub format_version: u16,
47    pub registration_position: CommittedLogPositionV1,
48    pub committed_at: ServerObservedCommitTimeV1,
49    pub sequence: SeqNo,
50    pub primary_term: PrimaryTerm,
51    pub mapping_generation: SchemaGeneration,
52    pub durability_class: DurabilityClass,
53    pub durability_boundary: DurabilityBoundary,
54}
55
56impl MutationCommitReceiptV1 {
57    pub fn validate(self) -> Result<(), MutationCommitReceiptError> {
58        if self.format_version != MUTATION_COMMIT_RECEIPT_FORMAT_VERSION
59            || self.registration_position.leader_term == 0
60            || self.registration_position.leader_node_id == 0
61            || self.registration_position.index == 0
62            || self.sequence.get() == 0
63            || self.primary_term.get() == 0
64            || self.mapping_generation.get() == 0
65            || self.durability_boundary != DurabilityBoundary::Registered
66            || self.durability_class.require_available().is_err()
67        {
68            return Err(MutationCommitReceiptError::InvalidReceipt);
69        }
70        self.committed_at.validate()
71    }
72}
73
74#[derive(Clone, Copy, Debug, Eq, PartialEq)]
75pub enum MutationCommitReceiptError {
76    InvalidCommitTime,
77    InvalidReceipt,
78}
79
80impl Display for MutationCommitReceiptError {
81    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
82        match self {
83            Self::InvalidCommitTime => formatter.write_str("mutation commit time is outside the current contract"),
84            Self::InvalidReceipt => formatter.write_str("mutation commit receipt is outside the current contract"),
85        }
86    }
87}
88
89impl Error for MutationCommitReceiptError {}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94
95    fn fixture() -> Result<MutationCommitReceiptV1, Box<dyn Error>> {
96        Ok(MutationCommitReceiptV1 {
97            format_version: MUTATION_COMMIT_RECEIPT_FORMAT_VERSION,
98            registration_position: CommittedLogPositionV1 {
99                leader_term: 7,
100                leader_node_id: 9,
101                index: 11,
102            },
103            committed_at: ServerObservedCommitTimeV1 {
104                format_version: SERVER_OBSERVED_COMMIT_TIME_FORMAT_VERSION,
105                unix_millis: 1_700_000_000_000,
106                uncertainty_millis: 5,
107            },
108            sequence: SeqNo::new(13),
109            primary_term: PrimaryTerm::new(15)?,
110            mapping_generation: SchemaGeneration::new(17),
111            durability_class: DurabilityClass::RegionalObject,
112            durability_boundary: DurabilityBoundary::Registered,
113        })
114    }
115
116    #[test]
117    fn registered_mutation_commit_receipt_validates() -> Result<(), Box<dyn Error>> {
118        fixture()?.validate()?;
119        Ok(())
120    }
121
122    #[test]
123    fn mutation_commit_receipt_refuses_nonregistration_boundary_and_invalid_position() -> Result<(), Box<dyn Error>> {
124        let mut receipt = fixture()?;
125        receipt.durability_boundary = DurabilityBoundary::IncarnationMemory;
126        assert_eq!(receipt.validate(), Err(MutationCommitReceiptError::InvalidReceipt));
127
128        let mut receipt = fixture()?;
129        receipt.registration_position.index = 0;
130        assert_eq!(receipt.validate(), Err(MutationCommitReceiptError::InvalidReceipt));
131        Ok(())
132    }
133}
lodestone_types/mutation_commit.rs

lodestone_types/
mutation_commit.rs
