Thalweg reference

The rendered prose documentation for Thalweg — the architecture decision records, the delivery blueprints, the operations notes and the testing standard, plus the publishing guide. It is generated from the markdown under docs/ and served at docs.thalweg.dev/reference.

The API reference (rustdoc for every crate) is a separate surface at docs.thalweg.dev/api. The overview, diagrams and support matrix are on the landing page.

This book is regenerated on commit from the docs tree; drop a markdown file into the right folder under docs/ and it appears here automatically.

Architecture Decision Records

Each ADR records one decision. Records follow the team house style and round-trip to Confluence via mark. Status flows Draft → For Review → Accepted (or Rejected / Superseded).

The decisions below were developed during design. Records are written per decision, not per crate — most decisions cut across several crates, and several crates carry no decision of their own.

Genuinely open rather than merely undrafted: the protobuf decode strategy (ADR-0046 — direction chosen, licence cleared as Apache-2.0, now gated only on a benchmark against the upstream parser on our hardware and schemas).

Noted for future consideration (not yet scoped)

Captured so they are not lost; no decision or crate yet, and not committed scope:

• DuckLake as a source and sink — treated like the other database targets (goes through the sink contract / a source adapter, primary-raw eligibility assessed on its durability + replayability like any other).
• REST and WebSocket (WSS) as a source and a sink — REST polling/webhook and WSS streaming, both directions. WSS-as-source is a streaming transport (fits connector-core); REST-as-source is poll-or-webhook; both as sinks are push/forward targets. Locator/recovery semantics to be worked through when scoped.

ADR-0001: Build a pure-Rust, sans-io streaming crate ecosystem with co-equal byte, native-value and Arrow representations

Reference

Foundational ADR for the stream-arrow workspace. Downstream ADRs (consumer-group protocol, neutral value model, Arrow union mapping, offset-correlation contract, record-metadata representation, table-catalog write semantics, connector flow-control model, transform DAG semantics, WASM sandbox contract) will reference this document.

Summary

Stub — to be written when the decision approaches Accepted.

Context

We ingest from Kafka across a multi-brand estate spanning Confluent, Redpanda, MSK, Azure Event Hubs and StreamNative, under demanding SLAs. Every current Rust option forces a compromise. rdkafka wraps librdkafka: mature and complete, but it is a C dependency with C memory-safety characteristics, a build-time burden in CI images (cmake, OpenSSL), and known cancel-safety hazards around StreamConsumer::recv() inside select!. The pure-Rust alternatives (rskafka, kafka-rust) deliberately omit consumer groups, which disqualifies them for our workloads.

Separately, no crate in the ecosystem offers a clean path from broker bytes to Arrow RecordBatch. Every available library is shaped by JVM conventions: decode message-at-a-time into an intermediate dynamic value type, then convert. That intermediate materialisation is precisely the cost Arrow exists to eliminate, and it sits directly on the hot path into our Lakehouse and Lakebase targets.

Arrow is not, however, the only representation we need. Regulated retention requires the original wire bytes be kept for audit and replay, unmodified and envelope-intact. Proxying, tee-to-archive and forward-to-third-party workloads want bytes and never want a schema resolved at all. And record-at-a-time consumers — request/response handlers, validators, small services — want a decoded value without paying for a columnar dependency they will not use. A design that treats Arrow as the sole exit forces all three through decode-and-re-encode, which is wasteful for the first two and lossy for the first.

Fidelity is a first-class requirement rather than a quality goal. Kafka’s record model carries more than key and value — ordered headers permitting duplicate keys and null values, a timestamp with a broker-authoritative type, leader epoch, and batch-level producer identity. Abstractions routinely collapse these (headers into a map, null into empty) and the loss is invisible until an audit or a replay needs them. The design commits to carrying the full inventory and to documenting the three fields that are broker-authoritative and therefore not round-trippable by equality.

Schema resolution also has two distinct sources in our estate. Message-schema registries (Confluent, Glue Schema Registry) answer “what shape is this payload”; table catalogs (Glue Data Catalog) answer “what shape is the table this feeds”. They are different APIs answering different questions, and both are needed — the second in both directions, since batches must register and evolve the tables they land in.

Pulsar is now in scope as a first-class transport rather than a later possibility, and as a fresh protocol implementation rather than a dependency on pulsar-rs. Our own evaluation found that crate’s limitations to be architectural rather than incidental: backpressure is implicit in whether the caller polls its Stream, with the permit window tunable only through batch size; there is no rate limiter and no explicit pause/resume; seek destroys and recreates the consumer; and selective partition subscription forfeits automatic discovery. Each is fixable in isolation, but fixing them inside a design shaped around implicit backpressure means working against its grain permanently.

That decision surfaces a second one. Sink and source semantics, flow control, observability, retry and dead-lettering are not transport-specific, and placing them in either client would deny them to the other. They belong in a transport-neutral connector-core that both implement — and the Kafka client is retrofitted onto it as soon as it exists, because an abstraction designed against a single transport is a guess.

A middle layer is also in scope, and it changes what the project is. Up to this point the system is a library ecosystem configured at compile time by its authors. A SQL-configured transform layer makes it additionally a deployable configured at runtime by people who did not write it — which introduces a threat model the rest of the design does not have, since user-supplied queries and user-supplied binaries then execute inside our process.

The shape required is a DAG, not a flat fan-out: raw records tap to one sink while the same source is decoded, unpacked, split into virtual tables for data quality and filtering, and further unnested into derived tables with their own sinks. Sinks attach at interior nodes. Two properties of that shape are load-bearing and silent when wrong — intermediate nodes with multiple dependents must be materialised or the plan re-executes per dependent and multiplies broker load, and nodes that change cardinality break the offset-correlation contract unless derived rows carry their originating locator.

The intent is an ecosystem, not a crate. Kafka is the first transport; Pulsar and MQ follow. XML/XSLT joins the codec layer later, reusing existing XML-to-Arrow work. That trajectory only pays off if the layering holds from the first commit — a transport-coupled codec cannot be reused by the second transport, and a codec-coupled transport cannot be reused by the second format.

Recommended option

Option 3 — pure-Rust ecosystem with sans-io codec crates exposing three co-equal representations. Options 1 and 2 both solve the immediate ingestion problem faster, but neither produces reusable assets. The decisive argument is not memory safety in isolation; it is that the codec crates are the durable value, and they are only durable if they are runtime-free, transport-free, and not mandatorily columnar.

Three representations are supported as peers rather than as a primary with fallbacks:

• Bytes — passthrough, envelope intact, no schema resolution. Available with no codec crate present at all.
• Native value — decoded into a neutral value model owned by codec-core, not the upstream libraries’ types.
• Arrow — batch-oriented decode straight into ArrayBuilders, no intermediate materialisation.

Record metadata travels with the decoded representations as reserved __kafka_-prefixed columns rather than out-of-band alongside the batch. The out-of-band alternative keeps schemas clean but does not survive an Arrow Flight boundary, which would lose every metadata field at precisely the point our Lakehouse writes need them.

Extension is provided by two mechanisms with deliberately distinct contracts rather than one general-purpose mechanism. In-process WASM modules serve pure transforms and close the real DataFusion gap — user-defined functions currently require compiling into our binary — with a fuel-metered sandbox granted no network or filesystem capability. Out-of-process Flight sidecars serve callouts, stateful work and model inference. Merging them was considered and rejected: granting network access to the embedded sandbox dissolves the isolation that justifies it and makes per-batch latency unpredictable inside a query plan, while forcing pure transforms through a network hop is needless cost. Inline model inference specifically belongs in the sidecar, since WASM-compiled inference is materially slower than native and GPU-backed inference is impossible in-sandbox.

Packaging takes the statically linked musl binary as the primary artefact, from which OCI images, systemd deployments and function packages all derive; an image-first approach yields only an image. OCI rather than LXC because Kubernetes and its runtimes consume OCI, and multi-arch rather than arm64-only because the second architecture costs a build-matrix entry while excluding most on-premise and much of Azure.

Observability is designed in rather than added later. Spans follow the record lifecycle and propagate through message headers so traces cross the broker; metric names and labels are defined in connector-core so both transports are interchangeable behind one dashboard. Retrofitting this after two clients exist means two incompatible instrumentation schemes and a migration.

Table catalogs are a separate crate (table-catalog) from message-schema registries, not an implementation behind the same trait. Conflating them would put table-format and partition-registration concerns behind an interface shaped for wire envelopes.

The neutral value model is chosen over re-exporting apache_avro::Value / prost_reflect::DynamicMessage / serde_json::Value deliberately. Re-exporting is cheaper initially but couples our public API to three upstream release cadences, makes format-agnostic consumers impossible to write, and means a format swap is a breaking change for users. Owning the model costs a mapping layer per codec and one hard design problem — a value model expressive enough for Avro unions, Protobuf oneof, JSON’s dynamism and XML’s mixed content, without becoming a lowest common denominator. We accept that cost because format-agnostic consumption is a stated requirement of the MQ and Pulsar phases, and retrofitting a neutral model after three codecs have shipped upstream types is a breaking change we would rather not schedule.

Arrow is a default-on feature of the codec crates. The common case is columnar and defaults should serve it; default-features = false yields a build with no arrow dependency for slim and embedded consumers. This has one consequence worth stating plainly: codec-core’s Arrow re-export — the version-pinning chokepoint that keeps arrow-rs churn contained — exists only when the feature is enabled, so the no-Arrow build has a materially smaller public API.

The cost is honest and large: consumer groups, the idempotent producer and transactions are months of work, and rebalance correctness is unforgiving. We accept it because that gap is exactly what keeps the Rust ecosystem dependent on librdkafka — the difficulty is the moat, not a reason to retreat. It is planned in full rather than deferred.

Two purity boundaries are accepted and documented rather than hidden. Zstd compression has no production-grade pure-Rust encoder; ruzstd decodes, so we decode natively and feature-gate zstd-encode off by default, to be developed when a producer path needs it. GSSAPI/Kerberos has no production-grade pure-Rust implementation; it is an optional feature over cross-krb5 that explicitly breaks the guarantee when enabled. Both are stated in the README rather than discovered by a user.

Options considered + consequences

Dimensions: end-to-end safety, time-to-first-ingest, reuse across transports and formats, ops burden, hot-path performance.

Option 1: Thin ergonomic layer over rdkafka

Description: Wrap librdkafka bindings, add a schema-registry client and Arrow decoding above it.

Consequences:

• Pros: Protocol correctness is solved — SASL mechanisms, all broker variants, consumer groups, transactions work today. Fastest route to production ingest, plausibly a working consumer in weeks.
• Cons: C dependency on the hot path defeats the end-to-end safety goal; CI images carry cmake and OpenSSL; cancel-safety footguns leak into every consumer we write; nothing built is reusable when Pulsar and MQ arrive. Zero-copy passthrough is constrained by the C buffer ownership model.
• Ops burden: low. Time-to-deliver: lowest. Reuse: none.

Option 2: Pure-Rust transport, Arrow-only codecs over upstream value types

Description: Build the Kafka client in pure Rust; decode via apache-avro/prost into their native types and convert to Arrow afterwards. Arrow is the only supported output.

Consequences:

• Pros: Achieves memory safety across the transport; meaningfully less codec work than Option 3; no neutral value model to design.
• Cons: Per-message intermediate materialisation on the hot path is the exact cost Arrow removes, and the conversion is where the CPU goes. Codecs stay coupled to upstream value types, so format-agnostic consumption is impossible and the reuse story for Pulsar/MQ/XML does not materialise. Passthrough and record-at-a-time consumers are unserved. Delivers the hard part of Option 3 without its main payoff.
• Ops burden: medium. Time-to-deliver: medium. Reuse: partial.

Option 3: Full pure-Rust ecosystem, sans-io codecs, three co-equal representations

Description: Layered workspace — shared transport primitives (wire-tls, wire-sasl, wire-compression), a Kafka client over kafka-protocol, a registry crate splitting wire envelope from schema resolution, and independent codec crates exposing bytes, neutral values and Arrow builders with no runtime dependency.

Consequences:

• Pros: End-to-end memory and type safety with two documented exceptions. Codec crates reusable by Pulsar, MQ, DataFusion TableProviders and Lambda without modification, and usable without Arrow at all. Passthrough is near-free given refcounted slices, serving audit-retention and proxy workloads with no codec present. Batch-oriented push/finish preserves the columnar benefit. Arrow Flight and IPC fall out of the pipeline layer.
• Cons: Consumer-group protocol, idempotent producer and transactions are substantial and correctness-critical. The neutral value model is a hard design problem with real risk of becoming a lowest common denominator, and every codec now owes three surfaces rather than one — roughly 30–40% more codec work per format, incurred up front rather than amortised. Full-fidelity round-tripping multiplies the test matrix by three representations. The catalog write path carries blast radius beyond this system, since a mis-evolved table affects every consumer of it. Long runway before the first Arrow ingest. Two protocol implementations to maintain rather than one, which is a sustained cost and the largest single commitment in the plan — mitigated but not removed by shared transport primitives and a shared connector layer. Runtime configurability introduces a threat model the library-only design avoids, and a dependence on DataFusion’s API stability confined to one crate.
• Ops burden: highest initially, lowest at steady state. Time-to-deliver: longest. Reuse: full.

Advice Received

Document version history

Note (repo import): This document predates the v2.2.0-informed revision pass. The confirmations, departures, and new decisions agreed during design review are tracked in docs/adr/README.md and will be folded in when the full ADR set is written. Treat the phase structure and crate layout as current; treat specific version pins and the single-ADR framing as pending update.

ADR-0003: Arrow version strategy across the workspace and DataFusion

Reference

Gates ADR-0023 (Iceberg/Delta crate maturity), which shares the binding constraint. Constrains every crate that exposes an Arrow type. Interacts with ADR-0047 (independent publishing), which makes arrow a public dependency.

Context

arrow-rs makes breaking API changes at most releases. Types from two Arrow versions are distinct types to the compiler even when structurally identical, so a RecordBatch cannot pass between them directly.

The workspace pins one Arrow version through a single re-export in twg-codec-core, which prevents internal skew. It cannot prevent skew with dependencies that pin Arrow themselves — DataFusion above all, and also delta-rs and iceberg-rust. DataFusion’s internal Arrow version is outside our control, so any batch entering or leaving a DataFusion plan crosses a version boundary whenever the workspace runs ahead.

Three forces pull against each other:

• Security and currency. The preference is to pin latest secure releases. Deliberately running an older Arrow means inheriting someone else’s upgrade cadence for security fixes.
• Cost per crossing. A predecessor system ran a newer workspace Arrow than its query engine and bridged via Arrow IPC, measuring roughly 5–10% CPU overhead — paid on every transform batch, and recorded as a regret.
• Public API. Because crates publish independently (ADR-0047), arrow is a public dependency of the codec crates. Its version is part of our API, an Arrow major bump is a breaking change for us, and nothing exposing an Arrow type can stabilise until this is settled.

Options considered + consequences

Dimensions: currency and security posture, cost per crossing, blast radius, maintenance burden.

Option 1: Pin the workspace to DataFusion’s Arrow version

Track whatever DataFusion depends on; no boundary exists, so nothing to bridge.

• Pros: zero crossing cost. Simplest possible dependency graph. No unsafe. Trivially compatible with delta-rs/iceberg-rust if they track similarly.
• Cons: currency is delegated. An Arrow security advisory cannot be acted on until DataFusion moves, which conflicts directly with the stated preference for latest secure releases. Newer Arrow features unavailable. Our published crates’ Arrow version — a public API commitment — is then set by a third party’s release cadence.
• Mitigation to evaluate: DataFusion’s cadence is frequent and tends to track Arrow closely, so the staleness window may be weeks rather than versions. Measure the actual historical lag before treating this as disqualifying.

Option 2: Newest Arrow workspace-wide, bridge via Arrow IPC

The predecessor’s choice.

• Pros: full currency and immediate security response. Wire format is stable across versions, so the bridge is straightforward and safe code.
• Cons: serialisation copies every buffer, so cost scales with bytes — the measured 5–10% CPU on every transform batch. Directly contradicts the no-intermediate-materialisation principle this project applies elsewhere. Recorded upstream as a regret.

Option 3: Newest Arrow workspace-wide, bridge via the Arrow C Data Interface

The C Data Interface is a stable, versioned ABI intended for exchange between Arrow implementations — including two versions of the same implementation. arrow-rs exposes to_ffi/from_ffi over #[repr(C)] structs, so the crossing becomes pointer passing plus a release callback rather than serialisation.

• Pros: full currency, and cost scales with the number of arrays rather than bytes — a different order of magnitude for wide or large batches. The machinery is already scoped: twg-ffi exists as the single audited unsafe crate for the C Data Interface (ADR-0042/0043), so this is existing code serving a second purpose rather than new surface.
• Cons: unsafe at the boundary, with release-callback lifetime discipline — contained in twg-ffi, but real. Both sides must implement compatible C Data Interface spec versions; this is a stable spec but the assumption needs verifying, not assuming. Not free: per-array export/import still costs something.
• To measure: crossing cost against Option 2 on representative batch widths. The hypothesis is a large win; it is a hypothesis.

Option 4: Minimise crossings rather than cheapen them

Unlike the predecessor, DataFusion here is the optional SQL transform stage. Passthrough, neutral-value and direct-Arrow paths bypass it entirely, so the bridge tax applies only to traffic that actually runs SQL.

• Pros: attacks the volume rather than the unit cost. Simple projections and filters could run on our own Arrow kernels, reserving DataFusion — and its boundary — for genuinely complex SQL.
• Cons: two transform paths to maintain and to keep semantically identical, which is its own correctness risk. Only helps if the traffic split is favourable.
• To measure: what fraction of realistic pipelines need full SQL versus a projection or filter that our own kernels could serve.

Option 5: Run DataFusion out of process, over Flight

Version skew is irrelevant across a process boundary.

• Pros: total decoupling; each side upgrades independently. twg-pipeline-flight already exists for sidecar transforms.
• Cons: a network hop and a lifecycle to manage for what is otherwise an in-process stage. Heavyweight as a default, but a legitimate escape hatch if the version conflict ever becomes intractable.

Option 6: Fork or patch DataFusion onto our Arrow

• Rejected. Sustained maintenance of a fork of a large, fast-moving project, for a version-alignment problem. The cost is unbounded and recurring.

Evidence

Measured against the crates.io registry on 2026-07-26.

DataFusion tracks Arrow closely

Median 28 days, range 9–46. Arrow ships a major roughly every three months, so DataFusion is typically one month behind and never a version generation behind.

The previous Arrow line keeps receiving patches

arrow 59.1.0 shipped 2026-07-07; arrow 58.4.0 shipped 2026-07-22 — after it. The N-1 major line is actively maintained, not abandoned on the next major.

This is decisive for Option 1’s central objection. DataFusion pins with a caret (arrow ^58.3.0), so any 58.x at or above that satisfies it. A fix landing in the 58 line is therefore available immediately, without waiting for DataFusion. Only a fix that lands exclusively in 59.x and is never backported would be blocked — a narrower exposure than “delegating security currency” implies.

The ecosystem has converged, which resolves ADR-0023’s binding constraint

DataFusion and iceberg-rust, released within a day of each other, pin the same Arrow major. Pinning 58 therefore removes the version boundary with both — so the question of which of them is the binding constraint dissolves: they agree.

Recommended option

Option 1 — pin the workspace to the Arrow major that DataFusion and the table crates share. Today that is Arrow 58.

The objection to Option 1 was that it delegates security currency to a third party’s cadence. The evidence narrows that materially: DataFusion’s adoption lag is a month, the N-1 Arrow line keeps receiving patches, and the caret pin means those patches are available to us immediately. The remaining exposure is a fix that lands only in the newest major and is never backported — real, but narrow, and detectable.

Choosing it buys the properties every other option spends something to approximate: no version boundary at all, so no bridge to build, no crossing cost to pay per batch, no unsafe at the boundary, and the simplest dependency graph. Options 2 and 3 both exist only to make a boundary cheaper; not having one is strictly better than having a cheap one.

Options 2, 3 and 4 are retained as a designed escape path, not discarded. If the ecosystem diverges — DataFusion and the table crates pinning different majors, or a security fix we cannot wait for — the workspace moves to the newer Arrow and bridges the DataFusion stage alone. Option 3 (the C Data Interface) is the bridge to reach for rather than Option 2 (IPC): its cost scales with array count rather than bytes, and twg-ffi already exists as the single audited unsafe crate for that interface, so the machinery is present rather than hypothetical. Option 4 (minimising crossings, available because DataFusion is the optional transform stage here) reduces whatever that bridge would cost.

Deferred measurements. The IPC-versus-C-Data-Interface crossing cost, and the traffic split between SQL-requiring and projection-only pipelines, are no longer gating: with no boundary, neither number changes the decision. They become the first work of the escape path, if it is ever taken.

Consequences

• The workspace pins Arrow 58, expressed once in twg-codec-core’s re-export.
• Moving Arrow major is a breaking change for our published crates. Because crates publish independently (ADR-0047), arrow is a public dependency and its version is part of the API. The supported version must be stated in each codec crate’s README, and a bump requires a major version bump of those crates.
• Arrow-major moves are scheduled behind DataFusion, roughly quarterly, rather than being taken as they land.
• cargo-audit in CI is the detector for the residual exposure — a Arrow advisory unfixed in the pinned line is the signal to take the escape path, and should be treated as such rather than waited out.
• ADR-0023 loses its Arrow-compatibility dimension entirely; what remains there is feature maturity, which needs hands-on evaluation rather than registry data.

Advice Received

Document version history

ADR-0058: Record-metadata columns — provenance naming and transport-neutral superset

Reference

Amends ADR-0005 (record metadata as reserved columns) and ADR-0051 (configurable metadata prefix, immutable once a target exists). Interacts with ADR-0004 (locator propagation through unnest), ADR-0044 (raw lane), ADR-0054 (content-hash dedup), and the Phase 5 connector-core retrofit and Phase 8 Pulsar work in the delivery plan.

Context

ADR-0005 established that record metadata rides with the record as reserved columns rather than out-of-band, because a sidecar (e.g. carried on OffsetSpan) is lost at the Arrow Flight boundary — precisely where metadata matters most for downstream Lakehouse writes. ADR-0051 made the column prefix configurable (default _twg_) and immutable once a target exists.

Those decisions left two things unresolved that only surfaced once the column set was written down concretely:

• The column names were Kafka-shaped and flat. The enumerated set (_twg_offset, _twg_leader_epoch, _twg_timestamp_type, …) mixes fields that only Kafka supplies with fields common to every transport, under one flat namespace. On a Pulsar stream, _twg_offset and _twg_leader_epoch are meaningless, and Pulsar’s own fields (message id, producer name, sequence id, event time, ordering key, redelivery count) have nowhere to live. The system is explicitly multi-transport, so a Kafka-only column vocabulary is incoherent.

• There was no distinction between metadata the source provides and metadata we produce, and no processing-time lineage at all. The only timestamp modelled was the source’s own (_twg_timestamp + _twg_timestamp_type). There was no record of when Thalweg read a record or when it wrote it, so a record carried no event-time-to-processing-time provenance — information operators need for latency accounting, and pipelines need for correctness when reasoning about lateness.

Options considered + consequences

Dimensions: transport neutrality, self-describing names, migration cost, collision safety.

Option 1: Keep flat Kafka-shaped names; add Pulsar fields ad hoc

Leave _twg_offset etc. as-is and bolt on Pulsar columns as needed.

• Pros: no rename; smallest diff today.
• Cons: the namespace stays Kafka-centric and a reader cannot tell a broker-supplied column from one Thalweg stamped. Pulsar fields accrete without a home. The incoherence ADR-0005 tried to avoid (metadata that doesn’t describe the actual transport) reappears per-transport.

Option 2: A transport-neutral _twg_source_* superset, with bare _twg_* reserved for Thalweg-stamped values

Rename every broker-supplied column to _twg_source_* and define one superset across transports (a column is null where a transport doesn’t supply it). Reserve bare _twg_* (no source infix) for values Thalweg itself produces, and add _twg_ingest_ts / _twg_emit_ts to complete the provenance clock.

• Pros: self-describing — the source infix is the authority marker. One schema describes Kafka and Pulsar. Event-time → processing-time lineage becomes first-class. Collision rules from ADR-0051 carry over unchanged.
• Cons: renames every metadata column — a larger schema change than the configurable-prefix case ADR-0051 governs; must land before any target exists. The Pulsar rows are net-new design until confirmed against a live broker.

Option 3: Namespace by transport (_twg_kafka_*, _twg_pulsar_*)

• Pros: unambiguous per transport.
• Cons: defeats the point of a neutral column set — a consumer would branch on transport to read topic/partition, and a stream migrated between transports would change schema. Rejected by the same reasoning that set the default prefix to transport-neutral _twg_ in ADR-0051.

Decision

Adopt Option 2.

Source-provided columns are named _twg_source_* and form one transport-neutral superset. A column is null for transports that do not supply it, so a single schema describes Kafka and Pulsar alike. This covers topic, partition, Kafka offset / Pulsar message id, leader epoch, key, source timestamp (+ Kafka timestamp type / Pulsar event time), headers (Kafka headers or Pulsar properties, as List<Struct<key, value>> since a MapArray cannot hold Kafka’s duplicate keys or null values), the Kafka batch fields (producer id / epoch / base sequence) and the Pulsar fields (producer name, sequence id, ordering key, redelivery count), and the source cluster id.

Thalweg-stamped columns keep the bare _twg_* form — the absent source infix is the signal that Thalweg, not the broker, is the authority:

• _twg_ingest_ts — stamped on read, at the Source::poll_batch boundary.
• _twg_emit_ts — stamped on write, when a BatchSink commits the batch.
• _twg_record_id — the deterministic content hash over {source cluster, topic, partition, offset, payload} that anchors dedup and recovery (ADR-0054).

_twg_source_timestamp → _twg_ingest_ts → _twg_emit_ts gives an event-time-to-processing-time provenance chain on every record.

The prefix (_twg_) remains configurable and immutable-once-set per ADR-0051; the reserved-namespace collision rule is unchanged (a payload field under the prefix is a hard decode error). MetadataPolicy continues to govern the decoded lanes only — None / Locator (default) / All; the raw lane’s Arrow form carries the full set by construction. The raw lane’s contract is an Arrow RecordBatch; the physical file schema (Parquet layout, opaque-envelope vs. exploded columns) is a sink decision, out of scope here.

Consequences

• Because this renames every source column, it must land before any target exists. Applying it to an existing target is a refused migration under ADR-0051 — a stronger version of the prefix-change case that ADR governs.
• The Pulsar rows are provisional. Pulsar’s message field model is not yet pinned in the repo (twg-pulsar-client is flow-control-only; twg-pulsar-arrow is a stub). The superset above is confirmed against a live Pulsar broker in Phase 8; column names may be adjusted then, which is cheap while the design is pre-implementation.
• RawRecord / Locator are Kafka-shaped today. Generalising them to carry this superset is part of the Phase 5 connector-core retrofit (delivery-plan risk F4).
• unnest propagation (ADR-0004) is specified for the locator only. How the full metadata superset propagates across a cardinality change is follow-up work once this naming is fixed.
• Status is Draft pending review; the delivery plan (v0.7) already reflects this model so the two stay in step.

ADR-0059: Public documentation rendering pipeline — self-hosted rustdoc and mdBook onto the Cloudflare site

Reference

Builds on ADR-0052 (documentation site on Cloudflare Pages at docs.thalweg.dev, apex reserved, build command none) and inherits its constraints wholesale. Interacts with ADR-0047 (independent per-crate publishing) — the crates that most need reference docs are the ones published standalone. Governed by the hard rule in AGENTS.md (nothing company-sensitive on the world-readable surface), enforced by scripts/check-public-surface.sh.

Context

The repository is private; only the Cloudflare-served site under docs/site/ is public (ADR-0052). The prose documentation — ADRs, blueprints, operations notes, the testing standard — lives as markdown under docs/, and API reference for the crates does not exist anywhere yet.

Two gaps follow from that:

• No published API reference. The crates are publish = false today (they are scaffolds), so nothing appears on docs.rs, and even once they publish, the four internal crates (twg-cli, the two *-arrow compositions, twg-e2e) never will. There is no rendered cargo doc for anyone — contributor or reader — without checking out the repository and building it.
• The prose docs are only readable as raw markdown. They render on GitHub, but the repository is private, so a reader arriving from the public site cannot reach them. Linking to GitHub blobs from docs.thalweg.dev would send public readers to a 404-behind-auth wall.

The binding constraint from ADR-0052 shapes every option: the Cloudflare build command is NONE. Cloudflare serves docs/site/ verbatim on push; it does not run mdbook, cargo, or any generator. So anything rendered has to be rendered before the push and committed as static output, or it does not exist on the site. A second, build-running deploy path was explicitly rejected in ADR-0052 because it would race Cloudflare’s app.

A further constraint: the site is world-readable and the repository is not, so whatever is rendered onto it is subject to the same secret-scanning gate as the hand-written pages. Generated output must pass check-public-surface.sh.

Options considered + consequences

API reference: how to publish cargo doc

Dimensions: works for unpublished crates, no company-sensitive leakage, no second deploy path, cost to keep current.

Option A1: Self-host rustdoc — build cargo doc locally, commit into docs/site/api/

• Pros: works regardless of publish status, so it covers the four internal crates and today’s all-unpublished workspace alike. No dependency on crates.io or docs.rs. Served from the same origin as the rest of the site, behind the same (optional) Access policy, scanned by the same gate. Cloudflare stays build-none.
• Cons: the rendered tree is committed, so it is large and it can drift from the source if not regenerated. rustdoc’s search index is a minified base64 blob that trips the AWS-key heuristic in the public-surface scan (a false positive to be handled narrowly, not by blinding the gate).

Option A2: Rely on docs.rs

• Rejected. docs.rs only builds published crates, and only builds them from crates.io. The workspace is entirely unpublished today, and the four internal crates are publish = false forever, so docs.rs can never carry them. It also puts the reference on a different origin from the rest of the docs, and outside the private-repo boundary and the public-surface gate.

Option A3: A CI job builds docs to a separate api. subdomain

• Rejected. This reintroduces exactly the second deploy path ADR-0052 refused: a GitHub Actions job publishing to its own target, racing Cloudflare and splitting the origin. It also runs the secret-bearing build in CI rather than behind the same gate as everything else. The whole point of build-none is that there is one publish mechanism.

Prose rendering: how to turn the markdown into a browsable site

Dimensions: static output (build-none compatible), auto-nav from the tree, handles the ADR/blueprint YAML frontmatter, Rust-toolchain-native.

Option B1: mdBook, rendered locally into docs/site/reference/

• Pros: produces a static tree that Cloudflare serves as-is. It is the Rust-native documentation renderer, so it is one cargo install for a Rust contributor and shares the toolchain already required. Navigation is a generated SUMMARY.md, which a committed generator can rebuild from the docs tree so a dropped-in file appears without hand-editing nav. Intra-doc .md links are rewritten to .html automatically.
• Cons: the rendered tree is committed (same drift risk as rustdoc, same mitigation — regenerate on commit). YAML frontmatter in the ADR/blueprint files is not markdown and renders as raw text unless stripped or preprocessed. mdBook’s own searchindex.js is a minified blob, the same false-positive class as rustdoc’s.

Option B2: MkDocs (or another Python/JS static-site generator)

• Rejected. It adds a non-Rust toolchain (Python + pip) to a Rust workspace purely for docs, which every contributor would then need. mdBook gives the same static-output property while staying inside the toolchain the project already mandates.

Option B3: Link to the raw markdown on GitHub

• Rejected. The repository is private, so public readers arriving from the site or a future crates.io listing hit an auth wall. This is the exact coupling ADR-0052 and DOCS-DEPLOY.md warn against.

Freshness: when the rendered output is produced

Option C1: Cloudflare runs the build

• Rejected by ADR-0052. The build command is NONE and a second deploy path races the first. Not available.

Option C2: A pre-commit hook regenerates incrementally and stages the output

• Pros: the rendered output lands in the same commit as the change that caused it — the discipline ADR-0052 already established for the graphify map and AGENTS.md already requires for docs. Incremental (only affected crates rebuild their rustdoc; the book rebuilds only when docs/** markdown changes) keeps commit latency bounded. Fail-soft, so a contributor without mdbook or the toolchain is warned, not blocked.
• Cons: contributors need mdbook and the Rust toolchain installed to regenerate locally; without them the hook no-ops and the output can go stale until someone with the tooling commits. Accepted: the tooling is already needed to work on the crates, and the fail-soft design matches the existing graphify hook.

Option C3: A periodic/manual regeneration

• Rejected. Exactly the “I’ll update the docs after” failure mode AGENTS.md calls out — it is how the site once carried a decision count of 44 against an actual 57. Same-commit regeneration is the established discipline here.

Decision

Self-host both reference surfaces, rendered locally and committed, served by the existing build-none Cloudflare project.

• API reference (Option A1). cargo doc --workspace --no-deps is rendered into docs/site/api/, served at docs.thalweg.dev/api/ and per-crate at /api/<crate>/. scripts/gen-api-docs.sh regenerates the docs for a given set of affected crates and syncs them into the tree.
  • The rustdoc SOURCE VIEW is stripped, never published. rustdoc’s default output includes a full syntax-highlighted source view — every crate rendered into api/src/**/*.rs.html, indexed by api/src-files.js, and reachable from a per-item Source link on every page. The repository is private, so publishing that would publish the entire private codebase onto the world-readable site — forbidden by the hard rule in AGENTS.md. rustdoc has no stable flag to suppress source generation, so the pipeline generates then strips: scripts/strip-rustdoc-src.sh deletes api/src/ and api/src-files.js and removes the relative Source anchors from every rendered page (upstream https://doc.rust-lang.org/…/src/… links, which reference Rust’s own source rather than ours, are preserved). gen-api-docs.sh invokes it after every render — whole-workspace and per-crate alike — so a future docs change can never re-publish the source view. Only the signatures and doc comments (the per-crate api/<crate>/ pages) reach the site.
• Prose reference (Option B1). An mdBook under docs/book/ renders the ADRs, blueprints, operations notes, testing docs and PUBLISHING.md into docs/site/reference/, served at docs.thalweg.dev/reference/. Its build-dir targets ../site/reference so mdbook build docs/book writes straight into the publish root. The SUMMARY.md navigation is generated from the docs tree by scripts/gen-reference-book.sh, which also strips the YAML frontmatter as a pre-render step so it does not appear as raw text.
• Freshness (Option C2). .githooks/pre-commit is extended, alongside the existing graphify step, to regenerate incrementally on commit: changed crates/<name>/** trigger gen-api-docs.sh for just those crates; changed docs/** markdown triggers a SUMMARY regeneration and mdbook build. Every step is fail-soft — a missing cargo/mdbook or any error warns to stderr and the commit proceeds; the hook always exits 0.
• The bespoke landing page stays the apex. docs/site/index.html remains hand-authored; it gains plain links into /reference/ and per-crate /api/ pages, without a new numbered section (which would renumber the gated nav).
• One theme and one way back across all three surfaces. The reference and the crate docs shipped with their own generator defaults (mdBook’s rust/ayu, rustdoc’s light/dark), and neither offered a route back to the landing page — a reader arriving at a crate page was stranded. Both are now re-skinned to the landing palette (paper / iron / verdigris) and carry the same persistent brand bar linking back to the main docs:
  • Reference (mdBook). default-theme/preferred-dark-theme are pinned to light, re-skinned by docs/book/custom/thalweg.css, and a brand bar is injected by docs/book/custom/thalweg.js — both loaded via additional-css/ additional-js and kept outside src/ so the SUMMARY/src/ regeneration never clobbers them.
  • Crate docs (rustdoc). docs/rustdoc/head.html (palette + layout) and docs/rustdoc/before.html (the brand bar) are injected on every page via RUSTDOCFLAGS=--html-in-header/--html-before-content, set in scripts/gen-api-docs.sh, so the theme is re-applied on every render rather than patched in after the fact.
• /api/ has a crate overview instead of a 404. cargo doc --workspace renders one tree per crate but no root index, so docs.thalweg.dev/api/ 404’d and per-crate docs were reachable only by guessing a URL. scripts/gen-api-index.sh writes docs/site/api/index.html — a landing page in the main theme listing and linking every rendered crate. The list is derived from the filesystem (each rustdoc crate root under api/), so it cannot drift from what was rendered, and it runs after the source-view strip, so it only ever links crate index pages, never source. gen-api-docs.sh invokes it after every render.
• The public-surface gate is rescoped minimally. check-public-surface.sh excludes only the minified search-index artifacts (rustdoc’s search-index*.js / search.index/*.js, mdBook’s searchindex.js / searchindex.json) — not the /api or /reference trees — so the rendered HTML (real ADR and rustdoc prose) is still scanned for secrets.

The crates stay publish = false. Self-hosting is precisely what lets the reference exist without publishing, so nothing here flips a crate to published.

Consequences

• Generated output is committed, and must be kept in step. The pre-commit hook is the mechanism; the fallback is scripts/gen-api-docs.sh and scripts/gen-reference-book.sh run by hand. Regeneration is deterministic, so a stale tree is a diff, not a mystery.
• Contributors need mdbook and the Rust toolchain to regenerate locally. This is documented in PUBLISHING.md, operations/DOCS-DEPLOY.md and CONTRIBUTING.md. Absent them the hook fail-soft no-ops, matching the graphify hook’s contract.
• The file count under docs/site/ grows substantially — rustdoc emits a few thousand files for a 44-crate workspace. This stays well under Cloudflare’s 20,000-files-per-deploy limit, but it is now a number to watch as crates gain real surface; the file count is reported in the pull request that introduces the pipeline.
• The public-surface exclusion is narrow by construction. It names the search-index files specifically; the rendered prose is still scanned. If a future generator emits a new minified blob that trips the gate, the fix is to name that file too, never to exclude a whole tree.
• rustdoc --no-deps documents the workspace crates only, not dependencies, so the tree does not balloon with third-party docs and does not leak dependency internals onto the public surface.
• The API surface is signatures and doc comments, not source. Stripping the source view (above) means the public reference shows what each crate exposes and how it is documented, but never the implementation — the appropriate boundary for a private codebase’s public reference. The strip is a script step, not a manual deletion, so it holds across every regeneration.
• Cloudflare stays build-none. No second deploy path is introduced; the site is still served verbatim, so ADR-0052’s ordering argument (gates protect the merge, branch protection enforces sequencing) carries over unchanged.

Advice Received

Document version history

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

ADR-0061: Regenerate rendered docs on main in CI, not in the pre-commit hook

Reference

Amends ADR-0059 (public documentation rendering pipeline: self-hosted rustdoc into docs/site/api/ and mdBook into docs/site/reference/, regenerated locally and committed). Interacts with ADR-0052 (Cloudflare Pages serves docs/site/ verbatim via Git integration, build command NONE) — unchanged by this ADR.

Context

ADR-0059 has the .githooks/pre-commit hook regenerate the rendered surfaces — docs/site/api/ (rustdoc, ~705 files) and docs/site/reference/ (mdBook, ~63 files) — and commit them in the same commit as the source change, so Cloudflare (which serves the tree verbatim) is always current.

That made every branch mutate ~768 generated files. Two branches that each touch a crate or a doc regenerate overlapping subsets, so they conflict on merge — and did: a routine feature PR hit 74 generated-file conflicts (api/*, reference/searchindex.js), dwarfing the 13 real code conflicts. The generated tree is a pure function of the source; a hand-resolved merge of it carries no information and is pure toil (and risks a corrupt, half-merged rendered page).

Decision

Stop regenerating the rendered surfaces in the pre-commit hook. Regenerate them once, on main, in CI after each merge, and commit them back.

• The .githooks/pre-commit hook keeps the graphify map sync (small, canonical) but no longer runs gen-api-docs.sh / gen-reference-book.sh. Branches therefore never mutate docs/site/api or docs/site/reference, so PRs cannot conflict on them.
• A new workflow .github/workflows/docs-regen.yml (push to main, guarded to the paths that affect the rendered output) runs the same two scripts whole-workspace, refreshes the hand-authored index.html counts that check-docs-sync.sh gates, and commits the result back to main with a [skip ci] message (so the bot commit does not re-trigger the workflow). A concurrency group serialises regens.
• The files stay tracked and Cloudflare’s Git-integration deploy (ADR-0052) is unchanged — it still serves the committed docs/site/ verbatim; only the writer moved from every developer’s commit to one post-merge CI job.

Options considered + consequences

Option 1 (chosen) — Regenerate on main in CI, keep the files tracked

• Pros: eliminates the entire class of generated-doc merge conflict with no change to the Cloudflare deploy model or any new secret; the rendered site stays in the repo (diffable, servable verbatim); index.html counts self-heal so a PR that adds a crate/ADR no longer has to hand-edit them.
• Cons: the ~768 files remain in the repo (history churn moves to a post-merge bot commit); the deployed site is briefly stale between a merge and the regen job finishing; the bot needs permission to push to main.
• Cost: one CI job per docs-affecting merge (~a couple of minutes).

Option 2 — Untrack the generated trees; build + deploy from CI via Wrangler

• Pros: the generated artifacts leave the repo entirely (no churn, no tracked build output).
• Cons: replaces ADR-0052’s Cloudflare Git-integration with a Wrangler direct-upload deploy, needs a CLOUDFLARE_API_TOKEN secret, and is a larger change to a working deploy path.
• Recommendation: rejected for now as disproportionate; revisit if the tracked generated tree becomes a repo-size problem. Recorded so the trade-off is explicit.

Option 3 — Do nothing; keep resolving the conflicts by hand

• Rejected: the conflicts are pure toil on a generated artifact and recur on every docs-touching PR.

Operational note

The regen job pushes to main, so it authenticates as a dedicated GitHub App (Contents:write, installed on this repo) rather than the default GITHUB_TOKEN: the job mints a short-lived installation token with actions/create-github-app-token and pushes as the App’s bot identity. That bot must be added to main’s ruleset / branch-protection bypass list, or the push to the protected branch is rejected. Secrets required: DOCS_APP_ID, DOCS_APP_PRIVATE_KEY. If the bypass is ever withdrawn, switch the final step to open a PR instead of pushing directly.

Document version history

ADR-0062: Unity Catalog as a table-catalog backend now; extract twg-catalog-unity on a trigger

Reference

Builds on ADR-0024 (catalog credential vending: UC, S3 Tables, S3/IAM, Iceberg REST) and ADR-0037 (catalog descriptive metadata: comments/properties/tags on create+evolution, idempotent re-sync). Interacts with ADR-0023 (single open table format — Iceberg; read Delta via generated Iceberg metadata), ADR-0019 (twg-type-map is the describe-only type authority), and ADR-0038/0057 (a sink cannot write, nor hold the primary-raw role, without a provisioned/registered table). Detailed delivery: docs/blueprints/zerobus-sink-implementation-plan.md §5.1 (the Unity Catalog / table-catalog stage).

Context

twg-table-catalog owns the transport-neutral TableCatalog trait — schema resolution, sink registration, credential vending, and descriptive metadata. Every sink depends on it: a sink only writes, so table creation, on-the-fly schema evolution, comments, and credential vending live in the catalog layer, and a sink (Zerobus in particular) is unusable without it.

Unity Catalog is the first concrete backend. Its surface is non-trivial — OAuth machine-to-machine auth with token refresh, the UC REST request/response models, credential-vending and Iceberg-REST endpoints, and retry/rate handling. That raises a crate-shape question: does UC belong inside twg-table-catalog as a backend module, or as its own client crate (twg-catalog-unity) that the catalog crate depends on — mirroring how transport clients (twg-kafka-client, twg-pulsar-client) are split out from twg-connector-core?

Deciding this prematurely cuts both ways. Extracting a crate before there is a second consumer or a large surface adds a published crate, a version cadence, and a trait/dependency boundary to maintain for no present benefit. Never extracting it risks twg-table-catalog accreting a large UC-specific client that other backends (S3 Tables, Glue, Iceberg REST) must compile past and that a would-be external consumer of “just the UC client” cannot take on its own.

Decision

Implement Unity Catalog as a backend module inside twg-table-catalog now. Extract it to a dedicated twg-catalog-unity client crate later, only when a trigger fires. The sink is written against the TableCatalog trait either way, so the extraction is a non-breaking internal refactor when it happens.

• The TableCatalog trait is the seam. Sinks (twg-sink-zerobus and others) call describe / reconcile / register_sink / vend_credentials and never name UC directly. twg-type-map remains the describe-only Arrow→dialect authority both the catalog and sinks execute against; neither hand-builds DDL.
• The UC backend delivers the ADR-0024/0037 duties driven by the incoming Arrow RecordBatch schema: create-on-absent, evolve-on-drift (additive safe / widening behind opt-in via twg-type-map::can_widen_to / else refuse loud), comments at create AND evolution (two authorities), idempotent re-sync (twg:managed/twg:comment_hash), and scoped temporary credential vending for both the write path and the Iceberg recovery read path.

Extraction trigger

Promote the UC backend module to a standalone twg-catalog-unity crate when any of these holds — record the promotion as its own follow-up when it fires:

1. A second consumer outside twg-table-catalog needs the raw UC client (e.g. a standalone provisioning CLI, or a non-sink component doing UC lookups).
2. Surface growth: the UC-specific code (auth/refresh + REST models + retry/rate handling) exceeds roughly a module’s worth and starts imposing UC-only dependencies on the catalog crate’s other backends.
3. Independent release need: UC needs to version or ship on a cadence independent of twg-table-catalog.

Until a trigger fires, one backend module is the smaller, correct commitment.

Options considered + consequences

Option 1 (chosen) — Backend module now, extract on a trigger

• Pros: smallest thing that works; no premature crate/boundary/version cost; the trait insulates every sink, so extraction is a later non-breaking refactor; matches thalweg’s deferred-work discipline (a trigger, not a guess).
• Cons: if a trigger fires soon, there is a (small, mechanical) extraction to do later.

Option 2 — Extract twg-catalog-unity up front

• Pros: clean client/backend split from day one; a would-be external consumer can take the UC client alone.
• Cons: a published crate, version cadence, and trait/dependency boundary to maintain before any second consumer or surface pressure justifies it — speculative structure. Rejected for now; revisited by the trigger.

Option 3 — UC logic in the sink

• Rejected outright. It would duplicate provisioning across every sink and couple the sink to a specific catalog, violating the trait seam and the crate-isolation standard.

Invariants pinned by this ADR

• Sinks depend on the TableCatalog trait, never on UC directly.
• twg-type-map is the only place Arrow→dialect type mapping lives; no DDL text is hand-built in the sink or the UC backend.
• Credential vending yields temporary, scoped credentials inheriting the caller’s privileges — never static secrets held by the sink.
• Extraction to twg-catalog-unity, if/when it happens, is non-breaking for sinks (same trait) and is gated on a stated trigger, not a guess.

Document version history

ADR-0063: Proto-source bundling — one FileDescriptorSet, per-package .proto, and a message-type manifest from any proto source

Reference

Builds on ADR-0001 (pure-Rust, no external toolchain on PATH) and the existing protobuf descriptor toolkit: twg-proto-schema already owns compile_proto_to_bytes (source .proto → FileDescriptorSet via protox, no protoc), build_arrow_schema (descriptor → Arrow schema), and twg-proto-flatten/twg-proto-decode consume the resulting FileDescriptorSet. Related to ADR-0046 (protobuf decode strategy).

This ADR adds the acquisition step in front of that toolkit: turning a third-party proto source tree (typically a Java/Maven/Gradle contracts repo with many src/main/proto roots) into the single FileDescriptorSet those crates already require, plus two human-facing artefacts. Implemented by a new tooling crate twg-proto-bundle and a twg proto bundle subcommand; the full build is in ../blueprints/proto-bundle-implementation-plan.md.

Context

Upstream contracts arrive as repositories of .proto files, not as compiled descriptors. The shape observed across representative sources — a small one (~9 files / 1 include root), a medium one (~130 files / 8 roots), and a large one (~530 files / 77 roots) — is a Java convention: each Maven/Gradle module carries its own .../src/main/proto include root, imports are written relative to those roots (import "example/common/id.proto"), and Google well-known types (timestamp, duration, wrappers, any) are imported. All observed files are proto3.

Three facts make naive compilation fail or mislead:

• Many include roots per repo. A single protoc/protox invocation needs the union of every src/main/proto directory on the include path, discovered, not hand-listed. 77 roots is not a config the caller should assemble by hand.
• Duplicate import paths. The same logical path resolves under multiple roots — one shared file (e.g. example/common/feed.proto) can appear under several module roots, and the large source carried 7 internal duplicates (e.g. a vendored build-staging tree alongside the canonical one). Today the colliding copies are byte-identical, but a descriptor keyed by import path cannot hold two definitions of one path, and silent “first root wins” would mask future drift into a wrong-but-compiling bundle.
• The consumer needs a type name, not just bytes. build_arrow_schema(pb, message_type, depth) and the decode path are driven by a fully-qualified message type (e.g. com.example.orders.Order). Nothing in a raw repo tells the operator which strings are valid.

Separately, “give me the merged .proto” is a common and reasonable ask, but a single compilable .proto cannot span multiple packages — the language allows one package per file, and a real source spans several (e.g. com.example.orders, example.common, example.foo). So the tool emits both: a single combined <name>.proto (compilable when the source is single-package; an explicitly-labelled combined view when multi-package) and one valid file per package. Neither replaces the .pb as the machine-consumable artefact.

Finally, the source is not always a repo. Operators also hold a single already-merged .proto, or a bare directory of .proto files with no src/main/proto convention. The tool must accept all three without a mode flag.

Decision

Add twg-proto-bundle: a pure-Rust tooling crate (and twg proto bundle subcommand) that resolves any proto source to one combined FileDescriptorSet (.pb), a single combined .proto, one proto3 file per package, and a fully-qualified message-type manifest. The .pb is authoritative and machine-consumable; the .proto renders and the manifest are derived, human-facing conveniences.

Concrete shape:

1. Source resolution — auto-detected, additive, prunable. The <SOURCE> positional resolves:

   1. a .proto file → its parent dir is the include root (only that file);
   2. a directory where --include-glob (default **/src/main/proto) matches ≥ 1 dir → repo mode, roots are the matches;
   3. otherwise → the directory itself is the lone include root (flat / bare package-tree mode).

   --include-root <DIR> (repeatable) adds extra roots on top of auto-detection — each resolved through the same glob detection — for a cross-repo dependency whose .proto live outside <SOURCE> (observed: one source imports shared types that ship in a separate repo’s contracts subtree). --exclude <name> (repeatable) prunes any subtree with that path component — needed for a vendored build-staging copy that diverges from the canonical tree (observed: a generated build directory mirroring the shared contracts).

2. Dedup by import path, identity-checked. Files are keyed by import-relative path. A path resolving under multiple roots is collapsed only if the copies are byte-identical; divergent copies are a hard error with a conflict report (path, competing roots, hashes), overridable via --on-conflict {error|first-wins}. Default error: silent drift must not corrupt the bundle.

3. Stage into one clean root, then compile. The deduped set is staged into a temp tree preserving import paths, compiled with that single dir as the sole include root via twg_proto_schema::compile_proto_to_bytes. One root removes all include-order ambiguity and makes the render deterministic. A leading UTF-8 BOM is stripped while staging — protox (unlike protoc) lexes a BOM as an invalid token, and a BOM-only difference must not read as a content conflict (observed: several source files are BOM-prefixed). protox bundles the Google well-known types and includes every reachable type (--include_imports semantics), so the .pb is self-contained.

4. .pb is the authoritative artefact. One FileDescriptorSet per invocation, holding every file in the source plus transitively-reachable WKTs. This is the exact input twg-proto-schema/twg-proto-decode already consume.

   Unresolved cross-repo dependencies are alerted, not opaque. When compilation fails because an import points at a file the source does not provide (and that is not a bundled Google WKT), the staged files’ import statements are scanned and every missing dependency is reported at once — each with the file(s) that require it — so the operator can supply them in one pass (e.g. an extra --include-root), rather than fixing them one protox-error-at-a-time. Auto-resolving those dependencies from a corpus of related repos remains deferred (D3).

5. .proto render is derived, in two forms. Both are rendered from the descriptor (maps un-synthesised from MapEntry, oneofs, proto3 optional, nested types, enums):

   • A single combined .proto — <name>.proto. When the source is single-package this is a valid, standalone-compilable file — verified end-to-end by feeding it back through the tool and diffing the recompiled type inventory (identical). When the source is multi-package, a single compilable .proto is impossible — the language allows one package per file — so the output is a combined, package-delimited view carrying a leading banner comment; it is for reading/grepping/diffing, and the .pb is the artefact to compile. The tool reports which form it produced.
   • One .proto per package — <name>/<package>.proto, importing sibling packages; Google imports centralised (a base file, import public) so no file redefines a WKT. Correctness gated by a round-trip test: recompile the render, diff the normalised type set against the source .pb.

   A per-package set only recompiles when the package dependency graph is acyclic (proto forbids circular imports; a single-package source round-trips exactly). Real multi-package sources are often cyclic (observed: two of the sources each have a mutually-recursive package pair). The bundler detects and reports the cycle; the render remains a readable view, and the .pb is authoritative regardless.

6. Message-type manifest. <name>.messages.json lists every fully-qualified message type in the set and flags root candidates (messages never used as a field type by any other message — the envelope/entry points), so the operator can read off the message_type string to hand to build_arrow_schema (e.g. com.example.orders.Order).

7. Tooling crate, not a runtime dependency. twg-proto-bundle sits above twg-proto-schema, is publish = false, and nothing on the ingest/runtime path depends on it. It is a build/operator-time artefact generator.

Options considered + consequences

Dimensions: correctness under duplicate paths, determinism, no external toolchain, operator ergonomics.

Option 1 (chosen) — Discover → dedup/identity-check → stage → compile; render per package; emit manifest

• Pros: single command for file/dir/repo sources; deterministic compile from a clean staged root; duplicate-path drift caught, not masked; pure Rust (no protoc); reuses the existing compile_proto_to_bytes; the .pb is exactly what downstream already consumes; round-trip test makes the render trustworthy.
• Cons: a stage copy per build (cheap — hundreds of small text files); a descriptor→.proto printer is real code with fidelity risk (bounded by the round-trip gate).

Option 2 — Hand the caller’s include roots straight to protox, no staging

• Pros: less code.
• Cons: duplicate import paths across roots make the result depend on include-order; a future divergence compiles silently to the wrong bytes. Loses the identity check. Rejected.

Option 3 — Shell out to protoc --include_imports --descriptor_set_out

• Pros: battle-tested compiler; free .pb.
• Cons: reintroduces an external toolchain on PATH, contra ADR-0001; still leaves discovery, dedup, per-package render, and the manifest unbuilt. Rejected.

Option 4 — Emit a single merged .proto instead of per-package

• Pros: matches the literal “one file” ask.
• Cons: a single compilable .proto is impossible for multi-package sources (one package per file); making it compile would force dropping packages and globally-unique renames across hundreds of files with colliding simple names (Metadata, Envelope, Status). Rejected as the sole form. Adopted as an additional output (<name>.proto): a genuine compilable file when the source is single-package (verified by recompiling it), and an explicitly-labelled combined view when multi-package. The per-package files and the authoritative .pb stand alongside it.

Invariants pinned by this ADR

• .pb is authoritative — the single combined .proto, the per-package .proto, and the manifest are all derived from it, never the reverse.
• Single .proto is honest about compilability — compilable only for a single-package source; a multi-package source produces a banner-labelled combined view, and the tool reports which form it emitted.
• No silent path collision — identical duplicates collapse; divergent duplicates error by default.
• Deterministic compile — from a single staged include root, not caller include order.
• No external toolchain — pure-Rust protox, consistent with ADR-0001; no protoc on PATH.
• Render is round-trip-verified for acyclic sources — an acyclic per-package render that does not recompile to the source type set is a test failure; a cyclic source is reported, and the .pb is the artefact to consume.
• Source-shape agnostic — a single .proto, a bare dir, and a src/main/proto repo all resolve through one command; cross-repo deps add roots, vendored mirrors are excluded.
• Missing dependencies are named, not opaque — an unresolved import fails with the complete list of missing files and their requirers, never a single first-encountered protox error.

Deferred work

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

• D1 — Custom option / extension fidelity in the render. The proto3 printer targets messages, enums, oneofs, maps, nested types, reserved, and standard field options. Trigger: a source using custom options (extensions) fails the round-trip diff.

• D2 — Compilable output for cyclic package graphs. A per-package render of a mutually-recursive package pair cannot recompile (proto forbids circular imports); today it is detected and reported, with the .pb as the artefact to consume. Trigger: an operator needs standalone-compilable .proto text for a cyclic source (would require preserving the original file boundaries, which were acyclic, rather than grouping by package).

• D3 — Auto-resolving cross-repo dependencies. Missing imports are now detected and reported (every unresolved import + its requirers), and roots/excludes are supplied explicitly. What remains deferred is resolving them automatically: given a corpus of related repos, find which one provides each missing import and add its root without the operator naming it. Trigger: a manifest/search-root of related repos exists that the tool could search transitively.

• D4 — proto2 support. All observed sources are proto3; the printer assumes it. Trigger: a proto2 source appears (adds required/groups/default-label rules).

Interaction with existing ADRs

• ADR-0001 (pure-Rust): honoured — protox, no protoc.
• ADR-0046 (protobuf decode): this ADR feeds that path; the .pb it emits and the message_type from its manifest are the inputs to build_arrow_schema / twg-proto-decode.

Document version history

ADR-0064: Cross-format schema bridge — a neutral hub, with protobuf → ODCS as the first spoke

Reference

Builds on ADR-0063 (proto-source bundling to a FileDescriptorSet), ADR-0016 (ODCS precedence / quality-rule merge), and ADR-0019 (twg-type-map as the describe-only, sans-io type-mapping authority). Consumes the .pb that twg-proto-bundle produces. Touches twg-contract-core (neutral schema hub), twg-contract-odcs (ODCS emit model), and a new twg-schema-bridge crate; wired into twg proto bundle --emit odcs.

Context

The bundler (ADR-0063) turns a proto source into a FileDescriptorSet and human artefacts. The next ask is to also emit an ODCS (Open Data Contract Standard) data contract from that schema, and — later — to go the other way (ODCS → .proto

• .pb) and to import/export further formats (JSON Schema, Avro). ODCS validation has other planned uses beyond this tool.

Two things shape the design:

• A hub already exists for leaf types. twg-type-map is the sans-io type-mapping authority (ADR-0019); it already carries json_schema and avro spokes and a common-ground type. Adding format-to-format schema conversion should extend that hub-and-spoke shape, not invent a parallel one.
• Formats differ in structure, not just leaf types. Protobuf carries packages, nested messages, oneof, map, and field numbers; ODCS carries a schema[] of objects with properties[], relationships[], and logicalTypeOptions. A conversion needs a neutral structural model as well as neutral leaf types.

A pairwise (format↔format) approach costs N² adapters as formats accumulate — the exact thing a hub prevents.

Decision

Convert every schema format through a neutral hub: structure through twg-contract-core’s schema model, leaf types through twg-type-map. Each format (protobuf, ODCS, JSON Schema, Avro) is a spoke to/from that hub. Build the protobuf → ODCS spoke now in a new twg-schema-bridge crate; defer the rest.

Concrete shape:

1. Neutral structural hub — twg-contract-core::schema. A SchemaModel is a flat set of SchemaObjects (fully-qualified name, is_root, fields). A Field has a neutral LogicalType (width/signedness preserved: Int32/Int64/UInt32/ UInt64/Float/Double/Bool/String/Bytes/Timestamp/Duration/Enum/Object) and a Cardinality (Single/Optional/Repeated/Map{key}). An object-typed field is an Object{fq-name} reference — objects reference each other by name, so a multi-package/multi-message source maps onto multiple cross-referencing objects with no inlining. This is the structural half of contract-core’s eventual resolved spec; quality rules / precedence / merge stay scaffold.

2. Leaf-type hub — twg-type-map. The cross-format type vocabulary. The protobuf spoke maps proto scalars → neutral LogicalType at the proto edge (proto’s type knowledge belongs there, mirroring twg-proto-schema owning proto’s Arrow derivation, ADR-0019); the JSON Schema / Avro spokes, when built, use type-map’s existing json_schema / avro mappings. type-map is thus the leaf hub in principle, invoked by each spoke as it lands.

3. Bridge crate — twg-schema-bridge. Depends on twg-contract-core (neutral model) and twg-contract-odcs (ODCS emit model); consumes a .pb as bytes, so it does not depend on twg-proto-bundle (keeping the bundler free of the contract stack) and the bundler does not depend on it. Ships: descriptor_set_to_schema_model, schema_model_to_odcs, and the composed descriptor_set_to_odcs.

4. ODCS emit model — twg-contract-odcs::model. A serde write-model for ODCS v3.1.0 (apiVersion/kind/id/version/status, schema[], properties[] with logicalTypeOptions, relationships[]). Parsing ODCS into the resolved spec remains a separate, scaffold concern.

5. protobuf → ODCS mapping (the built spoke). Each message → one schema[] object. A message-typed field → logicalType: object plus a schema-level relationships entry (from: Object.field, to: TargetObject, type: reference) — so packages/cross-references survive as ODCS’s own multi-schema references, not lost. Width/precision is preserved in logicalTypeOptions.format (i32/i64/u32/u64/f32/f64). Google well-known wrappers unwrap to their scalar; Timestamp/Duration map across; other google.protobuf.* degrade to string.

6. CLI. twg proto bundle --emit odcs writes <name>.odcs.json. It runs at the CLI layer (bundle → bridge on the produced .pb), preserving the dependency direction in (3).

Honest losses (pinned, like ADR-0063’s “.pb is authoritative”)

protobuf → ODCS is a faithful contract projection, not a wire-faithful mirror:

• Field numbers have no ODCS equivalent — dropped. So a round-trip back to proto cannot recover the original wire format.
• oneof grouping is not expressible — members render as ordinary (optional) fields; the mutual-exclusivity is lost.
• map<K,V> has no native ODCS type — rendered as an array of {key,value} objects.
• enums have no native ODCS type — rendered as string with the allowed values preserved in the property description.

The .pb remains the authoritative, wire-faithful artefact; the ODCS is the contract/documentation view.

Options considered + consequences

Option 1 (chosen) — Neutral hub (contract-core structure + type-map leaf), bridge crate, proto→ODCS first

• Pros: N adapters not N²; consistent with the existing type-map hub; JSON Schema/Avro slot in cheaply; bundler and contract crates stay dependency-clean; the neutral schema is the foundation contract-core needs anyway.
• Cons: more upfront than a direct bridge (a neutral model to define); the neutral model must grow as exotic features (oneof, custom options) demand.

Option 2 — Direct protobuf↔ODCS bridge, no neutral hub

• Pros: least code to ship the first two flows.
• Cons: N² adapters as JSON Schema / Avro arrive; duplicates leaf-type logic that type-map already owns. Rejected — the hub is the point.

Option 3 — Put proto↔ODCS inside the existing crates (emit in proto-bundle, parse in contract-odcs)

• Pros: fewer crates.
• Cons: couples the bundler to the contract stack and contract-odcs to proto descriptor traversal; neither stays independently publishable. Rejected in favour of a dedicated bridge.

Invariants pinned by this ADR

• One hub, many spokes — every format converts through the neutral schema model; no pairwise format↔format adapters.
• Dependency direction — bridge → {contract-core, contract-odcs}; bundler depends on neither the bridge nor the contract stack; the bridge consumes a .pb as bytes.
• .pb is authoritative — ODCS is a projection; lossy directions are documented, not silent.
• Leaf-type authority is type-map — spokes map through it (proto at its own edge, per ADR-0019); no parallel type vocabulary.

Deferred work

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

• D1 — ODCS → proto + .pb. The reverse spoke: ODCS → neutral → generate .proto(s) → compile via compile_proto_to_bytes. Field numbers are synthesised deterministically (sequential), so output will not wire-match a pre-existing proto. Trigger: an operator needs .proto/.pb generated from an ODCS contract.
• D2 — JSON Schema and Avro spokes. Import/export through the same hub, reusing type-map’s json_schema / avro leaf mappings. Trigger: a format beyond proto and ODCS is required.
• D3 — Richer ODCS fidelity. oneof as an ODCS construct, enum allowed-values as logicalTypeOptions rather than description text, primary-key / uniqueness from contract keys, column-level (not object-level) relationships. Trigger: a consumer needs one of these expressed structurally.
• D4 — ODCS parse → resolved spec. The read side feeding validation (ADR-0016). Trigger: the contract-validation phase is scheduled.

Interaction with existing ADRs

• ADR-0063 (bundling): this ADR consumes the .pb it emits; --emit odcs extends the same twg proto bundle command.
• ADR-0019 (type-map): honoured — type-map is the leaf-type hub; the proto spoke maps at the proto edge, as proto’s Arrow derivation already does.
• ADR-0016 (ODCS precedence/merge): the ODCS emit model here is independent of the parse/merge path that ADR-0016 governs; D4 connects them.

Document version history

ADR-0065: Zerobus sink — substrate seams, the Arrow-IPC version bridge, and a feature-gated SDK adapter

Reference

Implements the Zerobus sink designed in ../blueprints/zerobus-sink-implementation-plan.md, realising the pipelined fire-and-forget ack of ADR-0060, the source-ack invariant of ADR-0038, and the gap-aware coverage of ADR-0053. Builds on the catalog seam of ADR-0062 (twg-table-catalog, already implemented) and the type-mapping authority of ADR-0019 (twg-type-map::zerobus_sink). Honours the single-Arrow-major pin of ADR-0003.

Context

Building twg-sink-zerobus surfaced three decisions the blueprint did not settle, because they only arise at implementation time.

1. The substrate the sink implements did not exist. The blueprint has the sink implement BatchSink (twg-stream-arrow) and Sink (twg-connector-core) and call a coverage store (twg-offset-store). All three were 0-LOC scaffolds. The sink cannot be built without them.
2. The Databricks SDK is on a different Arrow major. databricks-zerobus-ingest-sdk uses Arrow 59; thalweg is pinned to Arrow 58 (ADR-0003, DataFusion’s major). The SDK’s ingest_batch(RecordBatch) therefore takes a different RecordBatch type than the sink holds — they cannot be passed directly.
3. The SDK is a heavy, network-bound dependency. Compiling it (with its Arrow-Flight/tonic stack and a second Arrow major) on every build, and requiring a live workspace to test the sink’s logic, is unacceptable. The correctness core — the ADR-0060 ack lane, coalescing, coverage — must be testable without it.

Decision

Build the minimal real substrate the sink needs, bridge the Arrow-major gap with Arrow IPC bytes, and invert the SDK behind a ZerobusTransport seam with a feature-gated real adapter and an in-memory fake.

Concrete:

1. Substrate, built minimally and for real.

   • twg-connector-core::sink — Sink, Receipt, OffsetSpan (source-position range, half-open [start, end)), SinkId, SinkError.
   • twg-stream-arrow::batch_sink — BatchSink, CommitInfo.
   • twg-offset-store::coverage — CoverageStore, CoveredRange, gap-aware union-merge and gap computation, and an InMemoryCoverage reference backend. Union-merge is a correctness requirement (ADR-0053: two rebalance writers both record what they wrote), not an optimisation.

   These are the seams the sink implements/calls, no more — the broader flow-control, recovery-role, and durable-backing machinery documented in those crates remains design intent for its phase. twg-sink-zerobus implements both traits.

2. ZerobusTransport — the inverted boundary. A single-stream async trait (submit(ipc_bytes) -> offset, wait_for_offset(offset), flush, close, unacked). The whole correctness core (ack lane, pool, coverage) is written and tested against a fake with scriptable ack latency / transient failures / poison. This is what makes the ADR-0060 properties — one wait_for_offset per coalesced burst, delivered monotonicity, confirm-only-after-ack, poison→exit, backoff-retry-on-existing — provable in CI without Databricks.

3. Arrow-IPC version bridge. The sink stays on Arrow 58 throughout. At the transport boundary a RecordBatch is serialised to Arrow IPC stream bytes (twg-sink-zerobus::ipc) and handed to the SDK’s ingest_ipc_batch(Bytes). IPC is forward/backward-compatible across 58↔59, so no Arrow-59 array/batch type ever enters the sink. Only the stream schema is rebuilt into the SDK’s Arrow-59 Schema at open time (a finite type map). This keeps ADR-0003’s single-major pin true for thalweg’s own code; Arrow 59 exists only inside the SDK, behind an optional feature.

4. Feature-gated real adapter. The Databricks ZerobusArrowStream adapter (databricks.rs) lives behind the databricks-sdk feature (off by default), so the SDK and its Arrow 59 are pulled in only when explicitly enabled. Auth uses a catalog-vended bearer token via a HeadersProvider; replay-on-failure is delegated to the SDK’s own recovery (.recovery(true)), so pool-level rebuild is a no-op for the real transport. CI’s --all-features compiles the adapter, so the binding is kept honest against the real API.

Options considered + consequences

Option 1 (chosen) — minimal real substrate + IPC bridge + inverted transport

• Pros: the sink is genuinely built and its hard parts are tested without a live service; Arrow stays single-major for our code; the SDK is opt-in; the substrate crates gain their real, reusable seams.
• Cons: the substrate crates are only as complete as the sink needs (flow control / recovery roles still pending); the real adapter is compiled-but-not executed in CI; IPC encode adds a serialise step per batch (cheap, and it is the version-safe boundary regardless).

Option 2 — bump thalweg to Arrow 59 and use ingest_batch directly

• Cons: ADR-0003 pins Arrow to DataFusion’s major; moving ahead of DataFusion is a breaking change across every published codec crate for one sink’s convenience. Rejected — the IPC bridge is local and reversible; a global version bump is neither.

Option 3 — put the real adapter in a separate out-of-workspace crate

• Cons: more moving parts, and it would not compile under the workspace’s --all-features gate, so the binding could rot undetected. Rejected in favour of a feature that CI exercises.

Option 4 — implement the full Phase 5a/5b substrate now

• Cons: flow control, DLQ, recovery roles, and durable coverage backings are large subsystems in their own right; building them to finish one sink over-reaches and would ship half-designed. Rejected — build the seams the sink needs; leave the rest to its phase.

Invariants pinned by this ADR

• Single Arrow major in our code — the sink and all its logic are Arrow 58; Arrow 59 is confined to the SDK behind a feature, reached only via IPC bytes.
• The correctness core is service-free-testable — the transport is inverted; the ADR-0060 properties are proven against a fake in CI.
• delivered, never submitted — coverage advances only from the ack lane, only after the durable WAL ack (ADR-0038/0053), preserved verbatim.
• Union-merge coverage — never last-write-wins (ADR-0053).
• The SDK is opt-in — default builds and tests do not depend on it.

Deferred work

Each entry carries a trigger.

• D1 — Emit-time auto-project transform. Dropped, not deferred. An earlier plan had the sink optionally auto-to_json a nested column landing in a STRING target. It is not built and not planned: silently JSON-encoding structure at the sink hides a schema mismatch the operator should see. The two explicit resolutions are strictly better and need no sink magic — declare the column with its real nested type (the catalog creates it), or write to_json(col) in the transform (the UDF already ships in twg-udf). The sink therefore keeps only the loud-named refusal, which points the operator at exactly those two fixes.
• D2 — Observability wiring. Metrics are collected in a local atomics struct; twg-observability is still a scaffold. Trigger: the observability surface lands.
• D3 — Full flow-control / recovery-role substrate. The connector-core / offset-store crates carry only the seams the sink needs. Trigger: Phase 5a/5b.
• D4 — Executed integration against a live workspace. The real adapter is compiled, not run, in CI. Trigger: a test Databricks workspace is available to the pipeline.

Interaction with existing ADRs

• ADR-0003 (single Arrow major): honoured via the IPC bridge; Arrow 59 never enters thalweg’s own code.
• ADR-0060 (pipelined ack): realised — the ack lane is the sink’s core, tested.
• ADR-0038/0053 (source-ack / coverage): the delivered watermark and union-merge coverage implement them.
• ADR-0062 (catalog): the sink calls the real TableCatalog; provisioning is the catalog’s, never the sink’s.
• ADR-0019 (type-map): ProtoFieldType describes the target columns; no protobuf descriptor is on the Arrow write path.

Document version history

Decode & Codec Learnings — prior-art hazards to design out

Hard-won lessons from a prior production streaming engine in this same domain — the system thalweg is being built to replace and hot-swap under. Each hazard here caused a real silver-lane outage or a sustained steady-state cost on that engine. They are recorded so the decode / codec / UDF crates bake the fix in as an invariant before they move from scaffold to implementation — not after the identical outage recurs on a thalweg-powered deploy.

Status. Design guidance, not shipped code. thalweg is pre-implementation; the codec/decode/UDF crates named below are scaffolds. Where a hazard is already latent in a thalweg scaffold (the same code lineage was carried forward), the file is named so it can be fixed in place. Capabilities that are intent rather than build are marked (planned). The sink/ack side of these lessons is already carried by how-a-sink-should-work.md and ADR-0038; this document is the wire → Arrow decode boundary counterpart, and cross-references rather than duplicates the sink side (§8).

The examples use neutral names (orders, events, bronze.example) per the repository’s no-sensitive-information rule. Nothing here is domain-specific.

0. The meta-lesson: reconcile shape at the boundary, and make the boundary inspectable

Nearly every incident below has one shape:

A value’s type or shape at a boundary (decode → transform, transform → sink, helper → destination column) was decided implicitly, and only became visible when a batch was rejected in production.

The feedback loop on the predecessor was: deploy → observe rejection storm → roll back → hotfix → repeat — one recovery train burned four sequential same-day hotfix PRs, each relocating the previous failure. The cheap fix in every case was the same: surface the boundary decision before it moves data. thalweg’s existing “loud, never silent” posture (how-a-sink-should-work §11) must extend back to the decode boundary.

Design rule: if a fix would have been prevented by an inspectable signal, the missing signal is the real defect — build the signal, not just the patch.

Two concrete surfaces every codec should expose (planned):

• twg … describe-schema <descriptor> — a twg subcommand (ADR-0045) that prints the exact Arrow schema each source field will decode to, so a transform author writes col['field'] vs a JSON-extract with certainty before deploy. twg-type-map (ADR-0019, the describe-only leaf) already owns describe-vs-execute; this is its read side extended to the decode schema.
• Startup INFO log of the resolved decode schema, once per descriptor at register time, so the shape is in the logs before the first record moves.

1. Decode shape must be deterministic and inspectable (Struct vs Utf8)

The hazard. A descriptor-driven decoder maps a sub-message to either a native Arrow Struct or a stringified Utf8-JSON, and which one a given field gets is opaque at authoring time (it depends on nesting depth, wrapper rules, map handling, and fallback caps). Both tails failed on the predecessor, on the same descriptor, in a single deploy:

• Tail A — author expected Struct, got Utf8. Native access (party['note']) fails at query-plan time: “type Utf8 is not Struct, Map, or Null.” The node errors, downstream starves, everything routes to DLQ, the DLQ writer is overwhelmed — full lane outage.
• Tail B — author expected Utf8, got Struct. Access succeeds, the batch carries a Struct, and the STRING-typed destination column rejects it at write: “expected LargeUtf8 but got Struct with N fields.” Worse than A — it fails at emit, after config validation passed and every row was enriched, so the blast radius is every record for the life of the deploy.

The real defect is not either mapping — it is that the mapping was invisible until a batch was rejected.

thalweg placement. twg-proto-decode (Arrow output stage, ADR-0046), twg-proto-schema (descriptor → schema), twg-type-map (target-type mapping). The decode-shape rules must be documented, stable per descriptor, and dumpable via §0’s describe-schema surface. twg-proto-decode/src/decode.rs today mirrors the predecessor’s typing pass “matching real behavior exactly” — including the Utf8 fallbacks — so it inherits this hazard verbatim until §0 lands.

2. Native container types, never a stringified JSON scalar

The hazard. Proto map<K,V> fields decoded to a single Utf8 JSON-object string ('{"a":{…},"b":{…}}'). Downstream could not index by key, lost the value type entirely (a map<string,double> became opaque STRING), and was forced back to re-parsing the raw payload N times per row (§7). The same applied to any container the decoder chose to stringify.

The fix. Emit native Arrow containers:

• Proto map<K,V> → Arrow Map<Utf8, V> (value type preserved — that’s the win; the key is Utf8 because the JSON intermediate stringifies keys). Downstream then uses element_at(m, key) / map_keys(m) / map_values(m), and it matches the target’s native MAP<K,V> type in twg-type-map.
• Repeated messages → List<Struct<…>>; singular sub-messages → Struct<…>.

A stringified container is acceptable only as an explicit, logged fallback (depth cap, §3), never as the default shape for a type the engine can model natively.

thalweg placement. twg-proto-decode (build the MapArray/ListArray), twg-type-map (proto/Arrow Map ↔ target MAP<K,V>; a map_fields.proto fixture already exists under twg-codec-protobuf). Latent hazard: twg-proto-decode/src/decode.rs stringifies map entries to Utf8 today.

Hot-swap note. Making a map column land as native Map instead of a JSON string is a behavior change for any config that writes that column straight into a STRING destination — it would then hit Tail B (§1). The escape hatch is §4’s to_json. Sequence the swap so both are available together.

3. The nesting-depth cap is a configured, observable knob — not a hidden constant

The hazard. Struct recursion was bounded by a hard-coded constant (= 12 on the predecessor). A schema deeper than the cap silently collapsed its whole sub-tree to Utf8, with no telemetry beyond a generic null-patch counter firing at write time — far from the cause. Operators could not raise it without a source patch and a redeploy.

The fix.

1. Expose the cap as config — twg-config (ADR-0028 TOML), overridable per deployment; default generously (well above realistic schema depth) so the fallback is genuinely exceptional.
2. Name the loss at the point it happens — a startup WARN listing every message that will trip the fallback under the current cap, and a per-type (deduped) WARN when it fires, so it is visible before the pipeline moves data. This is the decode-side face of how-a-sink-should-work §11 (“loud, not silent”).

Rule that generalises: any hard-coded limit that changes fidelity — a depth cap, a byte budget, a truncation threshold — is a config knob and emits a signal when it bites. A magic constant that silently drops type information is an incident waiting for the one schema that exceeds it.

thalweg placement. twg-proto-decode (the cap), twg-config (the knob), twg-observability (the WARN + a counter).

4. Symmetric serialise escape hatch: to_json alongside the JSON reader

The hazard. The predecessor shipped a reader for the stringified case (get_json_object(payload, '$.path')) but no writer — no way to serialise a native Struct/List/Map back to a string at emit time when the destination column is STRING (Tail B, §1). The only author-side unblock was CAST(NULL AS VARCHAR) — data loss disguised as a hotfix.

The fix. Register a to_json(<any>) -> STRING UDF as a first-class family member of get_json_object. If a boundary can be crossed one way, an author will need to cross it the other way.

Implementation notes that carry over:

• Signature::any(1, Immutable) — accept any Arrow input without coercion (coercion cannot handle Struct/List/Map).
• Delegate to the ecosystem JSON writer (arrow-json make_encoder) rather than a hand-walked value tree — full type coverage for free, no gaps on exotic types.
• SQL NULL round-trips as SQL NULL, not the string "null".
• Dependency pin: the arrow-json version must track DataFusion’s re-exported Arrow major (ADR-0003), or the FieldRef/dyn Array instances mismatch and it won’t compile.

End-state (planned, defer with a trigger): sink-side auto-projection — the sink already knows the destination’s declared type (twg-type-map), so nested-into-STRING can be reconciled transparently at emit. It touches the write hot path, retry-slice, and byte-budget accounting, so ship the UDF now and gate the auto-projection on a real trigger (e.g. “> N pipelines use to_json in a quarter” or “the same mismatch recurs on a second sink type”). Record the deferral with its trigger — a deferred item without a trigger is a wish, not a plan.

thalweg placement. twg-udf. Latent hazard: twg-udf registers the JSON reader but has no to_json — the escape hatch is absent today.

5. Type-producing helpers must not hide their output type

The hazard. A convenience UDF converting epoch-millis → timestamp was hard-coded to return a timezone-aware type (Timestamp(µs, Some("UTC"))). The tag is invisible at the SQL call site and propagates to everything derived from it (date_trunc('day', …)). Destination columns were timezone-naive (TIMESTAMP_NTZ, Timestamp(µs, None)), and the write layer rejects tz-aware-into-tz-naive:

Field 'event_ts_day' is incompatible - expected
  Timestamp(Microsecond, None) but got Timestamp(Microsecond, Some("UTC"))

A sibling pipeline that used the engine’s built-in millis→timestamp (tz=None) wrote fine — the two helpers looked interchangeable at the call site; the tag was the only difference, and it was hidden.

The fix.

• Prefer the query engine’s built-in conversion (tz-naive) for tz-naive targets.
• If a Spark-compat helper is provided, either return tz=None by default, or offer both variants with names that state the tag (…_ntz vs the UTC form), and doc-guard the tz-aware one: “timezone-aware — do NOT use for TIMESTAMP_NTZ destinations.”
• General rule: a type-producing helper makes its output type — tz tag, decimal precision/scale, nullability — obvious at the call site. Two helpers that differ only in a hidden type attribute guarantee that one of them causes an outage.

thalweg placement. twg-udf (the helpers), twg-type-map (tz-aware vs tz-naive target mapping). Latent hazard: twg-udf/src/local/spark_compat.rs hard-codes with_timezone("UTC") and a Some("UTC") return type for both millis_to_ts and to_timestamp_ms, with no tz-naive variant — the exact footgun, already carried forward.

6. Preserve wire binary as Arrow Binary end-to-end

The hazard. On a JSON-intermediate decode path, Avro bytes fields (and bytes + logicalType: decimal) were base64-encoded into Utf8 strings ("AA==") and stayed strings through expand and flatten. The Arrow Binary type was lost, so a downstream decode_avro_decimal UDF — which requires Binary — failed, and any future raw-bytes (non-decimal) use case was blocked. JSON has no binary type, so a JSON hop forces base64; the fault is routing binary through JSON at all.

The fix.

• Keep an Arrow-native decode path for binary. Decode Avro bytes/fixed directly to Arrow Binary/LargeBinary (and decimal logicalType to Decimal128), and preserve that type through expand and flatten rather than round-tripping through a JSON Value. thalweg’s three-co-equal-representations model (ADR-0001) makes this the natural path — Arrow is not a second-class hop.
• Where a JSON view is genuinely needed, base64 is the correct JSON representation for that view only — it must not become the column’s Arrow type.
• Make the consumer tolerant too: decode_avro_decimal should accept a base64-Utf8 input in addition to Binary/FixedSizeBinary, as a defensive fallback for any residual JSON-path value.
• Alternatively/additionally, thread the Avro logicalType into decode so decimal lands as Decimal128 in the bronze layer and no post-hoc UDF is needed.

thalweg placement. twg-codec-avro (Arrow-native bytes/decimal decode; ADR-0027 shares this for the OCF batch source), twg-proto-flatten / expand-columns (preserve Binary, don’t base64 nested bytes), twg-udf (decode_avro_decimal input tolerance). Latent hazard: twg-udf/src/local/avro_decimal.rs rejects anything but Binary/FixedSizeBinary today, and any JSON-intermediate expand will hand it a base64 string.

7. Native typed columns beat re-parsing a blob N times per row

When decode leaves values stringified (§1, §2, §6), transforms fall back to get_json_object(raw_payload, '$.…') — once per referenced path, per row. The query engine does not memoise across different paths, so a payload referenced 20 times per row is parsed 20 times per row (tens of thousands of redundant parses per second per pod on the predecessor), and it forces the raw blob to be retained through the transform graph purely as a re-parse source.

This is the steady-state tax of getting §1/§2/§6 wrong. Every field decoded as a native typed column is one the transform reads for free and the sink writes without a round-trip through string. Re-parse count per row is a good proxy metric for how much decode fidelity is being left on the floor — worth a twg-observability counter.

8. Streaming-sink concurrency & pipelined ack (cross-reference)

The predecessor’s largest single throughput loss was on the write side, and thalweg’s sink design already addresses it — this section only pins the failure signatures so they are recognisable, and points at the existing invariants.

Observed on the predecessor:

• Per-batch ack serialisation. Ingest awaited “wait until the server durably acked offset N” inline after every submit, serialising the sink to one in-flight batch — a ~20,000× loss against a pipelined SDK (~4 batches/sec against a 100 MB/s pipe). It looked like everything else: ack-timeout false-positives, backpressure trips at trivial throughput, batchers that never fill a batch.
• Stream churn. Opening a new sink stream per batch/worker tripped a “too many concurrent streams” limit (thousands of rejects/day); recovery felt slow because most flushes were single-record commits.

thalweg’s design already prevents these:

• Async, non-blocking ack releasing admission budget on arrival, off the clean path — how-a-sink-should-work §10, ADR-0038 (advance offset only on durable ack), ADR-0060 (pipelined fire-and-forget ack on a background lane).
• Advance on the delivered offset, never the submitted offset — the durable watermark is what recovery reads (ADR-0038, ADR-0053 “coverage may lag, never lead”).
• Few, long-lived streams per destination and honour server backoff on a concurrency reject (send on an existing stream; back off before reopening) — a self-protective path must not amplify the condition it detects.
• Size streams to bandwidth, not record count — large records hit a per-stream MB/s ceiling at low record rates; spread across a small fixed pool.

No new work here beyond keeping these signatures in the sink soak tests (docs/testing/SCENARIOS.md).

9. Decode/codec invariants (summary checklist)

A codec/decode/UDF crate built to this blueprint should satisfy:

• The Arrow type of every decoded field is inspectable before deploy (describe-schema surface) and logged at register time. (§0, §1)
• Modellable containers decode to native Arrow (Map/List/Struct), never a stringified JSON scalar; stringification is an explicit, logged fallback only. (§2)
• The nesting-depth cap is a config knob, defaulted high, and its fallback names the offending message loudly, at decode time not write time. (§3)
• Every JSON extractor has a matching serialiser (to_json(<any>)); no author is ever forced to CAST(NULL …) to unblock a STRING target. (§4)
• Type-producing helpers make their output type obvious at the call site (tz tag, precision, nullability); tz-naive is available and named. (§5)
• Wire binary stays Arrow Binary through decode/expand/flatten; base64 is a JSON-view detail, never the column’s type; binary consumers tolerate the base64 fallback. (§6)
• Native typed columns are preferred over repeated get_json_object re-parse; re-parse pressure is observable. (§7)
• Sink write path keeps the async non-blocking ack, few long-lived streams, delivered-offset watermark, and server-backoff honouring. (§8, ADR-0038)

Source grounding

Derived from production incidents on the predecessor engine thalweg replaces, abstracted to remove all environment- and domain-specific detail per the repository’s no-sensitive-information rule. Mapped onto thalweg’s current design artefacts — chiefly how-a-sink-should-work.md, AGENTS.md, and the doc-comment scaffolds of twg-proto-decode, twg-proto-flatten, twg-proto-schema, twg-codec-avro, twg-udf, and the implemented twg-type-map. Relevant ADRs: 0001 (three representations), 0003 (Arrow version pin), 0004 (unnest/offset cardinality), 0017 (unified DLQ), 0019 (type-map describe-only), 0027 (shared Avro decode), 0038 (source-ack invariant), 0045 (single twg binary/subcommands), 0046 (protobuf decode strategy), 0053 (coverage may lag, never lead). As the codec crates move from scaffold to implementation, retire the (planned) markers and fold each fixed hazard into the relevant crate README or a dedicated ADR.

How a Sink Should Work

A generic design blueprint for sinks — the destination/write side of a sync — grounded in thalweg’s current plans.

Status. This is a design document, not a description of shipped code. thalweg is pre-implementation: every sink crate is a scaffold (a README.md plus a doc-comment lib.rs — “the documented contract exists, the implementation does not”), and the only crate with real, tested code is twg-type-map. Landing order in the delivery plan: the BatchSink substrate and the first four sinks in Phase 5a, the coverage store + recovery in Phase 5b, Postgres/Zerobus + catalog + Iceberg in Phase 7, and the transform DAG in Phase 9. Where this document leans on a thalweg decision it names the crate/ADR; capabilities that are intent rather than build are marked (planned). The aim is a generic, reusable model of what a correct sink must do, using thalweg’s specification as the worked example.

1. What a sink is

A sink is the component that takes the pipeline’s record stream and writes it to a destination — a table, an object store, a message transport — durably, and reports back precisely what it committed so the source can safely advance. A sink is the last hop on the data path:

… ─▶ transform (DAG) ─▶ SINK(s)
                          │
                          └─ durable ack ─▶ source position advances
                                            per-sink coverage recorded

A sink treats every destination uniformly: a dead-letter / quarantine target is not a special downstream of the sink layer, it is just another sink with its own table and coverage. The sink itself is agnostic of what role a target plays.

A sink is responsible for four things, and this document expands each:

1. Accepting and writing a batch of Arrow RecordBatches with a well-defined lifecycle (§2). The write side is Arrow-only — upstream decode/transform normalise every representation into Arrow before a sink sees it.
2. Confirming durability and returning a precise record of what landed, so delivery is correct and replay-safe (§2, §5, §9).
3. Provisioning and evolving the destination table/schema, including comments/descriptions (§8).
4. Reporting progress, coverage, and errors back to the engine and operators without ever blocking the clean path (§10, §11).

2. The sink contract

thalweg models the write side at two layers, and a concrete sink usually implements against both.

2.1 BatchSink — the Arrow write substrate (planned; twg-stream-arrow, ADR-0001)

A sink consumes Arrow RecordBatches and nothing else — there is a single sink input type, not a family of per-representation sinks. Upstream decode/transform normalise raw bytes and the neutral value model into Arrow before the write side, so the write substrate is one abstraction over RecordBatch rather than a RawSink/ValueSink/ArrowSink/TeeSink split.

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

• Lifecycle: push-many → finish-once, returning (CommitInfo, OffsetSpan).
• The critical property: OffsetSpan returns identically from every sink, so commit-after-write, replay positioning, and the soak harness are written once and work across every sink.
• _twg_emit_ts is stamped on write, when a BatchSink commits the batch.

2.2 Sink — the transport-neutral delivery trait (planned; twg-connector-core, ADR-0012)

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

• Visible lifecycle: send → flush. send returns Vec<Receipt> — the per-batch acknowledgements.
• Note there is deliberately no open/commit/close/ack method in the trait. Commit and acknowledgement are modelled as data — the returned Receipt / OffsetSpan plus the per-sink coverage record in the offset store — not as extra trait methods. This keeps the trait minimal and pushes durability semantics into the values a sink returns.

2.3 Durable acknowledgement — defined per sink type

The governing invariant of the whole system is that the source position is committed only after a durable sink acknowledgement. “Durable” is defined per sink type:

2.4 Batch, transaction & materialisation model

• Batch-granular by default: the source advances to the batch high-water offset once the whole batch is durably acked. Per-record acknowledgement is the low-latency opt-in.
• Partial-failure recovery is per-sink and optional. Two strategies: the engine replays the whole batch and dedup on _twg_record_id absorbs the records that already landed; or, where a sink auto-retries internally (transaction or transport-level retry), the sink resolves the partial failure itself and engine-level replay is not needed. Whole-batch replay + dedup is a fallback for sinks that need it, not a mandatory step for every sink.
• Transaction model: 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, the sink does write-then-record with at-least-once semantics and lets dedup absorb the replay window.
• Materialisation is a property of the transform-DAG node that owns the sink, not a sink method — see §5.

2.5 The ReplayableSink capability (planned)

A sink that can hold the primary-raw role must satisfy a stronger contract: durable = true is mandatory (enforced at config validation) and the data it wrote must be recoverable — but not necessarily readable back through the sink or the sync itself. An external read path counts: Zerobus, for example, is write-only, yet its data is retrievable via Delta / Open Sharing or a Unity Catalog SQL-warehouse query. Together these are the ReplayableSink capability. What matters is that some durable, readable copy exists for recovery — not that the sink exposes a read API. A sink that cannot confirm durability before returning success, or whose writes cannot be read back by any path, is not eligible to anchor replay.

3. Inputs a sink needs

To be defined and run, a sink needs:

• A destination identity — table/catalog identifier, object-store URI, Flight endpoint, or database DSN.
• Credentials, vended not embedded. Sink/table credentials are obtained from the table catalog via credential vending (temporary, scoped credentials that inherit the requesting principal’s privileges) rather than held statically by the sink or its format crate. Transport auth (TLS/mTLS, SASL) is a separate, lower layer.
• The Arrow schema of the incoming RecordBatches, which the sink maps to the target table — an Arrow RecordBatch → sink-table conversion (DDL + type mapping, see §8). Schema resolution (contract / registry / inference) happens upstream on the read/decode side; by the time data reaches a sink it is already Arrow, so the sink’s only schema concern is the batch-to-table mapping.
• A role and durability setting — primary-raw vs secondary/derived, and whether durable is on (mandatory for primary-raw; an explicit, documented opt-out only for non-raw secondaries).
• Config-validation preconditions, checked at startup and re-checked rather than cached where a target property can change under a running pipeline (§6.2).
• A coverage/offset regime — where this sink’s write coverage is persisted (§9).

4. What a sink must support

1. Writing Arrow RecordBatches via its BatchSink write lifecycle (push-many → finish-once).
2. Durable acknowledgement appropriate to its type, and returning a precise OffsetSpan / Receipt.
3. Transactional coverage where the target supports it (exactly-once), or safe write-then-record (at-least-once) otherwise.
4. Idempotent replay where it relies on engine-level replay — tolerating whole-batch replays without duplicating landed data, via _twg_record_id (sinks that auto-retry internally may not need this; see §2.4).
5. Destination provisioning — create + evolve the table/schema and stamp comments (§8).
6. Backpressure participation — releasing admission-control budget on ack, never blocking the clean path (§6, §10).
7. Coverage reporting and error/status surfacing (§9, §10, §11).

5. Sink roles & the transform DAG

Sinks attach to a DAG, and their role determines whether they gate the source.

5.1 Primary-raw vs secondary/derived

• 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 asynchronously; 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.
• Durability is role-dependent. Primary-raw MUST be durable. Non-raw secondary sinks MAY disable durable to trade safety for throughput — because raw can rebuild them — but that opt-out is explicit and documented, never a silent default.
• Any replayable sink can hold primary-raw — e.g. Parquet-on-object-store, Postgres, or Zerobus where Iceberg reads are enabled.

5.2 Materialisation

A DAG node is { name, query, optional sink, materialisation }. Materialisation becomes automatic when a node has more than one dependent (DataFusion re-executes a plan per collect(), so an unmaterialised shared intermediate is re-derived per dependent). Sinks attach at interior nodes, not only leaves. Cardinality caveat (ADR-0004): an unnest node breaks the OffsetSpan range assumption, so that node’s span is the union of contributing locators.

5.3 Recovery-from-raw is a precondition, not later hardening

A multi-sink topology without recovery is silently lossy by construction, so configuration refuses one. Recovery is required before a second sink is allowed. Concurrency notes worth stating: concurrent live + recovery writes to a single sink are safe for dedup-on-id sinks; for append-only targets they produce out-of-order arrival with correct identity — acceptable, but it must be stated.

6. Concrete sink types

All five are scaffolds today. Durability follows §2.3; primary-raw eligibility follows §2.5 and §5.

6.1 twg-sink-postgres

Writes via binary COPY; durability is transaction commit; idempotency is the engine-wide _twg_record_id dedup. Because names are spliced into CREATE TABLE and binary COPY, it depends on the identifier-safety helpers in twg-type-map (§8).

6.2 twg-sink-zerobus

Writes Arrow RecordBatch over Flight (no protobuf descriptor is involved in the write path) into Delta; durability is server-side ack. Its primary-raw role is conditional and enforced as config-validation preconditions: the table must be registered in the catalog, Iceberg reads/column mapping must be enabled, reader/writer protocol versions must meet a minimum, and deletion vectors must NOT be enabled (switching them on later silently removes recovery’s read path). The engine must refuse the primary-raw role if any precondition is unmet, and re-check rather than cache — a table property can change under a running pipeline. Two knock-ons: a read-after-write hazard (Iceberg metadata generation is async, so raw-anchored recovery may see an older table state and silently replay an incomplete window — recovery must tolerate the lag or trigger generation synchronously), and a codec knock-on (Iceberg-enabled tables use Zstandard, not Snappy, so the recovery read path needs zstd decode — decode-only, purity-safe).

6.3 twg-sink-object-store

One sink, format backend selected by config, so routing, durable-ack, offset colocation and primary-raw eligibility are implemented once rather than per format. Durability per backend: parquet = plain files to object store, no table layer, durable on flush+close; iceberg = Iceberg table, durable on commit to the catalog. A format backend qualifies as ReplayableSink when it is both readable back for recovery and able to confirm durable ack before returning success — plain Parquet qualifies, making a Parquet raw lane a valid, cheap primary-raw target with no Delta/catalog dependency. Only one open table format (Iceberg) is in scope (ADR-0023); the backend seam keeps a future Delta backend additive.

6.4 twg-sink-flight and twg-sink-arrow-ipc

Header-only scaffolds. Durability from the core invariant: Flight = server-side ack; IPC file = flush+close. Both land in Phase 5a.

7. Scalability (sink side)

A sink participates in the engine’s flow control rather than owning its own:

• Ack-driven admission control. The engine’s byte-budget semaphore (sized in bytes, from the p95 post-decode expansion ratio) releases permits when a sink’s ack arrives. A slow sink therefore throttles intake naturally, without a separate mechanism, and without blocking the clean path (the permit is acquired and released in the same select! as ack processing) — see §10.
• Batch sizing follows backpressure. The backpressure state machine scales effective batch size as memory/channel/write-failure pressure rises; a sink should honour the batch size handed to it rather than buffering unboundedly.
• Secondary sinks scale independently. Because they ack async and recover from raw, a lagging derived sink never stalls the source; it can be scaled or rebuilt on its own cadence.
• Coverage partitions horizontally. Per-sink coverage is keyed by topic+partition, so multiple pods each own a disjoint slice of a sink’s write responsibility (§9).
• Streaming transports keep a small, fixed pool of long-lived streams per destination. Never open a stream per batch or per worker: on the predecessor that tripped a broker “too many concurrent streams” limit (thousands of rejects/day) and starved throughput by churning reconnections instead of sending. On a concurrency reject, honour the server’s suggested backoff and send on an existing stream rather than reopening immediately — a self-protective path must not amplify the condition it detects. Size the pool to bandwidth, not record count: a stream planned at ~X MB/s cannot carry records/sec × record_size above X, so large records saturate one stream at a low record rate — spread them across a few streams.

8. Creating & updating destination tables, schemas & comments

A sink must be able to provision and evolve its target, not just write into a pre-made table. This is a split responsibility between the table catalog and the type-mapping vocabulary.

8.1 Type mapping & identifier safety — the built foundation (implemented: twg-type-map)

Mapping to each target dialect is a hub-and-spoke through Arrow, and the goal is to convert straight from the Arrow RecordBatch to the target types wherever possible: each column’s Arrow type is mapped directly to its target type — SQL DDL text for Postgres, and Delta / Iceberg / Zerobus-proto types for the table formats. A decoded-JSON representation remains available as an intermediary only for the paths that genuinely need it (e.g. JSONB fallback); it is never a mandatory hop on the Arrow → target conversion. Per-dialect names are emitted correctly (e.g. Iceberg long, decimal(10,2), fixed[16]; Delta nested-JSON; proto BIGINT vs UC-API LONG). Identifier safety (strict vs hyphen-relaxed validation, quoting/escaping) is explicit so names are safe to splice into CREATE TABLE and binary COPY. This is the one piece that already exists and is tested.

8.2 Table registration & the describe-vs-execute split (planned: twg-table-catalog)

The TableCatalog trait owns table schema resolution, sink registration, credential vending, and descriptive metadata. Descriptive metadata is stamped at table CREATE and at schema EVOLUTION — a newly added column gets its comment too, because evolution is a write-path event just like creation — applied via a describe-vs-execute split (compute the desired state, then apply it).

8.3 Schema evolution

Policy: additive is safe, widening is conditionally safe behind an opt-in, anything else fails loudly. The widening vocabulary is already implemented (twg-type-map::can_widen_to; e.g. lossless int32 → int64 → double → string, int → f64 refused as lossy). The missing piece a full implementation still needs is the component that diffs a live table against the desired schema and issues the ALTERs.

8.4 Comments / descriptions

• Two comment authorities. Engine-owned columns (_twg_*, dq_results, offset-store columns) get fixed built-in descriptions; payload column descriptions flow from the ODCS contract — no contract means no payload comment.
• Table-level comment plus properties/tags (owner, contract URI + version, source topic) turn the catalog into a discovery surface.
• Backend degrade-gracefully matrix: Unity Catalog → COMMENT ON + tags + properties; Iceberg → table properties + column docs in the schema; Glue → column Comment fields + table Parameters.
• Idempotent re-sync. The sink marks its comments twg:managed=true and stores a twg:comment_hash; a re-sync writes only when the contract’s description changed versus the last synced hash — no churn on no-op deploys. Human edits to managed columns are detected via the managed flag; respect-vs- restore is a policy, not a blind stomp.

8.5 Credential vending

The catalog exposes a read-only Iceberg REST endpoint with credential vending — temporary scoped credentials inheriting the requesting principal’s privileges. Recovery reads resolve the table, obtain vended credentials, and read Parquet through the Iceberg reader (no Delta implementation required). Two metadata properties the sink should surface: whether Iceberg reads are enabled, and how far generated metadata trails (§6.2).

8.6 Emit-time type reconciliation — the write boundary rejects shape mismatches

Schema resolution is upstream (§3), but two type mismatches only bite at the write boundary, where the incoming Arrow RecordBatch meets the target column type — and on the predecessor each one took out a whole silver lane for the life of a deploy, because it fails at emit, after config validation passed:

• A nested Arrow type into a STRING column. A decoded Struct/List/Map landing in a column the target declares STRING (Arrow LargeUtf8) is rejected at write (“expected LargeUtf8 but got Struct with N fields”), for every record. The sink must reconcile, not reject: either the author serialises upstream with to_json(<any>) (decode-codec-learnings.md §4), or — the clean end-state — the sink auto-projects at emit, because it already knows the declared target type from twg-type-map. Auto-projection is a hot-path change (retry-slice, byte-budget — JSON runs 3–5× larger than the packed nested value), so it is gated behind a trigger; until then to_json is the first-class escape hatch, and a STRUCT-into-STRING mismatch is a loud failure (§11), never a silent NULL.
• A timezone-aware value into a timezone-naive column. A Timestamp(µs, Some("UTC")) into a TIMESTAMP_NTZ (Timestamp(µs, None)) column is rejected the same way. The fix is upstream — tz-naive helpers (decode-codec-learnings.md §5) — but the sink must surface the mismatch loudly, naming the offending column, not fail opaquely per batch.

The principle: the sink knows the target type, so any mismatch it could reconcile or name precisely must never surface as an opaque per-batch rejection.

9. Traceability (sink side)

• _twg_emit_ts is a sink event — stamped when a BatchSink commits — and closes the provenance chain _twg_source_timestamp → _twg_ingest_ts → _twg_emit_ts, giving end-to-end latency per record.
• Per-sink write coverage is the authoritative record of what each destination holds, kept in twg-offset-store as gap-aware ranges keyed (sink, topic, partition, start) → end, prefix-scanned, and merged by union, never last-write-wins (so concurrent writers during a rebalance cannot erase each other’s progress). Note this is a different quantity from the source position: once the source is acked, coverage is the only record of what a sink is still missing. The store is two-tiered — a durable/shared tier colocated with raw (never with the sink it describes) plus a pod-local write-behind cache checkpointed on an interval.
• Governing invariant: coverage may LAG, never LEAD. A lagging sink coverage costs a dedup-absorbed replay (safe); a leading one is silent permanent loss.
• The DQ audit is itself a first-class sink (receives = "dq_results"), append-only, with the same durability/recovery guarantees as data — so which records a sink accepted/quarantined/dropped, and under which versioned rule, is queryable per record.

10. Non-blocking ack back channel

Acknowledgements drive flow control and the source-position commit, but a caller is never blocked waiting on an individual ack:

• Async delivery ack. Sink::send returns Vec<Receipt> as a future; the byte-budget permit for that batch is released when the ack arrives, processed in the same select! loop as record flow. The sink keeps moving while acks stream back asynchronously.
• Secondary sinks never stall the source. They ack independently and async and are recovered from raw if they lag (§5), so a slow destination cannot gate intake.
• Ingest is fire-and-forget; the durable wait is a background lane. For a pipelined transport (Flight/Zerobus SDK), send returns as soon as the SDK accepts the batch; the “wait for durable server ack” runs on a per-stream background ack lane that drains a bounded channel, coalesces a burst, and confirms the whole burst with one round-trip on the largest offset (offset monotonicity ⇒ one wait confirms all lesser items). The at-least-once contract (ADR-0038) is preserved by where the confirmation fires, not whether — the source-position commit and per-sink coverage update still happen only on the durable ack, now from the ack lane. Backpressure is the channel filling: when full, send awaits, which the SDK’s own bounded buffer already models.
• Advance on the delivered offset, never the submitted offset. Submitted is an observable ingest high-water only; recovery (§9, ADR-0053) reads delivered. Any durability path that reads submitted is a bug.
• The dead-letter / quarantine target is itself just another sink, wired on a non-blocking lane — its own buffer and backpressure, genuinely independent of the clean write path. When it cannot keep up, quarantine sheds — never clean (overflow → drop-with-count loud metric, safe because raw is the system of record). Quarantine never gates: it does not sit between source and ack.

Anti-pattern to design out (prior-art). Making the async ack synchronous — awaiting the durable-offset wait inline after every submit — serialises the sink to one in-flight batch. On the predecessor engine this cost a ~20,000× throughput loss (≈4 batches/sec against a 100 MB/s pipe) and manifested as everything else: server-ack-timeout false-positives (each network hiccup fully exposed to the caller deadline, “fixed” by inflating the timeout — a bandaid that trades false positives for slow real-failure detection), backpressure trips at trivial throughput, and batchers that never fill a batch because they spend their time waiting between batches. The pipelined ack lane above absorbs transient hiccups in the SDK’s buffer instead of exposing them. This refinement is recorded as ADR-0060 (pipelined fire-and-forget sink ack); see also decode-codec-learnings.md §8.

11. Monitoring & surfacing errors/messages from a sink

The status/stats channel is read-side and never on the commit path, so observing a sink can never slow it down.

• Health & stats. /health exposes liveness/readiness including backpressure state; /stats exposes throughput, per-sink ack latency, coverage gaps, and quarantine/drop counts. (planned: twg-observability.)
• OpenTelemetry (OTLP) metrics/logs/traces, with per-batch spans through the write → ack path and latency derived from the _twg_ timestamp chain.
• Errors surface on three planes, by severity:
  1. Per-record — the unified DLQ tags every diverted record with a reason code (decode / contract / DQ / retry-exhausted), and the DQ audit records the disposition; deliberately-dropped junk never appears in the DLQ, keeping it high-signal.
  2. Aggregate — /stats counters and OTLP metrics (drop counts, quarantine overflow, retry exhaustion, coverage-gap alarms, per-sink ack latency).
  3. Liveness — /health plus backpressure state.
• Loud, not silent, on the dangerous cases. A sink must emit explicit, counted signals for: overflow drops, coverage gaps, unsafe schema-evolution attempts, unmet primary-raw preconditions, read-after-write recovery hazards, write-boundary type mismatches (nested-into-STRING, tz-aware-into-NTZ; §8.6), and type-lossy decode fallbacks (over-depth collapse; decode-codec-learnings §3) — never degrade quietly. Silent data loss is the one outcome the whole design exists to prevent.
• Discriminate self-inflicted from environmental on the ack lane. Two causes drive the same downstream symptom (delayed/absent durable acks, quarantine-drain timeouts): (a) the transport is transiently slow to ack — environmental, transient, recovered by a source/stream reconnect; and (b) our own ack lane died on an unrecoverable transport error — self-inflicted, permanent until the stream is recreated. Alert on the cause signal (a distinct ack-lane poisoned counter), never on the shared symptom, or minutes of sink poisoning get misread as broker slowness. Keep the two counters semantically separable at source; the runbook branching is an operations concern.

12. Sink design invariants (summary checklist)

A sink built to this blueprint should satisfy, at all times:

• Returns a precise OffsetSpan / Receipt; source position advances only after this sink’s durable ack (definition per §2.3).
• Coverage update is in the same transaction as the write for transactional sinks (exactly-once), else safe write-then-record (at-least-once) with _twg_record_id dedup absorbing replays.
• Primary-raw role requires the full ReplayableSink capability (durable and recoverable via some read path — the sink itself, Delta/Open Sharing, or a SQL warehouse), enforced at config validation and re-checked, not cached.
• Non-raw secondary sinks may disable durable only as an explicit, documented opt-out; they ack async and recover from raw, never gating the source.
• Per-sink coverage is gap-aware and union-merged; coverage may lag, never lead.
• Creates and evolves the destination table/schema — additive by default, widening behind opt-in, else fail loud — with identifier-safe DDL from twg-type-map.
• Comments are stamped at CREATE and EVOLUTION, from the correct authority (engine _twg_* vs ODCS payload), and re-synced idempotently (twg:managed / twg:comment_hash).
• Uses vended, scoped credentials from the catalog, not embedded secrets.
• Ack is asynchronous and releases admission budget; the DLQ/quarantine lane never backpressures the clean write path — quarantine sheds, never clean.
• The async ack is never made synchronous — ingest is fire-and-forget, the durable wait runs on a per-stream background lane, and the source advances on the delivered offset, never the submitted one. (§10)
• Streaming transports use a few long-lived streams per destination, honour server backoff on a concurrency reject, and are sized to bandwidth, not record count. (§7)
• Write-boundary type mismatches (nested-into-STRING, tz-aware-into-NTZ) are reconciled (to_json / auto-project) or surfaced loudly with the column named — never an opaque per-batch rejection or a silent NULL. (§8.6)
• Errors surface on three planes (per-record DLQ + audit, aggregate stats/OTLP, liveness health), and dangerous cases are loud; ack-lane poison is discriminated from environmental slow-ack at source. (§11)

Source grounding

Derived from thalweg’s current design artefacts — chiefly docs/blueprints/thalweg-delivery-plan.md, AGENTS.md, README.md, and the doc-comment scaffolds of twg-connector-core, twg-stream-arrow, twg-offset-store, twg-table-catalog, and the sink crates (twg-sink-postgres, twg-sink-zerobus, twg-sink-object-store, twg-sink-flight, twg-sink-arrow-ipc), plus the implemented twg-type-map. Relevant ADRs include 0001 (BatchSink/OffsetSpan), 0004 (unnest/span), 0007 (DAG), 0012 (Sink), 0022 (schema evolution), 0023 (single table format), 0024/0037 (catalog), 0038 (source-ack invariant), and 0057 (recovery-from-raw). As those crates move from scaffold to implementation, retire the (planned) markers and reconcile this document against the shipped behaviour.

The write-side hazards folded into §7 (stream pool / churn / bandwidth sizing), §8.6 (emit-time type reconciliation), §10 (pipelined fire-and-forget ack + the per-batch-wait anti-pattern), and §11 (loud type-mismatch cases + ack-lane-poison discrimination) are drawn from production incidents on the predecessor engine thalweg replaces, abstracted per the repository’s no-sensitive-information rule. Their decode-side counterparts live in decode-codec-learnings.md; the pipelined-ack refinement (§10) is recorded as ADR-0060.

Thalweg Plan-vs-Design Reconciliation

Inputs: docs/blueprints/thalweg-delivery-plan.md (v0.5, last-modified 2026-07-24) · docs/adr/README.md (ADRs 0001–0057) · drafted ADR files 0001 (Draft, v0.5) and 0003 (Accepted, v1.0) · crates/ tree (47 crates, all twg- prefixed).

Headline: The plan is one design generation behind. It was written when the whole system was framed as a single ADR (0001) covering ~18 crates; the design has since fragmented into 57 discrete decisions and 47 crates. The plan’s phase spine, layering rules, three-representation model, and fidelity approach remain correct and endorsed. What has drifted is (a) crate naming (unprefixed → twg-), (b) the crate matrix (sinks/sources/formats/contracts/DQ/observability now each have their own crate rather than being folded into a handful), and (c) three named subsystems the plan describes with stale specifics (recovery topology, metadata prefix, observability home). Roughly half the shipped crates have no phase, week budget, or exit gate in the plan at all.

Both ADR files carry an identical import note: “Treat the phase structure and crate layout as current; treat specific version pins and the single-ADR framing as pending update.” That is the reconciliation instruction in miniature — trust the plan’s shape, distrust its specifics.

A. CONFIRMED — plan decisions the ADR set still endorses unchanged

B. DEPARTED / SUPERSEDED — plan statements the design has since changed

B1. Recovery / coordination topology (the plan’s single biggest drift)

• Plan says (Phase 5, “Single-agent recovery only”): “One recovery worker needs no leases, no elections and no coordination substrate — it is the only claimant. Distributing recovery across agents is what needs coordination, and that is deferred … (ADR-0055, ADR-0056).” Recovery is framed as an internal capability of the single ingest process, unlocking the second sink.
• Current position — recovery is now a first-class deployment architecture, not an in-process worker:
  • ADR-0050: recovery is a deployment role (ingest / recovery / both); recovery windows are leased via the offset store; roles autoscale independently, to zero.
  • ADR-0057: reaffirms the plan’s gate (“recovery is a precondition for the second sink, not later hardening”) but narrows the deferral — only distributed recovery is deferred, not recovery itself.
  • ADR-0055: coordination substrate deferred, but with a chosen leading shape — an embedded replicated store, platform-independent, consensus library not written in-house.
  • ADR-0056: quorum is an additional worker role with a reduced throughput budget; modes 0 / 1 / 3+, two-node forbidden, stable assignment + graceful handoff.
  • ADR-0053: the coverage-store mechanics the plan attributes to twg-offset-store are now their own decision — position state may lag, never lead; gap-aware coverage stored with RAW, never in the sink it describes; local write-behind cache; source position only where no broker holds it.
• Impact: The plan’s ADR citations (0055/0056) are stale numbers pointing at the right idea. The rewrite must (a) split “recovery” into role-based recovery (0050, shippable in Phase 5) vs. distributed/quorum recovery (0055/0056/0057, genuinely deferred), and (b) attribute the coverage-store design to ADR-0053, not fold it into an unspecified twg-offset-store paragraph.

B2. Metadata column prefix — __kafka_ → configurable, default _twg_

• Plan says (Record-metadata section, repeatedly): reserved columns under a hard-coded __kafka_ prefix — __kafka_key, __kafka_timestamp, __kafka_offset, etc.; a payload field named __kafka_* is a hard decode error. The quality gate is named “Reserved-namespace collision” against __kafka_*.
• Current position — ADR-0051: the metadata column prefix is configurable, default _twg_; changing it on an existing target is a refused migration. The __kafka_ literal is gone (and was transport-specific, which conflicts with the transport-neutral connector goal — Pulsar metadata under a __kafka_ prefix is incoherent).
• Impact: Every __kafka_* string in the plan (metadata section, the collision gate, the examples) must become _twg_* (default) with a note that it is configurable and immutable-once-set. This is a pure find-and-replace-plus-semantics change, but it touches a named quality gate.

B3. Observability home — folded into connector-core → unified twg-observability crate

• Plan says (connector layer, Phase 5): “The instrumentation lives in connector-core, so both transports emit identical metric names.” OTel spans/metrics are a connector-core responsibility; the crate table lists connector-core deps as tower, opentelemetry, governor.
• Current position:
  • ADR-0032: a unified observability crate — metrics + OTLP + health share one surface → twg-observability.
  • ADR-0013: OTLP default, Prometheus scrape (disableable).
  • ADR-0031: health tree with three projections (/health, /ready, /live); readiness gated by source + primary-raw only.
  • ADR-0039: live per-stage memory/throughput digest — OTEL gauges + in-process /stats (30s window), honest per-stage attribution.
  • Confirmed by crate: twg-observability desc = “Prometheus metrics, OTEL/OTLP export, and the health tree with /health, /ready, /live projections.” twg-connector-core desc now reads “…flow control, observability, recovery” — it consumes observability, it no longer owns it.
• Impact: The plan must relocate OTel from a connector-core sub-bullet to a standalone crate + subsystem, and add the health tree and /stats digest, which have no presence in the plan at all (health/readiness is never mentioned). Cross-transport metric-parity gate stays, but now targets twg-observability.

B4. Sinks/sources — folded into stream-arrow → a modular source/sink crate matrix

• Plan says (crate table): stream-arrow = “BatchSink implementations, batching policy, IPC, Flight, offset correlation” — i.e. all sink concerns (including Flight and IPC) live in one crate; sources are implicit in kafka-client/pulsar-client.
• Current position — sinks and sources are each their own crate, and their existence is a decision (ADR-0012: all sources and sinks are optional and modular; replay-capable ones may hold primary-raw):
  • Sinks: twg-sink-object-store (ADR-0044), twg-sink-flight (ADR-0020), twg-sink-arrow-ipc (ADR-0026), twg-sink-postgres, twg-sink-zerobus.
  • Sources: twg-source-flight (ADR-0020), twg-source-arrow-ipc (ADR-0026), twg-source-batch (ADR-0015/0027/0033), twg-source-delta-sharing (ADR-0022).
  • twg-stream-arrow survives but is reduced to “batch sinks, IPC, Flight, offset correlation” as a shared substrate the modular sinks build on — not the home of every sink implementation.
• Impact: The plan’s Phase 5 “stream-arrow: … all four implementations (RawSink, ValueSink, ArrowSink, TeeSink)” is still valid as the core batching/correlation layer, but the plan has no concept of the 9 modular source/sink crates as separate deliverables. These need phase placement (see C and E).

B5. Single twg binary, not a library-only ecosystem

• Plan implies a library ecosystem plus a “deployable” that appears at the transform layer; there is no CLI/binary crate in the plan’s crate table, and packaging (Phase 9) produces “a binary” without naming its structure.
• Current position — ADR-0045: a single twg binary with subcommands (stream / batch / config), not separate executables. Crate: twg-cli.
• Impact: The binary is now a named crate with a defined command surface (including a whole batch mode, see C) and needs to appear in the workspace layout and packaging phase.

B6. Batch is now co-equal with streaming (not present in the plan)

• Plan says nothing — it is streaming-only end to end.
• Current position — ADR-0015: a batch CLI as a bounded source sharing the streaming pipeline; ADR-0027 Avro OCF as a batch source (shared Avro decode); ADR-0033 batch reader libraries (arrow-native + calamine + quick-xml; Polars rejected). Crate twg-source-batch reads Parquet/CSV/JSON/XML/Excel/Avro-OCF.
• Impact: This is a departure by omission — the plan asserts a streaming-only scope that the design has widened. See C for scope placement.

C. NEW / ABSENT-FROM-PLAN — subsystems with no phase, week budget, or exit criteria

Each entry: subsystem → ADR(s) → implementing crate(s) → recommended phase.

Codecs not in the plan’s Phase 7 list: twg-codec-custom (C17). Plan’s Phase 7 lists protobuf/json/xml only.

D. CRATE-LEVEL MAP — plan (~18, unprefixed) vs. actual (43, twg-)

Naming drift: The plan uses unprefixed names throughout (wire-tls, codec-core, stream-arrow). Every shipped crate is twg- prefixed. The plan already used the prefix in exactly one place — twg-offset-store in Phase 5 prose — showing the convention post-dates the plan and was applied inconsistently. The rewrite must prefix all 43.

Totals: 20 in-plan (19 renamed + 1 already-prefixed) · 26 NEW · 46 actual (of 47 workspace crates; this reconciliation table predates and omits twg-udf). The plan accounts for less than half the crate tree.

E. PER-PHASE IMPACT

Phase 0 — Foundations (2w)

• Wrong/incomplete: claims workspace-wide unsafe_code = "forbid" — contradicted by ADR-0043: twg-ffi is the sole audited unsafe exception. The rewrite must state “forbid with one crate-level allow, audited.”
• Absent scope to attach: publishing discipline (ADR-0047/0048, release-plz, dual-licence, semver checks), twg-config skeleton, twg-e2e harness scaffolding, docs-site infra (ADR-0052 — main branch protection is a gate precondition).
• Static-musl/aws-lc-rs proof stays correct.

Phase 1 — Transport primitives + value model (5w)

• Correct in spine.
• Absent scope: twg-type-map (ADR-0019) belongs here as a sans-io leaf beside the value model; twg-resource-loader (ADR-0029) foundational, pull in here or Phase 2; twg-config (ADR-0028/0030).

Phase 2 — Avro codec, all three surfaces (6w)

• Correct. Note ADR-0027 (Avro OCF) shares this decode — flag the reuse forward to batch mode.
• Absent scope: twg-contract-core/twg-contract-odcs first landing — contract-beats-inference (ADR-0035) interacts with codec schema resolution and should be designed alongside, not after.

Phase 3 — Kafka consumer, no groups (5w)

• Correct. Passthrough-ships-early claim stands.
• Absent scope: basic twg-observability metrics slice could start here (plan defers all OTel to Phase 5).

Phase 4 — Consumer groups (8w)

• Correct and unchanged. No drift. Still the hard phase per ADR-0001.

Phase 5 — Sinks, connector layer, coverage & recovery (11w) — most drift

• Wrong: “instrumentation lives in connector-core” → moved to twg-observability (B3). __kafka_ prefix → _twg_ (B2). “single-agent recovery only (ADR-0055/0056)” → role-based recovery per ADR-0050/0053/0057, distributed deferred (B1).
• Absent scope to attach: twg-observability + health tree + /stats (C7/C15); modular sink lane twg-sink-object-store/-arrow-ipc/-flight + source twg-source-flight/-arrow-ipc (C5/C6); DLQ reason-codes (C12); admission control (C14); dedup (C13); source-ack invariant hardening (ADR-0038). twg-offset-store mechanics → attribute to ADR-0053.
• Resequencing: Phase 5 is already 11 weeks and monolithic; this much added scope likely forces a split into 5a (single-sink + observability + connector-core) and 5b (coverage/recovery/multi-sink + object-store primary-raw lane).

Phase 6 — Producer, idempotence, transactions (6w)

• Correct. No drift.

Phase 7 — Ecosystem breadth (“ongoing”)

• Flagged as requested: Phase 7 is labelled “ongoing” with no exit gate — the only phase without one. This must be closed: either give it explicit exit criteria or split its contents into gated phases. It currently absorbs codecs (protobuf/json/xml), OAUTHBEARER crates, GSSAPI, and the whole AWS/Glue block, which is too much to leave ungated.
• Absent scope: twg-codec-custom (C17); the modular DB sinks twg-sink-postgres/twg-sink-zerobus (C5); the table-format block (C3) is far larger than the plan’s “Glue Data Catalog” bullet and probably deserves its own phase, not a Phase 7 sub-item.

Phase 8 — Pulsar (10w)

• Correct. Deepen with ADR-0049 (Pulsar flow control: staged subscription, topic-spread shedding, pod-seeded order, pressure-as-scale-signal) — the plan’s Pulsar gaps list predates this decision.

Phase 9 — Transform layer + packaging (12w)

• Correct in spine.
• Absent scope: twg-ffi (C8, the WASM C Data Interface boundary + ADR-0003 escape bridge); data-quality enforcement + audit (C2) run inside the DAG and have no home otherwise; twg-cli command surface + batch subcommand (B5/C4).

Cross-cutting resequencing gaps

• No headcount / parallelism assumption stated. The plan asserts “three critical paths” and “Phase 9 runs parallel to Phase 6 given separate ownership” but never states how many people/streams the timeline assumes. With 23 new crates (contracts, DQ, formats, batch, observability, 9 sink/source crates), the parallelism assumption is now load-bearing and must be made explicit — otherwise the Gantt is unfalsifiable.
• New subsystems have no Gantt bars at all: contracts, DQ, table formats, batch mode, observability/health, and the modular sink/source matrix appear nowhere in the timeline.

F. OPEN / PROVISIONAL — decisions the plan presents as settled but are genuinely open

Summary for the rewrite

1. Prefix everything twg- and expand the crate table from ~18 to 43 (§D).
2. Fix the three named departures verbatim: recovery topology (0050/0053/0055/0056/0057, not “single-agent only 0055/0056”), metadata prefix (_twg_ configurable, not __kafka_), observability home (twg-observability, not connector-core) (§B1–B3).
3. Add the absent half of the system: contracts/ODCS, data quality + audit, table formats (Iceberg-only), batch mode, the modular source/sink matrix, unified observability + health, FFI, type-map, resource-loader, config/secrets, admission control, dedup, DLQ reason-codes, releases/publishing, custom-decode (§C) — each needs a phase, budget, and exit gate.
4. Close Phase 7’s open-ended “ongoing” with an exit gate, and split the overloaded Phase 5.
5. State the parallelism/headcount assumption the timeline silently depends on.
6. Record ADR-0003 as Accepted (pin Arrow 58) and keep F1–F5 as explicitly open, not settled.
7. Reconcile the unsafe_code = "forbid" claim with the audited twg-ffi exception (ADR-0043).

Proto Bundle — Technical Implementation Plan

A crate-by-crate build plan for twg-proto-bundle and the twg proto bundle subcommand: turning any proto source (a single .proto, a bare directory of protos, or a Java/Maven/Gradle contracts repo with many src/main/proto roots) into one combined FileDescriptorSet (.pb), one proto3 file per package, and a fully-qualified message-type manifest. Grounds the decision in ADR-0063, reuses twg-proto-schema’s pure-Rust compile_proto_to_bytes (ADR-0001), and feeds the existing descriptor toolkit (build_arrow_schema, twg-proto-decode, ADR-0046).

Status. Design → build. The .pb path is the priority deliverable; the per-package .proto render and manifest are derived from it.

No sensitive information. All names in this plan are generic placeholders (com.example.*, example/common/*.proto). Worked examples describe sources only by scale — “the small single-package source”, “a large ~530-file source” — never by their real repo, package, or type names. Nothing domain-specific is embedded in the tool, its tests, or these docs.

0. What the operator runs

twg proto bundle <SOURCE> --out-dir <DIR>
     [--name <basename>]                     # default: source dir/file stem
     [--include-root <DIR>]                  # repeatable; EXTRA roots (cross-repo deps)
     [--exclude <name>]                      # repeatable; prune a subtree (e.g. a build-staging dir)
     [--include-glob '**/src/main/proto']    # repo-mode auto-detect pattern
     [--emit pb,single,proto,manifest]       # default: all four
     [--on-conflict error|first-wins]        # default: error
     [--wkt-imports central|per-file]        # default: central

Outputs in <DIR>:

Worked example — twg proto bundle <SMALL_SOURCE> --out-dir ./out --name orders (a single-package source) yields out/orders.pb, out/orders.proto (single-package ⇒ standalone-compilable), out/orders/com.example.orders.proto, and out/orders.messages.json listing com.example.orders.Order (and siblings), with the envelope/entry types flagged as roots. For a multi-package source <name>.proto is a combined view and the .pb is the artefact to compile.

1. Crate placement

New crate crates/twg-proto-bundle, publish = false (operator/build tooling, not a runtime dependency). Depends on twg-proto-schema with its compile feature. Sits above the descriptor toolkit in the layer graph; nothing depends back on it.

crates/twg-proto-bundle/
  Cargo.toml
  src/
    lib.rs        # BundleOptions, Bundle, bundle(), BundleError
    discover.rs   # source resolution -> include roots + ProtoFile{abs, import_path}
    reconcile.rs  # dedup by import path, identity check, staging tree
    compile.rs    # stage -> compile_proto_to_bytes -> FileDescriptorSet
    manifest.rs   # FileDescriptorSet -> {all, roots} FQ message types
    render.rs     # FileDescriptorSet -> one proto3 file per package
  tests/
    repos.rs      # path-gated integration over the three real repos + round-trip

Dependencies: prost / prost-types 0.14 (already the workspace proto version), walkdir, sha2 (content identity), serde + serde_json (manifest), thiserror (workspace), tempfile (staging). CLI adds clap.

2. discover — source resolution (ADR-0063 §Decision.1)

resolve_source(source, opts) -> Discovered { roots: Vec<PathBuf>, files: Vec<ProtoFile> }

• ProtoFile { abs: PathBuf, import_path: String, root: PathBuf } where import_path is abs relative to its root, forward-slashed.
• Resolution order:
  1. source is a file ⇒ root = parent (or --include-root), one ProtoFile.
  2. source is a dir + --include-root given ⇒ those roots verbatim.
  3. dir + --include-glob matches ≥ 1 dir ⇒ repo mode, roots = matches.
  4. else ⇒ source itself is the lone root.
• Walk each root with walkdir, skipping target/, build/, .git/, node_modules/; collect *.proto.
• Empty result ⇒ BundleError::NoProtoFiles.

Verified against the three repos: 1 root/9 files, 8/128, 77/534.

3. reconcile — dedup + stage (ADR-0063 §Decision.2–3)

• Group files by import_path.
• One path → many abs: read + sha256. All equal ⇒ keep first, count dropped. Divergent ⇒ BundleError::Conflict { import_path, roots, hashes } unless --on-conflict first-wins (then keep first from the first-listed root, log a warning — never silent).
• Stage: create a tempfile::TempDir; for each surviving import_path, copy its file to stage/<import_path> (creating parent dirs). Return Staged { dir: TempDir, files: Vec<String> /* import paths */ }.

Across the observed sources all collisions are byte-identical (a shared file present under two repos’ roots; ~7 internal dups in the large source) so the default error path passes; the check exists to catch future drift.

4. compile — staged tree → .pb (ADR-0063 §Decision.3–4)

• proto_paths = every staged file (absolute), include_paths = [stage.dir].
• Call twg_proto_schema::compile_proto_to_bytes(&proto_paths, &include_paths) (enable twg-proto-schema/compile).
• protox bundles the Google WKTs and includes all reachable types; the returned bytes are the FileDescriptorSet. Write to <name>.pb when pb is in --emit.
• Phase-1 assumption to confirm on first run: protox ships google/protobuf/{timestamp,duration,wrappers,any}.proto. If a compile reports a missing google/protobuf/* import, stage bundled WKT copies as a fallback (a localized change in compile.rs).

5. manifest — message-type inventory (ADR-0063 §Decision.6)

• Decode the .pb with prost_types::FileDescriptorSet.
• For each file, walk message_type recursively; FQ name = .<package>.<Msg>[.<Nested>…] (drop the leading dot for output).
• Root detection: collect every type_name referenced by any field across the set; a message whose FQ name is never referenced is a root (entry/envelope).
• Emit { "all": [...sorted...], "roots": [...sorted...] } to <name>.messages.json. Also returned in-process on the Bundle struct so tests/callers can assert on it (e.g. com.example.orders.Order present).

6. render — one .proto per package (ADR-0063 §Decision.5)

Rendered from the descriptor, not the source text, so it is canonical and package-grouped.

• Group FileDescriptorProtos by package.
• Per package emit: syntax = "proto3";, package <pkg>;, imports, then message and enum bodies.
• Imports: for each type referenced from another package, import that package’s generated file. Google WKTs: with --wkt-imports central (default), a generated base file holds the import "google/protobuf/*" lines and dependents pick them up via import public; with per-file, each file imports what it uses. No file ever redefines a WKT.
• Field rendering: scalar/message/enum types; repeated; proto3 optional (synthetic oneof unwrap); map<k,v> reconstructed from synthetic MapEntry nested messages (detect options.map_entry); oneof groups; nested messages and enums; reserved ranges/names.
• Correctness gate (test, not runtime): recompile the rendered files with protox, normalise both descriptors (sort files/fields, drop source-info), and assert equal to the source .pb. A render that does not round-trip is a test failure.

7. lib — public surface

#![allow(unused)]
fn main() {
pub struct BundleOptions {
    pub include_roots: Vec<PathBuf>,   // explicit override; empty = auto
    pub include_glob: String,          // default "**/src/main/proto"
    pub emit: EmitSet,                 // pb | proto | manifest
    pub on_conflict: OnConflict,       // Error (default) | FirstWins
    pub wkt_imports: WktImports,       // Central (default) | PerFile
    pub name: Option<String>,
}
pub struct Bundle {
    pub descriptor_set: Vec<u8>,       // the .pb bytes
    pub messages: MessageManifest,     // { all, roots }
    pub proto_files: Vec<RenderedProto>, // package -> text
    pub dropped_duplicates: usize,
}
pub fn bundle(source: &Path, out_dir: &Path, opts: &BundleOptions) -> Result<Bundle, BundleError>;
}

bundle() orchestrates discover → reconcile → compile → manifest → render, writes the --emit subset to out_dir, and returns Bundle for programmatic callers.

8. CLI wiring

twg-cli/src/main.rs is a scaffold (fn main() {}). Stand up a minimal clap command tree with proto bundle as the first real subcommand (streaming/batch stay scaffolded). main parses args into BundleOptions, calls twg_proto_bundle::bundle, prints a summary (N files, M packages, K messages, D duplicates dropped → out/…), maps BundleError to a non-zero exit.

9. Testing

• Unit: discovery per mode (file/override/repo/bare-dir); reconcile identity vs. divergence; map/oneof/optional/enum/nested rendering on small fixtures.
• Integration (tests/repos.rs, skipped unless an env var points at a source): driven by TWG_BUNDLE_TEST_SOURCE (+ optional TWG_BUNDLE_TEST_ROOTS, TWG_BUNDLE_TEST_EXCLUDE), so the suite carries no domain-specific names and runs against whatever tree the operator supplies. Assertions are structural invariants only: .pb decodes to a non-empty FileDescriptorSet; the manifest is non-empty; include-root count / dropped-duplicate count are surfaced; and a round-trip — recompile the rendered per-package set and compare the normalised type inventory to the source .pb (reported, not asserted, when the source has a package cycle).

10. Phasing

1. Phase 1 — .pb + manifest. discover → reconcile → stage → compile → manifest. Delivers the combined .pb and the message-type list (the priority). Settles the protox WKT question. Proven first on the smallest source.
2. Phase 2 — per-package render + round-trip test. The proto3 printer and its gate; run against sources of increasing size.
3. Phase 3 — CLI + docs. twg proto bundle, summary output, README/rustdoc.

11. Risks

• protox WKT bundling — Phase-1 first run confirms; cheap staged-WKT fallback.
• Render fidelity (maps, oneof, optional, custom options) — bounded by the round-trip gate; custom options deferred (ADR-0063 D1).
• Scale (~530 files / 77 roots on the largest observed source) — staging + one compile; well within protox.

Thalweg — Detailed Delivery Plan

Implements the ADR set 0001–0057 (with ADR-0058 amending 0005/0051 — record-metadata provenance naming), indexed in docs/adr/README.md. Where this plan and an ADR disagree, the ADR is authoritative; this document sequences the ADRs into deliverable phases and states the gates that defend them.

Revision note (v0.6). This revision folds the full 57-ADR design into the plan. The phase spine, the three-representation model, the sans-io layering, and the fidelity approach are unchanged and confirmed. What changed since v0.5: every crate is twg- prefixed and the workspace is 47 crates (not ~18); recovery is a deployment role with a coverage store that lives with raw (ADR-0050/0053/0057), not an in-process single agent; the metadata prefix is configurable and defaults to _twg_ (ADR-0051), not __kafka_; observability is its own crate twg-observability with a health tree (ADR-0013/0031/0032/0039), not folded into connector-core; sinks and sources are a modular crate matrix (ADR-0012/0020/0026/0044); data contracts, data quality, table formats, and batch mode are first-class subsystems with their own phases; and the workspace ships one twg binary with stream/batch/config subcommands (ADR-0045).

Workspace layout

Crates group into layers with a strictly downward dependency direction, enforced in CI. The single Arrow version is named once, in twg-codec-core’s re-export; nothing else names it (ADR-0003, Accepted — pin to DataFusion’s Arrow major, no bridge).

%% name: crate-layer-graph
graph BT
    subgraph transport[Transport — no runtime, no format]
        TLS[twg-wire-tls]
        SASL[twg-wire-sasl<br/><i>sans-io</i>]
        COMP[twg-wire-compression]
    end
    subgraph sources[Sources — modular, optional]
        KC[twg-kafka-client]
        PC[twg-pulsar-client]
        SB[twg-source-batch]
        SF[twg-source-flight]
        SI[twg-source-arrow-ipc]
        SD[twg-source-delta-sharing]
    end
    subgraph connector[Connector — transport-neutral]
        CN[twg-connector-core<br/><i>Source/Sink, Budget, retry, DLQ, recovery, dedup</i>]
        OBS[twg-observability<br/><i>OTLP, Prometheus, health tree, /stats</i>]
        OFF[twg-offset-store<br/><i>gap-aware coverage</i>]
    end
    subgraph framing[Framing & catalogs]
        SR[twg-schema-registry]
        TCAT[twg-table-catalog]
    end
    subgraph codec[Codecs — no transport, no runtime]
        CC[twg-codec-core<br/><i>Value model + Arrow re-export</i>]
        TM[twg-type-map]
        AVRO[twg-codec-avro]
        PROTO[twg-codec-protobuf]
        PFLAT[twg-proto-flatten]
        PSCH[twg-proto-schema]
        PDEC[twg-proto-decode]
        JSON[twg-codec-json]
        XML[twg-codec-xml]
        CUST[twg-codec-custom]
    end
    subgraph contracts[Contracts & DQ]
        KO[twg-contract-core]
        ODCS[twg-contract-odcs]
        DQE[twg-dq-enforcer]
        DQA[twg-dq-audit]
    end
    subgraph pipeline[Pipeline & transform]
        SA[twg-stream-arrow]
        PS[twg-pipeline-sql]
        PW[twg-pipeline-wasm]
        PF[twg-pipeline-flight]
        FFI[twg-ffi<br/><i>Arrow C Data Interface</i>]
    end
    subgraph formats[Table formats]
        FP[twg-format-parquet]
        FI[twg-format-iceberg]
    end
    subgraph sinks[Sinks — modular, optional]
        SKO[twg-sink-object-store]
        SKF[twg-sink-flight]
        SKI[twg-sink-arrow-ipc]
        SKP[twg-sink-postgres]
        SKZ[twg-sink-zerobus]
    end
    subgraph compose[Compose & runtime]
        KA[twg-kafka-arrow]
        PA[twg-pulsar-arrow]
        CFG[twg-config]
        RL[twg-resource-loader]
        CLI["twg-cli<br/><i>twg stream | batch | config</i>"]
    end

    KC --> TLS & SASL & COMP
    PC --> TLS & COMP
    KC --> CN
    PC --> CN
    SB --> CN
    CN --> OBS
    CN --> OFF
    AVRO & PROTO & JSON & XML & CUST --> CC
    CC --> TM
    ODCS --> KO
    DQE --> KO
    DQA --> DQE
    PS --> CN & CC & KO
    PW --> PS
    PF --> PS
    FFI --> CC
    SA --> CC & CN
    FP --> CC
    FI --> FP & TCAT
    SKO --> FP & FI
    KA --> KC & SR & SA & DQE
    PA --> PC & SA
    CLI --> KA & PA & SB & CFG
    CFG --> RL
    ODCS --> RL

Three rules make the layering work, all enforced in CI rather than by convention: nothing in the codec layer depends on the transport or connector layers; nothing in the codec layer or twg-wire-sasl depends on a runtime; twg-codec-* crates build clean with --no-default-features, i.e. with no arrow dependency present.

Crate inventory (44 crates)

Publishing is independent per crate under dual MIT OR Apache-2.0, via release-plz with semver checks and a codec version group (ADR-0047/0048); arrow is an intentional public dependency of twg-codec-core (ADR-0047), which is precisely the coupling the single-version chokepoint (ADR-0003) governs.

The three representations

The transport’s currency is bytes. Decoding is an optional stage applied to it, and the choice of representation is the consumer’s.

#![allow(unused)]
fn main() {
// twg-kafka-client yields this. No codec involvement, no schema resolution.
pub struct RawRecord {
    pub key: Option<Bytes>,              // None and Some(b"") are distinct
    pub value: Option<Bytes>,            // envelope intact
    pub headers: Headers,                // ordered, dup keys and null values permitted
    pub timestamp: Timestamp,            // carries CreateTime | LogAppendTime
    pub locator: Locator,                // topic, partition, offset, leader_epoch
    pub batch: BatchContext,             // producer_id, epoch, base_seq, txn/control flags
}

// Ordered sequence, NOT a map. Kafka permits duplicate keys and null values;
// a HashMap loses both before any test can catch it.
pub struct Headers(Vec<(Bytes, Option<Bytes>)>);

// twg-codec-core, always available — no arrow dependency
pub trait Decoder { type Error; fn decode(&self, payload: &[u8]) -> Result<Value, Self::Error>; }
pub trait Encoder { type Error; fn encode(&self, v: &Value, out: &mut BytesMut) -> Result<(), Self::Error>; }

// twg-codec-core, behind `arrow` feature (default-on)
pub trait ArrowDecoder: Send {
    fn schema(&self) -> SchemaRef;
    fn push(&mut self, payload: &[u8]) -> Result<()>;
    fn finish(&mut self) -> Result<RecordBatch>;
}
pub trait ArrowEncoder: Send {
    fn encode(&self, batch: &RecordBatch, out: &mut Vec<Bytes>) -> Result<()>;
}
}

ArrowDecoder goes from bytes to builders directly — it does not route through Value. The neutral model serves record-at-a-time consumers; the columnar path stays free of intermediate materialisation. Each codec therefore implements two decode paths over one schema-resolution core, which the type-mapping authority (twg-type-map, ADR-0019) describes once so both paths agree.

All three representations share one sink abstraction, so batching policy and offset correlation are written once:

#![allow(unused)]
fn main() {
pub trait BatchSink {
    type Output;
    fn push(&mut self, rec: &RawRecord) -> Result<()>;
    fn finish(&mut self) -> Result<(Self::Output, OffsetSpan)>;
}
}

RawSink → BytesBatch; ValueSink<D: Decoder> → Vec<Value>; ArrowSink<D: ArrowDecoder> → RecordBatch; TeeSink<A, B> → both, which is the audit-retention case (Arrow for analytics, raw bytes for regulated replay) and is cheap given refcounted slices. The producer side is symmetric: it accepts pre-formed Bytes, a Value, or a RecordBatch. OffsetSpan returns identically from every sink, so commit-after-write, replay positioning and the soak harness are written once and work regardless of representation.

Record metadata: the canonical _twg_ column set

Every record carries metadata beyond its payload — where it came from, when it happened, and what we did with it. That metadata is modelled once, as a canonical set of reserved columns under a configurable prefix (default _twg_, immutable once a target exists — changing it is a refused migration, ADR-0051), and reused across every lane. The raw lane’s contract is an Arrow RecordBatch carrying the payload bytes plus this column set; the physical file schema — Parquet layout, opaque-envelope vs. exploded columns — is a downstream sink decision and is deliberately not fixed here. The decoded Value and Arrow lanes carry a policy-selected subset of the same columns.

The columns split into two families by provenance, and the naming makes the split explicit (ADR-0005/0051, amended by proposed ADR-0058).

Source-provided — _twg_source_*. A transport-neutral superset of everything the upstream broker hands us. A column is null for transports that do not supply it, so one schema describes Kafka and Pulsar alike:

We-stamped — _twg_* (no source infix). Values we produce; the absent source infix is the tell that Thalweg, not the broker, is the authority:

Together, _twg_source_timestamp → _twg_ingest_ts → _twg_emit_ts give the full event-time → processing-time provenance chain the design previously lacked.

The rejected alternative was carrying metadata out-of-band in OffsetSpan. It keeps the batch schema clean, but a Flight consumer receives only the batch — the sidecar does not survive the wire — so every metadata field would be lost at exactly the boundary where it matters most for downstream Lakehouse writes. Columns cost schema noise and a namespace-collision risk with payload fields, which is why the prefix is reserved and validated: a payload field under the reserved prefix is a hard decode error rather than a silent shadow.

MetadataPolicy governs the decoded lanes only: None (payload only), Locator (topic/partition/offset/message-id — the common case for commit correctness), or All (the complete set above). Default is Locator. The raw lane’s Arrow form carries the full set by construction; the Value path carries the same fields in a reserved sibling struct under the identical naming convention.

Provenance-naming amendment (proposed ADR-0058, amends ADR-0005/0051). Earlier revisions used flat names (_twg_offset, _twg_leader_epoch, …). This revision draws the source-provided vs. we-stamped line into the names themselves (_twg_source_offset vs. _twg_ingest_ts) and generalises the source columns to a transport-neutral superset. Renaming every source column is a larger schema change than the configurable-prefix case ADR-0051 governs, so it is recorded as an explicit amendment rather than a silent rename and must land before any target exists. The Pulsar-specific rows are provisional: the Pulsar message field model is not yet pinned in the repo (twg-pulsar-client is flow-control-only; twg-pulsar-arrow is a stub), and is confirmed against a live broker in Phase 8. RawRecord/Locator as sketched above is Kafka-shaped today; generalising it to carry this superset is part of the Phase 5 connector-core retrofit (risk F4).

Contracts & data quality

A data contract describes what a stream is supposed to contain. Thalweg reads ODCS (Open Data Contract Standard) and uses it to decide both the shape of the data and what counts as acceptable. Contracts are optional: without one, schemas are inferred or taken from a registry and quality rules come from configuration; with one, the contract is authoritative on schema and its expectations merge with any configured manually (ADR-0016, ADR-0035).

• Schema authority (ADR-0035). Where a contract defines schema it wins over inference. For CSV/JSON/XML that is also faster — inference samples and guesses; a contract does not. twg-contract-core holds the resolved spec; twg-contract-odcs parses ODCS and translates it in, loaded local/HTTP/object-store through twg-resource-loader and validated at startup (ADR-0018/0029).
• Rules merge, tightest wins (ADR-0016/0036). A manual rule and a contract rule on the same field resolve to the more restrictive; neither source can silently loosen the other.
• Enforcement modes (ADR-0014). Strict fails the batch, quarantine diverts bad records (non-blocking; overflow drops-with-count, ADR-0041), annotate flags and passes through. Enforcement is twg-dq-enforcer, running inside the transform DAG.
• Drop rules (ADR-0036). Known junk — heartbeats, test traffic — is discarded deliberately, silent by default with opt-in audit, so the DLQ stays a signal about real failures.
• Unified DLQ with reason-codes (ADR-0017). Decode, contract, and quality failures land in one place tagged by reason (decode / contract / DQ / retry-exhausted); deliberately dropped traffic never appears there.
• Audited (ADR-0034). Every record’s verdict on every rule is recorded durably in twg-dq-audit, stamped with the rule version that produced it, as a first-class sink, on by default.
• Gates deduplication (ADR-0054). Content dedup is permitted only where a contract declares a natural key — dropping byte-identical payloads is unsafe without one, and validation refuses rather than warns.

The connector layer

twg-connector-core is transport-neutral and sits above every source and sink alike. Sink/source semantics, flow control, retry, DLQ, and recovery are cross-transport concerns; putting them in a single client would mean the other transports cannot reuse them.

#![allow(unused)]
fn main() {
pub trait Source {
    type Cursor: Cursor;                                  // Kafka offset | Pulsar MessageId
    fn poll_batch(&mut self, cx: &mut Context, budget: Budget)
        -> Poll<Result<Vec<RawRecord>>>;
    fn commit(&mut self, upto: Self::Cursor) -> BoxFuture<Result<()>>;
    fn pause(&mut self);                                  // explicit, not implicit
    fn resume(&mut self);
    fn seek(&mut self, to: Position<Self::Cursor>) -> BoxFuture<Result<()>>;
}

pub trait Sink {
    fn send(&mut self, batch: Payload) -> BoxFuture<Result<Vec<Receipt>>>;
    fn flush(&mut self) -> BoxFuture<Result<()>>;
}
}

Budget is the backpressure primitive: the caller states what it can absorb — max records, max bytes, deadline. A Kafka source translates that into fetch max_bytes/max_wait; a Pulsar source into permit-window replenishment via FLOW. Both expose the same explicit pause()/resume(). Admission control (ADR-0038/0040) sits on top: the byte budget self-corrects to a live measured p95 record size, with the static record_size_est_bytes in twg-config as a cold-start seed and floor. Rate limiting is a governor-backed layer in messages/sec and bytes/sec, composable per source.

Observability is its own crate (ADR-0032), not folded in here. twg-connector-core emits through twg-observability, which owns OTLP export (default), Prometheus scrape (disableable, ADR-0013), the per-stage /stats digest (30s window, honest per-stage attribution, ADR-0039), and the health tree with three projections /health, /ready, /live — readiness gated by source + primary-raw only (ADR-0031). Both transports emit identical metric names, so one dashboard works against either. Spans follow the record lifecycle (receive → decode → process → commit) with context propagated through message headers, so a trace crosses the broker.

Recovery, coverage & coordination

The correctness backbone is one rule applying to every position the system holds: position may lag, but must never lead (ADR-0053). A source position ahead of what was durably written resumes past records that never landed; a sink marked written when it wasn’t is never replayed. Both are silent loss; behind, in either case, is a redundant replay that dedup absorbs.

So the source-ack invariant (ADR-0038) holds: advance the offset only on a durable sink acknowledgement — batch-granular by default, per-record opt-in, durable mandatory for the primary-raw role. Any replayable sink can hold primary-raw (ADR-0012): twg-format-parquet (no catalog needed), Postgres, Zerobus where Iceberg reads are enabled.

Recovery is a precondition for the second sink, not later hardening (ADR-0057). Once the source is acknowledged the broker moves on and can no longer say what a downstream sink missed; from that moment the coverage store is the only record of it. A multi-sink topology without recovery is silently lossy by construction, so configuration refuses one.

• Coverage store (ADR-0053). twg-offset-store records gap-aware covered ranges per destination — not a high-water mark — stored with raw, never inside the sink it describes (a store in a failed sink cannot record that the sink failed). It may lag reality, costing a replay; it may never over-claim. A local write-behind buffer checkpoints on an interval.
• Deterministic IDs. Each record hashes to a stable ID including its source cluster, so bounded, deduplicated replay from raw restores a lagging sink and overlapping ranges merge harmlessly.
• Recovery is a deployment role (ADR-0050). A node runs as ingest, recovery, or both; recovery windows are leased via the offset store; the recovery role scales independently, to zero when nothing is behind, so a large backfill cannot throttle live ingest.
• Content dedup (ADR-0054). The record ID catches broker redelivery but not an upstream producer retry that republishes the same payload at a new offset; hashing the payload catches that. Opt-in, contract-gated on a declared natural key, window sized to the upstream retry interval.

Coordination is deliberately tiny (ADR-0055/0056). Workers scale freely on throughput; what they must agree on — which recovery window each agent claimed, which node drives recovery — is small, so the coordination layer is sized for that, not the data path. Quorum is a role, not a tier (ADR-0056): a subset of workers additionally hold quorum, lowering their own throughput budget to leave headroom for the consensus fsync. Modes are 0 (off), 1 (full function, no fault tolerance), 3+ (odd sizes only); two nodes is refused — quorum of two halts writes if either is lost, strictly worse than one. The coordination substrate itself is deferred (ADR-0055): the leading shape is an embedded replicated store, platform-independent, with the consensus library not written in-house; only distributed recovery waits on it (ADR-0057).

The transform layer

This is the first part configurable at runtime rather than compile time, which shifts what the project is: up to here a library ecosystem, from here also a deployable that non-authors configure. User-supplied SQL and user-supplied binaries now run inside the process, which changes the threat model.

SQL as sync configuration. twg-pipeline-sql exposes the source as a DataFusion TableProvider; extraction, mapping, casting, filtering and DQ rules are expressed as SQL. Fan-out is a DAG, not a flat set of sinks: raw taps straight out; the same source is decoded into an Arrow node; virtual tables derive for DQ, filtering, mapping; some unnest nested structures into further tables with their own sinks. Sinks attach at interior nodes, not only leaves. A node is { name, query, optional sink, materialisation }; materialisation is automatic when a node has more than one dependent (DataFusion re-executes a plan per collect(), so an unmaterialised shared intermediate is re-derived per dependent — the fan-out waste just moves down a level).

Two consequences, stated as contracts because they are silent when wrong:

• Cardinality changes break offset correlation (ADR-0004). An unnest node emits N rows per input record; the OffsetSpan range assumption no longer holds. Every derived row carries its originating record’s locator, and a node’s span is the union of contributing locators. Without this, commit-after-write acknowledges records whose derived rows never landed, and it fails quietly.
• Backpressure propagates through the DAG. The source Budget is the minimum across the sink set; per-node queue depth is an exported metric so a stalled leaf is diagnosable as such rather than as unexplained source throttling.

Two extension mechanisms, deliberately separate (ADR-0008).

The Arrow C Data Interface across the WASM boundary is twg-ffi (ADR-0042/0043), which is also the sole crate permitted unsafe — a single audited exception to the workspace-wide unsafe_code = "forbid". twg-pipeline-flight handles what WASM should not: inline model inference gets a native runtime, independent scaling, and a lifecycle the pipeline does not own.

Sources, sinks & table formats

Every source and sink is optional and modular (ADR-0012); a deployment wires only what it needs, and all implement the one Source/Sink contract so flow control, health, retry and recovery are shared.

• Sources: Kafka, Pulsar, batch files (twg-source-batch), Arrow Flight, Arrow IPC, Delta Sharing, and the inbound Arrow C Data Interface (twg-ffi).
• Sinks: object-store (Parquet/Iceberg by config, ADR-0044), Flight, Arrow IPC, Postgres, Zerobus.
• Table formats (ADR-0022/0023): one open table format is carried — Iceberg — read+write, catalog-mediated; Delta is read via generated Iceberg metadata and the catalog’s Iceberg REST endpoint rather than a second writer. Plain Parquet is the primary-raw-eligible raw lane needing no catalog. Catalog credential vending covers UC, S3 Tables, S3/IAM, and Iceberg REST (ADR-0024); descriptive metadata (comments/properties/tags) is written on create and evolution and re-synced idempotently (ADR-0037). Schema evolution against a live table is the sharp edge: additive is safe, widening conditionally safe, anything else fails loudly.

Batch mode

Batch is co-equal with streaming, not a separate tool (ADR-0015): twg-source-batch is a bounded source sharing the streaming pipeline, so contracts, DQ, transforms, sinks and recovery are the same code. It reads Parquet/CSV/JSON/XML/Excel via arrow-native + calamine + quick-xml (Polars rejected, ADR-0033), and Avro Object Container Files reuse the streaming Avro decode (ADR-0027). It is driven by twg batch.

Packaging

The primary artefact is a statically linked musl binary — the single twg command with stream/batch/config subcommands (ADR-0045) — not an image. From it, a distroless OCI image, a bare binary for systemd, and a Lambda package all follow. Multi-arch arm64 and amd64 manifests. aws-lc-rs needs care on musl and is the main integration risk, proven in Phase 0. Config is verbatim TOML carried by K8s (file-mount default, pointer option; no TOML↔YAML conversion, ADR-0028), with secret enumeration via twg config secrets (ADR-0030). Documentation ships to docs.thalweg.dev on Cloudflare Pages via Git integration, gates protecting the merge, so main must be protected (ADR-0052).

Fidelity: what “no information lost” means

“No information lost” is testable only if the places Kafka’s model is not round-trippable are stated up front. Three fields are broker-authoritative and cannot be asserted by equality:

Everything else round-trips exactly, proven by a corpus built to include the edge values abstractions usually swallow: null vs empty key, null (tombstone) vs empty value, zero vs 200 headers, duplicate header keys, null header values, negative and zero timestamps, unicode and max-length topic names, empty batches. Recompression is not byte-stable, so byte-identity is asserted on decompressed record bytes and the produce direction asserts semantic batch equality after a decompress, not wire equality.

Three test tiers run from Phase 3 onward as each representation lands: field-completeness (the edge-value corpus survives consume → each representation → produce → consume, with a per-field predicate), property-based (proptest generates arbitrary RawRecords against the same predicate), and differential (the same records via rdkafka and via twg-kafka-client, compared at the wire level after decompression). The twg-e2e crate owns the harness and encoded regressions (ADR-0021).

Phase plan

The critical-path timeline below assumes two to three parallel implementation streams under separate ownership (transport/codec, connector/sink, transform/breadth). Without that staffing the parallel tracks serialise and the calendar stretches accordingly; the assumption is stated so the Gantt is falsifiable.

Phase 0 — Foundations (2 weeks)

Workspace scaffold; CI from day one. Shared lint config; clippy::pedantic; unsafe_code = "forbid" at workspace level with one audited crate-level allow in twg-ffi (ADR-0043); per-file 85% coverage ratchet; cargo-deny for licence/advisory and the purity audit; cargo-public-api snapshots. Release discipline (ADR-0047/0048): release-plz, dual MIT OR Apache-2.0, semver checks, codec version group. twg-config, twg-resource-loader and twg-e2e skeletons land here. Docs-site infra with main branch protection (ADR-0052). A static musl build against aws-lc-rs is proven on both architectures.

Exit: empty crates build, all gates green, a deliberately-failing coverage commit is rejected, and a static musl binary linking aws-lc-rs runs on arm64 and amd64.

Phase 1 — Transport primitives + value model (5 weeks)

Two independent tracks. Transport: twg-wire-tls (all five broker variants), twg-wire-sasl (PLAIN, SCRAM-SHA-256/512 as sans-io state machines against RFC 5802 vectors), twg-wire-compression (gzip/snappy/lz4 encode+decode, zstd decode). Codec/core: twg-codec-core (neutral Value, Decoder/Encoder, Arrow traits, the re-export, the schema-evolution resolver) and twg-type-map (ADR-0019) beside it.

The Value model is the highest-risk design work, validated by paper exercise against real Avro/Proto/JSON/XML schemas before any codec is written. twg-resource-loader and twg-config foundations firm up here since contracts and SQL will need them.

Exit: SCRAM passes RFC vectors; TLS connects to Redpanda with mTLS; Value round-trips hand-built samples from all four formats; twg-codec-core builds with --no-default-features and no arrow in the tree.

Phase 2 — Avro codec + registry + contract core (6 weeks)

twg-schema-registry (Confluent client covering Redpanda/StreamNative/Apicurio, a static in-memory impl for tests, moka caching; Glue deferred to Phase 7). twg-codec-avro implementing both decode paths over shared resolution. WireFormat::peel is non-destructive so passthrough consumers keep the envelope. twg-contract-core lands here (ADR-0016/0035): contract-beats-inference interacts with codec schema resolution and must be designed alongside it, not bolted on later; twg-contract-odcs parsing follows once the resource loader is ready.

Two Avro decisions to settle and document: ["null","T"] unions map to nullable T while other unions map to a dense Arrow union / tagged Value variant; logical types map to Arrow equivalents with decimal precision/scale preserved.

Exit: decode a Confluent-framed Avro topic dump to RecordBatch byte-identical to an rdkafka+Java reference oracle; Value path round-trips the same corpus; a contract-driven schema overrides inference on one CSV/JSON sample; Criterion baselines established.

Phase 3 — Kafka consumer, no groups (5 weeks)

Connection pooling, metadata refresh, Fetch loop, manual assignment, explicit commit. Zero-copy passthrough is a design constraint of this phase: the decompression path writes into one buffer and hands out Bytes::slice views. Cancel-safety in select! is a first-class requirement defended with a test. A basic twg-observability metrics slice starts here rather than waiting for Phase 5.

Ships something usable: raw passthrough consume works at the end of this phase with no codec and no registry present, serving tee-to-archive, proxy and forwarding workloads.

Exit: consume a partitioned topic against Redpanda, MSK and Azure Event Hubs in integration CI; soak survives broker restart and leader election; passthrough byte-identity test green.

Phase 4 — Consumer groups (8 weeks)

The hard phase, not deferred or trimmed (ADR-0001). JoinGroup/SyncGroup/Heartbeat state machine; range, round-robin, sticky and cooperative-sticky assignors; incremental cooperative rebalance; offset-commit strategies that do not lose messages on rebalance; static membership. Modelled sans-io so it is testable without a broker, property-tested against protocol invariants, then run under a chaos harness that kills members mid-rebalance.

Exit: a 20-member group survives rolling restarts, member kills and partition-count changes without duplicate assignment or stuck rebalance; cooperative-sticky demonstrably avoids stop-the-world.

Phase 5 — Sinks, connector layer, observability (Phase 5a, 7 weeks)

twg-stream-arrow: BatchSink and all four implementations, batching policy, Arrow IPC file/stream writers, Flight do_get/do_put. twg-connector-core: Source/Sink, Budget + admission control, rate limiting, pause/resume, retry, unified DLQ with reason-codes (ADR-0017). twg-observability: OTLP, Prometheus, the health tree and /stats (ADR-0031/0032/0039). The modular sink lane lands: twg-sink-object-store (Parquet raw lane, primary-raw-eligible, ADR-0044), twg-sink-arrow-ipc, twg-sink-flight, plus twg-source-flight/twg-source-arrow-ipc. twg-kafka-client is retrofitted onto connector-core here — a transport-neutral abstraction designed against one transport is a guess, and the retrofit turns Kafka into the reference implementation Pulsar is measured against; if the traits prove Kafka-shaped, Phase 8 pays to fix them.

The offset-correlation contract is settled here and leaks everywhere downstream: each batch carries (topic, partition, offset_range) in a sidecar OffsetSpan, uniform across all four sinks (ADR-0004).

Config validation refuses more than one sink until Phase 5b lands — an honest constraint, not an unimplemented feature.

Exit (5a): twg-kafka-arrow end-to-end in all three representations, commit-after-successful-write demonstrated across a forced restart for each; Kafka source and sink through connector-core with OTel traces crossing the broker; a backpressure test showing a slow downstream throttling fetch rather than buffering unboundedly; /ready gates on source + primary-raw only.

Phase 5b — Coverage, recovery & multi-sink (4 weeks)

twg-offset-store with gap-aware covered ranges stored with raw (ADR-0053), the durable tier chosen by where raw lands, the local write-behind buffer. Bounded windowed replay from raw into a lagging sink, deduplicated by deterministic record id, with opt-in content dedup gated on a contract natural key (ADR-0054). Recovery as a deployment role (ingest/recovery/both) with leased windows and independent scale-to-zero (ADR-0050). This is the capability gate that unlocks the second sink: a non-raw sink that fails after the source is acked has permanently missed data, and the coverage store is the only record of what it missed (ADR-0057).

Exit (5b): a two-sink topology where one sink is failed mid-run, its gaps recorded, and a recovery pass restores it to parity without touching the source; a recovery-role pod scales from zero, drains a backfill window, and returns to zero.

Phase 6 — Producer, idempotence, transactions (6 weeks)

Producer with batching, accepting Bytes / Value / RecordBatch from the outset. Idempotent producer with producer-id/epoch and sequence numbers; transactional producer with the full InitProducerId/AddPartitionsToTxn/EndTxn cycle; exactly-once consume-transform-produce.

Exit: transactional throughput within a defined margin of the rdkafka baseline (margin recorded as a Criterion gate); fault-injection shows no duplicate commits across coordinator failover; all three input representations covered.

Phase 7 — Ecosystem breadth (10 weeks, gated)

No longer “ongoing” — it has an exit gate. twg-codec-protobuf (both surfaces; decode strategy per ADR-0046, still open — see Open decisions), twg-codec-json wrapping arrow-json, twg-codec-custom (batch-oriented WASM decode, ADR-0025), twg-codec-xml. OAUTHBEARER token providers as separate crates; GSSAPI behind the documented impurity flag. The database sinks twg-sink-postgres and twg-sink-zerobus.

AWS block (delivered together, ~5 weeks). SigV4 auth, Glue Schema Registry (its own WireFormat: header byte + 16-byte UUID) and Glue Data Catalog land as one unit — jointly testable since Glue and MSK share a credential path. The Phase 2 trait split means Glue Schema Registry is additive; the Data Catalog is twg-table-catalog, kept separate from schema-registry because a message-schema registry and a table catalog answer different questions.

Exit: Protobuf decodes a Confluent message-index-prefixed topic to RecordBatch byte-identical to a reference; JSON and XML corpora round-trip all three representations; a Postgres and a Zerobus sink each pass commit-after-write across a restart; the AWS block resolves and registers against a scratch Glue catalog.

Phase 8 — Pulsar (10 weeks)

A fresh implementation of the Pulsar binary protocol, informed by pulsar-rs but not derived from it (ADR-0010) — reusing twg-wire-* and connector-core directly, which a fork would duplicate. The specific defects it must not reproduce: implicit-Stream backpressure (here Budget + governor), seek that destroys and recreates the consumer (here in-place seek, ADR-0049), and partition subscription that forfeits discovery (here discovery is kept and intersected with an explicit selector for StatefulSet ordinals). Subscription modes, cumulative/negative ack, and Pulsar’s compression codecs — so twg-wire-compression gains any missing variants and Kafka inherits them. Flow control per ADR-0049: staged subscription, topic-spread shedding, pod-seeded order, pressure-as-scale-signal. twg-pulsar-arrow composes it with the codecs and sinks exactly as kafka-arrow does.

Exit: twg-codec-avro used by pulsar-arrow with zero changes; identical OTel metric names from both transports against one dashboard; the fidelity corpus adapted to Pulsar’s field model passes; a Kubernetes deployment consumes a partition subset and picks up a partition-count increase without restart.

Phase 9 — Transform layer, batch & packaging (12 weeks)

Sequenced after Phase 5b (the unnest locator contract interacts with the coverage store), can run parallel to Phase 6/7 given separate ownership.

twg-pipeline-sql (6 weeks). DataFusion TableProvider; the node DAG with automatic materialisation; interior sinks; unnest-aware locator propagation; Budget reduction across the sink set with per-node queue-depth metrics. DQ enforcement (twg-dq-enforcer) and the audit sink (twg-dq-audit) run inside the DAG (ADR-0014/0034). twg-pipeline-wasm + twg-ffi (3 weeks). wasmtime host, Component Model / WASI-P2, UDF registration, the Arrow C Data Interface boundary. Fuel/memory ceilings, no ambient capability. twg-pipeline-flight (2 weeks). Flight client for sidecar transforms with timeout, circuit breaker, connection pool; failures route through connector-core’s retry/DLQ. Batch & packaging (1 week). twg-source-batch + twg batch; static musl build for arm64/amd64; distroless OCI; multi-arch manifest.

Exit: a three-level DAG (raw sink, unpacked node with DQ leaves, nested unnest with its own sink) runs end to end with correct commit semantics across a forced restart; a user-supplied WASM UDF loads without rebuilding the binary and is fuel-limited; a stalled leaf is diagnosable from metrics alone; a batch file runs the same pipeline to the same sinks; the OCI image runs unmodified locally and in Kubernetes on both architectures.

Table formats (folded into Phases 7 & 9)

twg-format-parquet is required by the Phase 5a raw lane. twg-format-iceberg (read+write, Delta-via-Iceberg, ADR-0022/0023), catalog credential vending (ADR-0024) and descriptive metadata (ADR-0037) attach to the Phase 7 AWS/catalog block and are exercised by the Phase 9 sink DAG. twg-source-delta-sharing lands with the table-format work.

Timeline

%% name: delivery-gantt
gantt
    dateFormat YYYY-MM-DD
    axisFormat %b
    section Foundation
    P0 Scaffold + CI            :p0, 2026-08-03, 2w
    section Parallel tracks
    P1 Transport primitives     :p1a, after p0, 5w
    P1 Value model + core       :crit, p1b, after p0, 5w
    section Sequential
    P2 Avro + registry + contract:p2, after p1b, 6w
    P3 Consumer + passthrough   :milestone, p3, after p1a, 5w
    P4 Consumer groups          :crit, p4, after p3, 8w
    P5a Sinks + connector + obs  :p5, after p2, 7w
    P5b Coverage + recovery     :crit, p5b, after p5, 4w
    P6 Producer + txns          :p6, after p5b, 6w
    section Breadth
    P7 Codecs + auth + AWS/Glue :p7, after p5, 10w
    P8 Pulsar                   :crit, p8, after p6, 10w
    section Transform
    P9 SQL DAG + DQ + WASM + batch:p9, after p5b, 12w

Three critical paths: the Value model in Phase 1 gates every codec; Phase 4 gates the producer; connector-core in Phase 5a gates Pulsar. Passthrough ingest ships at end of Phase 3, Arrow ingest at end of Phase 5a, multi-sink + recovery at end of Phase 5b. Phase 8 runs after Phase 6 so Pulsar validates the layering before the ecosystem widens further; the AWS block can slip past it without consequence.

Quality gates

Standard gates apply — 85% per-file coverage ratchet, file-length and complexity caps, SSH-signed commits, git hooks mirroring CI. unsafe_code = "forbid" workspace-wide with the single audited twg-ffi exception (ADR-0043). These gates defend the ADR’s structural claims, since all erode silently under delivery pressure:

Layering enforcement. CI parses cargo metadata and fails if any twg-codec-* crate has a path dependency on twg-kafka-*/twg-wire-*, or if twg-codec-* or twg-wire-sasl depend transitively on tokio/async-std/smol.

No-Arrow build matrix. Every twg-codec-* crate is built and tested with --no-default-features, checking arrow appears nowhere in the resulting tree.

Arrow version discipline. cargo-public-api fails if a twg-codec-* crate exposes an arrow:: type not routed through twg-codec-core’s re-export.

Passthrough byte-identity. Record bytes consumed via RawSink are identical to what the broker sent, envelope included, compared after decompression.

Round-trip fidelity. The edge-value corpus and the proptest generator both run against all three representations on every commit; a field added to RawRecord without a corpus entry fails the build.

Reserved-namespace collision. A payload schema containing a field under the configured metadata prefix (default _twg_) is a hard decode error (ADR-0051).

DAG correctness under cardinality change. An unnest node’s locator union covers exactly the input records whose derived rows landed; commit-after-write never acknowledges a record with unlanded descendants (ADR-0004).

WASM sandbox capability. A loaded module cannot open a socket, read the filesystem, or exceed its fuel and memory ceilings (ADR-0008).

Position-never-leads. A test asserts the source is never acked ahead of durable primary-raw, and coverage never records a write that did not happen (ADR-0038/0053).

Contract merge is tightest-wins. A manual rule and a contract rule on one field resolve to the more restrictive; neither can loosen the other (ADR-0016/0036).

DQ audit completeness. Every record carries a durable per-rule verdict stamped with rule version; a rule change does not retro-alter prior verdicts (ADR-0034).

Dedup is contract-gated. Enabling content dedup without a declared natural key is refused at validation, not warned (ADR-0054).

Health readiness scope. /ready returns ready only when source + primary-raw are healthy, regardless of secondary-sink lag (ADR-0031).

Static-binary and multi-arch build. CI produces and smoke-tests static musl binaries and OCI images for arm64 and amd64 on every release; a dynamically-linked artefact fails the build.

Cross-transport metric parity. twg-kafka-client and twg-pulsar-client emit identical OTel metric names and label sets for the shared connector-level quantities (ADR-0032).

Codec reuse proof. From Phase 8, CI builds twg-pulsar-arrow against the unmodified twg-codec-* crates; any codec change made on behalf of Pulsar is a layering violation and fails review.

Purity audit. A cargo-deny rule fails on any C-linking dependency reaching the default feature set; zstd-encode and GSSAPI are exempted by name only, so a third exception requires a reviewed change to the deny list.

Risks

Open decisions

Genuinely open, not merely undrafted — the plan carries these as open rather than presenting them settled:

• F1 — Protobuf decode strategy (ADR-0046, Open). Direction chosen (selectively vendor a zero-copy parser, Apache-2.0 cleared), but gated on a benchmark against prost-reflect on our hardware and schemas. Phase 7’s Protobuf estimate is provisional until this resolves.
• F2 — Coordination substrate shape (ADR-0055, deferred). The leading shape is an embedded replicated store with an off-the-shelf consensus library; not yet chosen. Only distributed recovery waits on it.
• F3 — Arrow escape-path measurements (ADR-0003, Accepted). The pin (DataFusion’s Arrow major, no bridge) is decided; the IPC-vs-C-Data-Interface crossing cost and the SQL-vs-projection traffic split are designed but unmeasured, carried as open follow-ups rather than settled.
• F4 — connector-core trait shape (ADR-0006, Planned). Provisional until the Phase 5a Kafka retrofit and the paper review against Pulsar.
• F5 — Pulsar in-place seek feasibility (ADR-0049). In-place is the goal; destroy-and-recreate is the hedged fallback behind a flag.

Document version history

Sync note. The user-facing overview at docs/site/index.html maintains a summary view of this plan (its §09 Delivery-plan table, capability matrix, and layer diagram). This revision is aligned to that view. The index is now 58 ADRs (0001–0057 plus the ADR-0058 metadata amendment authored with plan v0.7): exactly one genuinely open (ADR-0046), one Accepted (ADR-0003), and one Draft (ADR-0058); the rest are decided and awaiting drafting. The site counts are corrected to match. A per-decision-vs-plan reconciliation is recorded in plan-reconciliation.md.

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.

Batch Reader Library Selection

Evaluated per format on four axes: pure Rust, maintained, decodes straight to Arrow, and version-coupling risk. “Pure Rust” is not one verdict — it splits per format, and two formats cannot be arrow-native by nature.

Why not Polars

The predecessor system used Polars for batch reading and had to vendor patches to compile on stable Rust, because polars-ooc/polars-stream depended on nightly-only APIs. The selection above removes the need for Polars entirely: parquet + arrow-csv + arrow-json cover the arrow-native formats and version in lockstep with the workspace Arrow through the codec-core re-export; calamine covers Excel; quick-xml plus our mapping covers XML. Every reader is pure Rust with no C dependency. Dropping Polars is not a preference — it removes the specific version-coupling failure the predecessor documented.

The two non-arrow-native readers

Excel (calamine) yields cell-by-cell values, not Arrow, so it needs a row→Arrow assembly step the others don’t. Excel’s loose typing (dates as serial floats, numbers-as-text) is a classic mapping minefield, so the cell→Arrow mapping is governed by type-map and the date-serial gotcha is fixed once, with a regression test.

XML splits into parsing (quick-xml — pure Rust, streaming, not in question) and mapping XML→Arrow, which is genuinely bespoke and ours. XML→Arrow is not a solved library problem the way Parquet→Arrow is: attributes vs elements, mixed content, repeated-elements-as-lists, namespaces and optional-everything make the mapping opinionated. codec-xml (streaming payloads) and the batch XML reader share one mapping core, as Avro registry and OCF share codec-avro.

Contract schema beats inference

CSV, JSON and XML inference is lossy — inherently for XML (everything is a string until typed), and sampling-based for CSV/JSON. Where a contract (ODCS) or XSD supplies a schema, batch reads use it and skip inference; inference is the fallback only when no schema is available. This makes the reads both safer and faster, and is another place contract-core pays for itself. Stated once as a cross-format source-batch rule (ADR-0035).

Configuration & GitOps

Thalweg is configured by TOML, layered TOML < env < CLI, validated at startup so nonsensical combinations fail to load rather than failing at runtime.

TOML is the source of truth; K8s only carries it

Kubernetes-native config is YAML, and a ConfigMap is a map[string]string. Rather than translate TOML↔YAML (which forces byte-stable round-tripping and leaks type and comment fidelity), Thalweg never lets K8s interpret the config — it only carries it. There is nothing to convert, so GitOps diffs stay clean: a config change is a one-line TOML diff, not a re-rendered YAML tree.

Two delivery modes, file-mount is the default:

File-mount (default). The ConfigMap holds the TOML (and each transform SQL) verbatim under its own key, mounted as a volume so it lands as a real file at a known path. The app loads it exactly as it loads a local core.toml — the same pattern used for transform SQL files.

Pointer (option). The ConfigMap holds a reference — an object-store URL or a path — and the TOML is fetched at startup via the shared resource loader. This decouples config lifecycle from deployment lifecycle (change config without a redeploy) at the cost of a boot-time fetch dependency; if the target is unreachable at startup, the process fails fast rather than starting on stale or absent config.

One loader for config, contracts, and SQL

Config (pointer mode), ODCS contracts, and transform SQL all resolve through twg-resource-loader: one local | http | object-store abstraction with one set of caching, retry, and fail-fast semantics. A resource is loaded and validated at startup; remote resources may opt into a refresh interval.

Secret enumeration

config secrets reads a resolved config and lists the secret references it requires, in text, json, or names form. It exists so a GitOps pipeline (for example ArgoCD plus a secrets operator) can provision exactly what a deployment needs without hand-maintaining a parallel list. It is pure inspection — it reads config and reports; it never resolves secret values.

This is the one piece of v2-era ConfigMap tooling worth keeping: it has nothing to do with TOML↔YAML translation, so dropping the conversion machinery does not cost it.

Deploying the documentation site

The site lives at docs/site/ and deploys to Cloudflare Pages.

Why not GitHub Pages

GitHub Pages does not serve from a private repository on the free tier. On paid tiers it will — but the published site is public regardless; only Enterprise offers access-controlled Pages. For a private repository whose docs should stay private, GitHub Pages cannot do the job at any tier below Enterprise.

Cloudflare Pages serves from a private repository, and Cloudflare Access can put the site behind SSO on the free Zero Trust tier. A private repo gets private docs.

How it deploys

Cloudflare’s GitHub App watches the repository and rebuilds on every push to main that touches the site. There is no deployment workflow in this repository and there should not be one — a second deploy path would race the first.

Project settings (Cloudflare dashboard → Workers & Pages → the project):

The ordering consequence, which matters

Cloudflare deploys independently of GitHub Actions. Its app reacts to the push directly, so the checks in docs.yml run alongside the deploy rather than before it.

That is fine for the correctness gates — a stale count merged and deployed is embarrassing, not dangerous. It matters for check-public-surface.sh, which exists to stop a credential or an internal identifier reaching a world-readable site. On a direct push to main, that scan runs while the content is already live.

So the ordering must come from branch protection rather than from workflow sequencing. Require pull requests into main, and make the docs checks required status checks. Then nothing reaches main without passing, and Cloudflare only ever sees a branch that has passed — the ordering is restored, just enforced a step earlier.

Without branch protection the gates are advisory for the published site. That is a legitimate choice for a small trusted team, but it should be a choice rather than an assumption, because the failure it stops is the one that cannot be undone by reverting: a secret that was briefly public has been public.

Visibility: public, deliberately

The site is published publicly. Cloudflare Access is deliberately not applied.

The repository stays private; the documentation does not. The site holds architecture, decisions and interfaces — no production identifiers, no credentials, no data, no commercial terms — and that is a property the hard rule in AGENTS.md exists to maintain rather than a happy accident.

It also resolves a coupling that would otherwise bite at first publish: crates are published to crates.io publicly and their READMEs link here, so Access-protected docs would strand every reader arriving from a crate listing.

What this changes

The sensitive-information rule is no longer precautionary. It is the control on a world-readable surface, and anything merged under docs/site/ is public on the next deploy. scripts/check-public-surface.sh runs in the Docs CI checks (docs.yml) as a backstop for the mechanical classes of mistake — credential shapes, internal system names, workspace hostnames, connection strings, private addresses. It is not a guarantee: it cannot catch a plainly-worded sentence naming something confidential, and the rule remains the actual control.

If this needs to change

Cloudflare dashboard → Zero Trust → Access → Applications → Add an application → Self-hosted, pointed at docs.thalweg.dev, with an email-domain or identity- provider policy. The free tier covers 50 users. Note that applying it strands crates.io readers unless a public landing page exists at the apex for the crate READMEs to point at instead.

Custom domain: docs.thalweg.dev

Cloudflare dashboard → the thalweg-docs Pages project → Custom domains → Set up a custom domain → docs.thalweg.dev. If the DNS is hosted by Cloudflare the record is created automatically; otherwise add a CNAME pointing at the project’s pages.dev hostname.

The apex thalweg.dev is deliberately left free for a future public landing page.

.dev is on the HSTS preload list, so browsers require HTTPS before the first request is sent. Cloudflare provisions the certificate automatically; there is nothing to configure, but a deploy is not reachable until the certificate is issued, which is usually a minute or two rather than instant.

Why a subdomain rather than the apex

Cloudflare Access is applied per-hostname. Putting the docs at the apex would put the entire domain behind whatever policy the docs need, leaving no room for a public landing page later without moving the docs — and moving them once links exist is the expensive version. A subdomain costs nothing now and keeps the apex free.

If the project goes public, nothing moves: Access comes off docs.thalweg.dev and the URL keeps working.

The coupling with publishing crates

Published crates are public on crates.io, and their READMEs link to docs.thalweg.dev. If the docs sit behind Access, a reader arriving from crates.io hits an authentication wall — and the repository is private too, so there is nothing else for them to reach.

Resolve this before the first publish, not after. Either:

• the docs are public and Access stays off — defensible, since the site holds architecture and decisions with no production identifiers, credentials or data, which the AGENTS.md rule exists to guarantee; or
• a public landing page goes at the apex and the crate READMEs point there, with docs.thalweg.dev staying behind Access; or
• crates are not published while the documentation is private.

Publishing public crates that link to private documentation is the one combination to avoid: it produces crates.io listings nobody can follow up on.

Validation and publishing

There is no deploy workflow in this repository. Publishing is Cloudflare’s GitHub App reacting to the push; validation is the Docs CI checks (docs.yml), run as required status checks on the pull request so nothing reaches main unvalidated:

• every HTML file parses — a broken page should not merge;
• the site’s crate list matches the workspace manifest — stale documentation should not merge either, which is the failure more likely to go unnoticed;
• check-public-surface.sh scans docs/site/ for sensitive content.

Once merged, Cloudflare’s app builds and publishes automatically. Deployments are immutable and Cloudflare keeps history, so rolling back is selecting a previous deployment in the dashboard.

The repository map (graphify)

The site also serves an interactive repository map at docs.thalweg.dev/graphify/, generated with graphify.com. It follows the same single-source-of-truth discipline as the rest of the site.

graphify-out/ is canonical. It lives at the repository root and is regenerated by graphify’s git hooks — never hand-edited. It holds the full graph plus caches, cost/telemetry and manifests, none of which belong on a public site.

docs/site/graphify/ is DERIVED — never hand-edit it. It is a published copy of exactly three world-safe files, produced by scripts/sync-graphify-site.sh: graph.html → index.html, graph.json, and GRAPH_REPORT.md. The script copies nothing else — no cache/, no cost.json, no manifest.json, no .graphify_* — so telemetry and internal metadata never reach the public surface.

The install: this repository uses tracked hooks. Run once, locally:

git config core.hooksPath .githooks

This supersedes graphify’s locally-installed .git/hooks/ post-commit and post-checkout — their rebuild machinery (Python discovery via graphify-out/.graphify_python, scan root via graphify-out/.graphify_root, the GRAPHIFY_OUT default, and the graphify.watch._rebuild_code call) is ported into .githooks/. .githooks/post-checkout keeps the detached branch-switch rebuild; there is deliberately no .githooks/post-commit, because the synchronous .githooks/pre-commit now owns rebuild-and-publish and a post-commit rebuild would be redundant.

.githooks/pre-commit does a synchronous, incremental rebuild of just the staged changes, then runs scripts/sync-graphify-site.sh to publish and stage the three files — so the map lands in the same commit as the changes it reflects. It is fail-soft: a graphify error or a graphify that is not installed prints a non-fatal warning and the commit proceeds, publishing the last-good map. A missing or broken graph never blocks a commit.

Rendered documentation: rustdoc and mdBook (ADR-0059)

Two reference surfaces are rendered locally and committed into docs/site/, so Cloudflare serves them verbatim and the build command stays NONE. They are decided in ADR-0059.

The crates are unpublished (ADR-0047), so docs.rs cannot carry them — self-hosting is what lets the API reference exist at all, and it covers the four permanently internal crates (twg-cli, the two *-arrow compositions, twg-e2e) too.

Navigation is generated, not hand-maintained. gen-reference-book.sh mirrors the docs tree into the book’s src/ (frontmatter stripped as a pre-render step so the ADR/blueprint YAML does not show as raw text), regenerates SUMMARY.md, and builds into docs/site/reference/ (build-dir in docs/book/book.toml). Drop a markdown file into the right folder and it appears on the site with no nav edit. docs/book/src/ is derived and git-ignored; only the rendered output is committed.

Regeneration is wired into the pre-commit hook. .githooks/pre-commit (which already rebuilds the graphify map) additionally, on each commit: regenerates the API docs for any changed crates/<name>/**, and rebuilds the reference book when any docs/** markdown changed — staging the output so it lands in the same commit. Every step is fail-soft: a missing cargo/mdbook or any error warns and the commit proceeds. This requires the tracked hooks to be active:

git config core.hooksPath .githooks

Prerequisites for local regeneration: the Rust toolchain (for cargo doc) and mdBook (cargo install mdbook). Without them the hook fail-soft no-ops and the rendered trees can go stale until someone with the tooling commits.

The generated search indexes are the one wrinkle for check-public-surface.sh: rustdoc’s search.index/ and mdBook’s searchindex.js are minified blobs, and rustdoc’s matches the AWS-key shape by coincidence. The gate excludes only those search-index artifacts, never the trees around them, so every rendered HTML page is still scanned for secrets.

Limits

The free tier allows 500 builds per month and 20,000 files per deployment, with unlimited bandwidth. The hand-authored landing page plus the graphify map is a handful of files; the rendered rustdoc and mdBook trees add a few thousand more — still an order of magnitude under the 20,000-file ceiling, but a number to watch as the crates gain real API surface. Run find docs/site -type f | wc -l to check.

Health & Observability

One crate, twg-observability, owns three surfaces that share an HTTP server and read from the same runtime signals: Prometheus metrics, OTEL/OTLP tracing, and health.

Health is a tree with three projections

Every stage reports into one registry — sources, decode, contract/DQ, transform DAG nodes, each sink, the recovery driver, the offset store. Each reports a state and the numbers behind it:

• ok — operating normally.
• degraded — working but impaired: a circuit breaker half-open, backpressure at Warning/Critical, elevated reject rate, rising commit latency. Visible and actionable by a human; not automatically acted upon.
• down — the stage is not functioning: source disconnected, sink circuit open, backpressure at Stuck.

Backpressure is not a separate signal. It is the connector-core four-state machine (Normal → Warning → Critical → Stuck) projected per stage, so the “growing backpressure” you want to see is the state machine’s state surfaced on each stage. Per-sink circuit-breaker state feeds sink stage health the same way.

Why three endpoints for one tree

Kubernetes probes read only the HTTP status code, never the body. The rich per-stage JSON is therefore invisible to the kubelet by design — which is what lets /health be richly honest without that honesty getting the pod killed.

curl -s /health | head -1 (or the emitted summary line) gives the overall code and a one-line story; the body underneath gives the per-stage breakdown so an operator can see immediately which stage is offline or building backpressure.

What gates what

Liveness (/live) fails only on genuinely unrecoverable in-process state. Never on a saturated sink or a rebalance — restarting does not fix a data-plane problem and wiring those in causes restart loops.

Readiness (/ready) is gated by exactly two things: source health and the primary-raw sink. If a source is down or raw cannot be written, the pod is not ready — because if raw cannot land, the source cannot safely acknowledge. Every other stage is display-only: a degraded secondary sink, a backed-up quarantine lane, a slow transform node, or lagging recovery all show in /health but never drop readiness.

Extension path, not built yet: a per-stage ignore | degrade | fail readiness policy — the monitoring twin of the sink block | isolate failure policy — for when some stage beyond source/primary-raw should gate readiness. Not needed on day one; readiness has two inputs until proven otherwise.

Metrics and tracing

Metrics export as Prometheus scrape (/metrics, disableable via config) and OTLP push (default destination Zerobus, any OTLP endpoint supported). Tracing follows the record lifecycle with context propagated through message headers, so a trace crosses the broker rather than stopping at it. Metric names and labels are defined in this crate so both transports emit identically — one dashboard works against either.

Live per-stage memory & throughput (/stats)

Beyond the continuous OTEL gauges, GET /stats serves an in-process 30-second rolling digest: per-stage throughput and average per-record / per-batch memory, as JSON. The two are not redundant — the gauges feed dashboards and alerting (Grafana computes its own windows); /stats is for the incident where you are on the pod and want it to tell you directly where its memory went in the last 30s, without depending on a metrics backend that may itself be degraded.

Per-stage memory is reported by honest attribution — queue depth × bytes, batch count × batch size, in-flight bytes — never by carving global RSS into per-stage fictions, since stages share an allocator.

The same live measurement feeds admission control: the budget uses the measured recent per-record size (p95 within the window, so a size spike does not blow the budget before a mean would catch up), with the record_size_est_bytes config as cold-start seed and floor. The static estimate self-corrects as real sizes are observed — an improvement over a fixed guess that is wrong the moment sizes drift.

Running the protobuf decode benchmark

crates/twg-codec-protobuf/benches/decode.rs is the measurement gating ADR-0046. It does not run today — every arm is a todo!(), because there is nothing behind it yet. This is the runbook for getting from here to a number.

Do this first: check the premise, outside the workspace

Before implementing anything, answer the question the whole plan rests on: does a zero-copy descriptor-driven parser actually beat prost-reflect by enough, on schemas like yours, to justify vendoring and maintaining a divergence?

That needs none of our code. In a scratch crate:

cargo new --lib pbbench && cd pbbench
# dev-deps: criterion, prost, prost-types, prost-reflect,
#           and the upstream SDK with its parser feature enabled
protoc --include_imports \
       --descriptor_set_out=wide.pb \
       -I ../thalweg/crates/twg-codec-protobuf/proto \
       ../thalweg/crates/twg-codec-protobuf/proto/wide_128.proto
cargo bench

Decode the same payload both ways and compare. An afternoon’s work that de-risks weeks: if zero-copy does not win decisively on a 128-field schema, the vendoring plan changes and you have saved the implementation.

Use wide_128.proto. A narrow schema flatters reflection-based decoding and will understate the gap enough to talk you out of the right answer.

What the in-repo benchmark needs, in order

1. Dev-dependencies in crates/twg-codec-protobuf/Cargo.toml: criterion, prost, prost-types, prost-reflect, and the upstream SDK behind its parser feature. The layering gate exempts [dev-dependencies] by design — a vendor SDK there is legitimate and ships to nobody; the same crate in [dependencies] would fail CI, correctly.
2. Compiled descriptors. The .proto fixtures exist in proto/. A build script or a committed artefact must turn them into a FileDescriptorSet:

   protoc --include_imports --descriptor_set_out=schemas.pb -I proto proto/*.proto
   

   --include_imports is not optional — without it the well-known types are absent and wellknown.proto fails to resolve, which is the very failure mode being tested.
3. Payload generation at the three benchmark sizes, deterministic and seeded so runs are comparable across machines and over time.
4. The four decode paths, replacing the todo!()s: prost-reflect (the floor), the upstream parser (the baseline to beat), and our two candidates — parse-then-walk and fused-into-builders.

Running it

cargo bench -p twg-codec-protobuf
cargo bench -p twg-codec-protobuf -- decode/wide_128   # the deciding case

Criterion writes to target/criterion/. Commit baselines once the numbers are trusted, and wire a CI failure on regression beyond a stated tolerance — a benchmark that gates a decision belongs in CI, not in a document (docs/testing/TESTING.md).

Reading the result

Two independent questions, deliberately measured separately:

Is vendoring worth it? Compare the upstream parser against prost-reflect on wide_128. If the margin is large, vendoring is justified; if not, take prost-reflect and drop the whole plan.

Is fusing worth it? Compare ours_fused against ours_parse_then_walk. If fusing does not beat walking by a real margin, drop the fusion and keep the simpler path — vendoring can still be right while fusion is not.

Bear in mind that fusing cannot eliminate per-record staging: protobuf fields arrive in arbitrary order and repeated fields interleave, so absent fields are only known at end of record. The upside is bounded, which is exactly why it is measured rather than assumed.

E2E Scenarios & Regression Catalogue

The authoritative list behind twg-e2e. Scenarios prove the system does what it should; regressions prove it never again does what it once did wrong.

Scenarios

Regressions (learned the hard way)