cockroachdb_changefeed: checkpoint on RESOLVED timestamps (CON-504) #4688
There are no files selected for viewing
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -59,8 +59,8 @@ func crdbChangefeedInputConfig() *service.ConfigSpec { | |
| ShortDescription("Cache resource storing the last delivered cursor, so restarts resume instead of re-reading the table."). | ||
| Optional(), | ||
| service.NewStringListField("options"). | ||
| Description("A list of options to be included in the changefeed (WITH X, Y...).\n\nNOTE: Both the CURSOR option and UPDATED will be ignored from these options when a `cursor_cache` is specified, as they are set explicitly by Redpanda Connect in this case. A RESOLVED option is also added (unless one is supplied here): the stored cursor only ever advances to resolved timestamps whose rows have all been acknowledged downstream, so a restart redelivers at most the changes since the last resolved timestamp (bounded by the `changefeed.min_checkpoint_frequency` cluster setting) and never skips data. Resolved records are internal cursor bookkeeping and are never emitted as messages; without a `cursor_cache` they are discarded."). | ||
| ShortDescription("Options to include in the changefeed. CURSOR and UPDATED are ignored when cursor_cache is set, and RESOLVED is added."). | ||
| Example([]string{`virtual_columns="omitted"`}). | ||
| Advanced(). | ||
| Optional(), | ||
|
|
@@ -72,6 +72,19 @@ type crdbChangefeedInput struct { | |
| statement string | ||
| cursorCache string | ||
| cursorCheckpointer *checkpoint.Capped[string] | ||
| // lastResolved is the most recent resolved timestamp seen in stream order. | ||
| // Data rows carry it as their tracker payload so an out-of-order ack can | ||
| // never mask a pending resolved timestamp behind an empty payload. Only | ||
| // touched from the Read goroutine. | ||
| lastResolved string | ||
| // persistMu serializes release+persistCursor pairs. Releases hand out | ||
| // monotonically increasing timestamps, but acks (pipeline goroutines) and | ||
| // resolved records (Read goroutine) persist concurrently: without a shared | ||
| // critical section two writes can land out of order and regress the cursor. | ||
| persistMu sync.Mutex | ||
| // resolvedDropWarning fires once when resolved records are discarded | ||
| // because no cursor_cache is configured. | ||
| resolvedDropWarning sync.Once | ||
|
|
||
| pgConfig *pgxpool.Config | ||
| pgPool *pgxpool.Pool | ||
|
|
@@ -124,16 +137,28 @@ func newCRDBChangefeedInputFromConfig(conf *service.ParsedConfig, res *service.R | |
| if c.cursorCache == "" { | ||
| options = tmpOptions | ||
| } else { | ||
| hasResolved := false | ||
| for _, o := range tmpOptions { | ||
| if strings.HasPrefix(strings.ToLower(o), "updated") { | ||
| continue | ||
| } | ||
| if strings.HasPrefix(strings.ToLower(o), "cursor") { | ||
| continue | ||
| } | ||
| if strings.HasPrefix(strings.ToLower(o), "resolved") { | ||
| hasResolved = true | ||
| } | ||
| options = append(options, o) | ||
| } | ||
| options = append(options, "UPDATED") | ||
| if !hasResolved { | ||
| // Only RESOLVED timestamps are safe cursors: every row of a | ||
| // transaction (and the entire initial backfill) shares one | ||
| // `updated` timestamp, and CURSOR resume is exclusive, so a | ||
| // row-level cursor would skip that timestamp's remaining rows on | ||
| // restart. A user-supplied resolved='interval' option is kept. | ||
| options = append(options, "RESOLVED") | ||
| } | ||
| if err := res.AccessCache(context.Background(), c.cursorCache, func(c service.Cache) { | ||
| cursorBytes, cErr := c.Get(context.Background(), cursorCacheKey) | ||
| if cErr != nil { | ||
|
|
@@ -244,6 +269,16 @@ func (c *crdbChangefeedInput) closeConnection() { | |
| } | ||
| } | ||
|
|
||
| // persistCursor writes a resolved cursor timestamp to the cursor cache. | ||
| func (c *crdbChangefeedInput) persistCursor(ctx context.Context, cursorTimestamp string) (cErr error) { | ||
| if err := c.res.AccessCache(ctx, c.cursorCache, func(cache service.Cache) { | ||
| cErr = cache.Set(ctx, cursorCacheKey, []byte(cursorTimestamp), nil) | ||
| }); err != nil { | ||
| return err | ||
| } | ||
| return | ||
| } | ||
|
|
||
| func (c *crdbChangefeedInput) Read(ctx context.Context) (*service.Message, service.AckFunc, error) { | ||
| c.dbMut.Lock() | ||
| defer c.dbMut.Unlock() | ||
|
|
@@ -252,64 +287,116 @@ func (c *crdbChangefeedInput) Read(ctx context.Context) (*service.Message, servi | |
| return nil, nil, service.ErrNotConnected | ||
| } | ||
|
|
||
| for { | ||
| // rows.Next() blocks until the next changefeed event. The mutex is held to | ||
| // prevent closeConnection() from calling rows.Close() concurrently. On | ||
| // shutdown, SoftStopCtx cancels the query context which unblocks this call. | ||
| if !c.rows.Next() { | ||
| err := c.rows.Err() | ||
| c.closeQueryLocked() | ||
|
|
||
| if c.shutSig.IsSoftStopSignalled() { | ||
| return nil, nil, service.ErrNotConnected | ||
| } | ||
| if err == nil { | ||
| err = service.ErrNotConnected | ||
| } else { | ||
| err = fmt.Errorf("row read: %w", err) | ||
| } | ||
| return nil, nil, err | ||
| } | ||
|
|
||
| values, err := c.rows.Values() | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("row values: %w", err) | ||
| } | ||
|
|
||
| rowBytes := values[2].([]byte) | ||
| gObj, gErr := gabs.ParseJSON(rowBytes) | ||
|
|
||
| // Resolved records carry NULL table/key columns and are bookkeeping, | ||
| // never emitted downstream. A resolved timestamp is CockroachDB's | ||
| // guarantee that nothing at or below it will be emitted again — the | ||
| // only safe cursor (and CockroachDB does not emit one until the | ||
| // initial scan completes, so a persisted cursor always covers the | ||
| // backfill). Register it behind the in-flight rows (immediately | ||
| // resolved marker): the timestamp persists once every row before it | ||
| // is acked, either right here or inside the last outstanding ack. | ||
| if gErr == nil { | ||
| if resolvedTs, _ := gObj.S("resolved").Data().(string); resolvedTs != "" { | ||
| if c.cursorCache == "" { | ||
| // Resolved records are cursor bookkeeping, never emitted as | ||
| // messages; without a cursor_cache there is no cursor to | ||
| // advance, so they are dropped. Warn once so a user who | ||
| // supplied resolved='...' expecting output can see why | ||
| // nothing surfaces. | ||
| c.resolvedDropWarning.Do(func() { | ||
| c.logger.Warnf("Discarding RESOLVED timestamp records: they are cursor bookkeeping and are never emitted as messages, and without a cursor_cache there is no cursor to advance") | ||
| }) | ||
| continue | ||
| } | ||
| releaseFn, err := c.cursorCheckpointer.Track(ctx, resolvedTs, 1) | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("tracking resolved cursor: %w", err) | ||
| } | ||
| c.lastResolved = resolvedTs | ||
| c.persistMu.Lock() | ||
| if cursorTimestamp := releaseFn(); cursorTimestamp != nil && *cursorTimestamp != "" { | ||
| if err := c.persistCursor(ctx, *cursorTimestamp); err != nil { | ||
| c.logger.Errorf("Failed to persist resolved cursor: %v", err) | ||
| } | ||
| } | ||
| c.persistMu.Unlock() | ||
| continue | ||
| } | ||
| } | ||
|
|
||
| var cursorReleaseFn func() *string | ||
| if c.cursorCache != "" { | ||
| // Data rows carry the last resolved timestamp seen before them as | ||
| // payload: they hold the ordered tracker's frontier (so no resolved | ||
| // timestamp can persist past an un-acked row) without masking one — | ||
| // if rows tracked with an empty payload resolved out of order, the | ||
| // frontier payload could regress to "" and a safe pending resolved | ||
| // timestamp would never persist. A row tracked after resolved T | ||
| // only becomes contiguously resolved once T and everything before | ||
| // it acked, so carrying T is always safe. Row-level `updated` | ||
| // timestamps are unsafe cursors: every row of a transaction — and | ||
| // the entire initial backfill — shares one, and CURSOR resume is | ||
| // exclusive. | ||
| if cursorReleaseFn, err = c.cursorCheckpointer.Track(ctx, c.lastResolved, 1); err != nil { | ||
| return nil, nil, fmt.Errorf("tracking row checkpoint: %w", err) | ||
| } | ||
| } | ||
|
Comment on lines
+354
to
370
There was a problem hiding this comment. Tracking data rows with an empty payload can silently swallow a pending resolved timestamp, so the cursor may stop advancing under out-of-order acks.
Consider the tracked sequence
Every row before Suggested fix: track data rows with the last-seen resolved timestamp as the payload rather than
Contributor
Author
There was a problem hiding this comment. Fixed in b335937. Data rows now carry the last resolved timestamp seen before them as their tracker payload (the mssqlserver/oracledb pattern suggested): a row tracked after resolved T only becomes contiguously resolved once T and everything before it acked, so the frontier payload can never regress to the empty string and a safe pending timestamp always persists. The same commit also serializes each release+persistCursor pair behind a persistMu — with carried payloads the concurrent ack/Read persist paths had the same out-of-order-write race just fixed in mssqlserver's batcher. Both integration tests re-verified green. |
||
|
|
||
| // Construct the new JSON | ||
| var jsonBytes []byte | ||
| if jsonBytes, err = json.Marshal(map[string]string{ | ||
| "table": values[0].(string), | ||
| "primary_key": string(values[1].([]byte)), // Stringified JSON (Array) | ||
| "row": string(rowBytes), // Stringified JSON (Object) | ||
| }); err != nil { | ||
| return nil, nil, err | ||
| } | ||
|
|
||
| msg := service.NewMessage(jsonBytes) | ||
| // The ack error is deliberately ignored: nacks are replayed by | ||
| // auto_replay_nacks (the default), and disabling that is a documented | ||
| // opt-in to DROP rejected messages, so the cursor must advance past | ||
| // them rather than pin the tracker. | ||
| return msg, func(ctx context.Context, _ error) error { | ||
| if cursorReleaseFn == nil { | ||
| return nil | ||
| } | ||
| c.persistMu.Lock() | ||
| defer c.persistMu.Unlock() | ||
| cursorTimestamp := cursorReleaseFn() | ||
| if cursorTimestamp == nil || *cursorTimestamp == "" { | ||
| return nil | ||
| } | ||
| return c.persistCursor(ctx, *cursorTimestamp) | ||
| }, nil | ||
| } | ||
| } | ||
|
|
||
| func (c *crdbChangefeedInput) Close(ctx context.Context) error { | ||
|
|
||
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
NOTE (observability/docs): when a user supplies
resolved='...'inoptionswithoutcursor_cache, this branch now silently discards every resolved record — no message emitted, no log at any level. Previously those records reached the row-assertion path (a crash, per the first commit's message), so going from panic to silent drop is an improvement, but the new behaviour is undocumented and unobservable:optionsdoc note added in this PR (input_changefeed.go#L62-L63) only explains thecursor_cache-is-set case; nothing tells a user thatresolvedrecords are consumed as bookkeeping and never surface downstream.Suggested fix: emit a debug (or one-shot warn) log when a resolved record is dropped because no
cursor_cacheis configured, and extend theoptionsfield description to state that resolved records are internal bookkeeping and are never emitted as messages.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed in c095c91 — a one-shot warning explains the discard when resolved records arrive with no cursor_cache configured, and the options field description (and regenerated docs) now state that resolved records are internal cursor bookkeeping, never emitted as messages, and are discarded without a cursor_cache.