diff --git a/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc b/docs/modules/components/pages/inputs/cockroachdb_changefeed.adoc index 5c7bc11ae8..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. +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/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..33bb2f46bf 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. 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,6 +137,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 +145,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 +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 } - // 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 == "" { + // 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 + } } - cursorTimestamp := cursorReleaseFn() - if cursorTimestamp == nil { - return nil + + 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) + } } - 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) + // 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 { diff --git a/internal/impl/cockroachdb/integration_test.go b/internal/impl/cockroachdb/integration_test.go index 98de998240..fa713ea7f2 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,168 @@ 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) + } + }() + + reached := assert.Eventually(t, func() bool { + seenMut.Lock() + 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)) + } +}