fix(engine): persist position before plugin ack — sev-0 Source.Ack ordering (Approach A)

[devarismeroxa]

Summary

Implements Approach A (ack-follows-durable-flush) for the sev-0 Source.Ack persist-ordering
bug, per DeVaris's sign-off on the approach in #2679's design doc
(docs/design-documents/20260723-source-ack-persist-ordering-fix.md) and postmortem
(docs/postmortems/20260723-source-ack-persist-ordering.md).

Source.Ack (pkg/connector/source.go) sent the ack to the plugin (stream.Send) before the
resulting position reached durable storage (persister.Persist only enqueues into an in-memory
batch, flushed asynchronously on a ~1s debounce). For a source backed by a pruning upstream (e.g. a
Postgres replication slot, whose ack advances confirmed_flush_lsn and frees WAL for recycling), a
crash in that window loses the durable position after the plugin was already told to commit — a
structural, unrecoverable gap. This was confirmed, not hypothetical: DBZ-1's chaos test (#2677)
drove this exact window and reproduced the gap on demand.

Supersedes #2677, folding in its chaos harness and flipping its central assertion from
"demonstrates the gap" to "asserts no gap."

The per-connector FIFO high-water-mark mechanism

Source.Ack no longer sends the plugin ack inline. It:

1. Updates Instance.State and appends (seq, positions) to a new pendingAcks FIFO queue
   (seq is a purely engine-internal, monotonically increasing counter — not derived from the
   opaque opencdc.Position bytes, which the engine cannot generically parse or compare; a
   Position's structure, e.g. per-partition offsets, is entirely connector-defined).
2. Registers persister.Persist(..., func(err) { s.onPersistFlushed(seq, err) }).

onPersistFlushed only sends the plugin ack once the persister confirms durability:

• On success, it advances durableAckSeq to max(durableAckSeq, seq) and drains pendingAcks
  from the front for every entry whose seq <= durableAckSeq, sending each in the exact order
  Ack originally queued them (invariant 4), then removes them from the queue.
• This is safe even when flush confirmations arrive out of order (a later-registered flush's
  transaction can finish before an earlier one still in flight, since Persister.flushNow's
  callback goroutines aren't ordered relative to each other): durableAckSeq only ever advances,
  and draining is idempotent — a stale, late callback finds nothing left to send.
• On failure, nothing is sent (invariant 1) and the error propagates via s.errs, same as before.

Graceful shutdown (invariant 7): Source.Teardown now forces a final persister.Flush and
waits for it (Persister.WaitPendingWrites, extended with a new callbackWg so it waits for the
callback — and therefore the deferred send — not just the store write) before canceling the
stream (stopStream). This ordering is load-bearing, not incidental: the in-memory stream (and a
real gRPC stream) shares one context between both directions, so canceling it first would make the
deferred ack's stream.Send race an already-closed ctx.Done() and very likely lose the message
instead of delivering it — I hit this as a real bug during implementation (see Adversarial
self-review below) before reordering fixed it.

Fails-without-fix verification

Per CLAUDE.md's "every bug fix ships with the test that would have caught it," I temporarily
reverted just the ordering change (reintroducing a synchronous stream.Send before
persister.Persist in Source.Ack, then restored it) and reran the new chaos assertion:

$ go test -race -v -run TestSIGKILL_PruningUpstream_NoGap -count=1 ./tests/chaos/...
=== RUN   TestSIGKILL_PruningUpstream_NoGap/mid-snapshot
    sigkill_test.go:181: not true: resumePos >= watermarkAtKill
=== RUN   TestSIGKILL_PruningUpstream_NoGap/mid-stream
    sigkill_test.go:181: not true: resumePos >= watermarkAtKill
--- FAIL: TestSIGKILL_PruningUpstream_NoGap (4.04s)

Red on both scenarios with the bug reintroduced; restored the fix and confirmed green
(diff against the pre-revert file showed byte-identical restoration).

Failure-mode analysis

Walking Source.Ack's sequence under Approach A (today's behavior in brackets):

1. Crash before the ack is enqueued. No change: nothing happened; restart re-reads/re-acks.
2. Crash mid-send of an already-durable deferred ack. Position is already on disk; restart
   resumes correctly. Plugin may or may not have received that specific message — benign duplicate
   if not. [today: this exact window is the sev-0, since the ack already went out before
   persistence.]
3. Crash after enqueue, before that batch's flush completes. No ack has been sent (deferred);
   restart resumes from the last durably-flushed position and reprocesses the gap as a duplicate,
   not a data-loss gap. [today: this is the exact scenario that produces the structural gap.]
4. Crash during flushNow's transaction commit. Unchanged, already verified atomic (badger's
   commit; chaos harness never observed CORRUPT_POSITION).
5. Crash after flush completes, before the deferred callback fires. Same as case 2: durable
   state is correct, ack is merely delayed to the next opportunity.
6. Per-partition/per-connector ordering (invariant 4) under interleaved out-of-order flush
   completion. Covered by the new unit test
   TestSource_OnPersistFlushed_OutOfOrderCompletionStillDeliversInOrder — simulates the
   highest-seq flush completing first and asserts all three positions are still delivered in
   order, with the two late/lower-seq callbacks arriving as safe no-ops.
7. Graceful (SIGTERM-style) vs. kill -9 shutdown. Covered by
   TestSource_Teardown_SendsPendingDeferredAckBeforeReturning: Teardown must not return until a
   pending ack has actually been sent. This is where I found and fixed the stream-context-ordering
   bug described above — without the fix, this specific test hung indefinitely (a real deadlock,
   not a flaky timing issue), because the deferred send raced an already-canceled stream context.
8. Cross-connector blast radius (inherited, not introduced). Persister.flushNow still flushes
   one shared batch across every connector pending in a debounce cycle; under this fix, a slow
   flush for one connector now also delays other connectors' deferred acks in the same cycle, not
   just their own write. Named in the design doc as inherited context, not a blocker.
9. MySQL/Mongo — not extended beyond what's proven here. See below.

What could this break / how would we know / rollback: the risk surface is entirely within
pkg/connector/{source,persister}.go; a regression would show up as either (a) the chaos suite
regressing to gap-producing (caught by TestSIGKILL_PruningUpstream_NoGap), (b) acks silently
never arriving at the plugin (caught by the FIFO-ack tests and would manifest as stalled upstream
commits / growing WAL in a real Postgres-slot deployment — no metric exists for this yet, tracked
as a follow-up per the design doc's Observability section), or (c) a deadlock on shutdown (caught
by TestSource_Teardown_SendsPendingDeferredAckBeforeReturning, which is exactly how I found the
stream-ordering bug in my own first draft). Rollback is a plain revert: no format/protocol change,
no migration.

MySQL/Mongo harness result

Per the design doc's explicit scope: this PR proves the fix gap-free against both a prune-class
synthetic upstream (TestSIGKILL_PruningUpstream_NoGap, modeling a Postgres replication slot) and a
retention-class synthetic upstream (TestSIGKILL_DurableUpstream_NoGap, modeling a durable,
replayable log such as Kafka) — both now pass with the identical assertion, proving the fix removes
the upstream-class-dependence entirely at the engine level. This does not verify the real MySQL
binlog or MongoDB oplog connectors (those don't exist yet per the Debezium-compete roadmap, DBZ-7)
— tracked as a required follow-up before claiming "fixed for all CDC," exactly as the design doc and
postmortem flag it. Not claiming more than this proves.

benchi / performance

• Committed, not run: benchi/ack-hot-path/ (config, pipeline, two tool variants for
  origin/main vs this branch, README). The sandbox this was authored in has benchi installable
  but no reachable Docker daemon, so the actual orchestrated run could not be executed — stated
  plainly, not glossed over.

• What I did run: pkg/connector/source_bench_test.go's BenchmarkSource_Ack, a same-sandbox
  Go microbenchmark of Source.Ack in isolation, at -benchtime=20000x -count=3, both on this
  branch and on the same temporary pre-fix revert used for the fails-without-fix check:

  # this branch (fix):      ~640-740 ns/op, 707 B/op, 6 allocs/op
  # reverted (pre-fix):     ~1180-1500 ns/op, 742-743 B/op, 8 allocs/op
  

  The fix is faster per call in this microbenchmark, not slower — pre-fix pays a synchronous
  stream Send (channel rendezvous with the plugin reader) on every call; the fix only enqueues
  and reuses the already-existing debounced persister. This is a single-function microbenchmark on
  one machine, not an end-to-end throughput/latency number — the benchi run above is what would
  actually gate a >10% regression claim per CLAUDE.md, and it has not been run. See
  benchi/ack-hot-path/README.md for full numbers and exact reproduction steps.

Adversarial self-review

• Re-read the diff hunting specifically for the class of bug this PR is about (ordering/timing
  around ack-vs-durability) and found a real one during implementation, not after: my first
  Teardown draft canceled the stream (stopStream) before forcing the final flush, which
  deadlocked (or, worse, silently dropped the ack) because the deferred send's stream.Send raced
  an already-canceled context. Caught by TestSource_Teardown_SendsPendingDeferredAckBeforeReturning
  hanging, not by inspection — fixed by reordering flush-then-cancel, documented at length in
  Teardown's doc comment so it isn't re-broken.
  Also required a Persister change (WaitPendingWrites), and I re-checked that this new
  callbackWg doesn't affect Destination's unrelated use of the same Persister.Persist — its
  callback is a trivial error-forward, so the extra wait is harmless there.
• Checked concurrency/lock ordering explicitly: onPersistFlushed runs from a goroutine spawned by
  Persister.flushNow and needs preparePluginCall's Instance.RLock — Teardown never holds
  Instance's exclusive lock across the flush-and-wait for exactly this reason (documented inline).
  pendingAcks/nextAckSeq/durableAckSeq are guarded by a dedicated ackMu, deliberately
  separate from Instance's lock, to avoid exactly this class of deadlock.
• Checked error paths: a flush failure leaves pendingAcks un-drained (nothing acked, matching
  invariant 1) and propagates via s.errs, unchanged from before this fix. A deferred send failure
  (e.g. a genuinely closed stream, not just a canceled context) is intentionally not escalated
  via s.errs — logged instead — because escalating it risks blocking that goroutine forever on an
  unbuffered channel nobody is guaranteed to still be draining mid-teardown; I hit exactly this as a
  second real deadlock while developing the chaos test (the chaos harness's runChild doesn't
  drain errs), fixed the same way.
  reader has finished draining before that.
• Checked resource cleanup: Source.Teardown's double-teardown defense (re-checking
  s.plugin == nil after each lock re-acquisition) is new relative to before, added because the
  reordering now releases Instance's lock mid-function; funnel.Worker's own teardownMu already
  serializes this in practice, but Teardown is a public method and shouldn't rely on that alone.
• Uncertainty I'm not resolving myself, surfaced for DeVaris: the cross-connector shared-batch
  blast radius (failure mode 8 above) now extends from "delays other connectors' writes" to "delays
  other connectors' acks too" — inherited from the existing Persister design, not introduced here,
  and already named in the design doc, but worth an explicit go/no-go rather than assuming it's
  fine by inheritance alone.

Gates

• make generate && git diff --exit-code: clean (no diff beyond this PR's own changes).
• golangci-lint run ./pkg/connector/... ./tests/chaos/...: 0 issues. (Repo-wide
  golangci-lint run ./... surfaces 36 pre-existing issues, all in
  pkg/plugin/processor/builtin/internal/diff/ — a vendored Go-stdlib diff package — and
  pkg/foundation/cerrors/conduiterr — none in files this PR touches.)
• go build ./...: clean.
• go test -race -count=1 ./... (whole repo): all packages green.
• go test -race -run TestSIGKILL -count=3 ./tests/chaos/...: green, 3x for flake-safety
  (~88s/run).
• New unit tests, all green under -race: TestSource_Ack_DeferredUntilDurablyFlushed,
  TestSource_OnPersistFlushed_OutOfOrderCompletionStillDeliversInOrder,
  TestSource_Teardown_SendsPendingDeferredAckBeforeReturning.
• Existing FIFO-ack tests still pass unchanged (TestWorker_Ack_DeliversPositionsFIFOToV1Plugin,
  TestWorker_Nack_DeliversPartialAckFIFOToV1Plugin) — now take ~1s each (waiting on the real
  debounce) instead of near-instant, since acks are genuinely deferred now.
• TestSource_Ack_Deadlock (pre-existing, concurrent-Ack stress test) still passes under -race.
• No new error codes added; TestAllCodesComplete unaffected (ran green as part of the full suite).

Risk tier

Tier 1 (data path: pkg/connector/source.go, pkg/connector/persister.go, plus the chaos
regression suite). Requires DeVaris Tier-1 sign-off before merge — per the standing rule, no
data-path change merges on automated review alone, and this session does not review or merge its
own PR.

Related

• Supersedes test(chaos): DBZ-1 SIGKILL crash-safety + engine FIFO-ack (first tests/chaos) #2677 (test(chaos): DBZ-1 SIGKILL crash-safety + engine FIFO-ack) — folds in its
  chaos harness and FIFO-ack test; this PR's TestSIGKILL_PruningUpstream_NoGap /
  TestSIGKILL_DurableUpstream_NoGap replace its ..._ProducesGap /
  ..._ProducesDuplicateNotGap tests.
• docs(design): Source.Ack persist-ordering fix + sev-0 postmortem #2679 — docs(design): Source.Ack persist-ordering fix + sev-0 postmortem (design doc + blameless
  postmortem this PR implements; approach sign-off required there first).
• Closes the DBZ-1 loop from the Debezium-compete roadmap
  (docs/design-documents/20260722-debezium-compete-roadmap.md) and closes Engine FIFO/in-order ack delivery to standalone connectors is unasserted (only a v1-client comment upholds it) #2672: the chaos suite
  now proves the sev-0 fixed (for the two upstream classes the harness models) rather than merely
  documenting it.

🤖 Generated with Claude Code

https://claude.ai/code/session_015GQFzakPShAYj8CcwajYDD

[devarismeroxa]

Pushed the two remaining items (commit 84b7185):

1. Bounded-Teardown regression test (was the missing piece). Added
TestSource_Teardown_BoundedWaitOnStuckFlush (pkg/connector/source_test.go): a blockingDB
wraps database.DB and gates NewTransaction on a channel the test controls, simulating a
stuck/slow flush deterministically (no real-sleep race). With a short teardownFlushTimeout
override, Teardown must return in well under 2s (vs. DefaultTeardownFlushTimeout's 10s, let
alone "forever") and log the bounded-wait warning. Verified it actually catches the bug: I
temporarily reverted the bounded WaitPendingWritesContext call back to the old unbounded
WaitPendingWrites, and the test hard-fails via the suite's own 5s deadline (goroutine dump,
not a graceful failure) — then restored the real fix and confirmed all three TestSource_Teardown_*
tests pass under -race. Added a fast-path complement,
TestSource_Teardown_FastFlushCompletesWithinBoundedTimeout (same short timeout, but a healthy
flush): proves the bound never fires as a false positive and the deferred ack is still delivered.

2. Doc alignment. Brought docs/design-documents/20260723-source-ack-persist-ordering-fix.md
(+ its companion postmortem) into this branch from #2679 and corrected it to match what actually
shipped here, rather than what was originally proposed:

• Connector-level FIFO seq, not per-source-partition HWM. The per-partition high-water-mark
  tracking the doc originally specified was never built, and isn't needed: Source.Ack
  unconditionally overwrites Instance.State.Position with the position of the last record in
  each call (source.go:394) — SourceState.Position is always the connector's full cumulative
  position, never a per-partition offset map. Because of that, whichever flush lands durably
  necessarily subsumes every earlier-queued seq, so a purely engine-internal, monotonically
  increasing counter is sufficient to preserve invariant 4. See
  TestSource_OnPersistFlushed_OutOfOrderCompletionStillDeliversInOrder for the shipped guarantee.
• Head-of-line blocking restated as process-wide, not merely cross-connector. Persister is a
  single instance shared across every pipeline in the process (see persister.go's
  WaitPendingWrites doc, ~184-186), so a slow flush for one connector delays deferred acks for
  every other connector on every other pipeline currently running — not just other connectors
  within the same pipeline, as the doc originally scoped it.
• Bounded-Teardown behavior added to the failure-mode analysis: graceful-shutdown-with-stuck-
  flush → bounded wait (WaitPendingWritesContext, DefaultTeardownFlushTimeout = 10s) → forced
  teardown → benign duplicate on restart, never a gap — same reasoning the SIGKILL chaos suite
  already established for the hard-crash path.

Both simplifications (FIFO seq vs. the originally-specified per-partition HWM, and the bounded
wait itself, which the doc didn't call for) are marked in the doc as pending DeVaris
re-affirmation at Tier-1 sign-off — argued to be safe simplifications of what was approved, not
assumed to be automatically covered by the original approach sign-off.

Gates run this session (pkg/connector, pkg/lifecycle-poc/funnel, tests/chaos):

• make generate && git diff --exit-code: clean, no generated-code drift beyond this commit's
  own two files.
• golangci-lint run ./pkg/connector/... ./tests/chaos/... ./pkg/lifecycle-poc/funnel/...: 0 issues.
• go build ./...: clean.
• go test -race -count=1 ./pkg/connector/... ./pkg/lifecycle-poc/funnel/...: all green.
• go test -race -run TestSIGKILL -count=3 ./tests/chaos/...: green, 3x
  (TestSIGKILL_PruningUpstream_NoGap and TestSIGKILL_DurableUpstream_NoGap, both scenarios,
  ~92s total).
• TestAllCodesComplete (cmd/conduit/internal/llmsgen): green.

Still Tier 1 — not merging. Needs DeVaris sign-off on (a) the approach as originally proposed and
(b) the two flagged simplifications above.