fix(logrepl): stop emitting unchanged TOASTed columns as NULL

[devarismeroxa]

Summary

First tests/chaos harness in the repo (v0.19 Workstream 7 / DBZ-1), per
docs/design-documents/20260722-debezium-compete-roadmap.md and the DBZ-1
implementation plan. Two independent deliverables:

1. SIGKILL crash-safety test on the source connector's offset↔position
   bridge (pkg/connector.Source.Ack / pkg/connector.Persister), mid-snapshot
   and mid-stream. Verdict: the ack-before-persist window is a real,
   structural GAP against a pruning (Postgres-replication-slot-like) upstream —
   not a hypothetical one. Against a durable (Kafka-like) upstream the exact
   same crash window in the exact same engine code produces only a benign
   duplicate. Per instructions, Source.Ack's ordering is not changed in
   this PR — this is the demonstration + escalation.
2. Engine-side FIFO/in-order ack test, closing #2672: a real
   conduit-connector-protocol v1 gRPC client (bufconn, no mock) proves
   funnel.Worker.Ack (full batch, worker.go:493) and Worker.Nack's
   partial-ack path (worker.go:513) both deliver acks to a v1-protocol plugin
   as individually ordered messages, matching input order.

Risk tier: Tier 1 (data path — ack/position/checkpoint logic). Design doc:
the DBZ-1 implementation plan (pre-code, dated 2026-07-23). Requires DeVaris
Tier-1 sign-off (wk4). Do not merge without it.

The SIGKILL verdict — gap vs. duplicate, and why it's conditional

pkg/connector/source.go:207-238 (Source.Ack) sends the ack to the plugin
(stream.Send, line 218) — which for a real connector like the Debezium/Kafka
Connect wrapper drives task.commitRecord/task.commit, advancing Postgres's
replication-slot confirmed LSN — before persisting its own position
(persister.Persist, lines 223-235). Persister.Persist
(persister.go:137-170) debounces: it batches and flushes asynchronously after
DefaultPersisterDelayThreshold (1s) or 10k items. So there is a real ~1s (or
10k-record) window where Conduit has told a plugin "you may durably commit"
before Conduit's own bookkeeping has reached durable storage.

This PR's chaos test targets exactly that window with two scenarios:

• mid-snapshot: a fast burst (no artificial pacing), killed ~30ms in — well
  before any automatic flush, so Conduit has persisted nothing when the kill
  lands (models Debezium's initial full-table snapshot).
• mid-stream: steady pacing, killed ~1.4s in — one automatic flush has
  already landed (a valid but stale checkpoint), and the kill lands inside the
  next debounce window before its flush fires.

Since the real conduit-kafka-connect-wrapper (Debezium/Postgres) isn't
available in this repo (separate JVM repo, not a Go module dependency here),
the test drives the real engine code (pkg/connector.Source, Persister,
real on-disk badger DB — same backend production uses) against a synthetic
upstreamStore with a prune toggle modeling the one behavior that actually
decides the outcome: can the plugin's backing system still supply data from
before its own last commit?

• prune=false (Kafka-like): same crash window, same engine code — restart
  resumes from the stale-but-valid checkpoint, re-delivers a few already-committed
  positions as harmless duplicates, then completes delivery through total.
  No gap, ever.

• prune=true (Postgres-slot-like: WAL segments recycled once confirmed):
  identical crash window, identical engine code — restart asks to resume
  from behind the already-committed/pruned watermark. chaosPlugin.Open
  surfaces this as a loud, structural error (modeling Postgres's real
  "requested WAL segment has already been removed", not a silent skip) in
  both the mid-snapshot and mid-stream variants. Confirmed by the test:

  SEV-0 FINDING confirmed for mid-snapshot: OPEN_GAP_ERROR: could not open source
  connector plugin: GAP: chaos upstream already committed/pruned through position 30,
  but Conduit asked to resume from position 0 — the 30 position(s) in between are no
  longer available upstream
  
  SEV-0 FINDING confirmed for mid-stream: OPEN_GAP_ERROR: could not open source
  connector plugin: GAP: chaos upstream already committed/pruned through position 95,
  but Conduit asked to resume from position 64 — the 31 position(s) in between are no
  longer available upstream
  

🔴 SEV-0 ESCALATION: Source.Ack's ack-before-persist ordering is not
crash-safe when the connected plugin's commit is irreversible from Conduit's
point of view (real Postgres replication slots included). This is a design
characteristic of pkg/connector/source.go today, not something this plan
decides to change — per instructions, fixing the ordering (e.g. persisting
before acking, or some other reordering) is explicitly out of scope for this
PR: it touches Source.Ack's core sequencing for every connector and needs
its own separate Tier-1 design doc. This PR is the demonstrating test and the
loud writeup, not the fix.

Failure-mode analysis (crash at each step of ack→commit→persist)

Walking pkg/connector/source.go:207-238's sequence:

1. Crash before the ack is sent (record read, never acked). Nothing
   changed anywhere. Restart re-reads and reprocesses. Correct — a
   duplicate, consistent with at-least-once. Metric that would show it: none
   needed, no operator-visible effect. Recovery: automatic (resume from last
   persisted position).
2. Crash mid-stream.Send (partial batch delivered to the plugin). The
   plugin has committed a prefix of the batch; Conduit resumes from before the
   entire batch (its own persist for this batch never ran). Structurally the
   same risk as case 3 below, just with a partial batch.
3. Crash after the ack is sent, before persister.Persist's async flush
   completes — the window this PR's SIGKILL test targets. Demonstrated: gap
   (pruning upstream) or benign duplicate (durable upstream) depending entirely
   on the plugin's own retention behavior. Metric that would show it in
   production, today: none — there is no replication-slot-lag metric, no
   heartbeat-staleness signal, no ack-vs-persist-lag gauge exposed anywhere in
   Conduit today. A production instance of this exact crash window would be
   invisible until a downstream data-quality incident surfaced it. This is a
   real observability gap this workstream inherits, not fixes (named for
   DBZ-3/DBZ-4, Phase 2). Recovery: operational only — restore from the last
   known-good destination/position snapshot, or replay from an earlier LSN if
   the replication slot still retains it; there is no automated self-healing
   (correctly, per CLAUDE.md's no-distributed-snapshot-machinery scope
   discipline).
4. Crash during the persister's own flushNow transaction commit
   (torn-write risk, invariant 5). Verified, not assumed: badger's
   transactional commit is atomic at the KV layer (all-or-nothing), so a
   SIGKILL mid-commit cannot produce a torn/half-written position — confirmed
   empirically by the chaos test's restart path (loadOrCreateInstance in
   tests/chaos/child.go treats any store.Get error other than "key not
   found" as CORRUPT_POSITION and hard-fails; this marker never fired across
   all runs). Invariant 2's "no corrupted position" holds structurally here —
   invariant 1/3 (ack-before-durable-persist ordering) does not, per the
   finding above.
5. Crash after the flush completes. Fully consistent state; same as the
   happy-path boundary between batches.

Cross-pipeline blast radius (per instructions, flagged explicitly):
Persister.flushNow (persister.go:238-271) flushes one shared batch across
every connector currently pending a write — Persist is called per-connector
but batched into a single map[string]persistData keyed by connector ID,
and triggerFlush/flushNow operate on that whole map in one transaction. A
slow or blocked write for one connector's state (or the transaction itself)
delays the flush for every other connector's pending position update sharing
that same debounce cycle, process-wide. This chaos test only exercises a
single connector, so it does not itself demonstrate the blast radius — flagging
it here because the fix DeVaris may eventually decide on for the sev-0 above
(if it touches persist timing/ordering) needs to account for this shared-batch
behavior, not just the single-connector case this PR tests.

FIFO-ack test coverage (closes #2672)

pkg/lifecycle-poc/funnel/worker_ack_order_test.go — real
conduit-connector-protocol v1 gRPC client (not mocked) over an in-memory
bufconn listener, talking to a fake SourcePluginServer that records ack
arrival order:

• TestWorker_Ack_DeliversPositionsFIFOToV1Plugin: full-batch ack
  (worker.go:493) — 5 positions acked together, observed as 5 separate,
  strictly ordered gRPC messages.
• TestWorker_Nack_DeliversPartialAckFIFOToV1Plugin: partial-batch ack
  following a partial DLQ nack (worker.go:513) — simulates the DLQ
  destination successfully writing 3-of-5 nacked records; asserts exactly the
  3-position prefix is acked, in order, and the other 2 are not acked
  (would violate invariant 1 — acking a record neither delivered nor durably
  DLQ'd).

This closes #2672: the FIFO guarantee standalone connectors depend on was
previously upheld only by a comment in conduit-connector-protocol's v1
client, unasserted anywhere in this repo.

Adversarial self-review findings (resolved before this PR)

• Off-by-one / race in the chaos harness itself (not production code):
  Source.Ack only guarantees the ack was handed off to the plugin
  (stream.Send rendezvous), not that the plugin's subsequent commit had
  finished — the child process could os.Exit before the last commit's fsync
  completed, truncating the observed run by one position and producing a
  false-positive gap unrelated to the engine code under test. Fixed with
  waitForUpstreamCommitted before the graceful-exit path; also fixed a bug
  where its timeout case silently fell through to DONE instead of failing.
• Process-leak risk: an assertion failing between spawnChild and the
  test's own sigkill/waitExit call would have left an orphaned child
  process running. Fixed via an idempotent reap() (guarded by sync.Once,
  since cmd.Wait() may only be called once) plus a t.Cleanup-registered
  fallback kill+reap.
• gosec/noctx findings on the harness's exec.Command/os.ReadFile/
  os.MkdirAll calls are taint-analysis false positives (self-controlled
  re-exec path and t.TempDir()-scoped fixture paths, not external input) —
  addressed with targeted //nolint comments explaining why, not blanket
  suppression.
• Checked error paths in child.go line by line: every os.Exit path prints a
  distinguishable marker (CORRUPT_POSITION, OPEN_GAP_ERROR, FATAL,
  DONE) so the parent test's assertions are never guessing from a bare exit
  code.
• Concurrency: chaosPlugin's produce/ack goroutines communicate only via the
  in-memory stream's channels and the upstreamStore's own mutex; verified
  clean under -race across repeated runs (go test -race -count=3, no
  reports).

Gates

• make generate && git diff --exit-code: clean (no diffs; new files only).
• golangci-lint run scoped to changed packages (./tests/chaos/...,
  ./pkg/lifecycle-poc/funnel/...): 0 issues. (A full-repo
  golangci-lint run ./... hit lock contention from concurrent agent sessions
  sharing this environment's lint cache and could not be cleanly re-run in
  isolation before submission — the scoped, diff-relevant run is clean, which
  is what's gated on here; go vet ./... and go build ./... were run
  full-repo without contention, see below.)
• go build ./...: clean.
• go vet ./...: clean except one pre-existing, unrelated finding
  (pkg/lifecycle-poc/funnel/worker.go:473, predates this PR, inside the
  already-commented-out multi-connector TODO block).
• go test -race ./tests/chaos/... ./pkg/lifecycle-poc/funnel/... ./pkg/connector/...:
  all green, including 3 repeated full runs of the chaos suite (-count=3)
  with no flakes observed.
• Per the process-maturity table in CLAUDE.md: this is a normal CI test, not
  a scheduled nightly chaos job (that gate is Phase 2). Coverage-floor and
  benchi regression gates are not yet live (Phase 1) and are not claimed here.

Requires DeVaris Tier-1 sign-off (wk4)

Open question carried from the DBZ-1 plan, unresolved by this PR (as
instructed): does fixing Source.Ack's ack-before-persist ordering become
its own Tier-1 design doc, or is there context that changes that? This PR
takes no position beyond demonstrating the gap is real and structural — please
confirm the escalation path before or alongside sign-off.

DO NOT MERGE without explicit sign-off.

🤖 Generated with Claude Code

https://claude.ai/code/session_015GQFzakPShAYj8CcwajYDD