Struct Checkpointer 

pub struct Checkpointer { /* private fields */ }

Aggregates batch resolutions into per-partition committable watermarks.

use spate_core::checkpoint::{AckStatus, Checkpointer};
use spate_core::record::PartitionId;

let mut cp = Checkpointer::new();
let p = PartitionId(0);
cp.begin_epoch(&[p], 1);
let mut issuer = cp.handle();

let ack = issuer.issue(p, 99); // batch covering offsets ..=99
drop(ack); // all records delivered

cp.drain();
assert_eq!(cp.take_watermarks(), vec![(p, 100)]);
assert_eq!(cp.take_watermarks(), vec![]); // idempotent until new acks

Implementations

impl Checkpointer

pub fn new() -> Self

A checkpointer with no assignment. Call begin_epoch when the source reports its first assignment.

pub fn handle(&self) -> AckIssuer

An issuer handle for a pipeline thread.

pub fn begin_epoch(&mut self, partitions: &[PartitionId], epoch: u32)

Start a new assignment epoch covering exactly partitions. Every rebalance bumps the epoch; in-flight batches from earlier epochs resolve as stale and their offsets are re-delivered by the source (at-least-once). Epochs must be strictly increasing.

Ordering contract: the runtime calls this before distributing the new assignment’s lanes to pipeline threads, so issuers observe the new epoch before issuing for it.

pub fn extend_epoch( &mut self, partitions: &[PartitionId], ) -> Result<(), FatalError>

Add partitions to the current epoch without disturbing existing trackers (additive lane gains — SourceEvent::LanesAdded). The epoch does not change, so in-flight batches for existing partitions keep resolving; only genuinely new partitions may be added — a partition revoked mid-epoch can only return in a new epoch, and re-adding a live partition would discard its ack state.

Ordering contract: as with Checkpointer::begin_epoch, call this before distributing the new lanes to pipeline threads.

Errors

Returns a FatalError if a partition was already admitted to this epoch, whether it is still live or has since been revoked. Both are source bugs, and the revoked case is the dangerous one: its tracker is gone, so it looks fresh, while issuers keep their sequence counters until the epoch changes. Admitting it would pair a mid-sequence registration with a tracker expecting zero, and PartitionTracker::register would panic on the controller thread — taking down the pipeline with a message naming neither this method nor the contract that was broken.

pub fn revoke(&mut self, partitions: &[PartitionId])

Drop tracking for revoked partitions mid-epoch (partial revocation or shutdown). Later resolutions for them are discarded as stale. A partition revoked this way can only return in a new epoch.

pub fn drain(&mut self) -> DrainStats

Apply all pending registrations and resolutions.

Two passes exploit the causal order guaranteed by AckIssuer (registration is sent before the batch’s AckRef exists): a resolution whose registration has not been drained yet is retried once after re-draining registrations; if it is still unknown, the driver is buggy and the resolution is counted and dropped.

pub fn take_watermarks(&mut self) -> Vec<(PartitionId, i64)>

Watermarks that advanced since the last call: (partition, committable offset) pairs ready for Source::commit. Empty when nothing moved — callers skip the commit entirely.

pub fn pending(&self, partition: PartitionId) -> usize

Unadvanced batches for one partition (backpressure trigger).

pub fn max_pending(&self) -> usize

The largest per-partition pending count.

pub fn stalled_partitions(&self) -> Vec<(PartitionId, Instant)>

Partitions whose watermark is permanently stalled behind a failed batch, with the stall start (health-probe input).

Trait Implementations

impl Debug for Checkpointer

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

Formats the value using the given formatter.

impl Default for Checkpointer

fn default() -> Self

Returns the “default value” for a type.