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

laminar_core/checkpoint/checkpoint_store/
mod.rs
