From 5682c5771b344677ffa58fd4100ca74f9a7e4674 Mon Sep 17 00:00:00 2001 From: Eric Moss Date: Sun, 4 Oct 2026 22:48:26 -0400 Subject: [PATCH] fix(backfill): "all" replays completed tables; status survives restarts POST /backfill {"all":true} called EnqueueAll, which skips tables that already completed, so on a running deployment it queued nothing and reported success. It now calls ReplayAll: every published table is replayed from the start, as the README and dashboard describe, and the response says how many were queued. The startup auto-backfill keeps EnqueueAll's resume semantics. Backfill status lived only in memory, so after a restart GET /backfill and the dashboard showed nothing although gtc_backfill_state recorded every table as done. The state table gains rows_copied (added in place on existing installs); the coordinator loads it at start and reports completed backfills with their row counts and completion time. The stats table list includes every backfilled table, so it is not empty until the first live change. Replaying a table cleared its persisted state in a goroutine after queueing it; for a small table that delete could land after the replay saved its completion. The state is now cleared before the table is queued. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 1 + README.md | 13 +- .../adapters/primary/backfill/coordinator.go | 123 ++++++++++++++---- .../primary/backfill/coordinator_test.go | 90 ++++++++++++- internal/adapters/primary/backfill/db.go | 34 +++-- internal/core/ports/backfill.go | 5 +- internal/infrastructure/server/server.go | 8 +- internal/infrastructure/server/stats.go | 10 ++ 8 files changed, 243 insertions(+), 41 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 1b8afc4..3586392 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -65,6 +65,7 @@ Consequences to keep in mind: re-describes tables with nothing to diff against, so a missed change cannot be redelivered. Observers must therefore never fail the WAL stream — `ports.SchemaObserver` returns no error by design. +- `POST /backfill` replays from the start: `{"table":...}` one table, `{"all":true}` every published table, completed or not (`ReplayAll`). The startup auto-backfill uses `EnqueueAll`, which only fills tables not yet done. `gtc_backfill_state` persists cursor, done and rows_copied; on start the coordinator loads it so completed backfills show in status after a restart. - Backfill emits `READ` events for existing rows, interleaved with the live stream via WAL watermarks (`pg_logical_emit_message`, prefix `gtc-backfill`). Chunk rows superseded by live events between the low and diff --git a/README.md b/README.md index 1e862a9..3b3f01b 100644 --- a/README.md +++ b/README.md @@ -523,18 +523,25 @@ Netflix's DBLog paper: streaming never pauses. Progress is persisted per table (in `gtc_backfill_state` in the source -database), so an interrupted backfill resumes where it left off. Tables -without a primary key are skipped with a warning. +database, including the rows copied), so an interrupted backfill resumes where +it left off and completed backfills still show in `GET /backfill` and the +dashboard after a restart. Tables without a primary key are skipped with a +warning. The HTTP API doubles as a **replay** mechanism — re-sync a table into the sinks at any time (e.g. after wiping a search index): ```bash curl -X POST localhost:8080/backfill -d '{"table":"public.products"}' -curl -X POST localhost:8080/backfill -d '{"all":true}' +curl -X POST localhost:8080/backfill -d '{"all":true}' # {"status":"enqueued","tables":N} curl localhost:8080/backfill # per-table progress ``` +Both forms replay from the beginning, including tables that already +completed; `all` covers every published table and reports how many it +queued. Tables already running or queued are left alone. (The automatic +backfill at slot creation only fills tables that have not completed.) + ## Delivery semantics GTC is **at-least-once**: diff --git a/internal/adapters/primary/backfill/coordinator.go b/internal/adapters/primary/backfill/coordinator.go index 5eacd50..8b0eb77 100644 --- a/internal/adapters/primary/backfill/coordinator.go +++ b/internal/adapters/primary/backfill/coordinator.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "log/slog" + "slices" "sync" "time" @@ -102,12 +103,26 @@ func (c *Coordinator) Start(ctx context.Context) { c.mu.Lock() c.state = state for name, ts := range state { - if !ts.Done { - if ref, ok := splitTableName(name); ok { - c.logger.Info("resuming interrupted backfill", slog.String("table", name)) - c.enqueueLocked(ref) + ref, ok := splitTableName(name) + if !ok { + continue + } + if ts.Done { + // Show backfills finished by earlier runs, so status does not + // read empty after every restart. + completed := ts.UpdatedAt.UTC() + c.status[name] = &domain.BackfillTableStatus{ + Schema: ref.Schema, + Table: ref.Table, + State: domain.BackfillDone, + RowsCopied: ts.RowsCopied, + CompletedAt: &completed, } + continue } + c.logger.Info("resuming interrupted backfill", slog.String("table", name)) + c.enqueueLocked(ref) + c.status[name].RowsCopied = ts.RowsCopied } c.mu.Unlock() } @@ -127,29 +142,80 @@ func (c *Coordinator) EnqueueTable(schema, table string) error { return fmt.Errorf("table %s is excluded from CDC", ref) } + c.mu.Lock() + busy := c.busyLocked(ref) + c.mu.Unlock() + if busy { + return fmt.Errorf("backfill for %s is already running or queued", ref) + } + + // Re-triggering a table is the replay primitive: start over from the + // beginning regardless of prior completion. The persisted state is cleared + // before the table is queued: clearing afterwards could delete the + // completion a small table records within milliseconds. + if err := c.db.ClearState(context.Background(), ref.String()); err != nil { + c.logger.Warn("failed to clear backfill state", slog.String("error", err.Error())) + } + c.mu.Lock() defer c.mu.Unlock() + if c.busyLocked(ref) { + return fmt.Errorf("backfill for %s is already running or queued", ref) + } + delete(c.state, ref.String()) + c.enqueueLocked(ref) + return nil +} - if c.current != nil && c.current.table == ref { - return fmt.Errorf("backfill for %s is already running", ref) +// ReplayAll re-backfills every published table from the beginning, including +// tables that already completed. Tables already running or queued are left as +// they are. It returns how many tables were queued. +func (c *Coordinator) ReplayAll(ctx context.Context) (int, error) { + tables, err := c.db.PublicationTables(ctx, c.cfg.PublicationName) + if err != nil { + return 0, err + } + + replay := make([]tableRef, 0, len(tables)) + c.mu.Lock() + for _, t := range tables { + ref := tableRef{Schema: t[0], Table: t[1]} + if c.isExcluded(ref) || c.busyLocked(ref) { + continue + } + replay = append(replay, ref) } - for _, queued := range c.queue { - if queued == ref { - return fmt.Errorf("backfill for %s is already queued", ref) + c.mu.Unlock() + + for _, ref := range replay { + if err := c.db.ClearState(ctx, ref.String()); err != nil { + c.logger.Warn("failed to clear backfill state", + slog.String("table", ref.String()), slog.String("error", err.Error())) } } - // Re-triggering a table is the replay primitive: start over from the - // beginning regardless of prior completion. - delete(c.state, ref.String()) - go func() { - if err := c.db.ClearState(context.Background(), ref.String()); err != nil { - c.logger.Warn("failed to clear backfill state", slog.String("error", err.Error())) + c.mu.Lock() + defer c.mu.Unlock() + queued := 0 + for _, ref := range replay { + if c.busyLocked(ref) { + continue } - }() + delete(c.state, ref.String()) + c.enqueueLocked(ref) + queued++ + } - c.enqueueLocked(ref) - return nil + c.logger.Info("backfill replay enqueued", slog.Int("tables", queued)) + return queued, nil +} + +// busyLocked reports whether the table is running or queued. Callers hold c.mu. +func (c *Coordinator) busyLocked(ref tableRef) bool { + if c.current != nil && c.current.table == ref { + return true + } + return slices.Contains(c.queue, ref) } func (c *Coordinator) EnqueueAll(ctx context.Context) error { @@ -347,12 +413,18 @@ func (c *Coordinator) backfillTable(ctx context.Context, ref tableRef) error { } c.mu.Lock() - cursor := c.state[ref.String()].Cursor + prev := c.state[ref.String()] c.mu.Unlock() + cursor := prev.Cursor + var copied int64 + if len(cursor) > 0 { + copied = prev.RowsCopied // resuming: keep the count from before + } now := time.Now().UTC() c.setStatus(ref, func(s *domain.BackfillTableStatus) { s.State = domain.BackfillRunning + s.RowsCopied = copied if s.StartedAt == nil { s.StartedAt = &now } @@ -370,10 +442,11 @@ func (c *Coordinator) backfillTable(ctx context.Context, ref tableRef) error { } if rowCount == 0 { + final := TableState{Done: true, RowsCopied: copied} c.mu.Lock() - c.state[ref.String()] = TableState{Done: true} + c.state[ref.String()] = final c.mu.Unlock() - if err := c.db.SaveState(ctx, ref.String(), nil, true); err != nil { + if err := c.db.SaveState(ctx, ref.String(), final); err != nil { c.logger.Warn("failed to persist backfill completion", slog.String("error", err.Error())) } done := time.Now().UTC() @@ -387,16 +460,18 @@ func (c *Coordinator) backfillTable(ctx context.Context, ref tableRef) error { } cursor = nextCursor + copied += int64(delivered) + progress := TableState{Cursor: cursor, RowsCopied: copied} c.mu.Lock() - c.state[ref.String()] = TableState{Cursor: cursor} + c.state[ref.String()] = progress c.mu.Unlock() - if err := c.db.SaveState(ctx, ref.String(), cursor, false); err != nil { + if err := c.db.SaveState(ctx, ref.String(), progress); err != nil { c.logger.Warn("failed to persist backfill progress", slog.String("error", err.Error())) } metrics.BackfillRowsTotal.WithLabelValues(ref.Schema, ref.Table).Add(float64(delivered)) c.setStatus(ref, func(s *domain.BackfillTableStatus) { - s.RowsCopied += int64(delivered) + s.RowsCopied = copied }) if c.cfg.ChunkDelay > 0 { diff --git a/internal/adapters/primary/backfill/coordinator_test.go b/internal/adapters/primary/backfill/coordinator_test.go index 3b95d2b..de268f4 100644 --- a/internal/adapters/primary/backfill/coordinator_test.go +++ b/internal/adapters/primary/backfill/coordinator_test.go @@ -19,6 +19,7 @@ type fakeDB struct { selects int watermarks chan watermarkPayload saved chan TableState + loaded map[string]TableState // returned by LoadState; nil means empty } func newFakeDB(pk []string, chunks [][]Row) *fakeDB { @@ -65,10 +66,13 @@ func (f *fakeDB) PublicationTables(context.Context, string) ([][2]string, error) func (f *fakeDB) EnsureStateTable(context.Context) error { return nil } func (f *fakeDB) LoadState(context.Context) (map[string]TableState, error) { + if f.loaded != nil { + return f.loaded, nil + } return map[string]TableState{}, nil } -func (f *fakeDB) SaveState(_ context.Context, _ string, cursor []string, done bool) error { - f.saved <- TableState{Cursor: cursor, Done: done} +func (f *fakeDB) SaveState(_ context.Context, _ string, state TableState) error { + f.saved <- state return nil } func (f *fakeDB) ClearState(context.Context, string) error { return nil } @@ -263,3 +267,85 @@ func TestNonReplicationURL(t *testing.T) { t.Errorf("URL without replication param must be unchanged, got %q", got) } } + +func TestStartShowsBackfillsCompletedBeforeRestart(t *testing.T) { + completedAt := time.Date(2026, 10, 5, 2, 19, 42, 0, time.UTC) + db := newFakeDB([]string{"id"}, nil) + db.loaded = map[string]TableState{ + "public.users": {Done: true, RowsCopied: 42, UpdatedAt: completedAt}, + } + c := NewCoordinator(db, Config{SlotName: "slot1"}, slog.Default()) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c.Start(ctx) + + status := c.Status() + if len(status) != 1 { + t.Fatalf("want the completed table in status after a restart, got %+v", status) + } + got := status[0] + if got.State != domain.BackfillDone || got.RowsCopied != 42 { + t.Errorf("want done with 42 rows, got %s with %d", got.State, got.RowsCopied) + } + if got.CompletedAt == nil || !got.CompletedAt.Equal(completedAt) { + t.Errorf("want completed_at %v, got %v", completedAt, got.CompletedAt) + } +} + +func TestReplayAllRequeuesCompletedTables(t *testing.T) { + db := newFakeDB([]string{"id"}, nil) + c := NewCoordinator(db, Config{SlotName: "slot1", PublicationName: "pub"}, slog.Default()) + c.state = map[string]TableState{"public.users": {Done: true, RowsCopied: 3}} + + // Resume semantics (startup): a completed table is not queued again. + if err := c.EnqueueAll(context.Background()); err != nil { + t.Fatal(err) + } + if len(c.queue) != 0 { + t.Fatalf("EnqueueAll must skip completed tables, queued %v", c.queue) + } + + queued, err := c.ReplayAll(context.Background()) + if err != nil { + t.Fatal(err) + } + if queued != 1 || len(c.queue) != 1 { + t.Fatalf("ReplayAll must queue the completed table, queued=%d queue=%v", queued, c.queue) + } + if _, ok := c.state["public.users"]; ok { + t.Error("ReplayAll must reset the table's progress") + } + + // A table already queued is not queued twice. + if again, _ := c.ReplayAll(context.Background()); again != 0 { + t.Errorf("want 0 tables queued while the replay is pending, got %d", again) + } +} + +func TestBackfillPersistsRowsCopied(t *testing.T) { + db := newFakeDB([]string{"id"}, [][]Row{{row(1, "ada"), row(2, "bob")}}) + c := NewCoordinator(db, Config{SlotName: "slot1", ChunkSize: 10}, slog.Default()) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c.Start(ctx) + if err := c.EnqueueTable("public", "users"); err != nil { + t.Fatal(err) + } + + low := waitWatermark(t, db, "low") + c.HandleWatermark("0/1", marshal(t, low)) + high := waitWatermark(t, db, "high") + c.HandleWatermark("0/2", marshal(t, high)) + c.WatermarkDelivered(marshal(t, high)) + + select { + case saved := <-db.saved: + if saved.RowsCopied != 2 || len(saved.Cursor) == 0 { + t.Errorf("want the chunk's progress saved with 2 rows copied, got %+v", saved) + } + case <-time.After(5 * time.Second): + t.Fatal("backfill never saved progress") + } +} diff --git a/internal/adapters/primary/backfill/db.go b/internal/adapters/primary/backfill/db.go index 5d73a3c..c2e11c5 100644 --- a/internal/adapters/primary/backfill/db.go +++ b/internal/adapters/primary/backfill/db.go @@ -6,6 +6,7 @@ import ( "fmt" "net/url" "strings" + "time" "github.com/emoss08/gtc/internal/adapters/primary/wal" "github.com/jackc/pgx/v5" @@ -38,7 +39,7 @@ type DB interface { EnsureStateTable(ctx context.Context) error LoadState(ctx context.Context) (map[string]TableState, error) - SaveState(ctx context.Context, table string, cursor []string, done bool) error + SaveState(ctx context.Context, table string, state TableState) error ClearState(ctx context.Context, table string) error Close() @@ -48,6 +49,10 @@ type DB interface { type TableState struct { Cursor []string Done bool + // RowsCopied counts rows delivered so far, so status survives restarts. + RowsCopied int64 + // UpdatedAt is when the state was last saved; loaded, never written. + UpdatedAt time.Time } type pgDB struct { @@ -231,6 +236,13 @@ func (d *pgDB) EnsureStateTable(ctx context.Context) error { done boolean NOT NULL DEFAULT false, updated_at timestamptz NOT NULL DEFAULT now() )`, quoteIdent(d.stateTable))) + if err != nil { + return err + } + // Added after the table shipped; existing installs gain it in place. + _, err = d.pool.Exec(ctx, fmt.Sprintf( + `ALTER TABLE %s ADD COLUMN IF NOT EXISTS rows_copied bigint NOT NULL DEFAULT 0`, + quoteIdent(d.stateTable))) return err } @@ -241,7 +253,7 @@ func (d *pgDB) LoadState(ctx context.Context) (map[string]TableState, error) { } rows, err := d.pool.Query(ctx, fmt.Sprintf( - "SELECT table_name, cursor, done FROM %s", quoteIdent(d.stateTable))) + "SELECT table_name, cursor, done, rows_copied, updated_at FROM %s", quoteIdent(d.stateTable))) if err != nil { return nil, err } @@ -251,10 +263,12 @@ func (d *pgDB) LoadState(ctx context.Context) (map[string]TableState, error) { var name string var cursorJSON []byte var done bool - if err := rows.Scan(&name, &cursorJSON, &done); err != nil { + var rowsCopied int64 + var updatedAt time.Time + if err := rows.Scan(&name, &cursorJSON, &done, &rowsCopied, &updatedAt); err != nil { return nil, err } - ts := TableState{Done: done} + ts := TableState{Done: done, RowsCopied: rowsCopied, UpdatedAt: updatedAt} if len(cursorJSON) > 0 { if err := json.Unmarshal(cursorJSON, &ts.Cursor); err != nil { return nil, fmt.Errorf("corrupt cursor for %s: %w", name, err) @@ -265,21 +279,21 @@ func (d *pgDB) LoadState(ctx context.Context) (map[string]TableState, error) { return state, rows.Err() } -func (d *pgDB) SaveState(ctx context.Context, table string, cursor []string, done bool) error { +func (d *pgDB) SaveState(ctx context.Context, table string, state TableState) error { if d.stateTable == "" { return nil } - cursorJSON, err := json.Marshal(cursor) + cursorJSON, err := json.Marshal(state.Cursor) if err != nil { return err } _, err = d.pool.Exec(ctx, fmt.Sprintf(` - INSERT INTO %s (table_name, cursor, done, updated_at) - VALUES ($1, $2, $3, now()) + INSERT INTO %s (table_name, cursor, done, rows_copied, updated_at) + VALUES ($1, $2, $3, $4, now()) ON CONFLICT (table_name) - DO UPDATE SET cursor = $2, done = $3, updated_at = now()`, + DO UPDATE SET cursor = $2, done = $3, rows_copied = $4, updated_at = now()`, quoteIdent(d.stateTable)), - table, cursorJSON, done, + table, cursorJSON, state.Done, state.RowsCopied, ) return err } diff --git a/internal/core/ports/backfill.go b/internal/core/ports/backfill.go index 8826dcf..cd94bf3 100644 --- a/internal/core/ports/backfill.go +++ b/internal/core/ports/backfill.go @@ -34,7 +34,10 @@ type BackfillManager interface { // progress (this is also the replay primitive). EnqueueTable(schema, table string) error // EnqueueAll schedules every table in the publication that has not - // already completed a backfill. + // already completed a backfill (resume semantics, used at startup). EnqueueAll(ctx context.Context) error + // ReplayAll re-backfills every table in the publication from the start, + // completed or not, and returns how many tables were queued. + ReplayAll(ctx context.Context) (int, error) Status() []domain.BackfillTableStatus } diff --git a/internal/infrastructure/server/server.go b/internal/infrastructure/server/server.go index 94c58b0..93465a1 100644 --- a/internal/infrastructure/server/server.go +++ b/internal/infrastructure/server/server.go @@ -259,10 +259,16 @@ func (s *Server) handleBackfillTrigger(w http.ResponseWriter, r *http.Request) { switch { case req.All: - if err := s.backfill.EnqueueAll(r.Context()); err != nil { + // "All" replays every published table, including completed ones; + // skipping those would make the request a silent no-op. + queued, err := s.backfill.ReplayAll(r.Context()) + if err != nil { s.writeBackfillError(w, err) return } + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(map[string]any{"status": "enqueued", "tables": queued}) + return case req.Table != "": schema, table := "public", req.Table if idx := strings.IndexByte(req.Table, '.'); idx > 0 { diff --git a/internal/infrastructure/server/stats.go b/internal/infrastructure/server/stats.go index ffa1bab..f1b1734 100644 --- a/internal/infrastructure/server/stats.go +++ b/internal/infrastructure/server/stats.go @@ -198,6 +198,16 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) { } sort.Slice(resp.Sinks, func(i, j int) bool { return resp.Sinks[i].Name < resp.Sinks[j].Name }) + // Event counters start at zero with the process; list every backfilled + // (published) table anyway, so a fresh restart still shows what is + // captured instead of an empty table. + for _, b := range resp.Backfill { + name := b.Schema + "." + b.Table + if _, ok := tables[name]; !ok { + tables[name] = &tableStats{Table: name} + } + } + for _, ts := range tables { resp.Tables = append(resp.Tables, *ts) }