laminar_core/checkpoint_decision/
committed.rs

1//! Immutable committed-checkpoint index creation, validation, and deletion.
2
3use super::{
4    AbortedCommittedCheckpointSeal, Bytes, CheckpointDecisionStore, CheckpointOutcome,
5    CommittedCheckpointIndex, CommittedCheckpointRef, DecisionError, ObjectStore, ObjectStoreExt,
6    PutMode, PutOptions, PutPayload, ABORTED_COMMITTED_CHECKPOINT_SEAL_VERSION,
7    MAX_COMMITTED_CHECKPOINT_INDEX_BYTES,
8};
9
10impl CheckpointDecisionStore {
11    async fn load_committed_checkpoint_bytes(
12        &self,
13        reference: &CommittedCheckpointRef,
14        expected_len: Option<u64>,
15    ) -> Result<Option<Bytes>, DecisionError> {
16        reference.validate().map_err(DecisionError::Conflict)?;
17        let record = format!("committed checkpoint '{}'", reference.sha256);
18        let Some(result) = self
19            .get_control_record(
20                &Self::committed_checkpoint_path(reference),
21                &record,
22                u64::try_from(MAX_COMMITTED_CHECKPOINT_INDEX_BYTES).unwrap_or(u64::MAX),
23            )
24            .await?
25        else {
26            return Ok(None);
27        };
28        Self::read_control_record_bytes(
29            result,
30            &record,
31            u64::try_from(MAX_COMMITTED_CHECKPOINT_INDEX_BYTES).unwrap_or(u64::MAX),
32            expected_len,
33        )
34        .await
35        .map(Some)
36    }
37
38    /// Create the canonical content-addressed body for a committed checkpoint index.
39    ///
40    /// Identical retries converge on the existing immutable body. The returned reference is safe
41    /// to publish only after this method succeeds.
42    ///
43    /// # Errors
44    /// Object-store I/O, malformed index content, deployment mismatch, or a conflicting body.
45    pub async fn create_committed_checkpoint(
46        &self,
47        index: &CommittedCheckpointIndex,
48    ) -> Result<CommittedCheckpointRef, DecisionError> {
49        let (encoded, reference) = index
50            .encode_and_reference()
51            .map_err(DecisionError::Conflict)?;
52        let deployment_id = self.load_or_create_deployment_id().await?;
53        if index.deployment_id != deployment_id {
54            return Err(DecisionError::Conflict(format!(
55                "committed checkpoint belongs to deployment {}, current deployment is {deployment_id}",
56                index.deployment_id
57            )));
58        }
59
60        let path = Self::committed_checkpoint_path(&reference);
61        let options = PutOptions {
62            mode: PutMode::Create,
63            ..PutOptions::default()
64        };
65        match self
66            .store
67            .put_opts(&path, PutPayload::from(Bytes::from(encoded)), options)
68            .await
69        {
70            Ok(_) => Ok(reference),
71            Err(put_error) => match self.load_committed_checkpoint(&reference).await {
72                Ok(stored) if stored == *index => Ok(reference),
73                Ok(_) => Err(DecisionError::Conflict(format!(
74                    "committed checkpoint '{}' differs from the proposed content",
75                    reference.sha256
76                ))),
77                Err(reconcile_error) => Err(DecisionError::Io(format!(
78                    "committed checkpoint write failed ({put_error}); reconciliation failed ({reconcile_error})"
79                ))),
80            },
81        }
82    }
83
84    /// Load and verify one exact content-addressed committed checkpoint index.
85    ///
86    /// Verification covers the object path, recorded and observed lengths, canonical JSON body,
87    /// SHA-256 reference, deployment identity, and committed-index invariants.
88    ///
89    /// # Errors
90    /// Object-store I/O, a missing object, malformed content, or any reference mismatch.
91    pub async fn load_committed_checkpoint(
92        &self,
93        reference: &CommittedCheckpointRef,
94    ) -> Result<CommittedCheckpointIndex, DecisionError> {
95        self.load_committed_checkpoint_optional(reference)
96            .await?
97            .ok_or_else(|| {
98                DecisionError::Conflict(format!(
99                    "committed checkpoint '{}' is missing",
100                    reference.sha256
101                ))
102            })
103    }
104
105    async fn load_committed_checkpoint_optional(
106        &self,
107        reference: &CommittedCheckpointRef,
108    ) -> Result<Option<CommittedCheckpointIndex>, DecisionError> {
109        let Some(bytes) = self
110            .load_committed_checkpoint_bytes(reference, Some(reference.len))
111            .await?
112        else {
113            return Ok(None);
114        };
115        let index: CommittedCheckpointIndex = serde_json::from_slice(&bytes).map_err(|error| {
116            DecisionError::Conflict(format!(
117                "committed checkpoint '{}': {error}",
118                reference.sha256
119            ))
120        })?;
121        let (canonical, actual_reference) = index
122            .encode_and_reference()
123            .map_err(DecisionError::Conflict)?;
124        if actual_reference != *reference || canonical.as_slice() != bytes.as_ref() {
125            return Err(DecisionError::Conflict(format!(
126                "committed checkpoint '{}' does not match its content-addressed reference",
127                reference.sha256
128            )));
129        }
130        let deployment_id = self.load_or_create_deployment_id().await?;
131        if index.deployment_id != deployment_id {
132            return Err(DecisionError::Conflict(format!(
133                "committed checkpoint belongs to deployment {}, current deployment is {deployment_id}",
134                index.deployment_id
135            )));
136        }
137        Ok(Some(index))
138    }
139
140    /// Permanently seal the exact content-addressed candidate for an aborted attempt.
141    ///
142    /// The seal occupies the candidate's existing path, so an in-flight conditional create can
143    /// either win and be replaced or lose to the seal. Identical retries converge.
144    ///
145    /// # Errors
146    /// The candidate is malformed or foreign, the path contains different content, or object-store
147    /// I/O cannot be reconciled to the exact seal.
148    pub async fn seal_aborted_committed_checkpoint_candidate(
149        &self,
150        index: &CommittedCheckpointIndex,
151    ) -> Result<(), DecisionError> {
152        let (candidate_bytes, reference) = index
153            .encode_and_reference()
154            .map_err(DecisionError::Conflict)?;
155        let deployment_id = self.load_or_create_deployment_id().await?;
156        if index.deployment_id != deployment_id {
157            return Err(DecisionError::Conflict(format!(
158                "committed checkpoint belongs to deployment {}, current deployment is {deployment_id}",
159                index.deployment_id
160            )));
161        }
162        let seal_bytes = crate::checkpoint::canonical_json_bytes(&AbortedCommittedCheckpointSeal {
163            version: ABORTED_COMMITTED_CHECKPOINT_SEAL_VERSION,
164            deployment_id: &deployment_id,
165            candidate: &reference,
166        })
167        .map_err(|error| DecisionError::Conflict(error.to_string()))?;
168        let path = Self::committed_checkpoint_path(&reference);
169
170        let observed = self
171            .load_committed_checkpoint_bytes(&reference, None)
172            .await?;
173        let mode = match observed.as_deref() {
174            None => PutMode::Create,
175            Some(bytes) if bytes == seal_bytes.as_slice() => return Ok(()),
176            Some(bytes) if bytes == candidate_bytes.as_slice() => PutMode::Overwrite,
177            Some(_) => {
178                return Err(DecisionError::Conflict(format!(
179                    "committed checkpoint '{}' contains neither the exact candidate nor its abort seal",
180                    reference.sha256
181                )));
182            }
183        };
184        let Err(mut write_error) = self
185            .store
186            .put_opts(
187                &path,
188                PutPayload::from(Bytes::from(seal_bytes.clone())),
189                PutOptions {
190                    mode: mode.clone(),
191                    ..PutOptions::default()
192                },
193            )
194            .await
195        else {
196            return Ok(());
197        };
198
199        let mut reconciled = self
200            .load_committed_checkpoint_bytes(&reference, None)
201            .await?;
202        if reconciled.as_deref() == Some(seal_bytes.as_slice()) {
203            return Ok(());
204        }
205        if matches!(mode, PutMode::Create)
206            && reconciled.as_deref() == Some(candidate_bytes.as_slice())
207        {
208            let Err(error) = self
209                .store
210                .put_opts(
211                    &path,
212                    PutPayload::from(Bytes::from(seal_bytes.clone())),
213                    PutOptions {
214                        mode: PutMode::Overwrite,
215                        ..PutOptions::default()
216                    },
217                )
218                .await
219            else {
220                return Ok(());
221            };
222            write_error = error;
223            reconciled = self
224                .load_committed_checkpoint_bytes(&reference, None)
225                .await?;
226            if reconciled.as_deref() == Some(seal_bytes.as_slice()) {
227                return Ok(());
228            }
229        }
230
231        if reconciled
232            .as_deref()
233            .is_some_and(|bytes| bytes != candidate_bytes.as_slice())
234        {
235            return Err(DecisionError::Conflict(format!(
236                "committed checkpoint '{}' changed to content other than its exact abort seal",
237                reference.sha256
238            )));
239        }
240        Err(DecisionError::Io(format!(
241            "committed checkpoint '{}' abort seal write failed and did not become durable: {write_error}",
242            reference.sha256
243        )))
244    }
245
246    /// Delete one exact committed index after validating any extant body.
247    ///
248    /// Missing objects and ambiguous deletes that removed the object are successful retries.
249    ///
250    /// # Errors
251    /// The reference, extant index, deployment identity, or object-store operation is invalid.
252    pub async fn delete_committed_checkpoint(
253        &self,
254        reference: &CommittedCheckpointRef,
255    ) -> Result<(), DecisionError> {
256        reference.validate().map_err(DecisionError::Conflict)?;
257        if self
258            .load_committed_checkpoint_optional(reference)
259            .await?
260            .is_none()
261        {
262            return Ok(());
263        }
264        let path = Self::committed_checkpoint_path(reference);
265        match self.store.delete(&path).await {
266            Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()),
267            Err(delete_error) => match self.store.head(&path).await {
268                Err(object_store::Error::NotFound { .. }) => Ok(()),
269                Ok(_) => Err(DecisionError::Io(delete_error.to_string())),
270                Err(reconcile_error) => Err(DecisionError::Io(format!(
271                    "committed checkpoint delete failed ({delete_error}); reconciliation failed ({reconcile_error})"
272                ))),
273            },
274        }
275    }
276
277    pub(crate) async fn validate_committed_checkpoint_for_outcome(
278        &self,
279        outcome: &CheckpointOutcome,
280    ) -> Result<CommittedCheckpointIndex, DecisionError> {
281        outcome.validate_shape(outcome.epoch)?;
282        let reference = outcome.committed_checkpoint.as_ref().ok_or_else(|| {
283            DecisionError::Conflict(format!(
284                "commit outcome for epoch {} requires a committed checkpoint",
285                outcome.epoch
286            ))
287        })?;
288        let index = self.load_committed_checkpoint(reference).await?;
289        if index.epoch != outcome.epoch
290            || index.checkpoint_id != outcome.checkpoint_id
291            || index.scope != outcome.scope
292            || index.assignment_fence.as_ref() != outcome.assignment_fence.as_ref()
293            || index.deployment_id != outcome.deployment_id
294        {
295            return Err(DecisionError::Conflict(format!(
296                "commit outcome for epoch {} does not match committed checkpoint '{}'",
297                outcome.epoch, reference.sha256
298            )));
299        }
300        Ok(index)
301    }
302}
laminar_core/checkpoint_decision/committed.rs

laminar_core/checkpoint_decision/
committed.rs
