Struct FileSystemCheckpointStore 

pub struct FileSystemCheckpointStore { /* private fields */ }

Filesystem-backed checkpoint store.

Writes checkpoint manifests as JSON files with atomic rename semantics. A latest.txt pointer (not a symlink) tracks the most recent checkpoint for Windows compatibility.

Implementations

impl FileSystemCheckpointStore

pub fn new(base_dir: impl Into<PathBuf>, max_retained: usize) -> Self

Creates a new filesystem checkpoint store.

The base_dir is the parent directory; checkpoints are stored under {base_dir}/checkpoints/. The directory is created lazily on first save.

The store’s vnode_count defaults to crate::checkpoint_manifest::DEFAULT_VNODE_COUNT. Hosts that run with a non-default value should chain Self::with_vnode_count so manifest validation checks the right invariant.

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

Override the vnode_count used during manifest validation.

Trait Implementations

impl CheckpointStore for FileSystemCheckpointStore

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_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 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 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.

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 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 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 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.