Crate twg_offset_store 

Sink write-coverage store. Scaffold only.

Two kinds of state, and one shared invariant

Source position — only for sources that need it

Where a broker holds the position (a Kafka consumer-group offset, a Pulsar subscription cursor) this crate stores nothing: we commit to the broker and it redelivers what is unacknowledged. Keeping a parallel copy would create a second source of truth able to disagree with the authoritative one.

Where no broker-side position exists — batch file sets, Arrow IPC and Flight streams, table shares, and the planned REST/WSS sources — the source’s own cursor is persisted here. Some transports have both modes (Kafka with manual assignment, Pulsar readers rather than consumers), so which regime applies is a property of the configuration, not of the transport, and the source declares it.

Sink write coverage — always

Per-sink write coverage: which batches have been durably written to which destination.

This matters because of what the source ack already did. Once raw is durable the broker is acknowledged and moves on, so the broker can no longer tell us what a downstream sink is missing. This store becomes the only record that a batch failed to reach a destination. Lose that record and the batch is silently absent from that sink forever, with nothing upstream able to notice.

This is not low-stakes state that happens to be useful for recovery. It is the mechanism by which “we cannot lose a message” is true at all.

The invariant: position may LAG, never LEAD

The same rule governs both kinds of state, for the same reason, and everything else here follows from the asymmetry:

• Lagging — coverage says a batch is unwritten when it was, or a source position sits behind what was durably written. Costs a redundant replay, absorbed by record-id dedup. Safe.
• Leading — coverage claims a write that did not happen, or a source position sits ahead of what landed. Silent, permanent loss: recovery will never replay it, and for a broker-managed source the ack is long gone. Catastrophic.

Therefore coverage is recorded after the sink’s durable acknowledgement, never before or concurrently. Where a sink is transactional, the coverage update belongs in the same transaction as the write — that is the exactly-once path. Where it is not, write-then-record with at-least-once semantics, and let dedup absorb the window.

Gaps, not a high-water mark

A watermark cannot express what this store exists to record. If a sink writes batches 1, 2, 4 and 5 while 3 fails, a watermark of 5 reports full coverage and batch 3 is lost with no trace. Coverage must therefore be gap-aware: a set of covered ranges per destination, so a hole in the middle is representable and recoverable.

Recovery plans from the gaps. Compaction of adjacent ranges keeps the representation small, but must never close a gap it has not verified.

Two tiers, and only one of them is shared

Being explicit, because the distinction carries the whole design:

Durable tier — shared, authoritative, colocated with RAW. Not with the sink whose coverage it records. This is the whole point and it is easy to get backwards: a store living in sink X cannot record that sink X is down, so it fails exactly when it is needed and the gaps go unrecorded.

Raw is the correct home because its availability already bounds the pipeline’s. If raw is unavailable the source is not being acked and no new gaps are being created, so a coverage store sharing raw’s fate can never be unavailable while there is something it needs to record. Any other destination can fail independently, which is precisely the case coverage exists to survive.

Its storage follows raw’s: a table in the same database where raw is relational, an Iceberg table in the same catalog where raw is a table format or object store. This is what a rebalancing owner reads, what a recovery agent plans from, and what survives losing a pod.

Local tier — pod-local, ephemeral, never authoritative. An embedded store on the pod’s own disk, holding coverage this pod has written since its last checkpoint. Never read by another node. Never the only copy of anything.

The embedded store therefore needs no clustering, no replication and no shared mode — those would be solving a problem it does not have. It is a write-behind buffer with an inspection surface, not a distributed store.

This is worth stating because the embedded stores in question are libraries, not services: local disk, single process, no cluster mode and no object-store backend. Sharing state between nodes would mean running a consensus layer over one — a Raft library with the embedded store as its state machine — which is something to build and operate rather than adopt. Coverage does not need it. Coverage needs a shared DURABLE LOCATION, and object storage already is one.

Note what is and is not deferred. This store is not — it ships with the second sink, because from the moment the source is acked on raw durability a non-raw sink that fails has permanently missed data, and this is the only record of what. A multi-sink topology without it is silently lossy.

What is deferred is distributed recovery. A single recovery worker is the only claimant and needs no leases, elections or coordination substrate at all; coordination becomes necessary only when backfill volume demands several agents working disjoint windows concurrently. That option space is in twg-connector-core.

What happens on rebalance

A partition moves to a new owner. The new owner reads the durable tier — never the previous owner’s local store, which it cannot see and has no business seeing. Coverage the previous owner had written but not yet checkpointed is lost, so the new owner resumes from the last checkpoint and replays forward; record-id dedup absorbs the overlap. That is the same bounded cost as losing a pod, and it is why the checkpoint interval is the knob that matters.

Why concurrent writers during a rebalance are safe

A rebalance can briefly leave two pods believing they own a partition, so both may write coverage for it. This is safe only because coverage merges by union, never by last-write-wins:

• each writer records ranges it actually wrote, so the union is true;
• overlapping ranges from redelivery merge harmlessly, the data having been deduplicated by record id;
• last-write-wins would silently discard one writer’s ranges, converting a survivable race into exactly the over-claim the invariant forbids.

Union merge is therefore a correctness requirement, not an implementation convenience.

The local write-behind cache

Committing coverage straight to a remote store costs a network round-trip per commit, and — more painful in practice — is opaque: when latency spikes there is no local state to inspect, so “is the sink slow or is the commit path slow” is inferred from outside rather than observed.

Coverage is therefore written to a local embedded store on the pod, cheaply and often, and checkpointed to the durable store on an interval. This is safe only because of the lag/over-claim asymmetry: a cache behind the durable store costs replay, so losing a pod re-replays at most one interval per sink.

It is safe only if the ordering holds. Recording coverage in the cache before the sink’s durable ack, then losing the pod, produces exactly the over-claim this design forbids — the batch is unwritten and now also unrecorded as missing. The cache does not soften the invariant; it inherits it.

Boundaries

The cache is never the only copy. Anything existing only in a pod-local store is already lost at the moment it matters.

Nothing outside the pod reads the local tier. Not another ingest pod on rebalance, not a recovery agent, not an operator tool. Recovery runs as a separate role on separate pods (ADR-0050) and plans from committed coverage only. The moment something external depends on the local tier it has become authoritative by accident, and the ephemerality that makes it cheap becomes a liability.

Diagnostics

A local store makes coverage inspectable on the host: what this pod believes each destination has covered, which gaps are outstanding, and how far the local view has drifted from the last checkpoint.

Data model, and why it decides the store

Gap-awareness changes the shape of the data, and the shape decides which store suits. Two ways to hold covered ranges:

One key per destination, value is the serialised range set. Tiny fixed keyspace, but every update is a read-modify-write of a value that grows with the number of gaps. Overwrite-in-place, which favours a B-tree.

One key per range — (sink, topic, partition, start) -> end, ordered by start. Chosen. Appending a range is a single write; reading a destination’s coverage is a prefix scan in key order; merging adjacent ranges writes one entry and tombstones the others.

The second is preferred because it makes the common path a plain append and the read a sorted scan, rather than deserialising and rewriting a whole range set on every batch. It also keeps the cost proportional to the number of gaps rather than to the number of updates.

Choosing the durable backing

Every source and sink is optional, so no particular technology can be assumed present. The durable tier is therefore pluggable, with an implementation for each kind of destination raw can land in — and the choice follows raw, because colocation with raw is the rule.

Relational — best where available. Many small updates to a small dataset, with in-place mutation when a retry closes a gap, is exactly a transactional database’s shape, and the coverage update can share the transaction with the raw write. Ideal, but only where the deployment includes one.

Object with conditional write — the strongest general answer. Coverage as a single small object per sink, updated by read-modify-write under a compare-and-swap precondition on the object’s current version. Object stores now offer conditional writes, which is enough for safe concurrent updaters without any table format at all. It avoids every objection to the alternatives: one object rewritten rather than a proliferation of small files; no catalog and so no dependency on a metadata service being warm; no table maintenance. A failed precondition means re-read, re-merge and retry, which converges because coverage merges by union. Available wherever raw lands in object storage, which is most deployments that are not relational.

Append-only — an event-sourced model. Where the destination cannot mutate, coverage becomes a log: append the gap, append the correction on retry, fold on read. Workable, and the cost should be chosen knowingly — answering “what is missing” means folding rather than selecting, so it wants a materialised view, which is another thing to maintain and to be stale.

Table format — viable, weakest fit. Frequent checkpoints produce many small files and metadata churn needing compaction, for a dataset that is tiny. And where the catalog’s metadata layer depends on a separate compute resource being warm, the coverage store inherits that availability — the same failure this design avoids by not living inside the sink it describes, arriving from another direction. Its advantage, being queryable by any engine with no service to run, is real and makes it a reasonable mirror for inspection. It is a poor primary.

Where raw is object-store or table-format backed, coverage as an Iceberg table alongside it earns more than durability:

Choice of embedded store

An LSM-tree store, pure Rust throughout, sharing the LZ4 implementation twg-wire-compression already uses.

Append-mostly writes, prefix scans in key order, and tombstones on merge are precisely an LSM’s strengths. An earlier draft of this design used a high-water mark per destination — one key, overwritten constantly — and warned that a small hot keyspace under repeated overwrite is the case an LSM handles least well. That concern was an artefact of the watermark model. A watermark cannot express a gap, so it was never a viable model, and the model that replaced it happens to be the one an LSM suits.

Purity also favours it: the B-tree alternative pulls libc, though that margin is narrow and either would serve.

Why not a networked store

A Redis-style store, however backed, would defeat the purpose. This cache exists to avoid a network round-trip per commit and to give a host-local path to inspect when latency spikes — a remote store reintroduces the round trip and moves the inspection point off the host, leaving you interrogating something remote to find out why something remote is slow.

Its weaker durability would not be the objection: lag is safe under the invariant above, so losing recent writes on a crash costs only a replay that dedup absorbs. The objection is simply that it is not local.

Nor is there shared state to justify one. Coverage is partitioned by topic and partition, so each pod owns its slice outright; when a partition moves on rebalance the new owner reads the durable checkpoint, and a stale cache on the previous owner costs a replay from that point. Cross-pod coordination — leases, elections — is a separate concern with a separate answer, and does not travel with this data.

The store stays swappable behind the trait regardless: the reasoning above is about fit, not about a property worth locking in.

Re-exports

pub use coverage::CoverageError;

pub use coverage::CoverageStore;

pub use coverage::CoveredRange;

pub use coverage::InMemoryCoverage;

Modules

coverage

Gap-aware per-sink write coverage: CoverageStore, CoveredRange, and an InMemoryCoverage backend.