gcp_spanner_cdc: advance partition watermarks only on ordered acks (CON-504) - #4686
Open
squiidz wants to merge 4 commits into
Open
gcp_spanner_cdc: advance partition watermarks only on ordered acks (CON-504)#4686squiidz wants to merge 4 commits into
squiidz wants to merge 4 commits into
Conversation
Each batch's ack wrote its own commit timestamp directly as the partition watermark, and the metadata store applies updates without a monotonic guard. With max_in_flight > 1 a later batch's ack could persist its timestamp while an earlier batch was still un-acked; a crash in that window skipped the earlier records on restart. Batches are now registered with a per-partition ordered tracker (checkpoint.Capped) at emission, and an ack persists only the resolved contiguous-prefix watermark. Mid-record flushes (zero watermark) carry the last safe watermark forward so they can never stall or regress the resolved value. Connect also discards partition batchers from a previous subscriber session instead of reusing their buffered rows and ack state.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
…e checkpoint_limit - The reconnect reset swapped the factory pointer while the previous session's periodic-flush goroutines could still hold it (a data race), and a straggler could re-create a partition entry in the new factory, leaving the new session's callback blocked on a nil request channel. The factory is now reset in place under its mutex, discarded batchers are closed (stopping their period timers and service.Batchers, previously leaked one per partition per reconnect), removal is identity-guarded, and the run goroutine cancels + waits for all flusher goroutines to exit BEFORE signalling stopped so the next Connect cannot race them. - The per-partition in-flight cap was a hardcoded 1024 counting messages while its comment said batches; it is now the fleet-standard checkpoint_limit config field (default 1024).
… dead state Four review follow-ups on the ordered-acks change: - resetPartitionBatchers hardcoded context.Background() while its only caller (Connect) has a context in hand, so a cancelled Connect could still block in batcher cleanup; the ctx is now threaded through. - The checkpoint_limit fallback duplicated the spec default as an unnamed 1024; both now use a defaultCheckpointLimit constant. - The stale gcp_spanner_cdc checkpoint_limit waiver is removed from the CDC conformance test's knownNonConformant map (the field is exposed under the canonical name since 2731ae1). - The never-read res field on spannerCDCReader is dropped.
Review finding: the flusher replies on an unbuffered errCh while the requester waits with a ctx escape - when the subscriber errgroup cancels mid-wait, the requester abandons the channel and the flusher wedges in the send forever, never returning to observe ctx.Done. Since the run goroutine now waits for the flushers before TriggerHasStopped, that wedge left the input stuck instead of reconnecting. A one-slot buffer makes the reply send always complete (exactly one reply per request).
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
1 participant
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).
The input had no ordered ack tracking: each batch's ack wrote its own commit timestamp directly as the partition watermark, and the metadata store applies updates without a monotonic guard. With
max_in_flight > 1, a later batch's ack could persist its timestamp while an earlier batch was still un-acked — a crash in that window skipped the earlier records on restart. Additionally, the per-partition batcher factory survived reconnects, carrying stale buffered rows and ack state into the new subscriber session.Changes:
emitregisters every batch with a per-partition ordered tracker (checkpoint.Capped, cap 1024 as backpressure); the ack persists only the resolved contiguous-prefix watermark. Mid-record flushes (zero watermark) carry the last safe watermark forward so they can never stall or regress the resolved value.ack.Oncenever invokes the watermark update on a non-nil ack error, and the nacked batch's unresolved tracker slot pins the watermark behind it.Connectdiscards partition batchers from the previous subscriber session; their un-acked rows are re-read from the persisted watermarks.updateWatermarkis a function seam on the reader (set to the subscriber'sUpdatePartitionWatermarkinConnect) so the ordering contract is unit-testable end-to-end.Proof of Work
gcp/enterprisepackage green under-race.Note for certification (tier C)
An adversarial kill-and-restart integration test was descoped: the Spanner emulator cannot serve change-stream queries (the subscriber's own integration tests pair the emulator's metadata store with a mock querier), and the reader-level integration tests require a real GCP Spanner instance (
changestreamstest.CheckSkipReal). The crash-window property is encoded at the unit level instead; a manual adversarial run against a real instance is recommended before merging.