Crate twg_connector_core

Expand description

Transport-neutral Source/Sink, flow control, observability, recovery.

The sink module implements the delivery seam (Sink, Receipt, OffsetSpan, SinkError); the flow-control, recovery, and coordination machinery described below remains design intent pending its delivery phase.

§Source-acknowledgement invariant

The source position — the broker’s consumer group offset or subscription cursor — is committed ONLY after a durable sink acknowledgement, never on send, buffered, or enqueued. Until it is committed the broker will redeliver, which is what makes the guarantee hold across a crash. “Durable” is per sink type: server-side ack (Zerobus/Flight), transaction commit (Postgres), flush+close (file/Parquet). A crash between send and durable must replay, not lose. This is the backbone the recovery story rests on.

§Which sink gates the source (multi-sink DAG)

Waiting for every sink would let the slowest throttle the source — the exact coupling raw-first exists to break. So the gate is the sink that owns replay authority:

Raw-first topology: the source acks when the PRIMARY-RAW sink acks durably. Secondary sinks ack independently and async; they are recovered FROM raw if they lag, so they never gate the source.
Single-sink topology: that sole sink’s durable ack gates the source.

§Granularity (configurable)

Default batch-granular: the source advances to the batch high-water offset once the whole batch is durably acked. A partial failure replays the whole batch; dedup on _twg_record_id absorbs the records that already landed (at-least-once + idempotent replay). Per-record is the low-latency opt-in for critical streams that cannot tolerate whole-batch replay.

§Durability requirement vs sink role

durable = true is mandatory for a primary-raw sink and enforced at config validation — a sink that cannot confirm durable acknowledgement cannot hold the primary-raw role (this is part of the ReplayableSink capability, alongside being readable for recovery). Non-raw secondary sinks MAY disable durable to trade safety for throughput, because raw can rebuild them; that opt-out is explicit and documented, never a silent default.

§Source position: who persists it is source-dependent

Not every source has a broker to hold its position, so a Source declares which of two regimes it is in:

Broker-managed. A consumer-group offset or subscription cursor lives server-side; we commit to the broker and it redelivers anything unacknowledged. Kafka consumer groups and Pulsar subscriptions. Note both have modes that opt OUT — Kafka with manual partition assignment, Pulsar readers rather than consumers — so this is a property of the configuration, not of the transport.

Store-managed. No server-side position exists, so we persist the source’s own Cursor ourselves: a file set plus a position within the current file; a table version and file list; a stream token; a page cursor or timestamp watermark. Batch, Arrow IPC, Flight, table-share and the planned REST/WSS sources are all in this regime.

§One invariant covers both, and covers sink coverage too

All position state may LAG. None may LEAD.

source position ahead of what was durably written to raw -> on restart we resume past records that never landed. Silent loss.
source position behind -> we re-read records already written. Duplicates, absorbed by record-id dedup. Safe.
sink coverage ahead (claiming an unwritten batch) -> recovery never replays it. Silent loss.
sink coverage behind -> a redundant replay. Safe.

So the rule is the same everywhere and for the same reason: commit position only after the durability it claims, never before, never concurrently. That is what makes a local write-behind cache safe for either — a lagging cache costs replay, a leading one loses data.

§Replayability of the origin decides the failure mode

What losing position actually costs depends on whether the source can be re-read, which is a third property worth stating separately:

Broker-managed — the broker redelivers. Losing our view costs nothing.

Store-managed, replayable origin — files, table shares, Iceberg reads. Losing position costs reprocessing from an earlier point; dedup absorbs it. The least critical case, because the origin is still there.

Store-managed, NON-replayable origin — a Flight or IPC stream, or a socket feed, where the producer cannot rewind. Losing position after the producer has moved on loses data outright, because nothing upstream retains it.

That last category is a genuine constraint rather than a caveat: a non-replayable source can offer at-least-once only as far as what has been durably written, and cannot recover beyond it. Such a source should not hold the primary-raw role unless its producer supports resumption, and config validation should say so rather than letting the topology look sound.

§Content dedup: catching duplicates our record id cannot

Opt-in per stream, contract-gated, best-effort. Never a default.

The record id hashes the source POSITION — cluster, topic, partition, offset, payload — which makes broker redelivery idempotent, because a redelivered record arrives at the same offset and hashes the same. It does nothing for a duplicate created UPSTREAM of the broker: a producer’s own at-least-once retry publishes the same logical record at a new offset, so it hashes differently and passes straight through.

Hashing the payload alone catches it. Seeing a content hash already seen within a window means the record is an upstream duplicate and can be dropped, which is the only place in this design that repairs someone else’s delivery semantics rather than our own.

§The hazard that makes this opt-in

Two genuinely distinct events can have byte-identical payloads. Two heartbeats. Two “clicked” events carrying no timestamp. Two identical sensor readings a second apart. Content dedup drops the second, and that is data loss CAUSED BY dedup — the worst kind, because the system reports success.

So it is enabled per stream and only where the payload is known to be unique, which the data contract is the right place to assert: a stream declaring a natural key or a uniqueness guarantee may enable it, and one that does not may not. Enabling it globally would eventually eat legitimate records from whichever stream first emits a repeated payload.

Where a payload is nearly-unique, hashing a declared subset — the natural key plus an event timestamp — is safer than hashing the whole body, and is the configuration to prefer.

§Best-effort by construction

The seen-set is bounded by time and size, so it can only catch duplicates arriving within the window; a retry hours later passes. It may also be unavailable, in which case the pipeline continues WITHOUT dedup rather than stopping — degrading a quality improvement, not a guarantee.

Nothing downstream may therefore depend on it. Sinks remain idempotent on record id regardless. Content dedup reduces duplicate VOLUME; it does not establish exactly-once, and describing it as though it did would invite exactly the reliance that makes its best-effort nature dangerous.

§Why it needs to be shared, when coverage does not

An upstream retry can be published to any partition — a different key, or a producer that round-robins — so it may be consumed by a different pod entirely. A pod-local seen-set would miss precisely the duplicates worth catching.

This is the one piece of per-record state that genuinely requires cross-pod visibility, and it is worth separating from the two that do not:

coverage needs a shared durable LOCATION, not shared memory — each pod owns disjoint partitions, so a local buffer plus a durable store suffices;
coordination (recovery leases, leader election, load tuning) needs LINEARISABILITY, not throughput — consensus-shaped, small, and separate;
content dedup needs a fast SHARED SET with high write volume and no durability requirement at all, since losing it costs only missed dedup.

Three different requirements. A single store chosen for one of them will over-serve the others; if one is run anyway, the others may ride along, but the justification should be named rather than assumed.

§What to use — deliberately not decided yet

This does not need deciding until recovery is DISTRIBUTED across agents. Recovery itself is not deferrable — it ships with the second sink, since a non-raw sink that fails after the source was acked has permanently missed data. But a single recovery worker is the only claimant of every window and needs no leases, elections or coordination substrate whatsoever. Coordination becomes necessary only when backfill volume demands several agents working disjoint windows at once.

Choosing before then means choosing before the shape is known, so the option space is recorded here to make the eventual decision informed rather than defaulted into.

Not platform-locked. A leases API provided by an orchestrator solves coordination neatly where one is present, but making a core capability depend on a particular deployment platform is wrong for an engine published as independent crates. Whatever is chosen must run wherever the binary runs.

Writing our own consensus is not on the table. That claim is about implementing Raft from scratch, which is among the hardest things to get right and where the failure mode is not that it does not work but that it works until a partition. It is emphatically NOT an argument against building on an existing consensus library: a pre-1.0 version in this ecosystem signals API-churn policy rather than immaturity, and the mature Rust implementation is used in production by real systems. Those are different claims and should not be conflated.

The leading candidate shape: embedded, replicated, self-contained. A library embedding a replicated SQL store with Raft consensus, offering distributed locks, TTL’d key-value caches and counters, with periodic encrypted backup to object storage and restore from it. That covers coordination and coverage checkpointing in one dependency, runs anywhere, and adds no separate service to operate — while remaining runnable AS a service where that suits the topology better.

Three risks to weigh rather than discover:

Write throughput is Raft-bounded — two network round-trips per write, so tens of thousands per second. Ample for coordination and for checkpointing coverage on an interval, and ample for content dedup too ONCE BATCHED: the pipeline is batch-oriented throughout, so a batch costs one bulk lookup and one bulk write rather than one of each per record, which divides the traffic by the batch size. Reads do not traverse consensus at all, being served locally from replicated state, so only writes cost anything. Still verify the batched rate against the store rather than assuming — but the per-record framing overstated the problem by three orders of magnitude.
Membership churn — resolved by sizing the tier rather than autoscaling it; see the topology modes below.
Durability across full shutdown — a cluster whose members all stop must have flushed to durable storage first, so graceful shutdown has to trigger a backup and startup has to be able to restore from it. This is the difference between a cache that can be rebuilt and state that is simply gone.

§Coordination topology: quorum as a role, not a tier

Rather than a separate coordination deployment, quorum membership is an additional role a subset of workers hold. Workers autoscale freely for throughput; membership is a slow-moving overlay on top of that, changing only when it must rather than on every scale event.

A node determines its roles on start by checking in: worker, recovery worker, and — where the cluster is short of members — additionally a quorum member.

A dual-role node lowers its throughput budget. Raft leadership means an fsync per commit and replication to followers, which contends with high-throughput ingest for the same disk and CPU. A node holding quorum therefore runs with a reduced admission budget and smaller batches, trading its own throughput for the cluster’s stability. This needs no new machinery: the byte budget and batch sizing are already per-pod, so it is a different configuration rather than a different code path.

Membership changes far less often than pod count, which is what makes this workable. A three-member cluster tolerates one member being gone, so scaling three workers down to two leaves a three-member cluster with one unreachable and quorum intact — no reconfiguration at all. Only dropping to a single worker genuinely loses quorum. Growth likewise: membership expands to three when a third worker is available and then stops, since beyond three the next useful size is five.

Scale-down of a quorum member is the hard case, and it is where this design earns or loses its keep. An autoscaler does not know which pods hold quorum and may terminate one; terminating two before reconfiguration completes leaves the cluster stuck with no quorum to reconfigure itself. Three things are needed:

stable assignment ordering so the role settles on the same nodes rather than shuffling — lowest stable identifier, join order, or an orchestrator-provided ordinal where one exists;
graceful handoff before termination, with a shutdown hook that transfers or relinquishes membership and a grace period long enough to complete it;
treating a terminating member as gone rather than failed, so the cluster does not wait out a failure detector for a departure it was told about.

Where the platform removes instances in a predictable order — highest ordinal first, as stateful workloads typically do — assigning quorum to the lowest ordinals means scale-down naturally takes non-members first, and the hard case mostly stops arising. That is an optimisation to use where available, not a dependency to rely on.

Bootstrap is the remaining edge: a node cannot ask the cluster what role it holds before a cluster exists. The first node self-elects a single-member cluster and subsequent nodes join it, which is the standard sequence and the reason mode 1 below is a genuine resting state rather than only a transition.

Four modes:

0 — not run. Coordination features are off. Recovery leases and content dedup are unavailable; everything else works. The correct choice for a single-pod deployment or where neither feature is wanted.

1 — single node, no fault tolerance. Fully functional and, counter to intuition, the FASTEST configuration: quorum of one means a write commits without a network round trip. Durability comes from periodic backup to object storage, so losing the node costs at most one backup interval.

2 — forbidden. Config validation must refuse it. Quorum of a two-node cluster is two, so both nodes must be up for any write: losing either halts writes entirely. That is strictly worse availability than a single node, which at least keeps serving alone, while also paying the replication cost. It is the one size that is worse than both its neighbours, and it is chosen by accident — by an autoscaler stepping through it, or by someone reasoning that two must be safer than one.

3 — full features. Quorum of two tolerates one node failing. The minimum genuinely highly-available configuration. Beyond it, odd sizes only: five, seven. An even size adds a voter without adding failure tolerance.

Transient two is not resting at two. Growing one to three passes through a two-member configuration during reconfiguration, which is unavoidable where membership changes one server at a time. That window is brief and taken deliberately while healthy; the prohibition is on two as a resting state, which is what an autoscaler would produce.

§Why single-node is genuinely viable here

Not a degraded compromise, because this workload tolerates losing coordination state — a property that falls out of decisions already made rather than one designed for:

leases expire anyway, so losing them costs waiting out a TTL, which is the same thing that happens when a lease-holder dies;
content dedup is best-effort, so losing the seen-set costs missed dedup, which is the tolerable direction;
coverage is durable elsewhere — this tier only ever buffers it.

Nothing here is the sole record of anything that matters. Fault tolerance therefore buys continuity of service rather than protection against loss, which is why one node is a legitimate choice and not merely a starting point.

§Batching makes dedup fit, and it is safe BECAUSE dedup is best-effort

Batching introduces two races, and both fall on the tolerable side of the asymmetry established above:

two records within one batch sharing a content hash both read as unseen at lookup time — resolved by deduplicating within the batch before the bulk write, which is local and free;
two pods processing the same hash concurrently both read unseen and both pass the record through — a MISSED dedup, which is a false negative, which is safe.

Neither race can produce a false positive, so neither can drop a legitimate record. A stricter guarantee would need per-record linearisable check-and-set, which is exactly what would exceed the throughput budget. The best-effort nature is therefore not a limitation that batching works around — it is the property that makes batching admissible at all.

§The binding constraint is memory, not throughput

With throughput resolved, what actually bounds content dedup is the size of the seen-set: ingest rate multiplied by the retention window. A window of minutes at high ingest is gigabytes; a window of hours is not affordable at any interesting rate, sharding notwithstanding.

So the window is sized to the expected upstream retry interval, not maximised. A producer’s at-least-once retry follows a timeout — seconds, occasionally minutes — so a window in minutes catches essentially all of what this feature exists to catch, and extending it buys diminishing returns at linear cost. Treating “longer is safer” as the default here is how the feature becomes unaffordable.

Whether to build it at all should still follow a measurement of duplicate rates rather than precede it: if upstream retries are rare, the capability may not justify its cost.

§One correctness constraint on any dedup structure

No probabilistic membership. A Bloom or similar filter answers “probably seen” with false positives — claiming a record was seen when it was not, which DROPS A LEGITIMATE RECORD. The tolerable error runs the other way: a bounded LRU or TTL cache has only false negatives, since an evicted entry reads as unseen, costing a missed dedup that a downstream idempotent sink absorbs.

Exact membership with eviction. The space saving from a probabilistic structure buys silent data loss, which is the one thing this design consistently refuses.

§Quarantine is non-blocking by construction

The quarantine / DLQ lane must NEVER apply backpressure to the clean path. Firing the write as async is not enough — if the two lanes share a bounded channel or budget, quarantine pressure still reaches clean downstream. So the lanes are genuinely independent: quarantine has its own buffer and its own backpressure, and a defined OVERFLOW policy so that when quarantine cannot keep up, quarantine sheds — never clean.

Live overflow → drop-with-count (loud metric). Safe because raw is the system of record: the rejected rows are in raw and recoverable; quarantine is a fast-inspection convenience view, not authoritative. The authoritative record of DQ outcomes is the dq-audit trail, which is a durable sink and unaffected.

Quarantine is NOT a gating sink — it never sits between source and ack (only primary-raw gates). So quarantine shedding under overflow cannot threaten the source-ack invariant; quarantine was never on the ack path.

Recovery interaction — two modes for the reconstruction pass:

rehydrate (default): recovery backfills the missing quarantine rows from raw, so the convenience view becomes complete for the recovered window.
skip-DLQ / drop (opt-in, for recovery under pressure): replay clean data only, do not re-quarantine rejects. Clean recovery is unaffected, but the quarantine view stays permanently incomplete for that window (rows remain in raw; nobody re-derives them unless recovery is re-run without skip). The flag surfaces this consequence — the quarantine table is not authoritative for windows recovered in skip mode. dq-audit is unaffected.

§Flow control: three coordinated subsystems, not one lever

Memory relief by a single mechanism does not hold under load. A production predecessor established that three subsystems are required together — a backpressure state machine that throttles the source, admission control that bounds in-flight bytes, and (per transport) staged subscription that bounds what the broker prefetches at all. Each covers a failure the others do not.

§Backpressure state machine

Four states — Normal, Degraded, Critical, Recovery — driven by memory ratio, channel fill and consecutive write failures, each reducing the effective batch size:

Normal memory <80%, channels <70%, no failures -> 100% batch Degraded memory 80-90% OR channels 70-90% OR failures -> 50% batch Critical memory >90% OR channels >90% -> 25% batch and sets a shared throttle flag pausing the source Recovery conditions clear for ~30s -> 75% batch, ~60s

Recovery is a state, not a transition. Returning straight to Normal the moment conditions clear re-saturates the pipeline and produces a throughput sawtooth. The hysteresis before entering Recovery, and the time spent there at reduced batch size, are what damp the oscillation — and are the reason this is four states rather than three.

§Working-set heuristic: do not trust the memory ratio alone

High memory with EMPTY channels is allocator fragmentation, not live data. Throttling on it is a false positive that costs throughput for nothing. When both source and decoded channels are near-empty (~~<5%) while memory is high (~~>80%), the pressure is stale and backpressure is NOT triggered. This is the difference between a memory signal that is actionable and one that merely correlates.

§Stuck detection, separate from backpressure

A distinct detector rather than a fifth state: memory high AND channels empty AND throughput zero, sustained. It escalates — warn, then report unhealthy so the orchestrator can recycle the pod, then force exit — and clears only after sustained progress (debounced) so it cannot oscillate. Thresholds are workload-tunable: they assume batch writes complete in minutes, so a pipeline whose sink routinely takes longer must raise them or reduce batch size.

This refines the liveness rule: a genuine deadlock IS a liveness failure and should fail /live. A merely saturated sink is not, and must not.

§Byte-budget admission control

A semaphore sized in bytes, not messages — message sizes vary by orders of magnitude, so a message count bounds nothing useful. Permits are acquired in proportion to a record’s estimated expanded size and released on downstream ack.

Estimate from the p95 expansion ratio (expanded/compressed), not the mean. A few highly compressible records drag the mean well above typical and waste budget on over-reservation. Use a conservative fixed ratio until enough samples exist. record_size_est_bytes in config is the cold-start seed and floor only.
Warm up. Start at a fraction of the budget, expand after N successful acks or a timeout, whichever first. Without it, a pod restarting against a deep backlog prefetches at full capacity and OOMs before reaching steady state.
Oversized records go to the DLQ, judged against grantable capacity, not total. During warm-up a record that would fit the full budget still cannot be admitted, and deferring it stalls everything behind it.
Acquire inside a select! alongside ack processing. Blocking on acquire outside it deadlocks when the budget is full, because the acks that would release permits never get to run.

Saturation sustained high while memory stays low means the SINK is the bottleneck, not memory — a distinct diagnosis the metric should make obvious rather than leaving to inference.

§Recovery runs as a role, not as a background thread

A pod runs in one of three modes, same binary, chosen by config:

ingest consume the source, write raw and the live sinks recovery read raw, replay bounded windows into lagging sinks; do NOT consume the source at all both the simple single-deployment case, and the default

§Why separate the roles

Recovery competes with live ingest for memory, CPU and sink write capacity. In a combined pod, recovery pressure trips the backpressure machine and throttles live ingest — which is precisely the coupling raw-first exists to break. Separating them lets live ingest hold its latency while recovery catches up on its own budget, and lets a large backfill run at whatever scale it needs without touching the live path.

both remains the default because a deployment with no lag has nothing to recover and should not need two deployments to say so.

§Distributing recovery across agents

Recovery already replays in bounded windows (time- and row-limited). That bound is what makes the work partitionable: a window is a unit of work, so multiple recovery pods can process different windows concurrently.

Coordination lives in the offset store, which is already colocated with the primary-raw sink and is therefore the one place with a consistent view of what each destination has covered. A recovery agent claims a window by lease, works it, and marks it complete. Leases expire, so an agent that dies mid-window does not strand it.

Lease expiry is safe because of deterministic record IDs. A re-claimed window may replay rows a dead agent already wrote; dedup absorbs them. Without idempotent replay this design would need distributed locking and would not be worth having.

§Autoscaling recovery independently

Ingest scales on its own pressure signal. Recovery needs a sibling: a recovery backlog metric — the volume of unclaimed windows, or the lag of the furthest-behind destination — that an autoscaler consumes to add agents, and that permits scale to zero when nothing is behind. Recovery capacity should cost nothing when there is nothing to recover, which is only possible because it is a separate deployment.

The steady state this produces: ingest pods serving the live stream at their SLA, and a recovery deployment that expands to drain a backlog and disappears when it is drained.

§Concurrent writes from both roles

Live and recovery writes reach the same sink at once. For sinks that dedup on record ID this is safe by construction. For append-only targets it produces out-of-order arrival with correct identity, which is acceptable but must be stated rather than assumed — a downstream consumer relying on arrival order is relying on something this design does not promise.

Also here: the DLQ reason-codes (decode | contract | dq | retry-exhausted), the primary-raw role, and offset-store colocation.

On terminology, since two different things get called “offsets”: the SOURCE position is the broker’s — a consumer group offset or subscription cursor, committed once raw is durable, with re-delivery of anything unacknowledged being the broker’s job. What twg-offset-store holds is per-SINK write COVERAGE, which is a different quantity and, once the source is acked, the only record of what a destination is missing.

It lives with RAW, never with the sink it describes — a store inside sink X cannot record that sink X is down, and would fail exactly when needed. Raw’s availability already bounds the pipeline’s, so a coverage store sharing raw’s fate is never unavailable while there is something to record.

Its invariant is asymmetric: coverage may LAG but must never OVER-CLAIM. A stale record costs a replay that dedup absorbs; a record claiming a write that did not happen is silent permanent loss, because the broker has moved on. That asymmetry is what makes a local write-behind cache safe — and it is also why coverage is recorded only AFTER the sink’s durable ack, never before.

§Considered: a consensus store for coordination

Not adopted; the reasoning is recorded because the instinct is right and the shape may return.

Recovery window leases, recovery-driver leader election and membership are genuinely distributed state needing linearisable compare-and-swap — claim a window, expire a lease, elect a driver. A Raft-backed key-value store is the textbook fit, and they currently live in the offset store by default rather than by design.

§Why not a per-record index

The tempting shape — key instance || record_id || sink, value a content hash — is a dedup index rather than an offset store, and it does not survive arithmetic. It is one entry per record per sink, so a pipeline at a few hundred thousand records per second across three sinks needs over a million consensus writes per second; Raft serialises every write through a leader and commits in the tens of thousands per second. It would become the pipeline bottleneck by two orders of magnitude, and it grows without bound absent a retention policy.

It is also redundant twice over. The record id is ALREADY a content hash over cluster, topic, partition, offset and payload, so a key containing that hash with a value holding a content hash is circular. And deduplication is already solved without a store: deterministic ids plus idempotent sink writes mean the sink’s own index answers “have I seen this”, which is why replay is safe.

Consensus is the right tool for the SMALL consistent state — leases, elections, membership — and the wrong tool for the high-volume per-record path.

§The tension to resolve before adopting one

The offset store is deliberately colocated with the primary-raw sink so that offset commit can be TRANSACTIONAL with the data write. Moving offsets into a separate consensus store breaks that and reintroduces two-phase commit between two systems. Leases and elections are additive and carry no such conflict; offsets are not. Adopt for the former, and treat any proposal to move the latter as a reversal of a decision made for a specific reason.

§If one is adopted

It must be pure Rust to survive the purity gate. Raft implementations pairing with an embedded C++ store fail it outright, as does anything pulling system TLS. A pure-Rust Raft over a pure-Rust embedded key-value store is the combination to look for, and maturity of both parts matters more than convenience of a pre-assembled crate.

Re-exports§

pub use sink::OffsetSpan;
pub use sink::Receipt;
pub use sink::Sink;
pub use sink::SinkError;
pub use sink::SinkId;

Modules§

sink: The transport-neutral delivery seam: Sink, Receipt, OffsetSpan, and SinkError (ADR-0012/0038).