Struct ObjectStoreCheckpointStore 

pub struct ObjectStoreCheckpointStore { /* private fields */ }

Object-store-backed checkpoint store.

Drives any object_store::ObjectStore backend (S3, GCS, Azure, local FS) directly over .await; no dedicated runtime. The app runtime’s HTTP connection pool is reused.

Object Layout

{prefix}/
  manifests/
    manifest-000001.json    # Checkpoint manifest (JSON)
    manifest-000002.json
    latest.json             # {"checkpoint_id": 2}
  checkpoints/
    state-000001.bin        # Optional sidecar state
    state-000002.bin

Manifest creation, finalization, and recovery-pointer publication are all conditional writes. The only mutable manifest transition is the exact Prepared to Finalized phase change.

Implementations

impl ObjectStoreCheckpointStore

pub fn new(store: Arc<dyn ObjectStore>, prefix: String) -> Self

Create a new object-store-backed checkpoint store.

prefix is prepended to all object paths (e.g., "nodes/abc123/"). It should end with / or be empty.

Embedded and single-node stores default to one key group. Cluster hosts must chain Self::with_key_group_count with their durable topology.

pub fn with_key_group_count(self, key_group_count: KeyGroupCount) -> Self

Override the stable key-group count used during manifest validation.

pub fn with_participant_id(self, participant_id: u64) -> Self

Bind this store to one runtime participant.

pub fn with_max_state_data_bytes( self, max_state_data_bytes: u64, ) -> Result<Self, CheckpointStoreError>

Override the aggregate raw logical-state and physical-sidecar bytes admitted per checkpoint.

Errors

Returns CheckpointStoreError::Invalid when the limit cannot safely bound an in-memory restore allocation and its one-byte overflow probe.

Trait Implementations

impl CheckpointStore for ObjectStoreCheckpointStore

fn max_state_data_bytes(&self) -> u64

Maximum aggregate raw logical operator-state bytes admitted for one checkpoint.

fn key_group_count(&self) -> KeyGroupCount

Runtime key-group count that manifests written by this store are expected to use. Consulted when validating loaded manifests. Embedded and single-node stores default to one key group.

fn participant_id(&self) -> u64

Participant whose manifests belong in this store namespace.

fn save<'life0, 'life1, 'async_trait>( &'life0 self, manifest: &'life1 CheckpointManifest, ) -> Pin<Box<dyn Future<Output = Result<(), CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Atomically persists a checkpoint manifest. Implementations must guarantee readers never observe a partial manifest.

fn finalize<'life0, 'async_trait>( &'life0 self, checkpoint_id: u64, ) -> Pin<Box<dyn Future<Output = Result<CheckpointManifest, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Atomically publish a prepared checkpoint as the latest recoverable cut.

fn load_latest<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<CheckpointManifest>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Loads the most recent checkpoint manifest, or Ok(None) when the recovery pointer does not exist in a fresh store.

fn load_by_id<'life0, 'async_trait>( &'life0 self, id: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<CheckpointManifest>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Loads a specific manifest, or Ok(None) if absent.

fn load_manifest_for_participant<'life0, 'async_trait>( &'life0 self, participant_id: u64, id: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<CheckpointManifest>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Load an exact checkpoint from a participant namespace.

fn list_ids<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<u64>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists all checkpoint IDs, sorted ascending. Unlike Self::list this enumerates corrupt manifests too (used by crash recovery). Callers rely on the ascending invariant.

fn list<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<(u64, u64)>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists all available checkpoints as (id, epoch) pairs, sorted ascending by ID. May read every manifest; callers that only need IDs should use Self::list_ids.

fn prune_before<'life0, 'async_trait>( &'life0 self, before_epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<usize, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Prunes manifests whose checkpoint epoch is strictly below before_epoch.

fn save_state_data<'life0, 'life1, 'async_trait>( &'life0 self, id: u64, chunks: &'life1 [Bytes], ) -> Pin<Box<dyn Future<Output = Result<(), CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Writes operator state sidecar bytes for a checkpoint.

fn load_state_data<'life0, 'async_trait>( &'life0 self, id: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Loads operator state sidecar bytes for a checkpoint, or Ok(None) if no sidecar was written.

fn load_state_data_for_participant<'life0, 'async_trait>( &'life0 self, participant_id: u64, id: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Load operator state sidecar bytes from a participant namespace.

fn state_data_len_for_participant<'life0, 'async_trait>( &'life0 self, participant_id: u64, id: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<u64>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Read only the durable sidecar object’s length from storage metadata.

fn ensure_manifest_participant( &self, manifest: &CheckpointManifest, ) -> Result<(), CheckpointStoreError>

Reject a manifest that was routed to another participant’s namespace.

fn ensure_manifest_valid( &self, manifest: &CheckpointManifest, ) -> Result<(), CheckpointStoreError>

Reject an invalid manifest before any sidecar, directory, or object is created.

fn load_checkpoint_artifacts<'life0, 'async_trait>( &'life0 self, id: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<CheckpointArtifacts>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Load the manifest and checksum-required sidecar for a local checkpoint exactly once.

fn load_checkpoint_artifacts_for_participant<'life0, 'async_trait>( &'life0 self, participant_id: u64, id: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<CheckpointArtifacts>, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Load the manifest and checksum-required sidecar from a participant namespace exactly once.

fn validate_checkpoint<'life0, 'async_trait>( &'life0 self, id: u64, ) -> Pin<Box<dyn Future<Output = Result<ValidationResult, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Validate a specific checkpoint’s integrity.

fn recover_latest_validated<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<RecoveryReport, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Walk backward from latest to find the first valid checkpoint.

fn save_with_state<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, manifest: &'life1 CheckpointManifest, state_data: Option<&'life2 [Bytes]>, ) -> Pin<Box<dyn Future<Output = Result<CheckpointManifest, CheckpointStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Atomically saves a checkpoint manifest with optional sidecar state data.