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    /// Load and verify one committed checkpoint index, or `None` once retention deleted it.
106    ///
107    /// # Errors
108    /// Object-store I/O, malformed content, or any reference mismatch.
109    pub async fn load_committed_checkpoint_optional(
110        &self,
111        reference: &CommittedCheckpointRef,
112    ) -> Result<Option<CommittedCheckpointIndex>, DecisionError> {
113        let Some(bytes) = self
114            .load_committed_checkpoint_bytes(reference, Some(reference.len))
115            .await?
116        else {
117            return Ok(None);
118        };
119        let index: CommittedCheckpointIndex = serde_json::from_slice(&bytes).map_err(|error| {
120            DecisionError::Conflict(format!(
121                "committed checkpoint '{}': {error}",
122                reference.sha256
123            ))
124        })?;
125        let (canonical, actual_reference) = index
126            .encode_and_reference()
127            .map_err(DecisionError::Conflict)?;
128        if actual_reference != *reference || canonical.as_slice() != bytes.as_ref() {
129            return Err(DecisionError::Conflict(format!(
130                "committed checkpoint '{}' does not match its content-addressed reference",
131                reference.sha256
132            )));
133        }
134        let deployment_id = self.load_deployment_id().await?.ok_or_else(|| {
135            DecisionError::Conflict(
136                "committed checkpoint deployment identity is missing; do not recreate or reset the namespace"
137                    .into(),
138            )
139        })?;
140        if index.deployment_id != deployment_id {
141            return Err(DecisionError::Conflict(format!(
142                "committed checkpoint belongs to deployment {}, current deployment is {deployment_id}",
143                index.deployment_id
144            )));
145        }
146        Ok(Some(index))
147    }
148
149    /// Permanently seal the exact content-addressed candidate for an aborted attempt.
150    ///
151    /// The seal occupies the candidate's existing path, so an in-flight conditional create can
152    /// either win and be replaced or lose to the seal. Identical retries converge.
153    ///
154    /// # Errors
155    /// The candidate is malformed or foreign, the path contains different content, or object-store
156    /// I/O cannot be reconciled to the exact seal.
157    pub async fn seal_aborted_committed_checkpoint_candidate(
158        &self,
159        index: &CommittedCheckpointIndex,
160    ) -> Result<(), DecisionError> {
161        let (candidate_bytes, reference) = index
162            .encode_and_reference()
163            .map_err(DecisionError::Conflict)?;
164        let deployment_id = self.load_or_create_deployment_id().await?;
165        if index.deployment_id != deployment_id {
166            return Err(DecisionError::Conflict(format!(
167                "committed checkpoint belongs to deployment {}, current deployment is {deployment_id}",
168                index.deployment_id
169            )));
170        }
171        let seal_bytes = crate::checkpoint::canonical_json_bytes(&AbortedCommittedCheckpointSeal {
172            version: ABORTED_COMMITTED_CHECKPOINT_SEAL_VERSION,
173            deployment_id: &deployment_id,
174            candidate: &reference,
175        })
176        .map_err(|error| DecisionError::Conflict(error.to_string()))?;
177        let path = Self::committed_checkpoint_path(&reference);
178
179        let observed = self
180            .load_committed_checkpoint_bytes(&reference, None)
181            .await?;
182        let mode = match observed.as_deref() {
183            None => PutMode::Create,
184            Some(bytes) if bytes == seal_bytes.as_slice() => return Ok(()),
185            Some(bytes) if bytes == candidate_bytes.as_slice() => PutMode::Overwrite,
186            Some(_) => {
187                return Err(DecisionError::Conflict(format!(
188                    "committed checkpoint '{}' contains neither the exact candidate nor its abort seal",
189                    reference.sha256
190                )));
191            }
192        };
193        let Err(mut write_error) = self
194            .store
195            .put_opts(
196                &path,
197                PutPayload::from(Bytes::from(seal_bytes.clone())),
198                PutOptions {
199                    mode: mode.clone(),
200                    ..PutOptions::default()
201                },
202            )
203            .await
204        else {
205            return Ok(());
206        };
207
208        let mut reconciled = self
209            .load_committed_checkpoint_bytes(&reference, None)
210            .await?;
211        if reconciled.as_deref() == Some(seal_bytes.as_slice()) {
212            return Ok(());
213        }
214        if matches!(mode, PutMode::Create)
215            && reconciled.as_deref() == Some(candidate_bytes.as_slice())
216        {
217            let Err(error) = self
218                .store
219                .put_opts(
220                    &path,
221                    PutPayload::from(Bytes::from(seal_bytes.clone())),
222                    PutOptions {
223                        mode: PutMode::Overwrite,
224                        ..PutOptions::default()
225                    },
226                )
227                .await
228            else {
229                return Ok(());
230            };
231            write_error = error;
232            reconciled = self
233                .load_committed_checkpoint_bytes(&reference, None)
234                .await?;
235            if reconciled.as_deref() == Some(seal_bytes.as_slice()) {
236                return Ok(());
237            }
238        }
239
240        if reconciled
241            .as_deref()
242            .is_some_and(|bytes| bytes != candidate_bytes.as_slice())
243        {
244            return Err(DecisionError::Conflict(format!(
245                "committed checkpoint '{}' changed to content other than its exact abort seal",
246                reference.sha256
247            )));
248        }
249        Err(DecisionError::Io(format!(
250            "committed checkpoint '{}' abort seal write failed and did not become durable: {write_error}",
251            reference.sha256
252        )))
253    }
254
255    /// Delete one exact committed index after validating any extant body.
256    ///
257    /// Missing objects and ambiguous deletes that removed the object are successful retries.
258    ///
259    /// # Errors
260    /// The reference, extant index, deployment identity, or object-store operation is invalid.
261    pub async fn delete_committed_checkpoint(
262        &self,
263        reference: &CommittedCheckpointRef,
264    ) -> Result<(), DecisionError> {
265        reference.validate().map_err(DecisionError::Conflict)?;
266        if self
267            .load_committed_checkpoint_optional(reference)
268            .await?
269            .is_none()
270        {
271            return Ok(());
272        }
273        let path = Self::committed_checkpoint_path(reference);
274        match self.store.delete(&path).await {
275            Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()),
276            Err(delete_error) => match self.store.head(&path).await {
277                Err(object_store::Error::NotFound { .. }) => Ok(()),
278                Ok(_) => Err(DecisionError::Io(delete_error.to_string())),
279                Err(reconcile_error) => Err(DecisionError::Io(format!(
280                    "committed checkpoint delete failed ({delete_error}); reconciliation failed ({reconcile_error})"
281                ))),
282            },
283        }
284    }
285
286    pub(crate) async fn validate_committed_checkpoint_for_outcome(
287        &self,
288        outcome: &CheckpointOutcome,
289    ) -> Result<CommittedCheckpointIndex, DecisionError> {
290        outcome.validate_shape(outcome.epoch)?;
291        let reference = outcome.committed_checkpoint.as_ref().ok_or_else(|| {
292            DecisionError::Conflict(format!(
293                "commit outcome for epoch {} requires a committed checkpoint",
294                outcome.epoch
295            ))
296        })?;
297        let index = self.load_committed_checkpoint(reference).await?;
298        if index.epoch != outcome.epoch
299            || index.checkpoint_id != outcome.checkpoint_id
300            || index.scope != outcome.scope
301            || index.assignment_fence.as_ref() != outcome.assignment_fence.as_ref()
302            || index.deployment_id != outcome.deployment_id
303        {
304            return Err(DecisionError::Conflict(format!(
305                "commit outcome for epoch {} does not match committed checkpoint '{}'",
306                outcome.epoch, reference.sha256
307            )));
308        }
309        Ok(index)
310    }
311}
laminar_core/checkpoint_decision/committed.rs

laminar_core/checkpoint_decision/
committed.rs
