laminar_core/checkpoint/checkpoint_store/
mod.rs

1//! Provider-neutral checkpoint persistence over [`object_store`].
2
3#![allow(clippy::disallowed_types)] // cold path: checkpoint metadata
4
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use bytes::Bytes;
9use object_store::{ObjectStore, PutPayload};
10
11use crate::checkpoint::checkpoint_manifest::{
12    checkpoint_artifact_intent_sha256, checkpoint_descriptor_sha256, ByteRange, CheckpointManifest,
13    NodeDataObject, PreparedSinkArtifactIntent, PreparedSinkDescriptor, StateChunkId,
14};
15use crate::checkpoint::{canonical_json_bytes, OutputSegmentRef};
16use crate::state::{KeyGroupCount, DEFAULT_KEY_GROUP_COUNT, LOCAL_NODE_ID};
17
18mod abort_seal;
19mod artifact_identity;
20mod artifact_intent;
21mod artifact_payload;
22mod conditional_probe;
23mod object_store_io;
24mod subscription_segments;
25mod validation;
26mod view_log;
27
28pub use artifact_identity::checkpoint_artifact_identity_sha256;
29pub use artifact_intent::{
30    CheckpointSinkArtifactIntent, MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_AGGREGATE_BYTES,
31    MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES,
32};
33pub use conditional_probe::{
34    probe_object_store_conditional_create, probe_object_store_conditional_update,
35};
36pub use subscription_segments::SubscriptionOrphanCleanup;
37pub use validation::validate_max_checkpoint_node_data_bytes;
38use validation::{
39    checkpoint_artifact_abort_seal_bytes, ensure_manifest_valid, normalize_prefix, sha256,
40    validate_abort_seal_request, validate_node_data_layout, ManifestAbortState,
41};
42
43const MAX_MANIFEST_BYTES: u64 = 16 * 1024 * 1024;
44const MAX_ABORT_SEAL_BYTES: u64 =
45    MAX_MANIFEST_BYTES + artifact_intent::MAX_CHECKPOINT_ARTIFACT_INTENT_RECORD_BYTES + 64 * 1024;
46const CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION: u32 = 3;
47const CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION_V2: u32 = 2;
48const CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION_V1: u32 = 1;
49
50/// Default maximum size of one participant's checkpoint data object.
51pub const DEFAULT_MAX_CHECKPOINT_NODE_DATA_BYTES: u64 = 512 * 1024 * 1024;
52
53/// Checkpoint persistence error.
54#[derive(Debug, thiserror::Error)]
55pub enum CheckpointStoreError {
56    /// Object-store operation failed.
57    #[error("object store: {0}")]
58    ObjectStore(#[from] object_store::Error),
59    /// Manifest serialization failed.
60    #[error("manifest serialization: {0}")]
61    Serde(#[from] serde_json::Error),
62    /// Persisted data violates the checkpoint contract.
63    #[error("invalid checkpoint: {0}")]
64    Invalid(String),
65    /// An exact checkpoint manifest was not found.
66    #[error("checkpoint {0} not found")]
67    NotFound(u64),
68}
69
70/// Exact canonical bytes persisted for a checkpoint manifest.
71///
72/// # Errors
73/// Returns an error when the manifest cannot be represented as JSON.
74pub fn checkpoint_manifest_bytes(
75    manifest: &CheckpointManifest,
76) -> Result<Vec<u8>, serde_json::Error> {
77    canonical_json_bytes(manifest)
78}
79
80#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
81#[serde(deny_unknown_fields)]
82struct CheckpointArtifactAbortSeal {
83    version: u32,
84    artifact_identity_sha256: String,
85    chunk: StateChunkId,
86    original_manifest: Option<CheckpointManifest>,
87    #[serde(default, skip_serializing_if = "is_false")]
88    sink_artifact_intent_protocol: bool,
89    #[serde(default, skip_serializing_if = "Vec::is_empty")]
90    open_sink_artifact_intents: Vec<CheckpointSinkArtifactIntent>,
91    #[serde(default, skip_serializing_if = "is_false")]
92    sink_cleanup_complete: bool,
93}
94
95const fn is_false(value: &bool) -> bool {
96    !*value
97}
98
99/// Durable state returned after sealing one aborted participant manifest.
100#[derive(Clone, PartialEq)]
101pub struct CheckpointManifestAbortSeal {
102    /// Original prepared manifest, when that participant completed persistence.
103    pub original_manifest: Option<(CheckpointManifest, Bytes)>,
104    /// Whether absence of an intent proves this participant never entered `begin_epoch`.
105    pub sink_artifact_intent_protocol: bool,
106    /// Begin-time intents retained when phase one never produced a participant manifest.
107    pub open_sink_artifact_intents: Vec<CheckpointSinkArtifactIntent>,
108    /// Whether exact connector cleanup completed before node data may be sealed.
109    pub sink_cleanup_complete: bool,
110}
111
112impl std::fmt::Debug for CheckpointManifestAbortSeal {
113    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114        formatter
115            .debug_struct("CheckpointManifestAbortSeal")
116            .field("has_original_manifest", &self.original_manifest.is_some())
117            .field(
118                "sink_artifact_intent_protocol",
119                &self.sink_artifact_intent_protocol,
120            )
121            .field(
122                "open_sink_artifact_intent_count",
123                &self.open_sink_artifact_intents.len(),
124            )
125            .field("sink_cleanup_complete", &self.sink_cleanup_complete)
126            .finish()
127    }
128}
129
130/// Immutable checkpoint storage contract.
131#[async_trait]
132pub trait CheckpointStore: Send + Sync {
133    /// Maximum bytes admitted for one node data object.
134    fn max_node_data_bytes(&self) -> u64;
135
136    /// Stable vnode count expected in manifests.
137    fn key_group_count(&self) -> KeyGroupCount {
138        DEFAULT_KEY_GROUP_COUNT
139    }
140
141    /// Node whose manifests this store writes.
142    fn participant_id(&self) -> u64 {
143        LOCAL_NODE_ID.0
144    }
145
146    /// Validate and conditional-create the immutable node data object, then its manifest. Active
147    /// inventory and exact Abort seals reconcile creates left ambiguous by caller cancellation.
148    async fn save_checkpoint(
149        &self,
150        manifest: &CheckpointManifest,
151        node_data: &[Bytes],
152    ) -> Result<Bytes, CheckpointStoreError>;
153
154    /// Persist every committable sink's bounded cleanup intent before `begin_epoch`.
155    async fn save_sink_artifact_intents(
156        &self,
157        chunk: StateChunkId,
158        expected_artifact_identity_sha256: &str,
159        intents: Vec<CheckpointSinkArtifactIntent>,
160    ) -> Result<(), CheckpointStoreError>;
161
162    /// Replace this exact aborted manifest path with a monotone seal. The protocol flag must come
163    /// from the exact active artifact inventory. If a valid manifest was already durable,
164    /// preserve it inside the seal and return it with its canonical bytes.
165    async fn seal_aborted_manifest(
166        &self,
167        chunk: StateChunkId,
168        expected_artifact_identity_sha256: &str,
169        sink_artifact_intent_protocol: bool,
170    ) -> Result<CheckpointManifestAbortSeal, CheckpointStoreError>;
171
172    /// Mark exact prepared-sink cleanup complete on an already sealed manifest.
173    async fn complete_aborted_sink_cleanup(
174        &self,
175        chunk: StateChunkId,
176        expected_artifact_identity_sha256: &str,
177    ) -> Result<CheckpointManifestAbortSeal, CheckpointStoreError>;
178
179    /// Replace this exact aborted node-data path with a monotone seal after sink cleanup is durable.
180    async fn seal_aborted_node_data(
181        &self,
182        chunk: StateChunkId,
183        expected_artifact_identity_sha256: &str,
184    ) -> Result<(), CheckpointStoreError>;
185
186    /// Load this node's exact manifest.
187    async fn load_manifest(
188        &self,
189        checkpoint_id: u64,
190    ) -> Result<Option<CheckpointManifest>, CheckpointStoreError> {
191        self.load_manifest_for_participant(self.participant_id(), checkpoint_id)
192            .await
193    }
194
195    /// Load an exact manifest from a known participant namespace.
196    async fn load_manifest_for_participant(
197        &self,
198        participant_id: u64,
199        checkpoint_id: u64,
200    ) -> Result<Option<CheckpointManifest>, CheckpointStoreError>;
201
202    /// Load a manifest only when its exact persisted length and digest match a committed index.
203    async fn load_manifest_verified(
204        &self,
205        participant_id: u64,
206        checkpoint_id: u64,
207        expected_len: u64,
208        expected_sha256: &str,
209    ) -> Result<Option<CheckpointManifest>, CheckpointStoreError>;
210
211    /// Read exact ranges after verifying the immutable object's declared length.
212    async fn load_node_data_ranges(
213        &self,
214        chunk: StateChunkId,
215        expected_object_length: u64,
216        ranges: &[ByteRange],
217    ) -> Result<Option<Vec<Bytes>>, CheckpointStoreError>;
218
219    /// Delete an explicitly identified manifest. GC must supply the identity from durable
220    /// checkpoint metadata, never from object listing.
221    async fn delete_manifest(&self, chunk: StateChunkId) -> Result<(), CheckpointStoreError>;
222
223    /// Delete an explicitly identified node object after its durable reference count reaches zero.
224    async fn delete_node_data(&self, chunk: StateChunkId) -> Result<(), CheckpointStoreError>;
225
226    /// Create this checkpoint's view-log object before its manifest, accepting an identical
227    /// retry only.
228    async fn save_view_log(
229        &self,
230        _object: &NodeDataObject,
231        _payload: &[Bytes],
232    ) -> Result<(), CheckpointStoreError> {
233        Err(view_log::unsupported_store_error())
234    }
235
236    /// Read exact view-log ranges after verifying the object's declared length.
237    async fn load_view_log_ranges(
238        &self,
239        _chunk: StateChunkId,
240        _expected_object_length: u64,
241        _ranges: &[ByteRange],
242    ) -> Result<Option<Vec<Bytes>>, CheckpointStoreError> {
243        Err(view_log::unsupported_store_error())
244    }
245
246    /// Delete a view-log object no retained manifest references, if it exists. A store without
247    /// view logs holds none.
248    async fn delete_view_log(&self, _chunk: StateChunkId) -> Result<(), CheckpointStoreError> {
249        Ok(())
250    }
251
252    /// Read and verify one prepared sink descriptor.
253    async fn load_prepared_sink_descriptor(
254        &self,
255        manifest: &CheckpointManifest,
256        descriptor: &PreparedSinkDescriptor,
257    ) -> Result<Option<Bytes>, CheckpointStoreError> {
258        artifact_payload::load_optional(
259            self,
260            manifest,
261            &descriptor.sink_name,
262            descriptor.payload,
263            &descriptor.sha256,
264            checkpoint_descriptor_sha256,
265            "prepared sink descriptor",
266        )
267        .await
268    }
269
270    /// Read and verify one begin-time sink artifact intent.
271    async fn load_sink_artifact_intent(
272        &self,
273        manifest: &CheckpointManifest,
274        intent: &PreparedSinkArtifactIntent,
275    ) -> Result<Option<Bytes>, CheckpointStoreError> {
276        artifact_payload::load_optional(
277            self,
278            manifest,
279            &intent.sink_name,
280            intent.payload,
281            &intent.sha256,
282            checkpoint_artifact_intent_sha256,
283            "sink artifact intent",
284        )
285        .await
286    }
287
288    /// Create an immutable subscription segment, accepting an identical retry only.
289    async fn save_subscription_segment(
290        &self,
291        _segment: &OutputSegmentRef,
292        _payload: Bytes,
293    ) -> Result<(), CheckpointStoreError> {
294        Err(subscription_segments::unsupported_store_error())
295    }
296
297    /// Load and verify an exact subscription segment without consulting object listing.
298    async fn load_subscription_segment(
299        &self,
300        _segment: &OutputSegmentRef,
301    ) -> Result<Option<Bytes>, CheckpointStoreError> {
302        Err(subscription_segments::unsupported_store_error())
303    }
304
305    /// Delete an explicitly unreachable subscription segment.
306    async fn delete_subscription_segment(
307        &self,
308        _object_key: &str,
309    ) -> Result<(), CheckpointStoreError> {
310        Err(subscription_segments::unsupported_store_error())
311    }
312
313    /// Delete grace-expired segment objects not present in an authoritative reachable set.
314    /// Object listing supplies candidates only; `reachable` and `through_checkpoint_id` are the
315    /// caller's committed-state authority.
316    async fn delete_subscription_orphans(
317        &self,
318        _reachable: &std::collections::BTreeSet<String>,
319        _through_checkpoint_id: u64,
320        _grace_before_ms: i64,
321    ) -> Result<SubscriptionOrphanCleanup, CheckpointStoreError> {
322        Err(subscription_segments::unsupported_store_error())
323    }
324}
325
326/// Checkpoint store backed by any [`ObjectStore`] implementation.
327pub struct ObjectStoreCheckpointStore {
328    store: Arc<dyn ObjectStore>,
329    prefix: String,
330    key_group_count: KeyGroupCount,
331    participant_id: u64,
332    max_node_data_bytes: u64,
333    exclusive_writer: bool,
334}
335
336#[async_trait]
337impl CheckpointStore for ObjectStoreCheckpointStore {
338    fn max_node_data_bytes(&self) -> u64 {
339        self.max_node_data_bytes
340    }
341
342    fn key_group_count(&self) -> KeyGroupCount {
343        self.key_group_count
344    }
345
346    fn participant_id(&self) -> u64 {
347        self.participant_id
348    }
349
350    async fn save_checkpoint(
351        &self,
352        manifest: &CheckpointManifest,
353        node_data: &[Bytes],
354    ) -> Result<Bytes, CheckpointStoreError> {
355        ensure_manifest_valid(
356            manifest,
357            self.participant_id,
358            self.key_group_count,
359            self.max_node_data_bytes,
360        )?;
361        validate_node_data_layout(manifest, node_data, self.max_node_data_bytes)?;
362
363        let encoded = Bytes::from(checkpoint_manifest_bytes(manifest)?);
364        if u64::try_from(encoded.len()).unwrap_or(u64::MAX) > MAX_MANIFEST_BYTES {
365            return Err(CheckpointStoreError::Invalid(format!(
366                "manifest exceeds the {MAX_MANIFEST_BYTES}-byte limit"
367            )));
368        }
369
370        let payload: PutPayload = node_data.iter().cloned().collect();
371        let path = self.node_data_path(manifest.node_data.chunk);
372        if !self.create_immutable(&path, payload).await?
373            && !self
374                .existing_object_matches(&path, &manifest.node_data)
375                .await?
376        {
377            return Err(CheckpointStoreError::Invalid(format!(
378                "node data for participant {} checkpoint {} already exists with different immutable content",
379                manifest.node_data.chunk.participant_id, manifest.node_data.chunk.checkpoint_id
380            )));
381        }
382
383        // The manifest is the readiness marker. Publishing it last prevents readers from polling
384        // an incomplete checkpoint while the larger node-data object is still being uploaded.
385        let path = self.manifest_path(manifest.node_data.chunk);
386        if !self
387            .create_immutable(&path, PutPayload::from_bytes(encoded.clone()))
388            .await?
389        {
390            match self
391                .load_manifest_or_abort_seal_bytes(manifest.node_data.chunk)
392                .await?
393            {
394                Some(existing) if existing == encoded => {}
395                Some(existing)
396                    if self
397                        .promote_artifact_intent_to_manifest(manifest, encoded.clone(), &existing)
398                        .await? => {}
399                Some(_) => {
400                    return Err(CheckpointStoreError::Invalid(format!(
401                        "checkpoint {} manifest already exists with different immutable content",
402                        manifest.checkpoint_id
403                    )));
404                }
405                None => {
406                    return Err(CheckpointStoreError::Invalid(format!(
407                        "checkpoint {} manifest create reported a conflict but no object exists",
408                        manifest.checkpoint_id
409                    )));
410                }
411            }
412        }
413        Ok(encoded)
414    }
415
416    async fn save_sink_artifact_intents(
417        &self,
418        chunk: StateChunkId,
419        expected_artifact_identity_sha256: &str,
420        intents: Vec<CheckpointSinkArtifactIntent>,
421    ) -> Result<(), CheckpointStoreError> {
422        self.save_artifact_intent_record(chunk, expected_artifact_identity_sha256, intents)
423            .await
424    }
425
426    async fn seal_aborted_manifest(
427        &self,
428        chunk: StateChunkId,
429        expected_artifact_identity_sha256: &str,
430        sink_artifact_intent_protocol: bool,
431    ) -> Result<CheckpointManifestAbortSeal, CheckpointStoreError> {
432        validate_abort_seal_request(chunk, expected_artifact_identity_sha256)?;
433        let empty_seal = CheckpointArtifactAbortSeal {
434            version: CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION,
435            artifact_identity_sha256: expected_artifact_identity_sha256.to_owned(),
436            chunk,
437            original_manifest: None,
438            sink_artifact_intent_protocol,
439            open_sink_artifact_intents: Vec::new(),
440            sink_cleanup_complete: false,
441        };
442        let empty_seal_bytes = checkpoint_artifact_abort_seal_bytes(&empty_seal)?;
443        let path = self.manifest_path(chunk);
444        if self
445            .create_immutable(&path, PutPayload::from_bytes(empty_seal_bytes.clone()))
446            .await?
447        {
448            return Ok(CheckpointManifestAbortSeal {
449                original_manifest: None,
450                sink_artifact_intent_protocol,
451                open_sink_artifact_intents: Vec::new(),
452                sink_cleanup_complete: false,
453            });
454        }
455
456        let current = self
457            .load_manifest_or_abort_seal_bytes(chunk)
458            .await?
459            .ok_or_else(|| {
460                CheckpointStoreError::Invalid(format!(
461                    "participant {} checkpoint {} manifest seal create conflicted but no object exists",
462                    chunk.participant_id, chunk.checkpoint_id
463                ))
464            })?;
465        match self.decode_manifest_abort_state(
466            chunk,
467            expected_artifact_identity_sha256,
468            &current,
469        )? {
470            ManifestAbortState::Sealed {
471                original_manifest,
472                sink_artifact_intent_protocol,
473                open_sink_artifact_intents,
474                sink_cleanup_complete,
475            } => Ok(CheckpointManifestAbortSeal {
476                original_manifest,
477                sink_artifact_intent_protocol,
478                open_sink_artifact_intents,
479                sink_cleanup_complete,
480            }),
481            ManifestAbortState::Manifest(manifest, canonical) => {
482                let sink_artifact_intent_protocol = !manifest.sink_artifact_intents.is_empty();
483                let seal = CheckpointArtifactAbortSeal {
484                    version: CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION,
485                    artifact_identity_sha256: expected_artifact_identity_sha256.to_owned(),
486                    chunk,
487                    original_manifest: Some(manifest.clone()),
488                    sink_artifact_intent_protocol,
489                    open_sink_artifact_intents: Vec::new(),
490                    sink_cleanup_complete: false,
491                };
492                self.replace_exact(
493                    &path,
494                    &current,
495                    checkpoint_artifact_abort_seal_bytes(&seal)?,
496                )
497                .await?;
498                Ok(CheckpointManifestAbortSeal {
499                    original_manifest: Some((manifest, canonical)),
500                    sink_artifact_intent_protocol,
501                    open_sink_artifact_intents: Vec::new(),
502                    sink_cleanup_complete: false,
503                })
504            }
505            ManifestAbortState::Intent(record) => {
506                let open_sink_artifact_intents = record.sink_intents().to_vec();
507                let seal = CheckpointArtifactAbortSeal {
508                    version: CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION,
509                    artifact_identity_sha256: expected_artifact_identity_sha256.to_owned(),
510                    chunk,
511                    original_manifest: None,
512                    sink_artifact_intent_protocol: true,
513                    open_sink_artifact_intents: open_sink_artifact_intents.clone(),
514                    sink_cleanup_complete: false,
515                };
516                self.replace_exact(
517                    &path,
518                    &current,
519                    checkpoint_artifact_abort_seal_bytes(&seal)?,
520                )
521                .await?;
522                Ok(CheckpointManifestAbortSeal {
523                    original_manifest: None,
524                    sink_artifact_intent_protocol: true,
525                    open_sink_artifact_intents,
526                    sink_cleanup_complete: false,
527                })
528            }
529        }
530    }
531
532    async fn complete_aborted_sink_cleanup(
533        &self,
534        chunk: StateChunkId,
535        expected_artifact_identity_sha256: &str,
536    ) -> Result<CheckpointManifestAbortSeal, CheckpointStoreError> {
537        validate_abort_seal_request(chunk, expected_artifact_identity_sha256)?;
538        let path = self.manifest_path(chunk);
539        let current = self
540            .load_manifest_or_abort_seal_bytes(chunk)
541            .await?
542            .ok_or_else(|| {
543                CheckpointStoreError::Invalid(format!(
544                    "participant {} checkpoint {} has no manifest abort seal",
545                    chunk.participant_id, chunk.checkpoint_id
546                ))
547            })?;
548        let ManifestAbortState::Sealed {
549            original_manifest,
550            sink_artifact_intent_protocol,
551            open_sink_artifact_intents,
552            sink_cleanup_complete,
553        } = self.decode_manifest_abort_state(chunk, expected_artifact_identity_sha256, &current)?
554        else {
555            return Err(CheckpointStoreError::Invalid(format!(
556                "participant {} checkpoint {} manifest is not sealed before sink cleanup",
557                chunk.participant_id, chunk.checkpoint_id
558            )));
559        };
560        if sink_cleanup_complete {
561            return Ok(CheckpointManifestAbortSeal {
562                original_manifest,
563                sink_artifact_intent_protocol,
564                open_sink_artifact_intents,
565                sink_cleanup_complete,
566            });
567        }
568        let seal = CheckpointArtifactAbortSeal {
569            version: CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION,
570            artifact_identity_sha256: expected_artifact_identity_sha256.to_owned(),
571            chunk,
572            original_manifest: original_manifest
573                .as_ref()
574                .map(|(manifest, _)| manifest.clone()),
575            sink_artifact_intent_protocol,
576            open_sink_artifact_intents: open_sink_artifact_intents.clone(),
577            sink_cleanup_complete: true,
578        };
579        self.replace_exact(
580            &path,
581            &current,
582            checkpoint_artifact_abort_seal_bytes(&seal)?,
583        )
584        .await?;
585        Ok(CheckpointManifestAbortSeal {
586            original_manifest,
587            sink_artifact_intent_protocol,
588            open_sink_artifact_intents,
589            sink_cleanup_complete: true,
590        })
591    }
592
593    async fn seal_aborted_node_data(
594        &self,
595        chunk: StateChunkId,
596        expected_artifact_identity_sha256: &str,
597    ) -> Result<(), CheckpointStoreError> {
598        validate_abort_seal_request(chunk, expected_artifact_identity_sha256)?;
599        let manifest_bytes = self
600            .load_manifest_or_abort_seal_bytes(chunk)
601            .await?
602            .ok_or_else(|| {
603                CheckpointStoreError::Invalid(format!(
604                    "participant {} checkpoint {} has no manifest abort seal",
605                    chunk.participant_id, chunk.checkpoint_id
606                ))
607            })?;
608        match self.decode_manifest_abort_state(
609            chunk,
610            expected_artifact_identity_sha256,
611            &manifest_bytes,
612        )? {
613            ManifestAbortState::Sealed {
614                sink_cleanup_complete: true,
615                ..
616            } => {}
617            ManifestAbortState::Sealed { .. } => {
618                return Err(CheckpointStoreError::Invalid(format!(
619                    "participant {} checkpoint {} sink cleanup is incomplete",
620                    chunk.participant_id, chunk.checkpoint_id
621                )));
622            }
623            ManifestAbortState::Manifest(..) => {
624                return Err(CheckpointStoreError::Invalid(format!(
625                    "participant {} checkpoint {} manifest is not sealed",
626                    chunk.participant_id, chunk.checkpoint_id
627                )));
628            }
629            ManifestAbortState::Intent(..) => {
630                return Err(CheckpointStoreError::Invalid(format!(
631                    "participant {} checkpoint {} sink intent is not sealed",
632                    chunk.participant_id, chunk.checkpoint_id
633                )));
634            }
635        }
636        let seal = CheckpointArtifactAbortSeal {
637            version: CHECKPOINT_ARTIFACT_ABORT_SEAL_VERSION,
638            artifact_identity_sha256: expected_artifact_identity_sha256.to_owned(),
639            chunk,
640            original_manifest: None,
641            sink_artifact_intent_protocol: false,
642            open_sink_artifact_intents: Vec::new(),
643            sink_cleanup_complete: false,
644        };
645        let encoded = checkpoint_artifact_abort_seal_bytes(&seal)?;
646        let path = self.node_data_path(chunk);
647        if self
648            .create_immutable(&path, PutPayload::from_bytes(encoded.clone()))
649            .await?
650            || self.node_data_is_exact_abort_seal(chunk, &encoded).await?
651        {
652            return Ok(());
653        }
654        self.overwrite(&path, encoded).await
655    }
656
657    async fn load_manifest_for_participant(
658        &self,
659        participant_id: u64,
660        checkpoint_id: u64,
661    ) -> Result<Option<CheckpointManifest>, CheckpointStoreError> {
662        if participant_id == 0 || checkpoint_id == 0 {
663            return Err(CheckpointStoreError::Invalid(
664                "manifest identity must use nonzero participant and checkpoint ids".into(),
665            ));
666        }
667        self.load_bounded_manifest(StateChunkId {
668            participant_id,
669            checkpoint_id,
670        })
671        .await
672    }
673
674    async fn load_manifest_verified(
675        &self,
676        participant_id: u64,
677        checkpoint_id: u64,
678        expected_len: u64,
679        expected_sha256: &str,
680    ) -> Result<Option<CheckpointManifest>, CheckpointStoreError> {
681        if participant_id == 0 || checkpoint_id == 0 {
682            return Err(CheckpointStoreError::Invalid(
683                "manifest identity must use nonzero participant and checkpoint ids".into(),
684            ));
685        }
686        let chunk = StateChunkId {
687            participant_id,
688            checkpoint_id,
689        };
690        let Some(bytes) = self.load_bounded_manifest_bytes(chunk).await? else {
691            return Ok(None);
692        };
693        let actual_len = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
694        let actual_sha256 = sha256(&bytes);
695        if actual_len != expected_len || actual_sha256 != expected_sha256 {
696            return Err(CheckpointStoreError::Invalid(format!(
697                "participant {participant_id} checkpoint {checkpoint_id} manifest differs from the committed reference"
698            )));
699        }
700        self.decode_manifest(chunk, &bytes).map(Some)
701    }
702
703    async fn load_node_data_ranges(
704        &self,
705        chunk: StateChunkId,
706        expected_object_length: u64,
707        ranges: &[ByteRange],
708    ) -> Result<Option<Vec<Bytes>>, CheckpointStoreError> {
709        self.load_object_ranges(&self.node_data_path(chunk), expected_object_length, ranges)
710            .await
711    }
712
713    async fn delete_manifest(&self, chunk: StateChunkId) -> Result<(), CheckpointStoreError> {
714        self.delete_exact(&self.manifest_path(chunk)).await
715    }
716
717    async fn delete_node_data(&self, chunk: StateChunkId) -> Result<(), CheckpointStoreError> {
718        self.delete_exact(&self.node_data_path(chunk)).await
719    }
720
721    async fn save_view_log(
722        &self,
723        object: &NodeDataObject,
724        payload: &[Bytes],
725    ) -> Result<(), CheckpointStoreError> {
726        view_log::save(self, object, payload).await
727    }
728
729    async fn load_view_log_ranges(
730        &self,
731        chunk: StateChunkId,
732        expected_object_length: u64,
733        ranges: &[ByteRange],
734    ) -> Result<Option<Vec<Bytes>>, CheckpointStoreError> {
735        self.load_object_ranges(&self.view_log_path(chunk), expected_object_length, ranges)
736            .await
737    }
738
739    async fn delete_view_log(&self, chunk: StateChunkId) -> Result<(), CheckpointStoreError> {
740        self.delete_exact(&self.view_log_path(chunk)).await
741    }
742
743    async fn save_subscription_segment(
744        &self,
745        segment: &OutputSegmentRef,
746        payload: Bytes,
747    ) -> Result<(), CheckpointStoreError> {
748        subscription_segments::save(self, segment, payload).await
749    }
750
751    async fn load_subscription_segment(
752        &self,
753        segment: &OutputSegmentRef,
754    ) -> Result<Option<Bytes>, CheckpointStoreError> {
755        subscription_segments::load(self, segment).await
756    }
757
758    async fn delete_subscription_segment(
759        &self,
760        object_key: &str,
761    ) -> Result<(), CheckpointStoreError> {
762        subscription_segments::delete(self, object_key).await
763    }
764
765    async fn delete_subscription_orphans(
766        &self,
767        reachable: &std::collections::BTreeSet<String>,
768        through_checkpoint_id: u64,
769        grace_before_ms: i64,
770    ) -> Result<SubscriptionOrphanCleanup, CheckpointStoreError> {
771        subscription_segments::delete_orphans(
772            self,
773            reachable,
774            through_checkpoint_id,
775            grace_before_ms,
776        )
777        .await
778    }
779}
780
781#[cfg(test)]
782mod tests;
laminar_core/checkpoint/checkpoint_store/mod.rs

laminar_core/checkpoint/checkpoint_store/
mod.rs
