laminar_core/checkpoint_decision/
retention.rs

1//! Durable checkpoint-retention state transitions.
2
3use super::{
4    CheckpointDecisionStore, CheckpointRetentionCursor, CheckpointRetentionState,
5    CheckpointRetentionUpdateResult, CheckpointScope, CommittedCheckpointRef, DecisionError,
6    DecisionStoreUpdateMode, DurableCheckpointRetentionHead, ObjectStore, PutMode, PutOptions,
7    PutPayload, UpdateVersion, VersionedCheckpointRetentionHead,
8    CHECKPOINT_RETENTION_HEAD_MAX_BYTES, CHECKPOINT_RETENTION_HEAD_VERSION,
9};
10
11impl CheckpointDecisionStore {
12    fn validate_retention_state_shape(
13        state: &CheckpointRetentionState,
14    ) -> Result<(), DecisionError> {
15        let cursor = match state {
16            CheckpointRetentionState::Idle { protected } => {
17                protected.validate().map_err(DecisionError::Conflict)?;
18                return Ok(());
19            }
20            CheckpointRetentionState::DeleteData { cursor }
21            | CheckpointRetentionState::DeleteMetadata { cursor } => cursor,
22        };
23        cursor
24            .protected
25            .validate()
26            .map_err(DecisionError::Conflict)?;
27        cursor.current.validate().map_err(DecisionError::Conflict)?;
28        if cursor.protected.epoch <= cursor.current.epoch {
29            return Err(DecisionError::Conflict(
30                "checkpoint retention cursor does not move behind its protected cut".into(),
31            ));
32        }
33        if let Some(next) = &cursor.next {
34            next.validate().map_err(DecisionError::Conflict)?;
35            if next.epoch >= cursor.current.epoch {
36                return Err(DecisionError::Conflict(
37                    "checkpoint retention next cursor is not an older cut".into(),
38                ));
39            }
40        }
41        if let Some(stop_before) = &cursor.stop_before {
42            stop_before.validate().map_err(DecisionError::Conflict)?;
43            if stop_before.epoch >= cursor.current.epoch {
44                return Err(DecisionError::Conflict(
45                    "checkpoint retention boundary is not older than its current cut".into(),
46                ));
47            }
48            let next = cursor.next.as_ref().ok_or_else(|| {
49                DecisionError::Conflict(
50                    "checkpoint retention crossed its exclusive lower boundary".into(),
51                )
52            })?;
53            if next.epoch < stop_before.epoch
54                || (next.epoch == stop_before.epoch && next != stop_before)
55            {
56                return Err(DecisionError::Conflict(
57                    "checkpoint retention next cursor crossed its exclusive lower boundary".into(),
58                ));
59            }
60        }
61        Ok(())
62    }
63
64    fn validate_retention_head_shape(
65        head: &DurableCheckpointRetentionHead,
66        deployment_id: &str,
67    ) -> Result<(), DecisionError> {
68        if head.version != CHECKPOINT_RETENTION_HEAD_VERSION || head.deployment_id != deployment_id
69        {
70            return Err(DecisionError::Conflict(
71                "checkpoint retention head has a foreign deployment or unsupported version".into(),
72            ));
73        }
74        Self::validate_retention_state_shape(&head.state)
75    }
76
77    async fn read_retention_head(
78        &self,
79        deployment_id: &str,
80    ) -> Result<Option<VersionedCheckpointRetentionHead>, DecisionError> {
81        let Some(result) = self
82            .get_control_record(
83                &Self::retention_head_path(deployment_id),
84                "checkpoint retention head",
85                CHECKPOINT_RETENTION_HEAD_MAX_BYTES,
86            )
87            .await?
88        else {
89            return Ok(None);
90        };
91        let update_version = UpdateVersion {
92            e_tag: result.meta.e_tag.clone(),
93            version: result.meta.version.clone(),
94        };
95        self.require_native_cas_token("checkpoint retention head", &update_version)?;
96        let bytes = Self::read_control_record_bytes(
97            result,
98            "checkpoint retention head",
99            CHECKPOINT_RETENTION_HEAD_MAX_BYTES,
100            None,
101        )
102        .await?;
103        let head: DurableCheckpointRetentionHead =
104            serde_json::from_slice(&bytes).map_err(|error| {
105                DecisionError::Conflict(format!("checkpoint retention head: {error}"))
106            })?;
107        Self::validate_retention_head_shape(&head, deployment_id)?;
108        let canonical = serde_json::to_vec(&head)
109            .map_err(|error| DecisionError::Conflict(error.to_string()))?;
110        if canonical.as_slice() != bytes.as_ref() {
111            return Err(DecisionError::Conflict(
112                "checkpoint retention head does not use its canonical body".into(),
113            ));
114        }
115        Ok(Some(VersionedCheckpointRetentionHead {
116            head,
117            update_version,
118        }))
119    }
120
121    async fn put_retention_head(
122        &self,
123        deployment_id: &str,
124        observed: Option<VersionedCheckpointRetentionHead>,
125        state: CheckpointRetentionState,
126    ) -> Result<CheckpointRetentionUpdateResult, DecisionError> {
127        Self::validate_retention_state_shape(&state)?;
128        let candidate = DurableCheckpointRetentionHead {
129            version: CHECKPOINT_RETENTION_HEAD_VERSION,
130            deployment_id: deployment_id.to_owned(),
131            state: state.clone(),
132        };
133        let mode = match (self.update_mode, observed.as_ref()) {
134            (_, None) => PutMode::Create,
135            (DecisionStoreUpdateMode::NativeCas, Some(current)) => {
136                PutMode::Update(current.update_version.clone())
137            }
138            (DecisionStoreUpdateMode::LocalSingleWriter, Some(_)) => PutMode::Overwrite,
139        };
140        let payload = Self::encode_control_record(
141            "checkpoint retention head",
142            &candidate,
143            CHECKPOINT_RETENTION_HEAD_MAX_BYTES,
144        )?;
145        let result = self
146            .store
147            .put_opts(
148                &Self::retention_head_path(deployment_id),
149                PutPayload::from(payload),
150                PutOptions {
151                    mode,
152                    ..PutOptions::default()
153                },
154            )
155            .await
156            .map(|_| ());
157        if result.is_ok() {
158            return Ok(CheckpointRetentionUpdateResult::Applied(state));
159        }
160
161        let winner = self.read_retention_head(deployment_id).await?;
162        if winner
163            .as_ref()
164            .is_some_and(|winner| winner.head == candidate)
165        {
166            return Ok(CheckpointRetentionUpdateResult::Unchanged(state));
167        }
168        let changed = winner.as_ref().map(|winner| &winner.head)
169            != observed.as_ref().map(|current| &current.head);
170        let error = result.expect_err("failed retention-head write has an error");
171        if changed
172            || matches!(
173                error,
174                object_store::Error::Precondition { .. }
175                    | object_store::Error::AlreadyExists { .. }
176                    | object_store::Error::NotFound { .. }
177            )
178        {
179            return Ok(CheckpointRetentionUpdateResult::Conflict {
180                current: winner.map(|winner| winner.head.state),
181            });
182        }
183        Err(DecisionError::Io(error.to_string()))
184    }
185
186    async fn begin_checkpoint_retention_inner(
187        &self,
188        protected: &CommittedCheckpointRef,
189    ) -> Result<CheckpointRetentionUpdateResult, DecisionError> {
190        protected.validate().map_err(DecisionError::Conflict)?;
191        let deployment_id = self.load_or_create_deployment_id().await?;
192        let decision = self
193            .read_decision_head(&deployment_id)
194            .await?
195            .and_then(|head| head.head.latest_commit)
196            .and_then(|outcome| outcome.committed_checkpoint)
197            .ok_or_else(|| {
198                DecisionError::Conflict(
199                    "checkpoint retention requires an authoritative local Commit".into(),
200                )
201            })?;
202        if decision != *protected {
203            return Err(DecisionError::Conflict(
204                "checkpoint retention protected cut is not the authoritative latest Commit".into(),
205            ));
206        }
207
208        let observed = self.read_retention_head(&deployment_id).await?;
209        if let Some(current) = observed.as_ref() {
210            match &current.head.state {
211                CheckpointRetentionState::Idle {
212                    protected: retained,
213                } if retained == protected => {
214                    return Ok(CheckpointRetentionUpdateResult::Unchanged(
215                        current.head.state.clone(),
216                    ));
217                }
218                CheckpointRetentionState::DeleteData { .. }
219                | CheckpointRetentionState::DeleteMetadata { .. } => {
220                    return Ok(CheckpointRetentionUpdateResult::Unchanged(
221                        current.head.state.clone(),
222                    ));
223                }
224                CheckpointRetentionState::Idle { .. } => {}
225            }
226        }
227
228        let protected_index = self.load_committed_checkpoint(protected).await?;
229        if protected_index.scope != CheckpointScope::Local {
230            return Err(DecisionError::Conflict(
231                "local checkpoint retention cannot protect a cluster checkpoint".into(),
232            ));
233        }
234        let stop_before = match observed.as_ref().map(|current| &current.head.state) {
235            Some(CheckpointRetentionState::Idle {
236                protected: retained,
237            }) => {
238                if retained.epoch >= protected.epoch {
239                    return Err(DecisionError::Conflict(
240                        "checkpoint retention cannot replace its retained cut with an older cut"
241                            .into(),
242                    ));
243                }
244                let retained_index = self.load_committed_checkpoint(retained).await?;
245                retained_index.predecessor
246            }
247            None => None,
248            Some(
249                CheckpointRetentionState::DeleteData { .. }
250                | CheckpointRetentionState::DeleteMetadata { .. },
251            ) => unreachable!("active retention returned above"),
252        };
253        let state = match protected_index.predecessor.as_ref() {
254            None if observed.is_none() => CheckpointRetentionState::Idle {
255                protected: protected.clone(),
256            },
257            None => {
258                return Err(DecisionError::Conflict(
259                    "authoritative Commit does not extend the retained checkpoint".into(),
260                ));
261            }
262            Some(current) => {
263                if let Some(CheckpointRetentionState::Idle {
264                    protected: retained,
265                }) = observed.as_ref().map(|head| &head.head.state)
266                {
267                    if current.epoch < retained.epoch
268                        || (current.epoch == retained.epoch && current != retained)
269                    {
270                        return Err(DecisionError::Conflict(
271                            "authoritative Commit does not extend the retained checkpoint".into(),
272                        ));
273                    }
274                }
275                let current_index = self.load_committed_checkpoint(current).await?;
276                protected_index
277                    .validate_predecessor_index(&current_index)
278                    .map_err(DecisionError::Conflict)?;
279                CheckpointRetentionState::DeleteData {
280                    cursor: CheckpointRetentionCursor {
281                        protected: protected.clone(),
282                        current: current.clone(),
283                        next: current_index.predecessor,
284                        stop_before,
285                    },
286                }
287            }
288        };
289        self.put_retention_head(&deployment_id, observed, state)
290            .await
291    }
292
293    /// Start or resume retention for the authoritative latest local Commit.
294    ///
295    /// # Errors
296    /// The protected cut is stale, its chain is invalid, or durable metadata cannot be updated.
297    pub async fn begin_checkpoint_retention(
298        &self,
299        protected: &CommittedCheckpointRef,
300    ) -> Result<CheckpointRetentionUpdateResult, DecisionError> {
301        if self.update_mode == DecisionStoreUpdateMode::LocalSingleWriter {
302            let lock = self.local_metadata_rmw_lock.as_ref().ok_or_else(|| {
303                DecisionError::Conflict(
304                    "local decision store is missing its namespace write lock".into(),
305                )
306            })?;
307            let _guard = lock.lock().await;
308            self.begin_checkpoint_retention_inner(protected).await
309        } else {
310            self.begin_checkpoint_retention_inner(protected).await
311        }
312    }
313
314    async fn advance_checkpoint_retention_inner(
315        &self,
316        expected: &CheckpointRetentionState,
317    ) -> Result<CheckpointRetentionUpdateResult, DecisionError> {
318        Self::validate_retention_state_shape(expected)?;
319        let deployment_id = self.load_or_create_deployment_id().await?;
320        let observed = self.read_retention_head(&deployment_id).await?;
321        if observed.as_ref().map(|head| &head.head.state) != Some(expected) {
322            return Ok(CheckpointRetentionUpdateResult::Conflict {
323                current: observed.map(|head| head.head.state),
324            });
325        }
326        let state = match expected {
327            CheckpointRetentionState::Idle { .. } => {
328                return Err(DecisionError::Conflict(
329                    "idle checkpoint retention has no destructive phase to advance".into(),
330                ));
331            }
332            CheckpointRetentionState::DeleteData { cursor } => {
333                CheckpointRetentionState::DeleteMetadata {
334                    cursor: cursor.clone(),
335                }
336            }
337            CheckpointRetentionState::DeleteMetadata { cursor }
338                if cursor.next == cursor.stop_before =>
339            {
340                CheckpointRetentionState::Idle {
341                    protected: cursor.protected.clone(),
342                }
343            }
344            CheckpointRetentionState::DeleteMetadata { cursor } => {
345                let current = cursor.next.as_ref().ok_or_else(|| {
346                    DecisionError::Conflict(
347                        "checkpoint retention ended before its exclusive lower boundary".into(),
348                    )
349                })?;
350                let current_index = self.load_committed_checkpoint(current).await?;
351                CheckpointRetentionState::DeleteData {
352                    cursor: CheckpointRetentionCursor {
353                        protected: cursor.protected.clone(),
354                        current: current.clone(),
355                        next: current_index.predecessor,
356                        stop_before: cursor.stop_before.clone(),
357                    },
358                }
359            }
360        };
361        self.put_retention_head(&deployment_id, observed, state)
362            .await
363    }
364
365    /// Advance one completed retention phase with an exact compare-and-swap.
366    ///
367    /// # Errors
368    /// The expected state is idle or invalid, or durable metadata cannot be updated.
369    pub async fn advance_checkpoint_retention(
370        &self,
371        expected: &CheckpointRetentionState,
372    ) -> Result<CheckpointRetentionUpdateResult, DecisionError> {
373        if self.update_mode == DecisionStoreUpdateMode::LocalSingleWriter {
374            let lock = self.local_metadata_rmw_lock.as_ref().ok_or_else(|| {
375                DecisionError::Conflict(
376                    "local decision store is missing its namespace write lock".into(),
377                )
378            })?;
379            let _guard = lock.lock().await;
380            self.advance_checkpoint_retention_inner(expected).await
381        } else {
382            self.advance_checkpoint_retention_inner(expected).await
383        }
384    }
385
386    /// Read the exact local retention state without listing storage.
387    ///
388    /// # Errors
389    /// Object-store I/O or malformed/foreign metadata.
390    pub async fn checkpoint_retention_state(
391        &self,
392    ) -> Result<Option<CheckpointRetentionState>, DecisionError> {
393        let deployment_id = self.load_or_create_deployment_id().await?;
394        Ok(self
395            .read_retention_head(&deployment_id)
396            .await?
397            .map(|head| head.head.state))
398    }
399}
laminar_core/checkpoint_decision/retention.rs

laminar_core/checkpoint_decision/
retention.rs
