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 writes use [PutMode::Create] for split-brain prevention (conditional PUT).

Implementations

impl ObjectStoreCheckpointStore

pub fn new( store: Arc<dyn ObjectStore>, prefix: String, max_retained: usize, ) -> 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.

The store’s vnode_count defaults to crate::checkpoint::checkpoint_manifest::DEFAULT_VNODE_COUNT. Hosts that run with a non-default value should chain Self::with_vnode_count.

pub fn with_vnode_count(self, vnode_count: u16) -> Self

Override the vnode_count used during manifest validation.

Trait Implementations

impl CheckpointStore for ObjectStoreCheckpointStore

fn vnode_count(&self) -> u16

Runtime vnode count that manifests written by this store are expected to use. Consulted when validating loaded manifests — a mismatch is reported as a manifest warning. Defaults to crate::checkpoint::checkpoint_manifest::DEFAULT_VNODE_COUNT when the implementation has no configured value.

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 update_manifest<'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,

Overwrites an existing manifest, bypassing the conditional-PUT fence used by Self::save. Used after a successful sink commit to record per-sink status transitions.

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) on 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 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<'life0, 'async_trait>( &'life0 self, keep_count: usize, ) -> Pin<Box<dyn Future<Output = Result<usize, CheckpointStoreError>> + Send + 'async_trait>>

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

Prunes old checkpoints, keeping at most keep_count recent ones. Returns the number of checkpoints removed.

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 cleanup_orphans<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, CheckpointStoreError>> + Send + 'async_trait>>

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

Delete orphaned state files that have no matching manifest.

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<(), 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.