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