From 55d27d86b03c3fb60f9258b75e21accedb5d3f0b Mon Sep 17 00:00:00 2001 From: Terry Tata Date: Sat, 5 Sep 2026 12:02:07 -0700 Subject: [PATCH] refactor(indexer): rename the replay tool to backfill --- .github/workflows/test-smoke.yaml | 4 +- ...cli_test.go => smoke_backfill_cli_test.go} | 86 +++++++++---------- indexer/Dockerfile | 4 +- indexer/Dockerfile.dev | 2 +- indexer/Justfile | 4 +- indexer/cmd/{replay => backfill}/main.go | 76 ++++++++-------- indexer/pkg/{replay => backfill}/README.md | 62 ++++++------- .../discovery.go} | 18 ++-- indexer/pkg/{replay => backfill}/engine.go | 36 ++++---- .../pkg/{replay => backfill}/engine_test.go | 52 +++++------ .../message_replay.go => backfill/message.go} | 12 +-- indexer/pkg/{replay => backfill}/store.go | 20 +++-- indexer/pkg/{replay => backfill}/types.go | 18 ++-- .../pkg/{replay => backfill}/types_test.go | 2 +- indexer/pkg/storage/postgres.go | 8 +- 15 files changed, 204 insertions(+), 200 deletions(-) rename build/devenv/tests/e2e/{smoke_replay_cli_test.go => smoke_backfill_cli_test.go} (82%) rename indexer/cmd/{replay => backfill}/main.go (81%) rename indexer/pkg/{replay => backfill}/README.md (57%) rename indexer/pkg/{replay/discovery_replay.go => backfill/discovery.go} (86%) rename indexer/pkg/{replay => backfill}/engine.go (81%) rename indexer/pkg/{replay => backfill}/engine_test.go (87%) rename indexer/pkg/{replay/message_replay.go => backfill/message.go} (87%) rename indexer/pkg/{replay => backfill}/store.go (93%) rename indexer/pkg/{replay => backfill}/types.go (84%) rename indexer/pkg/{replay => backfill}/types_test.go (99%) diff --git a/.github/workflows/test-smoke.yaml b/.github/workflows/test-smoke.yaml index b0cd90b26..c3fcb8600 100644 --- a/.github/workflows/test-smoke.yaml +++ b/.github/workflows/test-smoke.yaml @@ -142,8 +142,8 @@ jobs: pattern: TestE2ESmoke_RemoveRemotePool profile: standard.profile timeout: 10m - - name: TestE2ESmoke_Replay - pattern: TestE2ESmoke_Replay + - name: TestE2ESmoke_Backfill + pattern: TestE2ESmoke_Backfill profile: standard.profile timeout: 10m - name: TestE2EReorg diff --git a/build/devenv/tests/e2e/smoke_replay_cli_test.go b/build/devenv/tests/e2e/smoke_backfill_cli_test.go similarity index 82% rename from build/devenv/tests/e2e/smoke_replay_cli_test.go rename to build/devenv/tests/e2e/smoke_backfill_cli_test.go index ddacb3778..c33495ab4 100644 --- a/build/devenv/tests/e2e/smoke_replay_cli_test.go +++ b/build/devenv/tests/e2e/smoke_backfill_cli_test.go @@ -26,7 +26,7 @@ import ( "github.com/smartcontractkit/chainlink-testing-framework/framework" ) -const replayBinary = "/bin/indexer-replay" +const backfillBinary = "/bin/indexer-backfill" func execInContainer(ctx context.Context, containerName string, args ...string) (string, error) { cmd := exec.CommandContext(ctx, "docker", append([]string{"exec", containerName}, args...)...) @@ -56,13 +56,13 @@ func openIndexerDB(t *testing.T, in *ccv.Cfg) (*sql.DB, string) { return db, containerName } -func replayCLIArgs(subcommand string, extra ...string) []string { - return append([]string{replayBinary, subcommand}, extra...) +func backfillCLIArgs(subcommand string, extra ...string) []string { + return append([]string{backfillBinary, subcommand}, extra...) } -// TestE2ESmoke_ReplayCLI verifies the replay CLI subcommands work end-to-end: +// TestE2ESmoke_BackfillCLI verifies the backfill CLI subcommands work end-to-end: // migration check, list, status, and a discovery dry-run. -func TestE2ESmoke_ReplayCLI(t *testing.T) { +func TestE2ESmoke_BackfillCLI(t *testing.T) { if testing.Short() { t.Skip("skipping e2e test in short mode; requires a running devenv environment") } @@ -84,9 +84,9 @@ func TestE2ESmoke_ReplayCLI(t *testing.T) { }) t.Run("list empty", func(t *testing.T) { - out, err := execInContainer(t.Context(), containerName, replayCLIArgs("list")...) + out, err := execInContainer(t.Context(), containerName, backfillCLIArgs("list")...) require.NoError(t, err, "list should succeed; output: %s", out) - require.Contains(t, out, "No replay jobs found", "empty list should report no jobs; output: %s", out) + require.Contains(t, out, "No backfill jobs found", "empty list should report no jobs; output: %s", out) }) t.Run("seed and list", func(t *testing.T) { @@ -96,23 +96,23 @@ func TestE2ESmoke_ReplayCLI(t *testing.T) { VALUES ($1, 'messages', 'completed', false, NOW(), NOW())`, fakeJobID, ) - require.NoError(t, err, "seeding fake replay job") + require.NoError(t, err, "seeding fake backfill job") - out, err := execInContainer(ctx, containerName, replayCLIArgs("list")...) + out, err := execInContainer(ctx, containerName, backfillCLIArgs("list")...) require.NoError(t, err, "list should succeed; output: %s", out) require.Contains(t, out, fakeJobID, "list output must contain the seeded job ID; output: %s", out) }) t.Run("status", func(t *testing.T) { - out, err := execInContainer(t.Context(), containerName, replayCLIArgs("status", "--id", fakeJobID)...) + out, err := execInContainer(t.Context(), containerName, backfillCLIArgs("status", "--id", fakeJobID)...) require.NoError(t, err, "status should succeed; output: %s", out) require.Contains(t, out, fakeJobID, "status output must contain the job ID; output: %s", out) require.Contains(t, out, "completed", "status output must show completed status; output: %s", out) }) t.Run("discovery", func(t *testing.T) { - out, err := execInContainer(t.Context(), containerName, replayCLIArgs("discovery", "--since", "1")...) - require.NoError(t, err, "discovery replay should succeed with sequence 1; output: %s", out) + out, err := execInContainer(t.Context(), containerName, backfillCLIArgs("discovery", "--since", "1")...) + require.NoError(t, err, "discovery backfill should succeed with sequence 1; output: %s", out) }) } @@ -182,17 +182,17 @@ func sendAndWaitForIndexed( return sentEvt.MessageID, msgIDHex } -// TestE2ESmoke_ReplayForceOverwrite exercises the replay system end-to-end: +// TestE2ESmoke_BackfillForceOverwrite exercises the backfill system end-to-end: // // 1. Sends two messages with fast finality, waits for both to be indexed. -// 2. Replays message-1 only with `messages --ids --force` and verifies +// 2. Backfills message-1 only with `messages --ids --force` and verifies // only message-1's verifier_results and messages ingestion_timestamps changed. -// 3. Replays both with `discovery --since --force` and verifies both +// 3. Backfills both with `discovery --since --force` and verifies both // messages' timestamps are updated. -// 4. Replays again with `discovery --since ` (no --force, backfill-only) +// 4. Backfills again with `discovery --since ` (no --force, backfill-only) // and verifies neither message's timestamp changed because there is nothing // new to backfill. -func TestE2ESmoke_ReplayForceOverwrite(t *testing.T) { +func TestE2ESmoke_BackfillForceOverwrite(t *testing.T) { if testing.Short() { t.Skip("skipping e2e test in short mode; requires a running devenv environment") } @@ -254,9 +254,9 @@ func TestE2ESmoke_ReplayForceOverwrite(t *testing.T) { // ── Step 1: send two messages and wait for indexing ────────────────────── t.Log("Step 1: sending two messages with finality=1...") msgID1, msgHex1 := sendAndWaitForIndexed(t, ctx, src, dest, - []byte("replay-test-msg-1"), executorAddr, ccvAddr, receiver, &testCtx) + []byte("backfill-test-msg-1"), executorAddr, ccvAddr, receiver, &testCtx) msgID2, msgHex2 := sendAndWaitForIndexed(t, ctx, src, dest, - []byte("replay-test-msg-2"), executorAddr, ccvAddr, receiver, &testCtx) + []byte("backfill-test-msg-2"), executorAddr, ccvAddr, receiver, &testCtx) msg1VerifierTsBefore, err := getVerifierIngestionTimestamp(ctx, db, msgHex1) require.NoError(t, err, "read msg1 verifier_results ingestion_timestamp") @@ -266,17 +266,17 @@ func TestE2ESmoke_ReplayForceOverwrite(t *testing.T) { require.NoError(t, err, "read msg1 messages ingestion_timestamp") msg2MessageTsBefore, err := getMessageIngestionTimestamp(ctx, db, msgHex2) require.NoError(t, err, "read msg2 messages ingestion_timestamp") - t.Logf("Before replay: msg1 verifier=%s message=%s msg2 verifier=%s message=%s", + t.Logf("Before backfill: msg1 verifier=%s message=%s msg2 verifier=%s message=%s", msg1VerifierTsBefore.Format(time.RFC3339Nano), msg1MessageTsBefore.Format(time.RFC3339Nano), msg2VerifierTsBefore.Format(time.RFC3339Nano), msg2MessageTsBefore.Format(time.RFC3339Nano)) time.Sleep(2 * time.Second) - // ── Step 2: replay msg1 only with --force via --ids ───────────────────── - t.Log("Step 2: replaying msg1 with messages --ids --force...") + // ── Step 2: backfill msg1 only with --force via --ids ─────────────────── + t.Log("Step 2: backfilling msg1 with messages --ids --force...") out, err := execInContainer(ctx, containerName, - replayCLIArgs("messages", "--ids", msgHex1, "--force")...) - require.NoError(t, err, "messages replay failed; output: %s", out) + backfillCLIArgs("messages", "--ids", msgHex1, "--force")...) + require.NoError(t, err, "messages backfill failed; output: %s", out) msg1VerifierTsAfterIDs, err := getVerifierIngestionTimestamp(ctx, db, msgHex1) require.NoError(t, err) @@ -286,26 +286,26 @@ func TestE2ESmoke_ReplayForceOverwrite(t *testing.T) { require.NoError(t, err) msg2MessageTsAfterIDs, err := getMessageIngestionTimestamp(ctx, db, msgHex2) require.NoError(t, err) - t.Logf("After --ids replay: msg1 verifier=%s message=%s msg2 verifier=%s message=%s", + t.Logf("After --ids backfill: msg1 verifier=%s message=%s msg2 verifier=%s message=%s", msg1VerifierTsAfterIDs.Format(time.RFC3339Nano), msg1MessageTsAfterIDs.Format(time.RFC3339Nano), msg2VerifierTsAfterIDs.Format(time.RFC3339Nano), msg2MessageTsAfterIDs.Format(time.RFC3339Nano)) require.True(t, msg1VerifierTsAfterIDs.After(msg1VerifierTsBefore), - "msg1 verifier_results ingestion_timestamp must be updated after --ids --force replay") + "msg1 verifier_results ingestion_timestamp must be updated after --ids --force backfill") require.True(t, msg2VerifierTsAfterIDs.Equal(msg2VerifierTsBefore), - "msg2 verifier_results ingestion_timestamp must be unchanged after replaying only msg1") + "msg2 verifier_results ingestion_timestamp must be unchanged after backfilling only msg1") require.True(t, msg1MessageTsAfterIDs.After(msg1MessageTsBefore), - "msg1 messages ingestion_timestamp must be updated after --ids --force replay") + "msg1 messages ingestion_timestamp must be updated after --ids --force backfill") require.True(t, msg2MessageTsAfterIDs.Equal(msg2MessageTsBefore), - "msg2 messages ingestion_timestamp must be unchanged after replaying only msg1") + "msg2 messages ingestion_timestamp must be unchanged after backfilling only msg1") time.Sleep(2 * time.Second) - // ── Step 3: replay both with --force via discovery --since ─────────────── - t.Logf("Step 3: replaying both with discovery --since %s --force...", discoverySince) + // ── Step 3: backfill both with --force via discovery --since ────────────── + t.Logf("Step 3: backfilling both with discovery --since %s --force...", discoverySince) out, err = execInContainer(ctx, containerName, - replayCLIArgs("discovery", "--since", discoverySince, "--force")...) - require.NoError(t, err, "discovery force replay failed; output: %s", out) + backfillCLIArgs("discovery", "--since", discoverySince, "--force")...) + require.NoError(t, err, "discovery force backfill failed; output: %s", out) msg1VerifierTsAfterDisc, err := getVerifierIngestionTimestamp(ctx, db, msgHex1) require.NoError(t, err) @@ -330,11 +330,11 @@ func TestE2ESmoke_ReplayForceOverwrite(t *testing.T) { time.Sleep(2 * time.Second) - // ── Step 4: replay without --force (backfill-only, nothing to fill) ───── - t.Logf("Step 4: replaying with discovery --since %s (no --force)...", discoverySince) + // ── Step 4: backfill without --force (nothing new to fill) ────────────── + t.Logf("Step 4: backfilling with discovery --since %s (no --force)...", discoverySince) out, err = execInContainer(ctx, containerName, - replayCLIArgs("discovery", "--since", discoverySince)...) - require.NoError(t, err, "discovery backfill replay failed; output: %s", out) + backfillCLIArgs("discovery", "--since", discoverySince)...) + require.NoError(t, err, "discovery backfill failed; output: %s", out) msg1VerifierTsAfterBackfill, err := getVerifierIngestionTimestamp(ctx, db, msgHex1) require.NoError(t, err) @@ -349,13 +349,13 @@ func TestE2ESmoke_ReplayForceOverwrite(t *testing.T) { msg2VerifierTsAfterBackfill.Format(time.RFC3339Nano), msg2MessageTsAfterBackfill.Format(time.RFC3339Nano)) require.True(t, msg1VerifierTsAfterBackfill.Equal(msg1VerifierTsAfterDisc), - "msg1 verifier_results ingestion_timestamp must NOT change on backfill-only replay (already exists)") + "msg1 verifier_results ingestion_timestamp must NOT change on backfill-only run (already exists)") require.True(t, msg2VerifierTsAfterBackfill.Equal(msg2VerifierTsAfterDisc), - "msg2 verifier_results ingestion_timestamp must NOT change on backfill-only replay (already exists)") + "msg2 verifier_results ingestion_timestamp must NOT change on backfill-only run (already exists)") require.True(t, msg1MessageTsAfterBackfill.Equal(msg1MessageTsAfterDisc), - "msg1 messages ingestion_timestamp must NOT change on backfill-only replay (already exists)") + "msg1 messages ingestion_timestamp must NOT change on backfill-only run (already exists)") require.True(t, msg2MessageTsAfterBackfill.Equal(msg2MessageTsAfterDisc), - "msg2 messages ingestion_timestamp must NOT change on backfill-only replay (already exists)") + "msg2 messages ingestion_timestamp must NOT change on backfill-only run (already exists)") // ── Final: verify data integrity via indexer HTTP API ──────────────────── for _, tc := range []struct { @@ -366,8 +366,8 @@ func TestE2ESmoke_ReplayForceOverwrite(t *testing.T) { {"msg2", msgID2}, } { verifs, err := indexerMonitor.GetVerificationsForMessageID(ctx, tc.msgID) - require.NoError(t, err, "%s: failed to read verifications after all replays", tc.name) + require.NoError(t, err, "%s: failed to read verifications after all backfills", tc.name) require.GreaterOrEqual(t, len(verifs.Results), 1, - "%s: verifications must still be present after all replays", tc.name) + "%s: verifications must still be present after all backfills", tc.name) } } diff --git a/indexer/Dockerfile b/indexer/Dockerfile index 9a359eb97..c50afb680 100644 --- a/indexer/Dockerfile +++ b/indexer/Dockerfile @@ -12,7 +12,7 @@ COPY . . RUN --mount=type=cache,target=/root/.cache/go-build,id=ccv-go-build \ --mount=type=cache,target=/go/pkg/mod,id=ccv-go-mod \ cd indexer && CGO_ENABLED=0 go build -ldflags='-s -w' -o /bin/indexer cmd/main.go && \ - CGO_ENABLED=0 go build -ldflags='-s -w' -o /bin/indexer-replay cmd/replay/main.go + CGO_ENABLED=0 go build -ldflags='-s -w' -o /bin/indexer-backfill cmd/backfill/main.go FROM alpine:3.23 RUN apk --no-cache add ca-certificates @@ -21,7 +21,7 @@ RUN apk --no-cache add ca-certificates RUN addgroup -S indexer && adduser -S indexer -G indexer COPY --chown=indexer:indexer --from=builder /bin/indexer /bin/ -COPY --chown=indexer:indexer --from=builder /bin/indexer-replay /bin/ +COPY --chown=indexer:indexer --from=builder /bin/indexer-backfill /bin/ COPY --chown=indexer:indexer --from=builder /app/indexer/migrations ./migrations USER indexer:indexer CMD ["/bin/indexer"] diff --git a/indexer/Dockerfile.dev b/indexer/Dockerfile.dev index 6d88dd9a8..92228a73b 100644 --- a/indexer/Dockerfile.dev +++ b/indexer/Dockerfile.dev @@ -12,7 +12,7 @@ RUN --mount=type=cache,target=/go/pkg/mod,id=ccv-go-mod \ COPY . . RUN --mount=type=cache,target=/root/.cache/go-build,id=ccv-go-build \ --mount=type=cache,target=/go/pkg/mod,id=ccv-go-mod \ - CGO_ENABLED=0 go build -o /bin/indexer-replay ./indexer/cmd/replay/main.go + CGO_ENABLED=0 go build -o /bin/indexer-backfill ./indexer/cmd/backfill/main.go WORKDIR /app/indexer ENV DEVELOPMENT_LOG_PROFILE=true CMD ["air", "-c", "air.toml"] diff --git a/indexer/Justfile b/indexer/Justfile index 22a264d89..d1844b555 100644 --- a/indexer/Justfile +++ b/indexer/Justfile @@ -9,8 +9,8 @@ default: build: docker build -f Dockerfile -t indexer:latest .. -build-replay: - cd .. && go build -o indexer/bin/indexer-replay ./indexer/cmd/replay/main.go +build-backfill: + cd .. && go build -o indexer/bin/indexer-backfill ./indexer/cmd/backfill/main.go build-rc: docker build -f Dockerfile -t indexer:rc .. diff --git a/indexer/cmd/replay/main.go b/indexer/cmd/backfill/main.go similarity index 81% rename from indexer/cmd/replay/main.go rename to indexer/cmd/backfill/main.go index f224f3ce9..7d3d865bd 100644 --- a/indexer/cmd/replay/main.go +++ b/indexer/cmd/backfill/main.go @@ -19,12 +19,12 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/sqlutil/pg" ccvcommon "github.com/smartcontractkit/chainlink-ccv/common" + "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/backfill" "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/common" "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/config" "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/monitoring" "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/readers" "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/registry" - "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/replay" "github.com/smartcontractkit/chainlink-ccv/indexer/pkg/storage" "github.com/smartcontractkit/chainlink-ccv/internal/tablefmt" "github.com/smartcontractkit/chainlink-ccv/protocol" @@ -34,9 +34,9 @@ import ( func main() { app := cli.NewApp() - app.Name = "indexer-replay" - app.Usage = "Replay indexer data from upstream sources" - app.Commands = replayCommands() + app.Name = "indexer-backfill" + app.Usage = "Backfill indexer data from upstream sources" + app.Commands = backfillCommands() if err := app.Run(os.Args); err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) @@ -44,16 +44,16 @@ func main() { } } -func replayCommands() []cli.Command { +func backfillCommands() []cli.Command { return []cli.Command{ { Name: "discovery", - Usage: "Replay message discovery from a sequence number", + Usage: "Backfill message discovery from a sequence number", Action: discoveryAction, Flags: []cli.Flag{ cli.Uint64Flag{ Name: "since", - Usage: "Replay discovery since this aggregator sequence number", + Usage: "Backfill discovery since this aggregator sequence number", Required: true, }, cli.BoolFlag{ @@ -64,12 +64,12 @@ func replayCommands() []cli.Command { }, { Name: "messages", - Usage: "Replay CCV records for specific message IDs", + Usage: "Backfill CCV records for specific message IDs", Action: messagesAction, Flags: []cli.Flag{ cli.StringFlag{ Name: "ids", - Usage: "Comma-separated list of message IDs to replay", + Usage: "Comma-separated list of message IDs to backfill", Required: true, }, cli.BoolFlag{ @@ -80,7 +80,7 @@ func replayCommands() []cli.Command { }, { Name: "status", - Usage: "Show status of a replay job", + Usage: "Show status of a backfill job", Action: statusAction, Flags: []cli.Flag{ cli.StringFlag{ @@ -92,12 +92,12 @@ func replayCommands() []cli.Command { }, { Name: "list", - Usage: "List recent replay jobs", + Usage: "List recent backfill jobs", Action: listAction, }, { Name: "resume", - Usage: "Resume a failed or interrupted replay job", + Usage: "Resume a failed or interrupted backfill job", Action: resumeAction, Flags: []cli.Flag{ cli.StringFlag{ @@ -122,8 +122,8 @@ func discoveryAction(c *cli.Context) error { engine, cleanup := mustBuildEngine(ctx, true) defer cleanup() - req := replay.Request{ - Type: replay.TypeDiscovery, + req := backfill.Request{ + Type: backfill.TypeDiscovery, Since: int64(since), Force: c.Bool("force"), } @@ -131,12 +131,12 @@ func discoveryAction(c *cli.Context) error { jobID, err := engine.Start(ctx, req) if err != nil { if jobID != "" { - return fmt.Errorf("replay failed (job %s can be resumed): %w", jobID, err) + return fmt.Errorf("backfill failed (job %s can be resumed): %w", jobID, err) } - return fmt.Errorf("replay failed: %w", err) + return fmt.Errorf("backfill failed: %w", err) } - fmt.Printf("Replay completed successfully. Job ID: %s\n", jobID) //nolint:forbidigo // CLI user output + fmt.Printf("Backfill completed successfully. Job ID: %s\n", jobID) //nolint:forbidigo // CLI user output return nil } @@ -153,8 +153,8 @@ func messagesAction(c *cli.Context) error { engine, cleanup := mustBuildEngine(ctx, false) defer cleanup() - req := replay.Request{ - Type: replay.TypeMessages, + req := backfill.Request{ + Type: backfill.TypeMessages, MessageIDs: ids, Force: c.Bool("force"), } @@ -162,12 +162,12 @@ func messagesAction(c *cli.Context) error { jobID, err := engine.Start(ctx, req) if err != nil { if jobID != "" { - return fmt.Errorf("replay failed (job %s can be resumed): %w", jobID, err) + return fmt.Errorf("backfill failed (job %s can be resumed): %w", jobID, err) } - return fmt.Errorf("replay failed: %w", err) + return fmt.Errorf("backfill failed: %w", err) } - fmt.Printf("Replay completed successfully. Job ID: %s\n", jobID) //nolint:forbidigo // CLI user output + fmt.Printf("Backfill completed successfully. Job ID: %s\n", jobID) //nolint:forbidigo // CLI user output return nil } @@ -207,19 +207,19 @@ func resumeAction(c *cli.Context) error { return err } - engine, cleanup := mustBuildEngine(ctx, job.Type == replay.TypeDiscovery) + engine, cleanup := mustBuildEngine(ctx, job.Type == backfill.TypeDiscovery) defer cleanup() if err := engine.Resume(ctx, jobID); err != nil { return fmt.Errorf("resume failed: %w", err) } - fmt.Printf("Replay resumed and completed successfully. Job ID: %s\n", jobID) //nolint:forbidigo // CLI user output + fmt.Printf("Backfill resumed and completed successfully. Job ID: %s\n", jobID) //nolint:forbidigo // CLI user output return nil } -// renderJob prints a single replay job's details. -func renderJob(j *replay.Job) error { +// renderJob prints a single backfill job's details. +func renderJob(j *backfill.Job) error { table := tablefmt.NewKeyValue(os.Stdout) data := [][]string{ @@ -263,10 +263,10 @@ func truncateHash(h string) string { return h } -// renderJobList prints a table of replay jobs. -func renderJobList(jobs []replay.Job) error { +// renderJobList prints a table of backfill jobs. +func renderJobList(jobs []backfill.Job) error { if len(jobs) == 0 { - fmt.Println("No replay jobs found.") //nolint:forbidigo // CLI user output + fmt.Println("No backfill jobs found.") //nolint:forbidigo // CLI user output return nil } @@ -294,8 +294,8 @@ func renderJobList(jobs []replay.Job) error { return table.Render() } -// mustBuildEngine creates the full replay engine with all dependencies. -func mustBuildEngine(ctx context.Context, needsDiscoveryReader bool) (*replay.Engine, func()) { +// mustBuildEngine creates the full backfill engine with all dependencies. +func mustBuildEngine(ctx context.Context, needsDiscoveryReader bool) (*backfill.Engine, func()) { cfg := mustLoadConfig() lggr := mustCreateLogger(cfg) monitoring := monitoring.NewNoopIndexerMonitoring() @@ -324,9 +324,9 @@ func mustBuildEngine(ctx context.Context, needsDiscoveryReader bool) (*replay.En lggr.Warnf("Error closing migration database connection: %v", err) } - replayStore, err := replay.NewStoreFromConfig(ctx, lggr, pgCfg.URI, dbConfig, time.Duration(pgCfg.ConnMaxLifetime), time.Duration(pgCfg.ConnMaxIdleTime)) + backfillStore, err := backfill.NewStoreFromConfig(ctx, lggr, pgCfg.URI, dbConfig, time.Duration(pgCfg.ConnMaxLifetime), time.Duration(pgCfg.ConnMaxIdleTime)) if err != nil { - lggr.Fatalf("Failed to create replay store: %v", err) + lggr.Fatalf("Failed to create backfill store: %v", err) } indexerStorage, err := storage.NewPostgresStorage(ctx, lggr, monitoring, pgCfg.URI, pg.DriverPostgres, dbConfig, time.Duration(pgCfg.ConnMaxLifetime), time.Duration(pgCfg.ConnMaxIdleTime)) @@ -354,7 +354,7 @@ func mustBuildEngine(ctx context.Context, needsDiscoveryReader bool) (*replay.En } } - var aggFactory replay.AggregatorReaderFactory + var aggFactory backfill.AggregatorReaderFactory if needsDiscoveryReader && len(cfg.Discoveries) > 0 { disc := cfg.Discoveries[0] aggFactory = func(since int64) (*readers.ResilientReader, error) { @@ -366,7 +366,7 @@ func mustBuildEngine(ctx context.Context, needsDiscoveryReader bool) (*replay.En } } - engine := replay.NewEngine(replayStore, indexerStorage, verifierRegistry, aggFactory, lggr) + engine := backfill.NewEngine(backfillStore, indexerStorage, verifierRegistry, aggFactory, lggr) cleanup := func() { for _, c := range verifierCleanups { @@ -377,7 +377,7 @@ func mustBuildEngine(ctx context.Context, needsDiscoveryReader bool) (*replay.En return engine, cleanup } -func mustBuildStore(ctx context.Context) *replay.Store { +func mustBuildStore(ctx context.Context) *backfill.Store { cfg := mustLoadConfig() lggr := mustCreateLogger(cfg) @@ -389,7 +389,7 @@ func mustBuildStore(ctx context.Context) *replay.Store { LockTimeout: time.Duration(pgCfg.LockTimeout) * time.Second, } - store, err := replay.NewStoreFromConfig(ctx, lggr, pgCfg.URI, dbConfig, time.Duration(pgCfg.ConnMaxLifetime), time.Duration(pgCfg.ConnMaxIdleTime)) + store, err := backfill.NewStoreFromConfig(ctx, lggr, pgCfg.URI, dbConfig, time.Duration(pgCfg.ConnMaxLifetime), time.Duration(pgCfg.ConnMaxIdleTime)) if err != nil { lggr.Fatalf("Failed to create store: %v", err) } @@ -415,7 +415,7 @@ func mustCreateLogger(cfg *config.Config) logger.Logger { fmt.Fprintf(os.Stderr, "Failed to create logger: %v\n", err) os.Exit(1) } - return logger.Named(logger.Sugared(lggr), "indexer-replay") + return logger.Named(logger.Sugared(lggr), "indexer-backfill") } func createVerifierReader(ctx context.Context, lggr logger.Logger, vc *config.VerifierConfig, mon common.IndexerMonitoring) (*readers.VerifierReader, func(), error) { diff --git a/indexer/pkg/replay/README.md b/indexer/pkg/backfill/README.md similarity index 57% rename from indexer/pkg/replay/README.md rename to indexer/pkg/backfill/README.md index cf13d6f2c..38bd1abaf 100644 --- a/indexer/pkg/replay/README.md +++ b/indexer/pkg/backfill/README.md @@ -1,72 +1,74 @@ -# Replay +# Backfill -The replay module provides crash-recoverable data replay for the indexer. It re-fetches messages and CCV records from upstream sources to backfill missing data or overwrite stale records after bugs, deployments, or upstream outages. +The backfill module provides crash-recoverable data backfill for the indexer. It re-fetches messages and CCV records from upstream sources to backfill missing data or overwrite stale records after bugs, deployments, or upstream outages. + +> **Rename note:** this tool used to be called "replay" (`indexer-replay`, `indexer/pkg/replay`). It was renamed to "backfill" to avoid colliding with the verifier's unrelated message replay concept (re-verification of messages via `ccv jobqueue reschedule`). The Postgres table is still named `replay_jobs` — that name predates the rename and is kept for migration compatibility. ## Why a Separate Process -The replay runs as a standalone binary (`indexer-replay`) rather than inside the live indexer process. This guarantees complete isolation from the main polling and worker threads: +The backfill runs as a standalone binary (`indexer-backfill`) rather than inside the live indexer process. This guarantees complete isolation from the main polling and worker threads: -- **Own OS process** — separate goroutines, memory, and connection pools. A long-running replay cannot starve the live traffic of CPU, memory, or database connections. -- **Own gRPC/REST connections** — replay creates its own aggregator and verifier readers with independent circuit breakers, so replay-induced load never trips the live readers. +- **Own OS process** — separate goroutines, memory, and connection pools. A long-running backfill cannot starve the live traffic of CPU, memory, or database connections. +- **Own gRPC/REST connections** — backfill creates its own aggregator and verifier readers with independent circuit breakers, so backfill-induced load never trips the live readers. - **Own DB connection pool** — configurable independently; defaults to lower limits than the live indexer. ## Why CLI over HTTP -Both the CLI and an HTTP endpoint share the same replay engine, so switching later is straightforward. The CLI was chosen as the primary interface because: +Both the CLI and an HTTP endpoint share the same backfill engine, so switching later is straightforward. The CLI was chosen as the primary interface because: -- **Security** — no endpoint to protect. Whoever has `kubectl exec` access to the pod already has the right authorization level. An HTTP endpoint would require auth middleware, RBAC, and abuse protection (concurrent replay limits, rate limiting). -- **Operational fit** — replays are long-running (minutes to hours). HTTP would require an async pattern (accept → 202 → poll for status), which is essentially a CLI with extra ceremony. -- **Kubernetes Jobs** — for large replays the CLI can be launched as a Kubernetes Job with resource limits, timeouts, and automatic restart on failure (see below). +- **Security** — no endpoint to protect. Whoever has `kubectl exec` access to the pod already has the right authorization level. An HTTP endpoint would require auth middleware, RBAC, and abuse protection (concurrent backfill limits, rate limiting). +- **Operational fit** — backfills are long-running (minutes to hours). HTTP would require an async pattern (accept → 202 → poll for status), which is essentially a CLI with extra ceremony. +- **Kubernetes Jobs** — for large backfills the CLI can be launched as a Kubernetes Job with resource limits, timeouts, and automatic restart on failure (see below). -## Replay Modes +## Backfill Modes -### Discovery Replay +### Discovery Backfill Re-discovers messages from the aggregator starting at a given sequence number and gathers their CCV records from all configured verifiers. ``` -indexer-replay discovery --since 42 -indexer-replay discovery --since 42 --force +indexer-backfill discovery --since 42 +indexer-backfill discovery --since 42 --force ``` -The `--since` flag takes an aggregator sequence number (unsigned integer). All messages with a sequence number greater than or equal to the given value will be replayed. +The `--since` flag takes an aggregator sequence number (unsigned integer). All messages with a sequence number greater than or equal to the given value will be backfilled. -Without `--force` the replay backfills only — existing messages and CCV records are left untouched (`ON CONFLICT DO NOTHING`). With `--force`, existing records are overwritten (`ON CONFLICT DO UPDATE`). +Without `--force` the backfill fills gaps only — existing messages and CCV records are left untouched (`ON CONFLICT DO NOTHING`). With `--force`, existing records are overwritten (`ON CONFLICT DO UPDATE`). -### Message Replay +### Message Backfill Fetches CCV records from all configured verifiers for a specific set of message IDs. Does not re-run discovery. ``` -indexer-replay messages --ids "0xabc123,0xdef456" -indexer-replay messages --ids "0xabc123,0xdef456" --force +indexer-backfill messages --ids "0xabc123,0xdef456" +indexer-backfill messages --ids "0xabc123,0xdef456" --force ``` ### Job Management ``` -indexer-replay status --id # show details for a single job -indexer-replay list # list recent replay jobs -indexer-replay resume --id # resume a failed/interrupted job +indexer-backfill status --id # show details for a single job +indexer-backfill list # list recent backfill jobs +indexer-backfill resume --id # resume a failed/interrupted job ``` ## Crash Recovery -Replay jobs are persisted in a `replay_jobs` Postgres table. If the process crashes or the pod restarts mid-replay: +Backfill jobs are persisted in a `replay_jobs` Postgres table (the table name predates the backfill rename and is kept for migration compatibility). If the process crashes or the pod restarts mid-backfill: -1. **At-least-once checkpointing** — the progress cursor is periodically updated after replayed data is written. On a crash, the cursor may lag behind some already-committed rows, causing those rows to be replayed again, but no committed work is lost. +1. **At-least-once checkpointing** — the progress cursor is periodically updated after backfilled data is written. On a crash, the cursor may lag behind some already-committed rows, causing those rows to be backfilled again, but no committed work is lost. 2. **Advisory locks** — a PostgreSQL session-level advisory lock prevents two processes from running the same job concurrently. The lock is automatically released when the connection drops (crash, pod eviction). 3. **Automatic resumption** — on restart the engine detects the stale `running` job (via heartbeat timeout), re-acquires the lock, and resumes from the last persisted cursor, potentially reprocessing some already-written rows. ## Running as a Kubernetes Job -For large replays it is recommended to run the CLI as a Kubernetes Job rather than via `kubectl exec`. This gives you automatic retries, resource limits, and timeout control. +For large backfills it is recommended to run the CLI as a Kubernetes Job rather than via `kubectl exec`. This gives you automatic retries, resource limits, and timeout control. ```yaml apiVersion: batch/v1 kind: Job metadata: - name: indexer-replay-discovery + name: indexer-backfill-discovery spec: backoffLimit: 3 activeDeadlineSeconds: 7200 # 2 hour timeout @@ -74,10 +76,10 @@ spec: spec: restartPolicy: OnFailure containers: - - name: replay + - name: backfill image: indexer:latest command: - - /bin/indexer-replay + - /bin/indexer-backfill - discovery - --since - "42" @@ -106,11 +108,11 @@ With `restartPolicy: OnFailure` the pod is automatically restarted after a crash ``` ┌─────────────────────────────────┐ ┌──────────────────────────────────┐ -│ Live Indexer Process │ │ Replay CLI Process │ +│ Live Indexer Process │ │ Backfill CLI Process │ │ │ │ │ │ Discovery ──► Worker Pool │ │ CLI ──► Engine │ -│ │ │ │ ├── DiscoveryReplayer │ -│ ▼ │ │ └── MessageReplayer │ +│ │ │ │ ├── DiscoveryBackfill │ +│ ▼ │ │ └── MessageBackfill │ │ Scheduler │ │ │ └────────┬────────────────────────┘ └───────┬──────────────────────────┘ │ │ diff --git a/indexer/pkg/replay/discovery_replay.go b/indexer/pkg/backfill/discovery.go similarity index 86% rename from indexer/pkg/replay/discovery_replay.go rename to indexer/pkg/backfill/discovery.go index 934d03902..49f72b045 100644 --- a/indexer/pkg/replay/discovery_replay.go +++ b/indexer/pkg/backfill/discovery.go @@ -1,4 +1,4 @@ -package replay +package backfill import ( "context" @@ -9,9 +9,9 @@ import ( "github.com/smartcontractkit/chainlink-ccv/protocol" ) -func (e *Engine) runDiscoveryReplay(ctx context.Context, job *Job) error { +func (e *Engine) runDiscoveryBackfill(ctx context.Context, job *Job) error { if job.SinceSequenceNumber == nil { - return fmt.Errorf("discovery replay requires since_sequence_number") + return fmt.Errorf("discovery backfill requires since_sequence_number") } if e.aggregatorReaderFactory == nil { return fmt.Errorf("aggregator reader factory not configured") @@ -24,10 +24,10 @@ func (e *Engine) runDiscoveryReplay(ctx context.Context, job *Job) error { reader, err := e.aggregatorReaderFactory(sinceValue) if err != nil { - return fmt.Errorf("failed to create aggregator reader for replay: %w", err) + return fmt.Errorf("failed to create aggregator reader for backfill: %w", err) } - e.lggr.Infow("Discovery replay starting", + e.lggr.Infow("Discovery backfill starting", "jobID", job.ID, "sinceSequenceNumber", *job.SinceSequenceNumber, "resumeCursor", sinceValue, @@ -51,7 +51,7 @@ func (e *Engine) runDiscoveryReplay(ctx context.Context, job *Job) error { if len(responses) == 0 { consecutiveEmpty++ if consecutiveEmpty >= maxConsecutiveEmpty { - e.lggr.Infow("No more data from aggregator, discovery replay finished", + e.lggr.Infow("No more data from aggregator, discovery backfill finished", "jobID", job.ID, "totalProcessed", totalProcessed) return nil } @@ -61,7 +61,7 @@ func (e *Engine) runDiscoveryReplay(ctx context.Context, job *Job) error { consecutiveEmpty = 0 messages, verifications, _ := common.ConvertDiscoveryResponses(responses, time.Now(), e.registry) - e.lggr.Infow("Discovery replay batch", + e.lggr.Infow("Discovery backfill batch", "jobID", job.ID, "responses", len(responses), "messages", len(messages), @@ -76,7 +76,7 @@ func (e *Engine) runDiscoveryReplay(ctx context.Context, job *Job) error { if e.registry != nil { for _, resp := range responses { if err := e.gatherVerificationsForMessage(ctx, job, resp.Data); err != nil { - e.lggr.Warnw("Failed to gather verifications during discovery replay", + e.lggr.Warnw("Failed to gather verifications during discovery backfill", "jobID", job.ID, "messageID", resp.Data.MessageID, "error", err, @@ -104,7 +104,7 @@ func (e *Engine) persistDiscoveryBatch( ) error { encodable, skipped := common.FilterEncodableMessages(messages) for _, s := range skipped { - e.lggr.Warnw("Skipping non-encodable message in replay", "index", s.Index, "reason", s.Reason) + e.lggr.Warnw("Skipping non-encodable message in backfill", "index", s.Index, "reason", s.Reason) } if len(encodable) > 0 { diff --git a/indexer/pkg/replay/engine.go b/indexer/pkg/backfill/engine.go similarity index 81% rename from indexer/pkg/replay/engine.go rename to indexer/pkg/backfill/engine.go index 75a7e4c80..342571de9 100644 --- a/indexer/pkg/replay/engine.go +++ b/indexer/pkg/backfill/engine.go @@ -1,4 +1,4 @@ -package replay +package backfill import ( "context" @@ -15,7 +15,7 @@ import ( const cleanupTimeout = 5 * time.Second -// Storage is the subset of storage operations the replay engine needs. +// Storage is the subset of storage operations the backfill engine needs. type Storage interface { // UpsertVerifierResults inserts or overwrites verifier results based on force flag. UpsertVerifierResults(ctx context.Context, results []common.VerifierResultWithMetadata, force bool) error @@ -25,11 +25,11 @@ type Storage interface { GetMessage(ctx context.Context, messageID protocol.Bytes32) (common.MessageWithMetadata, error) } -// AggregatorReaderFactory creates a new aggregator reader for replay. +// AggregatorReaderFactory creates a new aggregator reader for backfill. // The caller is responsible for closing the returned reader. type AggregatorReaderFactory func(since int64) (*readers.ResilientReader, error) -// Engine orchestrates replay jobs with crash-recovery support. +// Engine orchestrates backfill jobs with crash-recovery support. type Engine struct { store *Store storage Storage @@ -58,7 +58,7 @@ func NewEngine( storage: storage, registry: reg, aggregatorReaderFactory: factory, - lggr: logger.Named(lggr, "ReplayEngine"), + lggr: logger.Named(lggr, "BackfillEngine"), batchThrottleDelay: 500 * time.Millisecond, } for _, opt := range opts { @@ -68,11 +68,11 @@ func NewEngine( } // Start creates a new job or resumes a stale one, acquires the advisory lock, -// and runs the replay to completion. +// and runs the backfill to completion. func (e *Engine) Start(ctx context.Context, req Request) (string, error) { job, err := e.findOrCreateJob(ctx, req) if err != nil { - return "", fmt.Errorf("failed to prepare replay job: %w", err) + return "", fmt.Errorf("failed to prepare backfill job: %w", err) } lockConn, err := e.store.AcquireAdvisoryLock(ctx, job.ID) @@ -81,7 +81,7 @@ func (e *Engine) Start(ctx context.Context, req Request) (string, error) { } defer func() { _ = lockConn.Close() }() - e.lggr.Infow("Starting replay", "jobID", job.ID, "type", job.Type, "force", job.ForceOverwrite, "cursor", job.ProgressCursor) + e.lggr.Infow("Starting backfill", "jobID", job.ID, "type", job.Type, "force", job.ForceOverwrite, "cursor", job.ProgressCursor) err = e.runJob(ctx, job) if err != nil { @@ -90,7 +90,7 @@ func (e *Engine) Start(ctx context.Context, req Request) (string, error) { if markErr := e.store.MarkFailed(cleanupCtx, job.ID, err.Error()); markErr != nil { e.lggr.Errorw("Failed to mark job as failed", "jobID", job.ID, "error", markErr) } - return job.ID, fmt.Errorf("replay job %s failed: %w", job.ID, err) + return job.ID, fmt.Errorf("backfill job %s failed: %w", job.ID, err) } cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout) @@ -100,7 +100,7 @@ func (e *Engine) Start(ctx context.Context, req Request) (string, error) { return job.ID, markErr } - e.lggr.Infow("Replay completed", "jobID", job.ID) + e.lggr.Infow("Backfill completed", "jobID", job.ID) return job.ID, nil } @@ -121,7 +121,7 @@ func (e *Engine) Resume(ctx context.Context, jobID string) error { } defer func() { _ = lockConn.Close() }() - e.lggr.Infow("Resuming replay", "jobID", job.ID, "type", job.Type, "cursor", job.ProgressCursor) + e.lggr.Infow("Resuming backfill", "jobID", job.ID, "type", job.Type, "cursor", job.ProgressCursor) err = e.runJob(ctx, job) if err != nil { @@ -130,7 +130,7 @@ func (e *Engine) Resume(ctx context.Context, jobID string) error { if markErr := e.store.MarkFailed(cleanupCtx, job.ID, err.Error()); markErr != nil { e.lggr.Errorw("Failed to mark job as failed", "jobID", job.ID, "error", markErr) } - return fmt.Errorf("replay job %s failed: %w", job.ID, err) + return fmt.Errorf("backfill job %s failed: %w", job.ID, err) } cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout) @@ -139,7 +139,7 @@ func (e *Engine) Resume(ctx context.Context, jobID string) error { return markErr } - e.lggr.Infow("Replay completed", "jobID", job.ID) + e.lggr.Infow("Backfill completed", "jobID", job.ID) return nil } @@ -163,19 +163,19 @@ func (e *Engine) findOrCreateJob(ctx context.Context, req Request) (*Job, error) job, err := e.store.CreateJob(ctx, req) if err != nil { - return nil, fmt.Errorf("failed to create replay job: %w", err) + return nil, fmt.Errorf("failed to create backfill job: %w", err) } - e.lggr.Infow("Created new replay job", "jobID", job.ID) + e.lggr.Infow("Created new backfill job", "jobID", job.ID) return job, nil } func (e *Engine) runJob(ctx context.Context, job *Job) error { switch job.Type { case TypeDiscovery: - return e.runDiscoveryReplay(ctx, job) + return e.runDiscoveryBackfill(ctx, job) case TypeMessages: - return e.runMessageReplay(ctx, job) + return e.runMessageBackfill(ctx, job) default: - return fmt.Errorf("unknown replay type: %s", job.Type) + return fmt.Errorf("unknown backfill type: %s", job.Type) } } diff --git a/indexer/pkg/replay/engine_test.go b/indexer/pkg/backfill/engine_test.go similarity index 87% rename from indexer/pkg/replay/engine_test.go rename to indexer/pkg/backfill/engine_test.go index 8f7cdfbd9..74f4304f4 100644 --- a/indexer/pkg/replay/engine_test.go +++ b/indexer/pkg/backfill/engine_test.go @@ -1,4 +1,4 @@ -package replay +package backfill import ( "context" @@ -16,8 +16,8 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" ) -// mockReplayStorage implements Storage for testing. -type mockReplayStorage struct { +// mockBackfillStorage implements Storage for testing. +type mockBackfillStorage struct { mu sync.Mutex messages []common.MessageWithMetadata verifications []common.VerifierResultWithMetadata @@ -27,11 +27,11 @@ type mockReplayStorage struct { forceUsed bool } -func newMockReplayStorage() *mockReplayStorage { - return &mockReplayStorage{} +func newMockBackfillStorage() *mockBackfillStorage { + return &mockBackfillStorage{} } -func (m *mockReplayStorage) UpsertMessages(_ context.Context, msgs []common.MessageWithMetadata, force bool) error { +func (m *mockBackfillStorage) UpsertMessages(_ context.Context, msgs []common.MessageWithMetadata, force bool) error { m.mu.Lock() defer m.mu.Unlock() if m.upsertMsgErr != nil { @@ -44,7 +44,7 @@ func (m *mockReplayStorage) UpsertMessages(_ context.Context, msgs []common.Mess return nil } -func (m *mockReplayStorage) UpsertVerifierResults(_ context.Context, results []common.VerifierResultWithMetadata, force bool) error { +func (m *mockBackfillStorage) UpsertVerifierResults(_ context.Context, results []common.VerifierResultWithMetadata, force bool) error { m.mu.Lock() defer m.mu.Unlock() if m.upsertCCVErr != nil { @@ -57,14 +57,14 @@ func (m *mockReplayStorage) UpsertVerifierResults(_ context.Context, results []c return nil } -func (m *mockReplayStorage) GetMessage(ctx context.Context, messageID protocol.Bytes32) (common.MessageWithMetadata, error) { +func (m *mockBackfillStorage) GetMessage(ctx context.Context, messageID protocol.Bytes32) (common.MessageWithMetadata, error) { if m.getMessageFunc != nil { return m.getMessageFunc(ctx, messageID) } return common.MessageWithMetadata{}, fmt.Errorf("message not found") } -func (m *mockReplayStorage) capturedMessages() []common.MessageWithMetadata { +func (m *mockBackfillStorage) capturedMessages() []common.MessageWithMetadata { m.mu.Lock() defer m.mu.Unlock() cp := make([]common.MessageWithMetadata, len(m.messages)) @@ -72,7 +72,7 @@ func (m *mockReplayStorage) capturedMessages() []common.MessageWithMetadata { return cp } -func (m *mockReplayStorage) capturedVerifications() []common.VerifierResultWithMetadata { +func (m *mockBackfillStorage) capturedVerifications() []common.VerifierResultWithMetadata { m.mu.Lock() defer m.mu.Unlock() cp := make([]common.VerifierResultWithMetadata, len(m.verifications)) @@ -126,7 +126,7 @@ func testVerifierResult(messageNumber int) common.VerifierResultWithMetadata { func TestPersistDiscoveryBatch(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() engine := &Engine{ @@ -157,7 +157,7 @@ func TestPersistDiscoveryBatch(t *testing.T) { func TestPersistDiscoveryBatch_ForceFlag(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() engine := &Engine{ @@ -187,7 +187,7 @@ func TestPersistDiscoveryBatch_ForceFlag(t *testing.T) { func TestPersistDiscoveryBatch_NoForceByDefault(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() engine := &Engine{ @@ -217,7 +217,7 @@ func TestPersistDiscoveryBatch_NoForceByDefault(t *testing.T) { func TestGatherAllVerifications_MessageNotFound(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() store.getMessageFunc = func(_ context.Context, _ protocol.Bytes32) (common.MessageWithMetadata, error) { @@ -244,7 +244,7 @@ func TestGatherAllVerifications_MessageNotFound(t *testing.T) { func TestGatherAllVerifications_NoCCVAddresses(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() vr := testVerifierResult(1) @@ -270,7 +270,7 @@ func TestGatherAllVerifications_NoCCVAddresses(t *testing.T) { func TestGatherAllVerifications_WithCCVAddressesButNoReaders(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() verifierAddr, _ := protocol.RandomAddress() @@ -302,7 +302,7 @@ func TestGatherAllVerifications_WithCCVAddressesButNoReaders(t *testing.T) { func TestGatherAllVerifications_CCVAddressesFromMessagesTable(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() ccvAddr, _ := protocol.RandomAddress() @@ -336,7 +336,7 @@ func TestGatherAllVerifications_CCVAddressesFromMessagesTable(t *testing.T) { func TestGatherAllVerifications_UpsertsMessage(t *testing.T) { lggr := logger.Test(t) - store := newMockReplayStorage() + store := newMockBackfillStorage() reg := registry.NewVerifierRegistry() vr := testVerifierResult(1) @@ -370,7 +370,7 @@ func TestGatherAllVerifications_UpsertsMessage(t *testing.T) { assert.True(t, store.forceUsed, "force flag should be passed to storage") } -func TestRunDiscoveryReplay_MissingSinceSequenceNumber(t *testing.T) { +func TestRunDiscoveryBackfill_MissingSinceSequenceNumber(t *testing.T) { lggr := logger.Test(t) engine := &Engine{ @@ -383,12 +383,12 @@ func TestRunDiscoveryReplay_MissingSinceSequenceNumber(t *testing.T) { Status: StatusRunning, } - err := engine.runDiscoveryReplay(context.Background(), job) + err := engine.runDiscoveryBackfill(context.Background(), job) require.Error(t, err) assert.Contains(t, err.Error(), "since_sequence_number") } -func TestRunDiscoveryReplay_MissingFactory(t *testing.T) { +func TestRunDiscoveryBackfill_MissingFactory(t *testing.T) { lggr := logger.Test(t) engine := &Engine{ @@ -403,12 +403,12 @@ func TestRunDiscoveryReplay_MissingFactory(t *testing.T) { SinceSequenceNumber: &sinceTS, } - err := engine.runDiscoveryReplay(context.Background(), job) + err := engine.runDiscoveryBackfill(context.Background(), job) require.Error(t, err) assert.Contains(t, err.Error(), "aggregator reader factory") } -func TestRunMessageReplay_EmptyMessageIDs(t *testing.T) { +func TestRunMessageBackfill_EmptyMessageIDs(t *testing.T) { lggr := logger.Test(t) reg := registry.NewVerifierRegistry() @@ -424,12 +424,12 @@ func TestRunMessageReplay_EmptyMessageIDs(t *testing.T) { MessageIDs: []string{}, } - err := engine.runMessageReplay(context.Background(), job) + err := engine.runMessageBackfill(context.Background(), job) require.Error(t, err) assert.Contains(t, err.Error(), "at least one message ID") } -func TestRunMessageReplay_NilRegistry(t *testing.T) { +func TestRunMessageBackfill_NilRegistry(t *testing.T) { lggr := logger.Test(t) engine := &Engine{ @@ -443,7 +443,7 @@ func TestRunMessageReplay_NilRegistry(t *testing.T) { MessageIDs: []string{"0x1234"}, } - err := engine.runMessageReplay(context.Background(), job) + err := engine.runMessageBackfill(context.Background(), job) require.Error(t, err) assert.Contains(t, err.Error(), "verifier registry") } diff --git a/indexer/pkg/replay/message_replay.go b/indexer/pkg/backfill/message.go similarity index 87% rename from indexer/pkg/replay/message_replay.go rename to indexer/pkg/backfill/message.go index efca9161b..d10842488 100644 --- a/indexer/pkg/replay/message_replay.go +++ b/indexer/pkg/backfill/message.go @@ -1,4 +1,4 @@ -package replay +package backfill import ( "context" @@ -10,9 +10,9 @@ import ( "github.com/smartcontractkit/chainlink-ccv/protocol" ) -func (e *Engine) runMessageReplay(ctx context.Context, job *Job) error { +func (e *Engine) runMessageBackfill(ctx context.Context, job *Job) error { if len(job.MessageIDs) == 0 { - return fmt.Errorf("message replay requires at least one message ID") + return fmt.Errorf("message backfill requires at least one message ID") } if e.registry == nil { return fmt.Errorf("verifier registry not configured") @@ -20,7 +20,7 @@ func (e *Engine) runMessageReplay(ctx context.Context, job *Job) error { startIdx := int(job.ProgressCursor) if startIdx > 0 { - e.lggr.Infow("Resuming message replay", "jobID", job.ID, "fromIndex", startIdx, "total", len(job.MessageIDs)) + e.lggr.Infow("Resuming message backfill", "jobID", job.ID, "fromIndex", startIdx, "total", len(job.MessageIDs)) } for i := startIdx; i < len(job.MessageIDs); i++ { @@ -38,7 +38,7 @@ func (e *Engine) runMessageReplay(ctx context.Context, job *Job) error { continue } - e.lggr.Infow("Replaying message", "jobID", job.ID, "messageID", msgIDHex, "index", i, "total", len(job.MessageIDs)) + e.lggr.Infow("Backfilling message", "jobID", job.ID, "messageID", msgIDHex, "index", i, "total", len(job.MessageIDs)) if err := e.gatherAllVerifications(ctx, job, msgID); err != nil { e.lggr.Warnw("Error gathering verifications for message", @@ -53,7 +53,7 @@ func (e *Engine) runMessageReplay(ctx context.Context, job *Job) error { } } - e.lggr.Infow("Message replay finished", "jobID", job.ID, "total", len(job.MessageIDs)) + e.lggr.Infow("Message backfill finished", "jobID", job.ID, "total", len(job.MessageIDs)) return nil } diff --git a/indexer/pkg/replay/store.go b/indexer/pkg/backfill/store.go similarity index 93% rename from indexer/pkg/replay/store.go rename to indexer/pkg/backfill/store.go index 86a540336..61d541658 100644 --- a/indexer/pkg/replay/store.go +++ b/indexer/pkg/backfill/store.go @@ -1,4 +1,4 @@ -package replay +package backfill import ( "context" @@ -15,13 +15,15 @@ import ( ) var ( - ErrJobNotFound = errors.New("replay job not found") - ErrJobLocked = errors.New("replay job is locked by another process") - ErrNoResumable = errors.New("no resumable replay job found") + ErrJobNotFound = errors.New("backfill job not found") + ErrJobLocked = errors.New("backfill job is locked by another process") + ErrNoResumable = errors.New("no resumable backfill job found") StaleJobTimeout = 5 * time.Minute ) // Store provides CRUD operations for the replay_jobs table. +// Note: the replay_jobs table name predates the backfill rename and is kept +// for migration compatibility. type Store struct { ds sqlutil.DataSource lggr logger.Logger @@ -36,7 +38,7 @@ func NewStore(ds sqlutil.DataSource, lggr logger.Logger) *Store { func NewStoreFromConfig(ctx context.Context, lggr logger.Logger, uri string, dbConfig pg.DBConfig, connMaxLifetime, connMaxIdleTime time.Duration) (*Store, error) { db, err := dbConfig.New(ctx, uri, pg.DriverPostgres) if err != nil { - return nil, fmt.Errorf("failed to open replay store connection: %w", err) + return nil, fmt.Errorf("failed to open backfill store connection: %w", err) } db.SetConnMaxLifetime(connMaxLifetime) db.SetConnMaxIdleTime(connMaxIdleTime) @@ -152,7 +154,7 @@ func (s *Store) AcquireAdvisoryLock(ctx context.Context, jobID string) (*sql.Con } var acquired bool - err = conn.QueryRowContext(ctx, `SELECT pg_try_advisory_lock(hashtext($1))`, "replay:"+jobID).Scan(&acquired) + err = conn.QueryRowContext(ctx, `SELECT pg_try_advisory_lock(hashtext($1))`, "backfill:"+jobID).Scan(&acquired) if err != nil { _ = conn.Close() return nil, fmt.Errorf("advisory lock query failed: %w", err) @@ -165,7 +167,7 @@ func (s *Store) AcquireAdvisoryLock(ctx context.Context, jobID string) (*sql.Con } // UpdateProgress atomically updates the job's cursor and heartbeat. This should -// be called within the same transaction that persists the replayed data. +// be called within the same transaction that persists the backfilled data. func (s *Store) UpdateProgress(ctx context.Context, tx sqlutil.DataSource, jobID string, cursor int64, processedItems int) error { query := ` UPDATE indexer.replay_jobs @@ -175,7 +177,7 @@ func (s *Store) UpdateProgress(ctx context.Context, tx sqlutil.DataSource, jobID now := time.Now() _, err := tx.ExecContext(ctx, query, cursor, processedItems, now, jobID) if err != nil { - return fmt.Errorf("failed to update replay progress: %w", err) + return fmt.Errorf("failed to update backfill progress: %w", err) } return nil } @@ -216,7 +218,7 @@ func (s *Store) ListJobs(ctx context.Context) ([]Job, error) { ` rows, err := s.query(ctx, query) if err != nil { - return nil, fmt.Errorf("failed to list replay jobs: %w", err) + return nil, fmt.Errorf("failed to list backfill jobs: %w", err) } defer func() { _ = rows.Close() }() diff --git a/indexer/pkg/replay/types.go b/indexer/pkg/backfill/types.go similarity index 84% rename from indexer/pkg/replay/types.go rename to indexer/pkg/backfill/types.go index 817b84e7b..07793f7f4 100644 --- a/indexer/pkg/replay/types.go +++ b/indexer/pkg/backfill/types.go @@ -1,4 +1,4 @@ -package replay +package backfill import ( "crypto/sha256" @@ -9,7 +9,7 @@ import ( "time" ) -// Type distinguishes between the two replay modes. +// Type distinguishes between the two backfill modes. type Type string const ( @@ -22,11 +22,11 @@ func ParseType(s string) (Type, error) { case TypeDiscovery, TypeMessages: return Type(s), nil default: - return "", errors.New("unknown replay type: " + s) + return "", errors.New("unknown backfill type: " + s) } } -// Status tracks the lifecycle of a replay job. +// Status tracks the lifecycle of a backfill job. type Status string const ( @@ -41,11 +41,11 @@ func ParseStatus(s string) (Status, error) { case StatusPending, StatusRunning, StatusCompleted, StatusFailed: return Status(s), nil default: - return "", errors.New("unknown replay status: " + s) + return "", errors.New("unknown backfill status: " + s) } } -// Job represents a persisted replay job stored in the replay_jobs table. +// Job represents a persisted backfill job stored in the replay_jobs table. type Job struct { ID string `json:"id"` Type Type `json:"type"` @@ -53,10 +53,10 @@ type Job struct { ForceOverwrite bool `json:"forceOverwrite"` RequestHash string `json:"requestHash"` - // Discovery replay params + // Discovery backfill params SinceSequenceNumber *int64 `json:"sinceSequenceNumber,omitempty"` - // Message replay params + // Message backfill params MessageIDs []string `json:"messageIds,omitempty"` // Progress tracking @@ -73,7 +73,7 @@ type Job struct { CompletedAt *time.Time `json:"completedAt,omitempty"` } -// Request is the input to start a new replay. +// Request is the input to start a new backfill. type Request struct { Type Type Since int64 diff --git a/indexer/pkg/replay/types_test.go b/indexer/pkg/backfill/types_test.go similarity index 99% rename from indexer/pkg/replay/types_test.go rename to indexer/pkg/backfill/types_test.go index 83c8a4628..2672e8f83 100644 --- a/indexer/pkg/replay/types_test.go +++ b/indexer/pkg/backfill/types_test.go @@ -1,4 +1,4 @@ -package replay +package backfill import ( "testing" diff --git a/indexer/pkg/storage/postgres.go b/indexer/pkg/storage/postgres.go index cdaf8a068..c8a466325 100644 --- a/indexer/pkg/storage/postgres.go +++ b/indexer/pkg/storage/postgres.go @@ -274,7 +274,7 @@ func buildBatchInsertCCVDataQuery(ccvDataList []common.VerifierResultWithMetadat } // buildBatchUpsertCCVDataQuery builds an INSERT ... ON CONFLICT DO UPDATE query -// for force-mode replay that overwrites existing CCV data. +// for force-mode backfill that overwrites existing CCV data. func buildBatchUpsertCCVDataQuery(ccvDataList []common.VerifierResultWithMetadata) (string, []any, error) { return buildBatchCCVDataQuery(ccvDataList, ccvDataConflictUpsert) } @@ -311,7 +311,7 @@ func (d *PostgresStorage) InsertVerifierResults(ctx context.Context, verifierRes } // UpsertVerifierResults inserts or overwrites verifier results depending on force. -// Used by the replay engine to support both backfill and force-overwrite modes. +// Used by the indexer backfill engine to support both its default and force-overwrite modes. func (d *PostgresStorage) UpsertVerifierResults(ctx context.Context, verifierResults []common.VerifierResultWithMetadata, force bool) error { if len(verifierResults) == 0 { return nil @@ -334,7 +334,7 @@ func (d *PostgresStorage) UpsertVerifierResults(ctx context.Context, verifierRes } // UpsertMessages inserts or overwrites messages depending on force. -// Used by the replay engine to support both backfill and force-overwrite modes. +// Used by the indexer backfill engine to support both its default and force-overwrite modes. func (d *PostgresStorage) UpsertMessages(ctx context.Context, messages []common.MessageWithMetadata, force bool) error { if len(messages) == 0 { return nil @@ -424,7 +424,7 @@ func buildBatchInsertMessagesQuery(messages []common.MessageWithMetadata) (strin } // buildBatchUpsertMessagesQuery builds an INSERT ... ON CONFLICT DO UPDATE query -// for force-mode replay that overwrites existing messages. +// for force-mode backfill that overwrites existing messages. func buildBatchUpsertMessagesQuery(messages []common.MessageWithMetadata) (string, []any, error) { return buildBatchMessagesQuery(messages, messagesConflictUpsert) }