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 partitions above every partition already admitted to the epoch 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 is still live, or is at or below the highest partition admitted to this epoch, which covers every partition revoked earlier in it. Both are source bugs. 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, so 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 pending_by_partition(&self) -> Vec<(PartitionId, usize)>

Unadvanced batches for every tracked partition, sorted by partition.

Every partition in the current epoch appears, including one with nothing unacknowledged, so the result doubles as the live set. A partition leaves it by being revoked or by a new epoch replacing it, never by falling quiet.

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

Partitions whose watermark is permanently stalled behind a failed batch, with the stall start on the checkpointer’s clock (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.