Zerobus Sink — Technical Implementation Plan

A crate-by-crate implementation plan for twg-sink-zerobus, the Databricks Zerobus (gRPC Direct Write → Delta) sink. It turns the generic sink contract in how-a-sink-should-work.md into a concrete build, grounded in the pipelined-ack decision (ADR-0060), the source-ack invariant (ADR-0038), and the recovery/coverage model (ADR-0053/0057), and shaped by the prior-art incidents in decode-codec-learnings.md §8.

Status. Core built (ADR-0065). twg-sink-zerobus implements the BatchSink/Sink substrate, the ADR-0060 fire-and-forget ack lane, the bandwidth-sized stream pool, catalog-driven provisioning, emit-time type reconciliation, and the primary-raw preconditions — all tested against a fake transport. The real Databricks Arrow-Flight adapter is behind the optional databricks-sdk feature (Arrow-IPC bridge; compiled in CI, executed only against a live workspace). The BatchSink/Sink/coverage seams were built minimally for this sink; the broader Phase 5a/5b flow-control and recovery-role machinery and observability wiring remain (planned) — see ADR-0065 deferred work. The emit-time auto-project transform (§5.2) was dropped, not deferred (ADR-0065 D1): nested-into-STRING fails loud-named, pointing at an explicit to_json(col) or a typed column. Crate/ADR references are to thalweg’s own artefacts.

No sensitive information. Product names that are part of the public target surface (Databricks, Zerobus, Delta, Iceberg, Unity Catalog) are used because the crate’s whole purpose is that target; environment-specific names (catalogs, schemas, tables, topics) are always generic — orders, bronze.example.

0. Sources this plan is built on

Two things anchor the design:

1. Zerobus’s own model, from the Databricks petabyte-scale write-ups and the databricks/zerobus-sdk:

   • A stream is a logical identity registered with the service; ordering is guaranteed per stream connection for the connection’s lifetime, regardless of which server pod processes it. There are no client-visible partitions — the stream, not a partition, is the unit of scale.
   • Bidirectional gRPC: one direction sends records, the other returns acknowledgements. The server acks the highest committed offset on the stream (not per-record); the client purges its in-flight buffer up to that offset. Durability is a latency-optimised write-ahead log — the ack means durable.
   • Demonstrated scale: 2,048 concurrent streams to a single table, ~12M rows/s and ~12 GB/s sustained, with up to 50,000 in-flight (unacked) records buffered per stream. Server-side hot routing spreads streams across a pod pool and drains gracefully on scale-down. Quotas are per-table and raised via the account team.
   • Wire formats: protobuf, Arrow, JSON. (thalweg writes Arrow — ADR-0001, ADR-0020 — so no protobuf descriptor is on the write path.)

2. thalweg’s own prior-art learnings (predecessor incidents, abstracted in ADR-0038/0060 and the sink blueprint):

   • Few, long-lived streams per destination. Opening a stream per batch or per worker trips the per-table concurrent-stream limit; recovery then stalls on reconnect churn instead of moving data.
   • Never make the async ack synchronous. A per-batch “wait for durable offset” inline after each submit serialises the sink to one in-flight batch — a ~20,000× throughput loss that looks like ack-timeout false-positives, trivial-throughput backpressure trips, and batchers that never fill.
   • Advance the source position on the delivered offset, never the submitted one — recovery reads delivered (ADR-0053: lag, never lead).
   • Size streams to bandwidth, not record count. A stream has a finite MB/s ceiling; large records saturate one stream at a low record rate, so spread across a small pool.

The plan below is the synthesis: a small pool of long-lived streams per table, fire-and-forget ingest, and a per-stream background ack lane that advances the delivered watermark — exactly ADR-0060, realised against the Zerobus SDK.

1. Crate boundaries — what lives where

twg-sink-zerobus is deliberately thin. It owns only what is Zerobus-specific; everything reusable is a dependency. This is the crate-isolation standard (strictly downward dependencies, one concern per crate).

What twg-sink-zerobus itself owns, and nothing else:

• The stream pool (open, health, reuse, graceful drain) keyed by target table.
• The fire-and-forget ingest path that submits an Arrow RecordBatch to a stream and hands a pending-ack token to the ack lane.
• The per-stream background ack lane (ADR-0060) that coalesces acks, waits on the largest offset, advances delivered, and fires the confirmation callback.
• The Zerobus error taxonomy and its retry/backoff/poison classification.
• The primary-raw precondition checks specific to Delta+Iceberg (ADR-0023).

Rule of thumb: if a piece of logic would be identical for the Postgres or Flight sink, it does not belong here — it belongs in twg-connector-core or twg-stream-arrow.

2. The two traits this sink implements

2.1 BatchSink (from twg-stream-arrow)

#![allow(unused)]
fn main() {
// substrate-owned; sink implements it
pub trait BatchSink {
    fn push(&mut self, batch: &RecordBatch) -> Result<()>;
    fn finish(&mut self) -> Result<(CommitInfo, OffsetSpan)>;
}
}

• push maps the RecordBatch columns to the target table’s declared Arrow types (via twg-type-map, §5), applies emit-time type reconciliation (§5.2), and submits to a pooled stream fire-and-forget — it does not wait for a durable ack.
• finish flushes the current batching window and returns the OffsetSpan for the submitted high-water. It does not block on durability; the delivered watermark advances asynchronously (§4).
• _twg_emit_ts is stamped here, on write (ADR provenance chain).

2.2 Sink (from twg-connector-core)

#![allow(unused)]
fn main() {
pub trait Sink {
    fn send(&mut self, batch: Payload) -> BoxFuture<Result<Vec<Receipt>>>;
    fn flush(&mut self) -> BoxFuture<Result<()>>;
}
}

• send returns as soon as the SDK accepts the batch (fire-and-forget). The Receipt carries the submitted offset span plus a handle the ack lane resolves.
• flush awaits the SDK’s own flush / the largest-outstanding offset for a drain-to-durable, used at shutdown and at explicit checkpoint boundaries.
• No open/commit/close/ack methods — commit and ack are modelled as data (Receipt, coverage record), per the trait’s minimalism.

Durable ack definition for this sink: server-side WAL ack of the offset (the Zerobus write-ahead-log confirmation). This is the “durable” cell for the Zerobus row of the sink-family table in the blueprint.

3. Stream pool

3.1 Topology

• A small, fixed pool of long-lived streams per target table, opened once at sink start and reused for the sink’s life. Default pool size is a config knob (§7), defaulting low (e.g. 2–4) and sized to bandwidth (§3.3), not to record rate. This directly answers the predecessor’s stream-churn incident and stays far under Zerobus’s per-table concurrency headroom (2,048 demonstrated).
• Each stream is an Arc-shared SDK stream behind an async lock; read (submit) locks are non-blocking against each other, a write lock only on open/close/rebuild.
• Per-stream isolation is load-bearing (ADR-0060): each stream owns its own ack channel, ack task, submitted/delivered atomics, and {table,stream} metric labels. No shared runtime state between two streams, and — where a deployment runs both a raw lane and a derived lane — no shared state between lanes (a derived-lane stall must not delay the raw anchor’s acks).

3.2 Stream selection

• A batch is assigned to a pool stream by a stable, low-cardinality policy so ordering-sensitive records stay on one stream (Zerobus guarantees order per stream). Default: round-robin across the pool for order-agnostic tables; a key-hash to a fixed stream index when the config declares an ordering key.
• Selection never opens a new stream on the hot path. If a stream is mid-rebuild, the batch waits (bounded) for a healthy pool member rather than minting a stream — the anti-churn rule.

3.3 Sizing to bandwidth

• Databricks documents a hard per-stream limit of ~100 MB/s. A single stream cannot carry more, so the pool size target is ceil(peak_MB_per_s / per_stream_mb_ceiling), floored at the configured minimum, with per_stream_mb_ceiling defaulting to 100 MB/s (the documented limit).
• Large-record tables (e.g. ~15 KB rows) therefore get more streams at a lower record rate than small-record tables — the sizing is by bytes. Worked example: 15 KB rows at a 10,000 rows/s target is ~150 MB/s, which exceeds one stream’s 100 MB/s — so that table needs at least two streams, whereas a small-record table at the same row rate fits on one.
• The default matches the documented ceiling; it stays a config value (operators may set it lower to leave headroom) and the derived pool size is logged at startup.

3.4 Lifecycle & graceful drain

• Open lazily at first write per table, then keep alive.
• Rebuild on poison (§6): tear the stream down within a bounded budget, drain its outstanding acks to the extent possible, then re-open. The pool serves other streams throughout.
• Shutdown: stop accepting new submits, flush each stream to its largest outstanding offset, advance coverage, then close. Matches Zerobus’s own graceful-drain-on-scale-down behaviour.

4. The ingest + ack pipeline (ADR-0060, realised)

This is the heart of the sink and the single most important thing to get right.

push(batch)
  └─ map + reconcile types (§5)
  └─ submit to pooled stream           ← fire-and-forget; returns on SDK accept
  └─ submitted.fetch_max(offset)
  └─ send PendingAck{offset, on_confirm, submitted_at} to this stream's ack channel
        (channel full ⇒ push awaits — this is the backpressure, not a round-trip)

[per-stream ack lane task]  (spawned at stream open)
  loop:
    first = ack_rx.recv().await            ← block for one item
    burst = drain_nonblocking(ack_rx)      ← coalesce everything queued
    wait_for_offset(max(burst.offset))     ← ONE round-trip confirms all ≤ items
    for item in burst.sorted_by_offset():
        item.on_confirm()                  ← advance coverage / commit source pos
    delivered.fetch_max(max(burst.offset))
    observe coalesce_ratio, delivery_latency

Key properties, each mapped to a learning:

• Fire-and-forget ingest — the SDK’s own bounded in-flight buffer (up to ~50k records/stream) provides backpressure; when our per-stream ack channel fills, push awaits. No inline per-batch durability wait (the ~20,000× trap).
• Coalesced ack — Zerobus acks the highest committed offset, and offsets are monotonic per stream, so one wait_for_offset on the burst maximum confirms every lesser item. The coalesce-ratio histogram is the “are we actually pipelining?” signal (≫1 under load).
• Advance on delivered, never submitted — on_confirm (coverage update / source-position commit) fires only from the ack lane, only after the WAL ack. submitted exists solely to expose an ingest high-water and compute pending = submitted − delivered. Any durability path reading submitted is a bug (ADR-0038/0053).
• Confirmation token is a typed enum, not a boxed closure — compile-time lane separation so a derived-lane stream cannot hold a raw-anchor commit token (ADR-0060 option 1). Variants: RawAnchor{…}, DerivedCoverage{…}, None.
• Backpressure is a bounded channel — per-stream capacity is a config knob (default a few thousand), and channel-full surfaces as source-side awaiting, which the batcher’s throttle observability already sees.

5. Schema, provisioning (catalog), and emit-time type reconciliation

5.1 Unity Catalog & table-catalog delivery — a co-equal prerequisite

Zerobus only writes; it does not create the target table, stamp comments, evolve the schema, or vend the credentials the write and recovery paths need. A Zerobus sink is therefore useless without a catalog layer, so that layer is a co-equal deliverable of this plan, detailed here in full. It is a separate, reusable subsystem every sink shares — not sink code — and a hard prerequisite: the sink’s role/precondition checks (§8) and every write assume the table already exists with the right schema, comments, and grants.

5.1.1 Crate shape & the trait seam

twg-table-catalog owns the transport-neutral TableCatalog trait (schema resolution, sink registration, credential vending, descriptive metadata; ADR-0024/0037). The Unity Catalog backend is the one this sink needs; other backends (S3 Tables, Glue, plain Iceberg REST) implement the same trait, and the sink is written against the trait, never against UC directly. twg-type-map (built) is the describe-only Arrow→dialect authority the catalog executes against — neither sink nor catalog hand-builds DDL text.

Decision (ADR-0062): implement UC as a backend module inside twg-table-catalog now; extract to a dedicated twg-catalog-unity client crate later, only on a stated trigger — the UC surface (OAuth M2M auth + token refresh, REST models, retry/rate handling) outgrows a module, a second consumer outside the catalog crate needs the raw UC client, or UC needs an independent release cadence. Until a trigger fires, one module keeps it simple; the TableCatalog trait insulates the sink from the choice either way, so the extraction is a non-breaking refactor when it happens.

#![allow(unused)]
fn main() {
// twg-table-catalog — the seam every sink depends on (sketch)
pub trait TableCatalog {
    // Resolve current live schema + table properties (None if absent).
    fn describe(&self, table: &TableRef) -> BoxFuture<Result<Option<TableState>>>;

    // Reconcile the target to `desired` (from the incoming RecordBatch schema +
    // contract). Computes a plan, then executes it. Idempotent; safe to call every
    // first-write and on drift.
    fn reconcile(&self, table: &TableRef, desired: &DesiredTable)
        -> BoxFuture<Result<ReconcileOutcome>>;

    // Register this sink as a writer of the table (ownership/coverage bookkeeping).
    fn register_sink(&self, table: &TableRef, sink: SinkId) -> BoxFuture<Result<()>>;

    // Temporary, scoped credentials inheriting the caller's privileges.
    fn vend_credentials(&self, table: &TableRef, access: Access)
        -> BoxFuture<Result<VendedCredentials>>;
}
}

reconcile is describe-vs-execute: build a ReconcilePlan (a pure value — CreateTable, AddColumn, WidenColumn, SetComment, SetProperty, or Refuse), then apply it. The plan is logged and unit-testable without a live UC.

5.1.2 The five duties, driven by the incoming Arrow RecordBatch schema

1. Create-on-absent. Table missing → create from the incoming Arrow schema: every payload column mapped Arrow→Delta/UC via twg-type-map, plus the reserved _twg_* metadata columns, identifier-safe names. Table properties/tags (owner, contract URI + version, source topic) set at create. For a table intended as primary-raw, create it Iceberg-reads-enabled and deletion-vectors-off so it satisfies §8 up front rather than being refused at first write.
2. Evolve-on-drift, on the fly. A later batch whose schema differs from live → the catalog diffs live-vs-desired and issues the ALTERs, inline on the first drifted batch (then cached until the next drift — not an offline migration):
   • Additive (new column) → ADD COLUMN + its comment. Always safe.
   • Widening → allowed only behind an opt-in, gated by twg-type-map::can_widen_to (lossless int32→int64→double→string; int→f64 refused as lossy).
   • Anything else (narrowing, type conflict, column removal) → Refuse, fail loud, named — never a silent drop or a lossy coercion.
3. Comments / descriptions — two authorities. Engine-owned columns (_twg_*, dq_results) get fixed built-in descriptions; payload column comments flow from the ODCS contract (no contract → no payload comment). Stamped at create AND evolution (a newly added column gets its comment in the same plan), applied via the describe-vs-execute split.
4. Idempotent re-sync. Comments carry twg:managed=true + a twg:comment_hash; a re-sync writes only when the contract description changed — no churn on no-op deploys. Human edits to managed columns are detected via the flag; respect-vs-restore is a policy, not a blind stomp. Backend degrade-gracefully matrix: UC → COMMENT ON + tags + properties; Iceberg → table properties + column docs; Glue → column Comment + table Parameters.
5. Credential vending. UC vends temporary, scoped credentials that inherit the requesting principal’s privileges — never static secrets in the sink. The same vending serves both the write path and the Iceberg recovery read path (§8.1), via UC’s Iceberg-REST endpoint. Auth to UC is OAuth (machine-to-machine) with token refresh handled in the backend, behind vend_credentials.

5.1.3 Catalog-stage build order (delivered before the sink writes)

Runs as Phase 0 of §11 (the sink cannot write without it), independently testable:

• C1 — trait + type mapping. TableCatalog trait, DesiredTable from an Arrow schema + contract via twg-type-map, ReconcilePlan as a pure value. No network.
• C2 — UC auth + describe. OAuth M2M + token refresh; describe returning live TableState. Read-only against UC.
• C3 — create + comments. Plan→execute for CreateTable incl. _twg_* columns, properties/tags, and create-time comments (two authorities).
• C4 — evolve (ALTER diff engine). Additive + gated widening + refuse; evolution-time comments; drift cache.
• C5 — credential vending. Scoped temporary credentials for write and Iceberg read; principal-privilege inheritance.
• C6 — idempotent re-sync. twg:managed/twg:comment_hash; managed-edit detection; degrade-gracefully matrix.

5.1.4 Catalog-stage tests

• Plan-level unit tests (no network): additive→AddColumn; lossless→WidenColumn under opt-in; lossy/narrowing/removal→Refuse loudly; comment-hash unchanged→no-op.
• UC integration (recorded/replayed or a test workspace): create-from-schema, drift→ALTER, comment sync, credential vend + expiry/refresh.
• Idempotency: reconcile twice → second call is a no-op (no churn).
• Primary-raw shape: a table created for the raw role comes back Iceberg-reads-enabled, deletion-vectors-off (satisfies §8 without a later refuse).

The sink’s only job against all of this is to call reconcile before/at first write and on drift, and vend_credentials for the write/recovery paths; every DDL, comment, tag, grant, and auth mechanic lives in twg-table-catalog.

5.2 Emit-time type reconciliation — the write-boundary guard

This is where the predecessor lost whole lanes (decode-codec-learnings §1/§4/§5), so the sink must handle two mismatches at the write boundary rather than let Arrow Flight reject the batch:

• Nested-into-STRING. A decoded Struct/List/Map landing in a column the target declares STRING (Arrow LargeUtf8). The JSON serialiser already exists: to_json(<any>) → STRING is shipped in twg-udf (the manual, author-invoked path — an SQL author writes to_json(col) today). What the sink adds is the automatic emit-boundary wiring: because the sink already knows the declared target type (from twg-type-map), when incoming is Struct/List/Map and the target is LargeUtf8 it calls that same serialiser transparently, symmetric to the decoder’s Utf8-JSON fallback. This auto-project is gated by config (§7, default off until soaked) — with it off, an unreconciled column fails loud, named, never opaquely per batch; the author’s to_json(...) remains the manual escape hatch either way. No new serialiser is needed — only the sink-side wiring to invoke the existing one at emit.
• tz-aware-into-NTZ. A Timestamp(µs, Some("UTC")) into a TIMESTAMP_NTZ (Timestamp(µs, None)) column. The fix is upstream (tz-naive helpers, learnings §5), but the sink must surface the mismatch loudly with the column named.

Reconciliation is a describe-time comparison of incoming_arrow_type vs declared_target_type; anything the sink could reconcile or could name precisely must never surface as an opaque per-batch rejection.

6. Error taxonomy, retry, and poison

A single ZerobusError enum with an is_retryable() (transient/backoff) vs poison (permanent-until-rebuild) split, mirroring ADR-0060’s classifier:

Rules:

• Self-protective paths must not amplify. A concurrency reject triggers backoff
  • send-on-existing, never an immediate reopen (that is what caused the predecessor’s storm).
• Discriminate self-inflicted from environmental at the metric source: a distinct ack_lane_poisoned counter (our stream died) vs the shared slow-ack symptom (broker transiently slow). The runbook branches on cause, not symptom.
• Retries never re-advance offsets. A retried burst still advances delivered only once, on genuine WAL ack.

7. Configuration

All knobs live in twg-config (TOML, ADR-0028-style), namespaced under the sink; each has a conservative default and is logged at startup.

Deliberately not a knob: whether to wait per-batch. Fire-and-forget is structural, not configurable.

8. Primary-raw eligibility (Delta + Iceberg)

Zerobus can hold the primary-raw role only under the conditions the scaffold already documents, enforced as config-validation preconditions, re-checked not cached (a table property can change under a running pipeline):

• the table is registered in the catalog (managed or external);
• Iceberg reads / column mapping enabled;
• reader/writer protocol versions meet the feature minimum;
• deletion vectors NOT enabled (they silently remove recovery’s read path).

If role = primary_raw and any precondition is unmet → refuse the role at config validation (loud, ADR-0057: recovery is a precondition for a second sink, not later hardening). durable is mandatory for primary-raw (ADR-0038).

8.1 Recovery read path

• Recovery reads the written data back through Iceberg via twg-format-iceberg
  • twg-table-catalog (vended, scoped credentials) — the sink itself is write-only, which is allowed (ReplayableSink needs a read path, not a sink-native one).
• Read-after-write hazard: Iceberg metadata generation is asynchronous, so a recovery pass reading immediately after a write may see an older table state. Recovery must tolerate the lag (bounded trailing windows make this natural) or trigger metadata generation synchronously before reading. Do not assume read-after-write consistency across the format boundary.
• zstd knock-on: Iceberg-enabled tables use Zstandard Parquet; the read path uses twg-wire-compression decode-only (purity-safe, no encode exception).

8.2 Coverage & the delivered watermark

• Per-sink coverage is gap-aware ranges in twg-offset-store, keyed (sink, topic, partition, start) → end, union-merged never last-write-wins, colocated with raw, never with the sink it describes (ADR-0053), and advanced from the ack lane’s on_confirm (delivered), so coverage may lag, never lead.

9. Observability

All {table,stream}-labelled, no shared axis between lanes (twg-observability):

• zerobus_ack_coalesce_ratio — burst-size histogram; ≫1 under load = pipelining healthy; steady 1 = not pipelining (alert p50<2 over 5 min).
• zerobus_pending_offsets — submitted − delivered; climbs under load.
• zerobus_delivery_latency — emitted from the ack lane, off the hot path.
• zerobus_ack_lane_retry_total / zerobus_ack_lane_poisoned_total — transient vs permanent; poisoned is the cause signal the runbook branches on.
• zerobus_stream_rebuilds_total, zerobus_concurrency_rejects_total — churn/limit visibility (the predecessor’s blind spot).
• zerobus_type_reconcile_total{kind} and a loud counter for unreconcilable nested-into-STRING / tz-into-NTZ mismatches, column named (§5.2).

Health tree exposes stream-pool liveness and backpressure state; none of this sits on the commit path.

10. Testing plan

• Unit — error classification (transient vs poison), stream-selection policy, type-reconciliation describe-time comparison, config defaults/derivation.
• Ack-lane property tests — coalescing correctness (one wait_for_offset confirms all ≤ items), delivered monotonicity, and the invariant that on_confirm fires only after a modelled durable ack (never on submit). A fake SDK stream with injectable ack latency drives this.
• Anti-regression harness for the ~20,000× trap — a soak test asserting in-flight depth > 1 under load (i.e. pending_offsets climbs), so any re-introduction of an inline per-batch wait fails CI. Lives in twg-e2e / docs/testing/SCENARIOS.md.
• Poison → rebuild — inject an unrecoverable stream error; assert bounded drain, stream rebuild, no offset re-advance, other pool streams unaffected.
• Concurrency-reject backoff — assert honour-backoff + send-on-existing, no new-stream storm.
• Primary-raw precondition matrix — each unmet precondition refuses the role loudly; deletion-vectors-enabled is explicitly rejected.
• Recovery read-back — write via the sink, read back through twg-format-iceberg, including the async-metadata lag path (tolerate or force).
• Coverage semantics — union-merge under simulated concurrent writers; assert lag-never-lead.

11. Phased build order

Each phase is independently testable and leaves the crate compiling.

0. Unity Catalog / table-catalog subsystem (twg-table-catalog + UC backend) — the co-equal prerequisite, delivered as its own C1–C6 sub-build (§5.1.3): trait + type mapping → UC auth + describe → create + comments → evolve (ALTER diff engine) → credential vending → idempotent re-sync. Reusable by every sink; must exist before the sink writes anything. Full detail in §5.1.
1. Skeleton BatchSink + Sink over a single stream, fire-and-forget + inline ack (temporarily), Arrow→Delta mapping via twg-type-map, table provisioning via the phase-0 catalog. Proves the write path end-to-end.
2. Per-stream ack lane (ADR-0060): move the wait off the hot path, coalesce, advance delivered, typed confirmation enum. Add the anti-regression soak test.
3. Stream pool: pool-per-table, selection policy, bandwidth sizing (100 MB/s ceiling), graceful drain; bounded rebuild on poison.
4. Error taxonomy + backoff: concurrency-reject handling, poison→rebuild, metric cause/symptom split.
5. Emit-time type reconciliation: loud mismatch surfacing; then the gated auto-project.
6. Primary-raw preconditions + recovery read path (twg-format-iceberg, zstd decode, async-metadata lag), coverage in twg-offset-store.
7. Observability full set + health tree; runbook stub for the poison-vs-slow-ack branch.

Phases 1–2 are the critical correctness core; 3–7 harden and scale.

12. Invariants this sink must always satisfy

• Ingest is fire-and-forget; the durable wait is a per-stream background lane; the source advances on delivered, never submitted. (§4, ADR-0060/0038)
• A small pool of long-lived streams per table, sized to bandwidth; never a stream per batch/worker; honour server backoff on a concurrency reject and send on an existing stream. (§3, §6)
• Per-stream (and per-lane) isolation in the type system — no shared channel/task/atomic/label; cross-lane confirmation bleed is a compile error. (§3.1, §4)
• Durable ack = WAL server ack; coverage is gap-aware, union-merged, stored with raw, and may lag never lead. (§2.2, §8.2)
• The target table is provisioned and evolved by the catalog from the incoming RecordBatch schema (create-on-absent, additive/widening on the fly, comments at create+evolution) — the sink calls, never hand-builds DDL. (§5.1)
• Write-boundary type mismatches (nested-into-STRING, tz-into-NTZ) are reconciled or surfaced loudly with the column named — never an opaque rejection or a silent NULL; the JSON serialiser is the shipped twg-udf to_json, not a new one. (§5.2)
• Primary-raw role enforces its preconditions at config validation, re-checked not cached, and refuses the role loudly when unmet; recovery reads back via Iceberg, tolerating async-metadata lag. (§8)
• Errors surface on three planes; poison is discriminated from environmental slow-ack at source. (§9)

Source grounding

Built from thalweg’s own artefacts — how-a-sink-should-work.md, decode-codec-learnings.md, the twg-sink-zerobus scaffold, and the delivery plan — plus the Databricks Zerobus petabyte-scale write-ups and the databricks/zerobus-sdk model (stream-as-identity, highest-offset async ack, ~50k in-flight/stream, 2,048 streams/table). Relevant ADRs: 0001 (BatchSink/OffsetSpan), 0012 (Sink, primary-raw eligibility), 0020 (Flight/IPC transport), 0023 (single table format; Delta-via-Iceberg reads), 0024 (credential vending), 0037 (catalog metadata), 0038 (source-ack invariant), 0053 (position may lag never lead; coverage with raw), 0057 (recovery precedes the second sink), and 0060 (pipelined fire-and-forget ack). Predecessor incidents are abstracted per the no-sensitive-information rule. Retire (planned) markers as Phases 5a/5b/7 land.