From 03005faa75082c9f40f3c65a9555e570b842b8ad Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 10 Aug 2026 15:27:44 -0400 Subject: [PATCH 1/5] cockroachdb_changefeed: checkpoint on RESOLVED timestamps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cursor was persisted from each acked row's own `updated` timestamp, but every row of a transaction — and the ENTIRE initial backfill — shares one timestamp, and CURSOR resume is exclusive. Acking a single backfill row then crashing persisted a cursor that skipped every other backfill row on restart: unbounded silent loss. When cursor_cache is set the changefeed now runs WITH RESOLVED (a user-supplied resolved='interval' is preserved) and only resolved timestamps — CockroachDB's guarantee that nothing at or below them will be emitted again, never issued before the initial scan completes — are persisted, gated through the ordered tracker so a resolved timestamp only persists once every row before it is acked. Data rows are tracked with an empty payload; nacks never resolve (logged, checkpoint pinned). Resolved records are consumed as bookkeeping rather than emitted downstream, which also fixes a latent panic: their NULL key crashed the row assertions whenever resolved was supplied via options. Restarts now redeliver at most the changes since the last resolved timestamp (duplicates, never loss). --- .../pages/inputs/cockroachdb_changefeed.adoc | 2 +- internal/impl/cockroachdb/config_test.go | 41 ++++ internal/impl/cockroachdb/input_changefeed.go | 162 ++++++++++----- internal/impl/cockroachdb/integration_test.go | 188 ++++++++++++++++++ 4 files changed, 341 insertions(+), 52 deletions(-) diff --git a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc index 5c7bc11ae8..08e0ed75c2 100644 --- a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc +++ b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc @@ -275,7 +275,7 @@ A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resou A list of options to be included in the changefeed (WITH X, Y...). -NOTE: 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. +NOTE: 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. *Type*: `array` diff --git a/internal/impl/cockroachdb/config_test.go b/internal/impl/cockroachdb/config_test.go index 25beb7873b..30ecc24ce4 100644 --- a/internal/impl/cockroachdb/config_test.go +++ b/internal/impl/cockroachdb/config_test.go @@ -15,6 +15,7 @@ package crdb import ( + "context" "testing" "github.com/stretchr/testify/assert" @@ -46,3 +47,43 @@ options: assert.Equal(t, "EXPERIMENTAL CHANGEFEED FOR strm_2 WITH UPDATED, CURSOR='1637953249519902405.0000000000'", selectInput.statement) require.NoError(t, selectInput.Close(t.Context())) } + +func TestCRDBConfigParseWithCursorCache(t *testing.T) { + spec := crdbChangefeedInputConfig() + env := service.NewEnvironment() + + parse := func(t *testing.T, conf string) *crdbChangefeedInput { + t.Helper() + selectConfig, err := spec.ParseYAML(conf, env) + require.NoError(t, err) + selectInput, err := newCRDBChangefeedInputFromConfig(selectConfig, service.MockResources(service.MockResourcesOptAddCache("mycache"))) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, selectInput.Close(context.Background())) }) + return selectInput + } + + t.Run("adds RESOLVED and strips CURSOR/UPDATED", func(t *testing.T) { + selectInput := parse(t, ` +dsn: postgresql://root@localhost:26257/defaultdb?sslmode=disable +tables: + - strm_2 +cursor_cache: mycache +options: + - UPDATED + - CURSOR='1637953249519902405.0000000000' +`) + assert.Equal(t, "EXPERIMENTAL CHANGEFEED FOR strm_2 WITH UPDATED, RESOLVED", selectInput.statement) + }) + + t.Run("preserves a user-supplied resolved interval", func(t *testing.T) { + selectInput := parse(t, ` +dsn: postgresql://root@localhost:26257/defaultdb?sslmode=disable +tables: + - strm_2 +cursor_cache: mycache +options: + - resolved='5s' +`) + assert.Equal(t, "EXPERIMENTAL CHANGEFEED FOR strm_2 WITH resolved='5s', UPDATED", selectInput.statement) + }) +} diff --git a/internal/impl/cockroachdb/input_changefeed.go b/internal/impl/cockroachdb/input_changefeed.go index 74f4ad87c3..a9fdb3aa17 100644 --- a/internal/impl/cockroachdb/input_changefeed.go +++ b/internal/impl/cockroachdb/input_changefeed.go @@ -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."). - ShortDescription("Options to include in the changefeed. CURSOR and UPDATED are ignored when cursor_cache is set."). + 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,6 +124,7 @@ 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 @@ -131,9 +132,20 @@ func newCRDBChangefeedInputFromConfig(conf *service.ParsedConfig, res *service.R 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 } - // 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() + 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) + 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 } - return nil, nil, err - } - - values, err := c.rows.Values() - if err != nil { - return nil, nil, fmt.Errorf("row values: %w", err) - } - - var cursorReleaseFn func() *string - rowBytes := values[2].([]byte) - if gObj, err := gabs.ParseJSON(rowBytes); err == nil { - if cursorTimestamp, _ := gObj.S("updated").Data().(string); cursorTimestamp != "" { - cursorReleaseFn, _ = c.cursorCheckpointer.Track(ctx, cursorTimestamp, 1) + values, err := c.rows.Values() + if err != nil { + return nil, nil, fmt.Errorf("row values: %w", err) } - } - - // 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, _ error) (cErr error) { - if cursorReleaseFn == nil { - return nil + 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 + } } - cursorTimestamp := cursorReleaseFn() - if cursorTimestamp == nil { - return nil + + 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) + } } - if err := c.res.AccessCache(ctx, c.cursorCache, func(c service.Cache) { - cErr = c.Set(ctx, cursorCacheKey, []byte(*cursorTimestamp), nil) + + // 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 err + return nil, nil, err } - return - }, nil + + 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) + return err + } + cursorTimestamp := cursorReleaseFn() + if cursorTimestamp == nil || *cursorTimestamp == "" { + return nil + } + return c.persistCursor(ctx, *cursorTimestamp) + }, nil + } } func (c *crdbChangefeedInput) Close(ctx context.Context) error { diff --git a/internal/impl/cockroachdb/integration_test.go b/internal/impl/cockroachdb/integration_test.go index 98de998240..356c844447 100644 --- a/internal/impl/cockroachdb/integration_test.go +++ b/internal/impl/cockroachdb/integration_test.go @@ -18,7 +18,12 @@ import ( "context" "errors" "fmt" + "os" + "path/filepath" + "strconv" + "strings" "sync" + "sync/atomic" "testing" "time" @@ -86,6 +91,9 @@ cockroachdb_changefeed: tables: - foo cursor_cache: foocache + options: + - resolved='1s' + - min_checkpoint_frequency='1s' `, port) cacheConf := fmt.Sprintf(` @@ -130,6 +138,23 @@ file: return len(outBatches) == 1000 }, time.Second*5, time.Millisecond*100) + // The cursor only advances to RESOLVED timestamps whose rows are all + // acked. Wait for a resolved checkpoint that postdates the moment every + // row above was received, so the restart below resumes without + // redelivery. Cursor values are "." HLC timestamps. + cutoffNanos := time.Now().UnixNano() + require.Eventually(t, func() bool { + b, err := os.ReadFile(filepath.Join(tmpDir, "crdb_changefeed_cursor")) + if err != nil { + return false + } + nanos, err := strconv.ParseInt(strings.SplitN(string(b), ".", 2)[0], 10, 64) + if err != nil { + return false + } + return nanos > cutoffNanos + }, time.Second*30, time.Millisecond*100, "cursor never advanced past the delivered rows") + require.NoError(t, streamOut.StopWithin(time.Second*10)) //-------------------------------------------------------------------------- @@ -177,3 +202,166 @@ file: require.NoError(t, streamOut.StopWithin(time.Second*10)) } + +// TestIntegrationCRDBBackfillAckCrash verifies that acknowledging a single +// backfill row never persists a cursor that skips the rest of the backfill: +// every row of the initial scan shares one `updated` timestamp and CURSOR +// resume is exclusive, so the pre-fix per-row cursor lost the entire backfill +// after ack-one-then-crash. Only fully-acknowledged RESOLVED timestamps may +// persist. See CON-504. +func TestIntegrationCRDBBackfillAckCrash(t *testing.T) { + integration.CheckSkip(t) + + tmpDir := t.TempDir() + + ctr, err := testcontainers.Run(t.Context(), "cockroachdb/cockroach:latest", + testcontainers.WithCmd("start-single-node", "--insecure"), + testcontainers.WithExposedPorts("8080/tcp", "26257/tcp"), + testcontainers.WithWaitStrategy( + wait.ForHTTP("/health").WithPort("8080/tcp").WithStartupTimeout(time.Minute), + ), + ) + testcontainers.CleanupContainer(t, ctr) + require.NoError(t, err) + + mappedPort, err := ctr.MappedPort(t.Context(), "26257/tcp") + require.NoError(t, err) + port := mappedPort.Port() + + var pgpool *pgxpool.Pool + require.Eventually(t, func() bool { + if pgpool == nil { + if pgpool, err = pgxpool.New(t.Context(), fmt.Sprintf("postgresql://root@localhost:%v/defaultdb?sslmode=disable", port)); err != nil { + return false + } + } + if _, err = pgpool.Exec(t.Context(), "SET CLUSTER SETTING kv.rangefeed.enabled = true;"); err != nil { + return false + } + _, err = pgpool.Exec(t.Context(), "CREATE TABLE bar (a INT PRIMARY KEY);") + return err == nil + }, time.Minute, time.Second) + t.Cleanup(func() { + pgpool.Close() + }) + + const rowCount = 100 + for i := range rowCount { + _, err := pgpool.Exec(t.Context(), fmt.Sprintf("INSERT INTO bar VALUES (%v);", i)) + require.NoError(t, err) + } + + template := fmt.Sprintf(` +cockroachdb_changefeed: + dsn: postgres://root@localhost:%v/defaultdb?sslmode=disable + tables: + - bar + cursor_cache: barcache + options: + - resolved='1s' + - min_checkpoint_frequency='1s' +`, port) + + cacheConf := fmt.Sprintf(` +label: barcache +file: + directory: %v +`, tmpDir) + + readCursor := func() string { + b, err := os.ReadFile(filepath.Join(tmpDir, "crdb_changefeed_cursor")) + if err != nil { + return "" + } + return string(b) + } + + // Run 1: acknowledge exactly ONE backfill row, then block every other + // delivery, and crash. The pre-fix code persisted the acked row's own + // `updated` timestamp here, which on restart skipped the rest of the + // backfill (all rows share that timestamp and CURSOR is exclusive). + { + streamOutBuilder := service.NewStreamBuilder() + require.NoError(t, streamOutBuilder.SetLoggerYAML(`level: OFF`)) + require.NoError(t, streamOutBuilder.AddCacheYAML(cacheConf)) + require.NoError(t, streamOutBuilder.AddInputYAML(template)) + + received := make(chan struct{}, 1) + var acked atomic.Bool + require.NoError(t, streamOutBuilder.AddBatchConsumerFunc(func(ctx context.Context, _ service.MessageBatch) error { + if acked.CompareAndSwap(false, true) { + return nil // ack the first row only + } + select { + case received <- struct{}{}: + default: + } + <-ctx.Done() + return ctx.Err() + })) + + streamOut, err := streamOutBuilder.Build() + require.NoError(t, err) + + runCtx, crash := context.WithCancel(t.Context()) + runDone := make(chan struct{}) + go func() { + defer close(runDone) + _ = streamOut.Run(runCtx) + }() + + select { + case <-received: + case <-time.After(time.Minute): + t.Fatal("backfill rows were never delivered") + } + // Give the input time to (wrongly) persist a cursor from the single + // acked row before crashing. + time.Sleep(3 * time.Second) + require.Empty(t, readCursor(), "no cursor may be persisted while backfill rows are unacknowledged") + crash() + select { + case <-runDone: + case <-time.After(30 * time.Second): + t.Fatal("run 1 did not stop after the simulated crash") + } + } + + // Run 2: restart with a free-flowing consumer. Every backfill row must be + // delivered. + { + streamOutBuilder := service.NewStreamBuilder() + require.NoError(t, streamOutBuilder.SetLoggerYAML(`level: OFF`)) + require.NoError(t, streamOutBuilder.AddCacheYAML(cacheConf)) + require.NoError(t, streamOutBuilder.AddInputYAML(template)) + + var seenMut sync.Mutex + seen := map[string]struct{}{} + require.NoError(t, streamOutBuilder.AddBatchConsumerFunc(func(_ context.Context, mb service.MessageBatch) error { + msgBytes, err := mb[0].AsBytes() + require.NoError(t, err) + seenMut.Lock() + seen[string(msgBytes)] = struct{}{} + seenMut.Unlock() + return nil + })) + + streamOut, err := streamOutBuilder.Build() + require.NoError(t, err) + go func() { + if err := streamOut.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + + var got int + require.Eventually(t, func() bool { + seenMut.Lock() + got = len(seen) + seenMut.Unlock() + return got == rowCount + }, time.Minute, time.Millisecond*100, "backfill rows were skipped after ack-one-then-crash: got %v of %v", got, rowCount) + + require.NoError(t, streamOut.StopWithin(time.Second*10)) + } +} From e50c83e800f9a41d2784c685c904bf37d4576a2d Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 10 Aug 2026 16:17:19 -0400 Subject: [PATCH 2/5] cockroachdb_changefeed: document the nack-pinning trade-off With cursor_cache set and auto_replay_nacks disabled, a rejected row deliberately pins the cursor and eventually stalls the input; the field docs now state this and recommend keeping auto_replay_nacks enabled. --- .../modules/components/pages/inputs/cockroachdb_changefeed.adoc | 2 ++ internal/impl/cockroachdb/input_changefeed.go | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc index 08e0ed75c2..ec3bb173c4 100644 --- a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc +++ b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc @@ -267,6 +267,8 @@ tables: A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resource^] to use for storing the current latest cursor that has been successfully delivered, this allows Redpanda Connect to continue from that cursor upon restart, rather than consume the entire state of the table. +NOTE: with a cursor cache configured and `auto_replay_nacks` disabled, a row that is rejected downstream permanently pins the cursor before that row (no later cursor can be persisted, and the input eventually stops delivering once its in-flight limit fills). This is deliberate — advancing past a rejected row would silently lose it — so keep `auto_replay_nacks` enabled unless rejections are handled by restarting the pipeline. + *Type*: `string` diff --git a/internal/impl/cockroachdb/input_changefeed.go b/internal/impl/cockroachdb/input_changefeed.go index a9fdb3aa17..a93d6b9366 100644 --- a/internal/impl/cockroachdb/input_changefeed.go +++ b/internal/impl/cockroachdb/input_changefeed.go @@ -55,7 +55,7 @@ func crdbChangefeedInputConfig() *service.ConfigSpec { Description("CSV of tables to be included in the changefeed"). Example([]string{"table1", "table2"}), service.NewStringField("cursor_cache"). - Description("A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resource^] to use for storing the current latest cursor that has been successfully delivered, this allows Redpanda Connect to continue from that cursor upon restart, rather than consume the entire state of the table."). + Description("A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resource^] to use for storing the current latest cursor that has been successfully delivered, this allows Redpanda Connect to continue from that cursor upon restart, rather than consume the entire state of the table.\n\nNOTE: with a cursor cache configured and `auto_replay_nacks` disabled, a row that is rejected downstream permanently pins the cursor before that row (no later cursor can be persisted, and the input eventually stops delivering once its in-flight limit fills). This is deliberate — advancing past a rejected row would silently lose it — so keep `auto_replay_nacks` enabled unless rejections are handled by restarting the pipeline."). ShortDescription("Cache resource storing the last delivered cursor, so restarts resume instead of re-reading the table."). Optional(), service.NewStringListField("options"). From 134f43bfddb92ed900cc4f899b6c0247ed6d19b9 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Tue, 11 Aug 2026 10:02:44 -0400 Subject: [PATCH 3/5] cockroachdb_changefeed: nacks resolve the cursor (auto_replay_nacks off is an opt-in drop) Unwinds the nack-pinning behavior and its doc note. 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, so the cursor advances past them; pinning produced permanent backpressure once the checkpoint limit filled. RESOLVED-based cursor checkpointing is unchanged. --- .../pages/inputs/cockroachdb_changefeed.adoc | 2 -- internal/impl/cockroachdb/input_changefeed.go | 16 ++++++---------- 2 files changed, 6 insertions(+), 12 deletions(-) diff --git a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc index ec3bb173c4..08e0ed75c2 100644 --- a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc +++ b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc @@ -267,8 +267,6 @@ tables: A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resource^] to use for storing the current latest cursor that has been successfully delivered, this allows Redpanda Connect to continue from that cursor upon restart, rather than consume the entire state of the table. -NOTE: with a cursor cache configured and `auto_replay_nacks` disabled, a row that is rejected downstream permanently pins the cursor before that row (no later cursor can be persisted, and the input eventually stops delivering once its in-flight limit fills). This is deliberate — advancing past a rejected row would silently lose it — so keep `auto_replay_nacks` enabled unless rejections are handled by restarting the pipeline. - *Type*: `string` diff --git a/internal/impl/cockroachdb/input_changefeed.go b/internal/impl/cockroachdb/input_changefeed.go index a93d6b9366..d1bd9c701f 100644 --- a/internal/impl/cockroachdb/input_changefeed.go +++ b/internal/impl/cockroachdb/input_changefeed.go @@ -55,7 +55,7 @@ func crdbChangefeedInputConfig() *service.ConfigSpec { Description("CSV of tables to be included in the changefeed"). Example([]string{"table1", "table2"}), service.NewStringField("cursor_cache"). - Description("A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resource^] to use for storing the current latest cursor that has been successfully delivered, this allows Redpanda Connect to continue from that cursor upon restart, rather than consume the entire state of the table.\n\nNOTE: with a cursor cache configured and `auto_replay_nacks` disabled, a row that is rejected downstream permanently pins the cursor before that row (no later cursor can be persisted, and the input eventually stops delivering once its in-flight limit fills). This is deliberate — advancing past a rejected row would silently lose it — so keep `auto_replay_nacks` enabled unless rejections are handled by restarting the pipeline."). + Description("A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resource^] to use for storing the current latest cursor that has been successfully delivered, this allows Redpanda Connect to continue from that cursor upon restart, rather than consume the entire state of the table."). ShortDescription("Cache resource storing the last delivered cursor, so restarts resume instead of re-reading the table."). Optional(), service.NewStringListField("options"). @@ -351,18 +351,14 @@ func (c *crdbChangefeedInput) Read(ctx context.Context) (*service.Message, servi } msg := service.NewMessage(jsonBytes) - return msg, func(ctx context.Context, err error) error { + // 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 } - 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) - return err - } cursorTimestamp := cursorReleaseFn() if cursorTimestamp == nil || *cursorTimestamp == "" { return nil From b335937b05fab271d73f54e9f7637664824833af Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Fri, 14 Aug 2026 11:32:01 -0400 Subject: [PATCH 4/5] cockroachdb_changefeed: carry resolved timestamps on row slots and serialize cursor persists Two review findings on the RESOLVED cursor path: - Data rows were tracked with an empty payload, and the ordered tracker's release returns the payload of the highest contiguously-resolved entry. Under out-of-order acks the frontier payload could therefore regress to the empty string and mask a pending resolved timestamp - the cursor only advanced in the narrow case where the row tracked right after a resolved marker was still un-acked, so restarts replayed far more than necessary. Rows now carry the last resolved timestamp seen before them: a row is only contiguously resolved once that timestamp and everything before it acked, so carrying it forward is always safe. - Releases hand out monotonic timestamps, but acks (pipeline goroutines) and resolved records (Read goroutine) persisted concurrently with no shared ordering, so two cache writes could land out of order and regress the cursor. Each release+persist pair now runs under a shared mutex - the same defect class fixed in mssqlserver's batcher on this ticket. Also fixes the new integration test's failure diagnostic (the message args were evaluated before Eventually ran, so it always reported 0 rows). Both integration tests re-verified green. --- internal/impl/cockroachdb/input_changefeed.go | 34 +++++++++++++++---- internal/impl/cockroachdb/integration_test.go | 14 ++++---- 2 files changed, 35 insertions(+), 13 deletions(-) diff --git a/internal/impl/cockroachdb/input_changefeed.go b/internal/impl/cockroachdb/input_changefeed.go index d1bd9c701f..6e616dfa5f 100644 --- a/internal/impl/cockroachdb/input_changefeed.go +++ b/internal/impl/cockroachdb/input_changefeed.go @@ -72,6 +72,16 @@ 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 pgConfig *pgxpool.Config pgPool *pgxpool.Pool @@ -318,24 +328,32 @@ func (c *crdbChangefeedInput) Read(ctx context.Context) (*service.Message, servi 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 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 { + // 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) } } @@ -359,6 +377,8 @@ func (c *crdbChangefeedInput) Read(ctx context.Context) (*service.Message, servi if cursorReleaseFn == nil { return nil } + c.persistMu.Lock() + defer c.persistMu.Unlock() cursorTimestamp := cursorReleaseFn() if cursorTimestamp == nil || *cursorTimestamp == "" { return nil diff --git a/internal/impl/cockroachdb/integration_test.go b/internal/impl/cockroachdb/integration_test.go index 356c844447..fa713ea7f2 100644 --- a/internal/impl/cockroachdb/integration_test.go +++ b/internal/impl/cockroachdb/integration_test.go @@ -354,13 +354,15 @@ file: } }() - var got int - require.Eventually(t, func() bool { + reached := assert.Eventually(t, func() bool { seenMut.Lock() - got = len(seen) - seenMut.Unlock() - return got == rowCount - }, time.Minute, time.Millisecond*100, "backfill rows were skipped after ack-one-then-crash: got %v of %v", got, rowCount) + defer seenMut.Unlock() + return len(seen) == rowCount + }, time.Minute, time.Millisecond*100) + seenMut.Lock() + got := len(seen) + seenMut.Unlock() + require.True(t, reached, "backfill rows were skipped after ack-one-then-crash: got %v of %v", got, rowCount) require.NoError(t, streamOut.StopWithin(time.Second*10)) } From c095c917072f2f246080ea780f2c6bc070658f3d Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Sat, 15 Aug 2026 20:48:46 -0400 Subject: [PATCH 5/5] cockroachdb_changefeed: surface and document resolved-record discard without cursor_cache Review note: with resolved='...' supplied in options but no cursor_cache, resolved records were consumed as bookkeeping with no output and no log at any level. A one-shot warning now explains the discard, and the options field description states that resolved records are internal cursor bookkeeping, never emitted as messages, and are discarded without a cursor_cache. Docs regenerated. --- .../pages/inputs/cockroachdb_changefeed.adoc | 2 +- internal/impl/cockroachdb/input_changefeed.go | 13 ++++++++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc index 08e0ed75c2..4dfaedde8f 100644 --- a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc +++ b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc @@ -275,7 +275,7 @@ A https://docs.redpanda.com/redpanda-connect/components/caches/about[cache resou A list of options to be included in the changefeed (WITH X, Y...). -NOTE: 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. +NOTE: 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. *Type*: `array` diff --git a/internal/impl/cockroachdb/input_changefeed.go b/internal/impl/cockroachdb/input_changefeed.go index 6e616dfa5f..33bb2f46bf 100644 --- a/internal/impl/cockroachdb/input_changefeed.go +++ b/internal/impl/cockroachdb/input_changefeed.go @@ -59,7 +59,7 @@ 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."). + 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(). @@ -82,6 +82,9 @@ type crdbChangefeedInput struct { // 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 @@ -322,6 +325,14 @@ func (c *crdbChangefeedInput) Read(ctx context.Context) (*service.Message, servi 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)