ADR-0060: Pipelined fire-and-forget sink ack — durable confirmation on a background lane

Reference

Refines ADR-0038 (source-ack invariant: advance offset only on a durable sink ack) with the mechanism by which that ack is awaited, so the invariant does not serialise the sink. Builds on ADR-0001 (BatchSink / OffsetSpan), ADR-0012 (Sink delivery trait), and interacts with ADR-0053 (position may lag, never lead), ADR-0057 (recovery-from-raw as a precondition for a second sink), ADR-0049 (Pulsar flow control / pressure-as-scale-signal), and ADR-0039 (per-stage throughput digest). Formalises §10 of ../blueprints/how-a-sink-should-work.md.

Lands with the BatchSink substrate and the first sinks in Phase 5a of the delivery plan; the Zerobus/Flight streaming sinks that most need it arrive in Phase 7. Touches twg-stream-arrow (BatchSink substrate), twg-connector-core (Sink), twg-offset-store (delivered-offset watermark), and twg-observability (ack-lane metrics).

Context

The governing invariant (ADR-0038) is that the source position advances only after a durable sink acknowledgement. The naive implementation of that invariant — for a pipelined transport whose SDK acknowledges asynchronously — is to await “wait until the server has durably acked offset N” inline, after every submit, and advance the offset when it returns. That is correct but catastrophically slow: it serialises the sink to one in-flight batch per stream.

On the predecessor engine thalweg replaces, this cost a ~20,000× throughput loss (≈4 batches/sec against a 100 MB/s pipe, in-flight depth pinned at 1). It did not present as “the sink is slow.” It presented as everything else:

• Server-ack-timeout false-positives. Each transient network hiccup was fully exposed to the caller-side deadline; the “fix” was inflating the timeout, a bandaid that trades false positives for slow detection of real outages. Under pipelined use the SDK’s own buffer absorbs the hiccups.
• Backpressure trips at trivial throughput. The batcher could not push through the ack, so source-side pending grew, the throttle fired, and partition parallelism was shed — the transport looked throttled while the real bottleneck was a per-batch .await at the sink.
• Adaptive batching fighting the wrong constraint. Batches flushed on time, not size, because the batcher spent its time waiting between batches rather than accumulating records.

Naively “just make it async” is not enough: the offset advance, the offset-store write, and the per-sink coverage update (ADR-0053) currently fire on the write-success path, which under the serial shape meant “durably acked.” Firing them on submit-only would advance the recovery watermark past records the sink has not confirmed — a crash-recovery data-loss bug. The fix must preserve when those calls fire (only on durable confirmation) while removing where they fire from (off the ingest hot path).

Decision

Ingest is fire-and-forget; the durable wait runs on a per-stream background ack lane; the source advances on the delivered offset.

Concrete shape:

1. Fire-and-forget ingest. send / the BatchSink write returns as soon as the SDK accepts the batch. It advances a per-stream submitted high-water and hands a PendingAck { offset, on_confirm, submitted_at } to a bounded per-stream channel. Channel-full ⇒ send awaits — the SDK’s designed “block if too many messages queued” backpressure, surfacing as source-side awaiting rather than round-trip serialisation.
2. Per-stream background ack lane. Drains the channel in bursts (blocking recv for the first item, then non-blocking try_recv until empty), coalesces the burst, and confirms it with one round-trip on the largest offset (offset monotonicity ⇒ that single wait confirms every lesser item). It then fires each on_confirm in ascending-offset order, advances a per-stream delivered high-water, and records per-batch delivery latency from submitted_at.elapsed().
3. The confirmation callback carries the destination-specific commit, and cross-lane bleed is a compile error, not a review checklist item — model it as a typed enum of lane tokens (raw-anchor commit vs. secondary/derived coverage vs. none), so a secondary-lane stream statically cannot hold a raw-anchor token and vice-versa. Adding a future destination is an exhaustive-match enum arm.
4. delivered, not submitted, is the durable watermark. Recovery (ADR-0053, ADR-0057) reads delivered; submitted exists only to expose an observable ingest-side high-water and to compute pending = submitted − delivered. Any durability path that reads submitted is a bug.
5. Poison ⇒ bounded-graceful teardown. A permanent transport error exits the ack task and drops the receiver; the next send sees the closed channel, the stream is treated as stale, and the existing supervisor rebuild path recovers it. No hard abort, no process exit.

The at-least-once contract of ADR-0038 is preserved verbatim — the same commit / coverage calls, moved to fire from the ack lane, still only on durable confirmation.

Per-lane isolation (load-bearing)

A deployment that runs both a primary-raw lane and a secondary/derived lane must keep them independent in runtime state: each stream owns its own channel, task, atomics, and per-{destination} metric labels. A secondary-lane stall cannot delay raw-lane ack processing (which would degrade the recovery anchor), and a raw-lane stall cannot mask secondary-lane latency on its own dashboard. There is no shared channel, task, atomic, or metric label between lanes.

Metrics (twg-observability)

All {destination}-labelled, no shared axis between lanes:

• coalesce-ratio histogram — burst sizes; ≫ 1 under load means we are pipelining, a steady 1 means we are not.
• ack-lane retry / poisoned counters — transient backoff vs. permanent lane death; the poisoned counter is the cause signal that the runbook branches on (never the shared symptom signal — see §11 of the sink blueprint).
• pending gauge (submitted − delivered) — climbs ≫ 1 under load.
• delivery-latency histogram — emitted from the ack lane, off the hot path.

Options considered + consequences

Dimensions: throughput ceiling, at-least-once safety, cross-lane isolation, review cost.

Option 1 (chosen) — Two-lane per stream, typed enum lane callbacks

Fire-and-forget ingest + per-stream ack lane, confirmation as a typed enum token.

• Pros: matches the SDK’s designed pipelined semantics; preserves ADR-0038 verbatim; per-stream isolation and compile-time lane separation make cross-lane bleed impossible; small and independently testable.
• Cons: one background task per stream (small scratch buffer); the SDK stream moves behind an Arc<RwLock<…>> (non-blocking read locks on the hot path); metric-label cardinality grows by a small constant × destinations.

Option 2 — Trait-object confirmation callback (Box<dyn AckCallback>)

• Pros: extensibility across crate boundaries.
• Cons: loses the compile-time lane-separation guarantee (a secondary stream could hold a raw callback); vtable dispatch + allocation on the hot ack loop; buys nothing for in-crate destinations. Rejected.

Option 3 — Coalescing loop without type-level lane separation (bare closure)

• Pros: fewer types.
• Cons: same isolation loss as Option 2; every new call site must be hand-checked to bind the right destination’s state. The per-lane invariant is load-bearing enough to encode in the type system. Rejected.

Option 4 — Do nothing; raise the server-ack timeout

• Pros: zero code change.
• Cons: does not fix the throughput ceiling; each timeout bump trades false-positive ack-timeouts for slower real-outage detection; the sink stays orders of magnitude off nameplate. Rejected as a permanent stance (acceptable only as a temporary bandaid).

Invariants pinned by this ADR

• Per-stream isolation — no shared runtime state between two streams.
• Compile-time lane separation — cross-lane confirmation bleed is a compile error.
• At-least-once preserved — commit / coverage fire only on durable confirmation, from the ack lane; never optimistically on submit.
• Bounded backpressure — a fixed per-stream channel; the SDK never sees more than the cap in-flight-unacked from us; a poisoned lane tears the stream down gracefully.
• delivered is the watermark — recovery never reads submitted.
• Distinct primary signals — self-inflicted ack-lane poison and environmental slow-ack remain semantically separable at source; the runbook branches on cause, not symptom.

Deferred work

Each entry carries a trigger; an entry without one is a wish, not deferred work.

• D1 — Post-rollout tuning of the quarantine drain budget. Under two-lane the drain-before-drop tail latency is now bounded by the sink’s durable-ack tail, not the broker prefetch. Trigger: the drain-timeout counter fires > 1×/day per pod for 3+ consecutive days and the ack-lane-poisoned counter is 0 (i.e. genuine tail latency, not lane failure).
• D2 — Ack-lane depth-aware backpressure signal. Channel-full send.await is invisible to the throttle-trigger axis; a saturation gauge + a dedicated throttle reason would distinguish “SDK slow” from “ack lane deep.” Trigger: pending steady-states at ≥ 90% of the channel cap for a rolling 15 minutes.
• D3 — Configurable per-stream channel capacity. One compile-time cap today. Trigger: two destinations in one deploy justifiably want caps differing by ≥ 4×.
• D4 — Coalescing tuning telemetry / alerting. An alert on p50 coalesce-ratio < 2 over 5 minutes (the “not actually pipelining” signature). Trigger: first post-rollout runbook update.

Interaction with existing ADRs

• ADR-0038 (source-ack): unchanged in contract; this ADR only moves where the durable-ack wait happens. delivered is the durable-in-target watermark.
• ADR-0053 (position may lag, never lead): coverage advances from the ack lane’s on_confirm, preserving the lag-not-lead guarantee.
• ADR-0057 (recovery-from-raw): recovery reads delivered; a lagging secondary lane never gates the source and is rebuilt from raw.
• ADR-0049 (Pulsar flow control): channel-full backpressure surfaces as source-side awaiting, which the existing throttle observability already sees (until D2 refines it).

Document version history