oracledb_cdc: gate post-snapshot checkpoint on downstream acks (CON-504) - #4675
Open
squiidz wants to merge 24 commits into
Open
oracledb_cdc: gate post-snapshot checkpoint on downstream acks (CON-504)#4675squiidz wants to merge 24 commits into
squiidz wants to merge 24 commits into
Conversation
…pping the publisher
Track order defines the ordered checkpoint sequence, but Track was called after releasing the batcher mutex, so the count-triggered flush (Publish) and the timed-flush loop could register batches out of order and persist a regressing SCN on ack. Track now happens under the same lock as the flush. Also guards the loop's UntilNext call, which read batcher state concurrently mutated by Publish (a data race confirmed by the new stress test under -race). Same fixes as the mssqlserver batcher, which shares this lifted pattern.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
Mirrors the mssqlserver review fixes (#4677): a nacked batch no longer resolves its checkpoint slot, and a nacked snapshot batch fails waitSnapshotAcks so the post-snapshot SCN is not persisted over undelivered rows (auto_replay_nacks is user-toggleable, so a nack can be terminal). publishBatch had no production callers left after the flush/track refactor; deleted, with the batcher tests rewritten to drive the production Publish/flushCurrent paths.
Contributor
Author
|
Two commits added since the initial review round:
|
A terminal nack (auto_replay_nacks disabled) deliberately pins the checkpoint and eventually stalls the input behind checkpoint_limit, but that consequence was invisible: nothing was logged anywhere on the nack path. Emit an error identifying the batch's checkpoint SCN, whether it was a snapshot batch, and the pinned-checkpoint consequence so operators can connect a stalled input to the downstream rejection.
This was referenced Aug 10, 2026
Merged
The snapshot ack gate collapses soft-stop cancellation and downstream rejection into one Info line whose wording only describes the former. A rejection discards the post-snapshot SCN and re-runs the whole snapshot - an unexpected, data-affecting outcome - so it now logs at error level with wording that names it, mirroring the cancellation/error split used by the surrounding branches.
Aligns with the mssqlserver review outcome (#4677): a terminal nack (auto_replay_nacks disabled) pinned the ordered checkpoint tracker permanently — the publisher and tracker were built once and reused across Connect retries, so after one nack no SCN could ever be persisted again and the input eventually wedged behind checkpoint_limit. A nack now triggers a restart, and Connect rebuilds the publisher (batcher + tracker) per attempt, sealing the old one so late acks from the previous session can neither persist stale positions nor trigger spurious restarts. The restart resumes from the last durable SCN and redelivers.
Contributor
Author
|
Added 88c7f78, aligning the terminal-nack behavior with the #4677 review outcome: a nack now triggers a restart, and Connect rebuilds the publisher (batcher + ordered tracker) per attempt — sealing the old one so late acks from the previous session can neither persist stale positions nor trigger spurious restarts. Previously a terminal nack (auto_replay_nacks disabled) pinned the shared tracker across reconnects, permanently wedging checkpointing. Verified by new unit tests (terminal nack invokes the restart hook; sealed publishers neither persist nor restart) plus the snapshot-ack-barrier and resume-from-checkpoint integration tests against the rebuilt Connect path. |
…opt-in drop)
Unwinds the nack-pinning and terminal-nack-restart changes from the review
rounds. Per the framework's documented contract for auto_replay_nacks
("If set to false these messages will instead be deleted"), disabling
replay is an explicit opt-in to drop rejected messages - typically because
failures are routed to a DLQ, which acks. Pinning the checkpoint (or
restarting to force redelivery) contradicted that contract: pinning
produced permanent backpressure once checkpoint_limit filled, and the
restart variant turned a persistently-failing message into an infinite
redelivery loop. The snapshot ack gate still guards the crash window; a
nack now simply settles its slot and the stream continues.
Contributor
Author
|
The nack-handling changes from the earlier review rounds have been unwound in the latest commit. The framework's documented contract for Unwound here (c25b42f): the terminal-nack restart machinery and per-connect publisher rebuild. Kept: the snapshot ack barrier before the post-snapshot SCN persists, the flush/track race fixes, and the guarded UntilNext. Barrier and resume integration tests re-verified green after the unwind. |
…blisher after a failed send Two review findings on the flush/track path: - The flush-order/track-order atomicity fix moved checkpoint.Track under batcherMu, so a Track blocked on checkpoint_limit (slow downstream during bulk snapshot load) froze every concurrent Publish and the timed-flush ticker instead of just its own flusher. Each flush now takes an order ticket under batcherMu - atomically with the Flush, keeping the user's batching policy exact - and Track+send admission happens in ticket order outside batcherMu, so the checkpoint sequence still matches flush order exactly while a blocked Track stalls only the ticket queue. New blocked-Track buffering test proven red against the old locking. - A failed batch send rolled back the snapshot gate but never resolved the checkpoint slot, permanently pinning the constructor-lifetime tracker. Resolving the slot instead would be unsafe (another flusher may already have delivered a later-tracked batch, whose ack would then persist an SCN past the undelivered rows), so the publisher is marked poisoned and Connect rebuilds it with a fresh tracker; the session resumes from the last durable SCN, which is necessarily before the orphaned rows. cacheSCN is now serialized and monotonic so a previous session's late acks can never regress the durable position. Also renames trackedBatch.msg to msgs (review suggestion). Full 16-test integration suite green.
Review finding on the ticket refactor: the snapshot gate Add moved out of the flush critical section (it now runs in trackBatch, after ticket admission and after checkpoint.Track, which can park on checkpoint_limit). The timed-flush loop is an independent flusher, so at the handoff it could already hold the final snapshot rows - flushed, ticketed, but not yet counted on the gate - while flushCurrent saw an empty batcher and returned with no barrier. waitSnapshotAcks could then release early and the post-snapshot SCN persist ahead of undelivered rows: the exact crash window this PR closes (plus a WaitGroup Add-during-Wait misuse hazard). flushCurrent now takes its ticket unconditionally, so its admission is a sequence barrier: every earlier flush has finished trackBatch+send before it returns, and the gate counts every published snapshot batch. New test encodes the parked-flusher interleaving and is proven red against the unbarriered code.
Same finding as mssqlserver (991bcc9), fixed proactively: a graceful stop landing in the snapshot handoff window exits flushCurrent via softCtx cancellation but logged at ERROR. Cancellation without a hard stop now logs at Info like the adjacent handoff branches; genuine flush failures keep the error level.
Ports the mssqlserver fixes from the same review round, plus the oracle-specific finding that motivated them: the timed-flush loop parks its send under hardStopCtx by design, so a soft stop during the snapshot handoff left flushCurrent(softCtx) wedged in admit behind it - Close burned both shutdown timeouts and logged a spurious error before the publisher's hard stop finally unwound the queue. admit(ctx) now escapes via the caller's context with an abandonment protocol (release skips abandoned tickets, keeping the sequence intact; abandoned batches were never tracked, so their rows re-read from the last durable checkpoint), and the batcher teardown in Close runs under batcherMu with a closed flag guarding the flush paths. New test encodes the exact parked-holder/queued-flusher scenario.
Review finding on the abandonment protocol: an abandoned ticket's batch had already left the batcher but was never tracked, so nothing pinned the ordered tracker for those rows - a later ticket (the timed loop survives a soft stop under hardStopCtx by design) could still track, deliver, and ack rows after the gap, persisting an SCN past the dropped ones: silent loss on restart, the exact hazard the failed-send path poisons against. The admit doc claimed re-read safety the code did not establish. Admission order is what makes the fix sound: at abandon time nothing after the gap has been tracked yet. An abandon whose ticket owned a non-empty batch now seals the queue - every later admission is refused with errQueueSealed, so nothing can ever be tracked past the gap - and poisons the publisher; Connect rebuilds and the session resumes from the last durable SCN, genuinely re-reading the dropped rows. Empty-ticket abandons (the flushCurrent barrier, mssql's window markers) stay benign. Same change on mssqlserver. New test encodes the exact interleaving: rows-owning abandon behind a parked holder must poison and refuse later flushers.
Two review findings from the mssqlserver sibling, applied here in the same round: - A downstream rejection with auto_replay_nacks disabled dropped rows and advanced the checkpoint past them with no log anywhere (the drop logging was removed wholesale with the nack-pinning unwind). The ack function now warns with the batch size, snapshot flag, and checkpoint SCN when it advances past rejected rows, and the failed-send poison path logs the rebuild it schedules. - Connect's poisoned rebuild made the publisher field mutable while ReadBatch and Close read it unguarded on other goroutines - a Close racing a rebuild could soft-stop the OLD publisher and leave the new one wedged. The field is now an atomic.Pointer: readers can never observe a stale pointer, and the session captures its own generation as a local.
Review finding, the abandon fix's sibling: a trackBatch failure after admission (parked in checkpoint.Track when the context cancels) strands rows that already left the batcher with nothing registered in the tracker, while the deferred release advances the queue - a later ticket could track, deliver, and ack rows past the gap, persisting an SCN that skips the dropped ones on restart. All three flush paths now seal (and thereby poison) on a track failure with rows in hand, and flushCurrent also seals when Flush itself errors alongside a non-empty batch. New test parks an admitted flusher in Track at capacity, cancels it, and asserts the poison plus refusal of later flushers.
Review finding completing the flushed-but-untracked table: flushCurrent sealed on a Flush error but loop() discarded the error entirely (rows drained by the failed Flush vanished with no log, no seal) and Publish returned without sealing - in both cases a concurrent flusher could still track and ack past the dropped rows. Flush runs the user's batching.processors chain, so the error is reachable in any config with batch processors. Both paths now seal (and thereby poison) and the loop surfaces the error at error level instead of discarding it.
Review ask from the mssqlserver sibling, applied here too: the rebuild block is extracted into rebuildPublisherIfPoisoned and unit-tested (old generation closed, fresh tracker stored, late ack from the abandoned generation is a no-op on the durable position), and cacheSCN's advance/equal/regress/invalid semantics are locked in directly.
…tcherMu Two review findings on the seal placement, both TOCTOU windows: - Abandon and seal were separate steps (admit marked abandoned under ticketMu, the caller sealed afterwards): the moment abandoned[ticket] is visible, a release from the previous holder can skip it and admit the next ticket, which can track, deliver, and ack past the dropped rows before the caller's seal lands. admit now takes ownsRows and seals+poisons in the same ticketMu critical section that records the abandonment; row-less abandons stay benign. - The Flush-error seal ran after batcherMu was released: in that gap another flusher could flush, take the next ticket, and be admitted past the discarded rows. All three paths now seal before the unlock (batcherMu before ticketMu is the established order, so the nesting is safe), and flushCurrent seals on any Flush error for consistency.
Review finding: after the seal moved under batcherMu, admit always refuses with errQueueSealed when Flush has failed, so flushCurrent's post-admit error check was dead and the operator-facing log described a sealed queue instead of the actual batching.processors failure. The flush error now returns before admit (the seal is already applied, so the barrier is unaffected), matching how Publish and loop() surface it.
Investigating the review's test ask showed the branch is unreachable through the current public API: service.Batcher.Flush never assigns its error return (the internal policy batcher returns only a batch, and processor failures surface as errored messages) - verified empirically with a registered hard-failing BatchProcessor and confirmed in the wrapper source. The seals stay because the signature declares the error and sealing is the correct handling if a future benthos version does fail here; the timed-loop branch now documents the unreachability.
Ports the mssqlserver log-level split: sendTracked's cancellation under a signalled soft stop is the expected graceful-shutdown path and logs at debug; the warning stays for a send that fails while the publisher is meant to be live.
…agates Review finding on the log-level split: the check read the publisher's own shutSig, which on the streaming path is only triggered by pub.Close() AFTER the session contexts have already been cancelled - so the graceful unwind still warned. Close now sets an explicit stopping flag on the publisher before triggering the input's soft stop, and sendTracked checks it alongside shutSig.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Part of CON-504 (CDC at-least-once / ack-gated progress).
After the snapshot read loop completed, the input persisted the post-snapshot SCN immediately — while snapshot batches could still be un-acked downstream, and while a partial batch could still be sitting unflushed in the batcher. A crash in that window meant the restart saw a cached SCN, skipped the snapshot, and silently lost the un-delivered rows. #4670 fixed the ack-path leak; this closes the remaining read-time write.
Mirrors the postgres_cdc barrier from #4584: the publisher counts in-flight snapshot batches, and the snapshot→streaming handoff now flushes the remaining partial batch, blocks until every snapshot batch is acknowledged (escapable by soft-stop), and only then persists the SCN.
Proof of Work
flushCurrentpublishes buffered rows without stopping the publisher loop; all oracledb_cdc: prevent checkpointing during snapshot #4670 regression tests unchanged and green.TestIntegrationOracleDBCDCSnapshotAckBarrier): blocked consumer + simulated crash → asserts no checkpoint row exists → restart re-runs the snapshot and re-delivers every row. Verified it fails against the pre-fix code at exactly the checkpoint assertion: