Struct CheckpointCoordinator 

pub struct CheckpointCoordinator { /* private fields */ }

Checkpoint coordinator.

Implementations

impl CheckpointCoordinator

pub fn new(config: CheckpointConfig) -> Self

Create a new checkpoint coordinator without durable storage.

Checkpoints still capture registered operator state into their metadata but are not persisted to disk. Use CheckpointCoordinator::with_storage (or a storage_path in the config) for durable, recoverable checkpoints.

pub fn with_storage( config: CheckpointConfig, storage: Arc<dyn CheckpointStorage>, ) -> Self

Create a coordinator backed by a durable CheckpointStorage.

pub fn from_config(config: CheckpointConfig) -> Result<Self>

Create a coordinator whose storage is derived from config.storage_path.

When config.storage_path is set, a FileCheckpointStorage rooted at that path is created (the directory is created if missing). When it is None, this is equivalent to CheckpointCoordinator::new.

pub async fn register_operator( &self, name: impl Into<String>, state: Arc<dyn DynOperatorState>, )

Register an operator state to be captured on every checkpoint.

The name uniquely identifies the operator across snapshot/restore; the same name must be used when restoring so the correct bytes are routed back to the operator. Registering an existing name replaces it.

pub async fn operator_count(&self) -> usize

Number of registered operators.

pub async fn trigger_checkpoint(&self) -> Result<u64>

Trigger a new checkpoint, capturing the state of all registered operators.

This snapshots every registered DynOperatorState, aggregating the captured bytes into the checkpoint metadata (operator_states) and recording the real total size_bytes. The checkpoint is left active (not yet completed/persisted); call CheckpointCoordinator::complete_checkpoint to finalize it, or use the combined CheckpointCoordinator::checkpoint which triggers, persists, and completes in one step.

If any operator snapshot fails, no active checkpoint is registered and the error is propagated — a failed capture never masquerades as success.

pub async fn checkpoint(&self) -> Result<u64>

Trigger, persist, and complete a checkpoint in one durable step.

Captures registered operator state, persists it through the configured storage, and only marks the checkpoint successful once the state has been durably written. If capture or persistence fails, the checkpoint is completed as a failure (removed from the active set, not added to the completed set) and the error is returned — there is no silent success.

pub async fn restore_checkpoint(&self, checkpoint_id: u64) -> Result<()>

Restore registered operators from a persisted checkpoint.

Loads the checkpoint through the configured storage and routes each stored operator-state blob back to the operator registered under the same name. Operators without a stored blob are left untouched.

pub async fn complete_checkpoint( &self, checkpoint_id: u64, success: bool, ) -> Result<()>

Complete a checkpoint.

pub async fn active_count(&self) -> usize

Get active checkpoint count.

pub async fn active_metadata( &self, checkpoint_id: u64, ) -> Option<CheckpointMetadata>

Get a copy of an active (triggered-but-not-completed) checkpoint’s metadata, if present.

pub async fn completed_count(&self) -> usize

Get completed checkpoint count.

pub async fn latest_checkpoint(&self) -> Option<u64>

Get the latest completed checkpoint ID.

pub async fn clear_old_checkpoints(&self, keep_count: usize)

Clear old checkpoints.

pub async fn start_periodic_checkpointing(self: Arc<Self>)

Start periodic checkpointing.

On every config.interval, this runs the full capture → persist → complete pipeline via CheckpointCoordinator::checkpoint, bounded by config.timeout. A checkpoint is only reported as successful once the registered operator state has actually been captured (and, when storage is configured, durably written). Failures and timeouts are logged and the loop continues; a checkpoint is never marked successful without real state having been captured.