laminar_core/checkpoint/
prepared_witness.rs

1//! Compact inventory records for unresolved prepared checkpoints.
2
3use super::{PipelineIdentity, PIPELINE_IDENTITY_VERSION};
4use crate::state::CheckpointAttempt;
5
6/// Maximum unresolved prepared attempts one process may include in a stopped report.
7///
8/// The inventory is recovery evidence, not history. Retaining a bounded recent set keeps the
9/// control-plane record small and makes an unexpectedly growing backlog fail closed.
10pub const MAX_PREPARED_CHECKPOINT_WITNESSES: usize = 64;
11
12/// One process's compact evidence that an exact checkpoint attempt reached `Prepared`.
13///
14/// This is deliberately not a commit signal. A recovery driver must resolve every witness through
15/// the immutable checkpoint-outcome authority before allowing the cluster to restart.
16#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
17#[serde(deny_unknown_fields)]
18pub struct PreparedCheckpointWitness {
19    /// One nonzero canonical checkpoint ID, mirrored in both retained fields.
20    pub attempt: CheckpointAttempt,
21    /// Stable logical participant recorded by the prepared manifest.
22    pub participant_id: u64,
23    /// Canonical non-nil deployment UUID recorded by the prepared manifest.
24    pub deployment_id: String,
25    /// Exact logical pipeline and state-ABI identity recorded by the prepared manifest.
26    pub pipeline_identity: PipelineIdentity,
27}
28
29impl PreparedCheckpointWitness {
30    /// Construct and validate a compact prepared-checkpoint witness.
31    ///
32    /// # Errors
33    /// Returns an error when any persisted identity is zero, non-canonical, or unsupported.
34    pub fn new(
35        attempt: CheckpointAttempt,
36        participant_id: u64,
37        deployment_id: String,
38        pipeline_identity: PipelineIdentity,
39    ) -> Result<Self, String> {
40        let witness = Self {
41            attempt,
42            participant_id,
43            deployment_id,
44            pipeline_identity,
45        };
46        witness.validate()?;
47        Ok(witness)
48    }
49
50    /// Validate the persisted identities carried by this witness.
51    ///
52    /// # Errors
53    /// Returns an error describing the first non-canonical field.
54    pub(crate) fn validate(&self) -> Result<(), String> {
55        if !self.attempt.is_canonical() {
56            return Err(
57                "prepared checkpoint attempt must use one nonzero canonical checkpoint ID".into(),
58            );
59        }
60        if self.participant_id == 0 {
61            return Err("prepared checkpoint participant must be nonzero".into());
62        }
63        let deployment = uuid::Uuid::parse_str(&self.deployment_id)
64            .map_err(|error| format!("prepared checkpoint deployment ID is invalid: {error}"))?;
65        if deployment.is_nil() || deployment.to_string() != self.deployment_id {
66            return Err(
67                "prepared checkpoint deployment ID must be a canonical non-nil UUID".into(),
68            );
69        }
70        if !self.pipeline_identity.is_canonical() {
71            return Err(format!(
72                "prepared checkpoint pipeline identity must use canonical version {PIPELINE_IDENTITY_VERSION} and a lowercase SHA-256 digest"
73            ));
74        }
75        Ok(())
76    }
77
78    /// Whether all persisted identities have their canonical production shape.
79    #[must_use]
80    pub fn is_canonical(&self) -> bool {
81        self.validate().is_ok()
82    }
83
84    #[cfg(feature = "cluster")]
85    pub(crate) const fn ordering_key(&self) -> (u64, u64) {
86        (self.attempt.checkpoint_id, self.participant_id)
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93
94    fn witness() -> PreparedCheckpointWitness {
95        PreparedCheckpointWitness::new(
96            CheckpointAttempt::canonical(7),
97            3,
98            uuid::Uuid::from_u128(9).to_string(),
99            PipelineIdentity::empty(),
100        )
101        .unwrap()
102    }
103
104    #[test]
105    fn witness_rejects_zero_and_noncanonical_identities() {
106        let valid = witness();
107        assert!(valid.is_canonical());
108
109        let mut invalid = valid.clone();
110        invalid.attempt.epoch = 0;
111        assert!(!invalid.is_canonical());
112
113        let mut invalid = valid.clone();
114        invalid.attempt.checkpoint_id = 0;
115        assert!(!invalid.is_canonical());
116
117        let mut invalid = valid.clone();
118        invalid.attempt.checkpoint_id = invalid.attempt.epoch + 1;
119        assert!(!invalid.is_canonical());
120
121        let mut invalid = valid.clone();
122        invalid.participant_id = 0;
123        assert!(!invalid.is_canonical());
124
125        let mut invalid = valid.clone();
126        invalid.deployment_id = uuid::Uuid::nil().to_string();
127        assert!(!invalid.is_canonical());
128
129        let mut invalid = valid;
130        invalid.pipeline_identity.canonical_version = PIPELINE_IDENTITY_VERSION + 1;
131        assert!(!invalid.is_canonical());
132    }
133
134    #[test]
135    fn witness_wire_rejects_unknown_fields() {
136        let mut value = serde_json::to_value(witness()).unwrap();
137        value["unknown"] = serde_json::json!(true);
138        assert!(serde_json::from_value::<PreparedCheckpointWitness>(value).is_err());
139    }
140}
laminar_core/checkpoint/prepared_witness.rs

laminar_core/checkpoint/
prepared_witness.rs
