Backfill all replays completed tables; backfill status survives restarts

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

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**:

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 {

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")
+	}
+}