Struct CheckpointCoordinator 

pub struct CheckpointCoordinator { /* private fields */ }

Orchestrates the checkpoint lifecycle across sources, sinks, and operator state.

Implementations

impl CheckpointCoordinator

pub async fn new( config: CheckpointConfig, store: Box<dyn CheckpointStore>, ) -> Result<Self, DbError>

Create a coordinator seeded from the highest stored checkpoint.

Errors

Returns a store read failure rather than silently starting at epoch 1 and clobbering on-disk state.

pub fn set_cluster_controller(&mut self, controller: Arc<ClusterController>)

Activate cluster-mode 2PC. Without this the coordinator runs single-instance semantics.

pub fn set_state_backend(&mut self, backend: Arc<dyn StateBackend>)

Wire a state backend to enable per-vnode markers and the durability gate.

pub fn set_decision_store(&mut self, store: Arc<CheckpointDecisionStore>)

Wire the durable commit-marker store.

pub fn set_assignment_version(&mut self, version: u64)

Set the assignment generation forwarded to write_partial for the split-brain fence.

pub fn set_local_watermark_ms(&mut self, watermark: Option<i64>)

Set the local watermark reported in BarrierAck so the leader can fold it into the cluster-wide minimum. None disables this instance’s contribution.

pub fn set_vnode_set(&mut self, vnodes: Vec<u32>)

Set the owned vnodes. Also the default gate set until set_gate_vnode_set is called.

pub fn set_gate_vnode_set(&mut self, vnodes: Vec<u32>)

Set the vnodes the durability gate checks (the full registry in cluster mode).

pub async fn begin_initial_epoch(&self) -> Result<(), DbError>

Begin the initial epoch on all exactly-once sinks.

Must be called once after all sinks are registered and before any writes. Subsequent epochs are started automatically after each successful checkpoint commit.

Errors

Returns the first sink error.

pub fn set_metrics(&mut self, prom: Arc<EngineMetrics>)

Wire Prometheus engine metrics.

pub async fn checkpoint( &mut self, request: CheckpointRequest, ) -> Result<CheckpointResult, DbError>

Run a full checkpoint cycle.

Barrier propagation and operator snapshots (steps 1-2) are handled by the caller and passed in via CheckpointRequest.

Errors

Returns DbError::Checkpoint if any phase fails.

pub async fn reconcile_prepared_on_init(&self)

On startup, reconcile any Pending sinks in the last manifest.

Commit marker present → drive local commit; marker absent → rollback. In cluster mode the leader re-announces the decision.

pub fn phase(&self) -> CheckpointPhase

Current checkpoint phase.

pub fn epoch(&self) -> u64

Next epoch to be allocated.

pub fn next_checkpoint_id(&self) -> u64

Next checkpoint ID to be allocated.

pub fn config(&self) -> &CheckpointConfig

Checkpoint configuration.

pub fn stats(&self) -> CheckpointStats

Checkpoint performance statistics.

pub fn store(&self) -> &dyn CheckpointStore

The underlying checkpoint store.

pub async fn checkpoint_with_offsets( &mut self, request: CheckpointRequest, ) -> Result<CheckpointResult, DbError>

Run a full checkpoint using pre-captured source offsets.

Non-empty source_offset_overrides bypass the live snapshot call — required for barrier-aligned checkpoints where source positions must match operator state exactly.

Errors

Returns DbError::Checkpoint if any phase fails.

pub async fn follower_checkpoint( &mut self, request: CheckpointRequest, ann: BarrierAnnouncement, decision_timeout: Duration, ) -> Result<bool, DbError>

Follower checkpoint: ack the capture, run the durable prepare, then wait for the leader’s commit/abort. Returns Ok(true) = committed, Ok(false) = aborted.

The ack means “aligned + captured”; the leader verifies prepare completion through the restorable gate (partials written last imply the full prepare finished).

Errors

Propagates sink pre-commit, manifest save, or marker-write failures.

pub async fn recover(&mut self) -> Result<Option<RecoveredState>, DbError>

Recover from the latest stored checkpoint.

Returns Ok(None) for a fresh start (no checkpoint found).

Errors

Returns DbError::Checkpoint if the store read fails.

pub async fn load_latest_manifest( &self, ) -> Result<Option<CheckpointManifest>, DbError>

Load the latest manifest from the store.

Errors

Returns DbError::Checkpoint on store errors.

Trait Implementations

impl Debug for CheckpointCoordinator

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.