cockroachdb_changefeed: checkpoint on RESOLVED timestamps (CON-504) #4688
There are no files selected for viewing
| 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."). | ||
| 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(), | ||
|
|
@@ -124,16 +124,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 +256,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 +274,102 @@ 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 == "" { | ||
| continue | ||
| } | ||
| releaseFn, err := c.cursorCheckpointer.Track(ctx, resolvedTs, 1) | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("tracking resolved cursor: %w", err) | ||
| } | ||
| if cursorTimestamp := releaseFn(); cursorTimestamp != nil && *cursorTimestamp != "" { | ||
| if err := c.persistCursor(ctx, *cursorTimestamp); err != nil { | ||
| c.logger.Errorf("Failed to persist resolved cursor: %v", err) | ||
| } | ||
| } | ||
| continue | ||
| } | ||
| } | ||
|
|
||
| var cursorReleaseFn func() *string | ||
| if c.cursorCache != "" { | ||
| // Data rows are tracked with an empty payload: they hold the | ||
| // ordered tracker's frontier (so no resolved timestamp can persist | ||
| // past an un-acked row) but never advance the cursor themselves. | ||
| // 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, "", 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) | ||
| return msg, func(ctx context.Context, err error) error { | ||
| if cursorReleaseFn == nil { | ||
| return nil | ||
| } | ||
| if err != nil { | ||
| // auto_replay_nacks is user-toggleable, so a nack can be | ||
| // terminal. Never resolve: the cursor stays pinned before this | ||
| // row so no resolved timestamp can be persisted past its | ||
| // undelivered data. | ||
| c.logger.Errorf("Row rejected downstream: the cursor is now pinned before this row and the input will stall once the checkpoint limit is reached, unless the row is redelivered (auto_replay_nacks) or the pipeline restarts: %v", err) | ||
|
There was a problem hiding this comment. This introduces a new user-visible failure mode that isn't documented, which CONTRIBUTING.md §1.2.3 requires ("Known limitations and edge cases are documented"). Before this change a nack only meant that row's own cursor wasn't persisted — later rows still advanced the cursor. Now the nacked row's checkpoint slot is never released, so:
The comment here acknowledges the stall, but it's only reachable via a supported, user-settable option ( Suggested fix: extend the
Contributor
Author
There was a problem hiding this comment. Fixed in e50c83e — the cursor_cache description (and regenerated docs) now state that with auto_replay_nacks disabled a rejected row pins the cursor and eventually stalls the input, that this is deliberate (advancing would silently lose the row), and recommend keeping auto_replay_nacks enabled. |
||
| return err | ||
| } | ||
| cursorTimestamp := cursorReleaseFn() | ||
| if cursorTimestamp == nil || *cursorTimestamp == "" { | ||
| return nil | ||
| } | ||
| return c.persistCursor(ctx, *cursorTimestamp) | ||
| }, nil | ||
| } | ||
| } | ||
|
|
||
| func (c *crdbChangefeedInput) Close(ctx context.Context) error { | ||
|
|
||
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.