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 fn new(config: CheckpointConfig, store: Box<dyn CheckpointStore>) -> Self

Creates a new checkpoint coordinator.

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_counters(&mut self, counters: Arc<PipelineCounters>)

Sets the shared pipeline counters for checkpoint metrics emission.

When set, checkpoint completion and failure will update the counters automatically.

pub fn register_wal_manager(&mut self, wal_manager: PerCoreWalManager)

Registers a per-core WAL manager for checkpoint coordination.

When registered, prepare_wal_for_checkpoint() will write epoch barriers and sync all segments, and truncate_wal_after_checkpoint() will reset all segments.

pub fn register_changelog_drainer(&mut self, drainer: ChangelogDrainer)

Registers a changelog drainer to flush before checkpointing.

Multiple drainers may be registered (one per core or per state store). All are flushed during prepare_wal_for_checkpoint().

pub fn prepare_wal_for_checkpoint( &mut self, ) -> Result<WalPrepareResult, DbError>

Prepares the WAL for a checkpoint.

1. Flushes all registered ChangelogDrainer instances (Ring 1 → Ring 0 catchup)
2. Writes epoch barriers to all per-core WAL segments
3. Syncs all WAL segments (fdatasync)
4. Records and returns per-core WAL positions

Call this before checkpoint() and pass the returned positions into that method.

Errors

Returns DbError::Checkpoint if WAL operations fail.

pub fn truncate_wal_after_checkpoint( &mut self, current_positions: Vec<u64>, ) -> Result<(), DbError>

Truncates (resets) all per-core WAL segments after a successful checkpoint.

Call this after checkpoint() returns success. WAL data before the checkpoint position is no longer needed.

Errors

Returns DbError::Checkpoint if truncation fails.

pub fn wal_manager(&self) -> Option<&PerCoreWalManager>

Returns a reference to the registered WAL manager, if any.

pub fn wal_manager_mut(&mut self) -> Option<&mut PerCoreWalManager>

Returns a mutable reference to the registered WAL manager, if any.

pub fn changelog_drainers(&self) -> &[ChangelogDrainer]

Returns a slice of registered changelog drainers.

pub fn changelog_drainers_mut(&mut self) -> &mut [ChangelogDrainer]

Returns a mutable slice of registered changelog drainers.

pub async fn checkpoint( &mut self, operator_states: HashMap<String, Vec<u8>>, watermark: Option<i64>, table_store_checkpoint_path: Option<String>, source_watermarks: HashMap<String, i64>, pipeline_hash: Option<u64>, ) -> 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 as operator_states.

Arguments

• operator_states — serialized operator state from DagCheckpointCoordinator
• watermark — current global watermark
• table_store_checkpoint_path — table store checkpoint path
• source_watermarks — per-source watermarks
• pipeline_hash — pipeline identity hash

Errors

Returns DbError::Checkpoint if any phase fails.

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 smoothed_duration_ms(&self) -> f64

Returns the current smoothed checkpoint duration (milliseconds).

Returns 0.0 if no checkpoints have been completed or adaptive mode is not enabled.

pub fn unaligned_checkpoint_count(&self) -> u64

Returns the number of unaligned checkpoints completed.

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_extra_tables( &mut self, operator_states: HashMap<String, Vec<u8>>, watermark: Option<i64>, table_store_checkpoint_path: Option<String>, extra_table_offsets: HashMap<String, ConnectorCheckpoint>, source_watermarks: HashMap<String, i64>, pipeline_hash: Option<u64>, ) -> Result<CheckpointResult, DbError>

Performs a full checkpoint cycle with additional table offsets.

Identical to checkpoint() but merges extra_table_offsets into the manifest’s table_offsets field. This is useful for ReferenceTableSource instances that are not registered as SourceConnector but still need their offsets persisted.

Errors

Returns DbError::Checkpoint if any phase fails.

pub async fn checkpoint_with_offsets( &mut self, operator_states: HashMap<String, Vec<u8>>, watermark: Option<i64>, table_store_checkpoint_path: Option<String>, extra_table_offsets: HashMap<String, ConnectorCheckpoint>, source_watermarks: HashMap<String, i64>, pipeline_hash: Option<u64>, source_offset_overrides: HashMap<String, ConnectorCheckpoint>, ) -> Result<CheckpointResult, DbError>

Performs a full checkpoint with pre-captured source offsets.

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