Struct CheckpointCoordinator 

pub struct CheckpointCoordinator { /* private fields */ }

Unified checkpoint coordinator.

Orchestrates the full checkpoint lifecycle across sources, sinks, and operator state, persisting everything in a single CheckpointManifest.

Implementations

impl CheckpointCoordinator

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

Creates a new checkpoint coordinator, seeded from the latest stored checkpoint.

Errors

Returns DbError::Checkpoint if store.load_latest() fails. A store read error is surfaced rather than silently starting at (1, 1) and clobbering existing on-disk state. Ok(None) is the fresh-start path and is not an error.

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

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

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

Wired with a non-empty vnode_set to enable per-vnode markers and the epoch_complete durability gate.

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

Install the shared commit-marker store.

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

Record the assignment generation this coordinator is writing with. Forwarded to backend.write_partial so the split-brain fence can reject stale writers. Host sets this whenever a fresh AssignmentSnapshot rotates in.

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

Record this instance’s current local watermark, reported in every subsequent BarrierAck so the leader can compute the cluster-wide minimum. None disables the per-follower contribution — leader falls back to its own watermark (and the other followers’).

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

Vnodes this instance owns; drives marker writes. Also the default gate set until Self::set_gate_vnode_set is called.

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

Set the vnodes the leader’s durability gate checks (the full registry in cluster mode). Defaults to vnode_set when unset.

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

Begins the initial epoch on all exactly-once sinks.

Must be called once after all sinks are registered and before any writes occur. This starts the first Kafka transaction for exactly-once sinks. Subsequent epochs are started automatically after each successful checkpoint commit.

Errors

Returns the first error from any sink that fails to begin the epoch.

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

Inject prometheus engine metrics.

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

Performs a full checkpoint cycle (steps 3-7).

Steps 1-2 (barrier propagation + operator snapshots) are handled externally by the DAG executor 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 against the durable commit marker. Marker present → drive local commit (idempotent); marker absent → rollback. Runs on every node; the leader additionally re-announces the decision.

pub fn phase(&self) -> CheckpointPhase

Returns the current phase.

pub fn epoch(&self) -> u64

Returns the current epoch.

pub fn next_checkpoint_id(&self) -> u64

Returns the next checkpoint ID.

pub fn config(&self) -> &CheckpointConfig

Returns the checkpoint config.

pub fn stats(&self) -> CheckpointStats

Returns checkpoint statistics.

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

Returns a reference to the underlying store.

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

Performs a full checkpoint with pre-captured source offsets.

When CheckpointRequest::source_offset_overrides is non-empty, those sources skip the live snapshot_sources() call and use the provided offsets instead. This is essential for barrier-aligned checkpoints where source positions must match the operator state at the barrier point.

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 half of a cluster checkpoint: pre-commit + save + markers + ack, then wait for the leader’s commit/abort. Ok(true) = committed, Ok(false) = aborted/timed out.

Errors

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

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

Attempts recovery from the latest checkpoint.

Creates a RecoveryManager using the coordinator’s store and delegates recovery to it. On success, advances self.epoch past the recovered epoch so the next checkpoint gets a fresh epoch number.

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

Errors

Returns DbError::Checkpoint if the store itself fails.

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

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