Module checkpoint 

Checkpointing: acknowledgement tracking and watermark commits.

One AckRef is created per source poll batch (by an AckIssuer on the polling thread) and cloned into every record derived from that batch — filter drops, flat_map fan-out, and multi-sink routing all compose through plain Clone/Drop. When the last clone drops, the batch resolves with the worst status observed and a message flows to the Checkpointer, which advances per-partition contiguous watermarks (PartitionTracker) and hands Source::commit its positions on an interval.

Invariants (see docs/DESIGN.md and CLAUDE.md):

• the tracker stays synchronous and tokio-free (loom-tested);
• the ack path never blocks (unbounded channels, atomics only);
• a watermark never advances past an unacknowledged or failed batch, including across rebalances (assignment epochs make stale acknowledgements harmless).

The concurrency-bearing primitives are model-checked with loom:

RUSTFLAGS="--cfg loom" cargo test -p etl-core --release checkpoint::loom_tests

Structs

AckIssuer

Creates acknowledgement handles on pipeline threads.

AckMsg

Message sent to the checkpointer when the last AckRef clone of a batch drops.

AckRef

Refcounted handle to a batch’s acknowledgement state. Clone = one atomic increment, no allocation; records hold one each.

AckSet

A fail-safe collection of acknowledgement handles for data that has not been durably written yet (encoded chunks, sink batches).

BatchId

Identity of one source poll batch within a partition and assignment epoch. Epochs are bumped on every rebalance so acknowledgements from revoked assignments are recognized as stale and discarded.

Checkpointer

Aggregates batch resolutions into per-partition committable watermarks.

DrainStats

Counters from one Checkpointer::drain call, for metrics.

PartitionTracker

Contiguity tracker for one partition within one assignment epoch.

Enums

AckStatus

Resolution status of a batch. Forms a “worst-of” lattice: once any record of the batch fails, the batch is failed.

ResolveOutcome

Result of applying one resolution to the tracker.