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.