feat(status): expose row-copy progress as a structured field on Progress

docs/migrate.md

@@ -669,10 +669,10 @@ Note that the whole report is a single log record containing newlines. Spirit's
 
 | Field | Meaning |
 | --- | --- |
-| `%` | Rows copied out of the estimated total. The total comes from table statistics, so the percentage can drift slightly and is not a row count you should reconcile against. |
-| `n/m` | The figures the percentage is derived from. |
+| `%` | Rows settled out of the estimated total. The total comes from table statistics, so the percentage is approximate and not a row count you should reconcile against. Rows the binlog applier wrote before the copy reached them are not counted (the copy inserts with `INSERT IGNORE`), so on a busy table the percentage finishes short of 100%. A resume from checkpoint continues the count from where the previous run left it. |
+| `n/m` | The figures the percentage is derived from: rows settled so far and the estimated table cardinality, the same numbers the status API reports per table and as `Copy`. |
 | `chunk-size` | Rows in the most recently claimed chunk. The chunker sizes chunks dynamically to hit the [`--target-chunk-size`](#target-chunk-size) byte budget, so this number moving is normal and healthy — it is how Spirit adapts to row width. A chunk size that has collapsed to its floor and stayed there means the rows are too wide to fit the budget even at the floor, i.e. lower `--target-chunk-size` than the data wants, not a struggling server. |
-| `eta` | Remaining rows divided by the recently measured copy rate. `TBD` for the first minute (no rate measured yet) and `DUE` past 99.99%. It is computed from a single 10-second sample, so early on it swings a lot; treat a large jump as noise unless it persists. |
+| `eta` | Remaining copy divided by the recently measured copy rate. On a table with an auto_increment key the remaining copy is distance through the key range, so `DUE` (past 99.99% of the range) need not coincide with the `%` column when the ids are sparse; on other keys it is the same rows-against-estimate ratio as `%`, and the two move together. `TBD` for the first minute (no rate measured yet). It is computed from a single 10-second sample, so early on it swings a lot; treat a large jump as noise unless it persists. |
 | `throttled` | Whether the copy is currently paused by a throttler (replica lag, commit latency, or load). A migration that is throttled is behaving as designed — it is protecting the server, not stalling. |
 
 The `checksum` row that replaces this one during the checksum phase has the same shape — including its own `chunk-size`, since the checksum sizes chunks dynamically as well, though against a fixed 5s time budget rather than the byte budget — plus `threads=` and `throttled=` for the checksum's own pacing.

pkg/copier/README.md

@@ -36,11 +36,14 @@ The trade-offs are higher network transfer and CPU for serialization, which the
 type Copier interface {
     Run(ctx context.Context) error
     GetETA() string
+    GetETAState() status.ETA
     GetChunker() table.Chunker
     SetThrottler(throttler throttler.Throttler)
     GetThrottler() throttler.Throttler
     StartTime() time.Time
     GetProgress() string
+    CopyProgress() status.CopyProgress
+    ChunkSize() uint64
 }
 ```
 
@@ -58,7 +61,9 @@ type ChunkCopier interface {
 
 - **`Run(ctx)`**: Starts the copy process and blocks until completion or error. Spawns multiple worker goroutines based on the configured concurrency level.
 - **`GetETA()`**: Returns estimated time to completion as a human-readable string. Returns "TBD" during the initial warmup period (1 minute), "DUE" when >99.99% complete, or a duration like "2h30m15s".
-- **`GetProgress()`**: Returns progress as "copied/total percentage%" (e.g., "1000000/5000000 20.00%").
+- **`GetETAState()`**: The same estimate as a `status.ETA{State, Duration}`, for callers that branch on whether an estimate exists yet. `GetETA()` is its `String()`.
+- **`CopyProgress()`**: Returns the copier's own progress as `status.CopyProgress{RowsCopied, RowsTotal}`. This is the measure the copier paces on: for the optimistic chunker it is keyspace distance against the auto_increment max, not a row count, so the runners report settled rows from the chunker instead (see `status.CopyFromTables`). `GetProgress()` is its rendered form, kept for interface compatibility; Spirit itself no longer calls it.
+- **`ChunkSize()`**: Rows in the most recently claimed chunk, the `chunk-size=` field of the status log block. The chunker sizes chunks dynamically to hit the target byte budget, so this moves during a copy.
 - **`GetChunker()`**: Returns the underlying chunker for accessing detailed progress information.
 - **`SetThrottler(throttler)`**: Updates the throttler used to control copy rate.
 - **`GetThrottler()`**: Returns the current throttler.
@@ -167,8 +172,10 @@ for {
     case <-ctx.Done():
         return
     case <-ticker.C:
-        progress := copier.GetProgress()
-        eta := copier.GetETA()
+        // Settled rows against the row estimates, summed over the tables.
+        // CopyProgress() is the copier's own pacing measure, not a row count.
+        progress := status.CopyFromTables(status.TablesFromChunker(copier.GetChunker()))
+        eta := copier.GetETAState()
         fmt.Printf("Progress: %s, ETA: %s\n", progress, eta)
     }
 }

pkg/copier/buffered.go

@@ -643,20 +643,9 @@ func (c *buffered) ChunkSize() uint64 {
 	return c.chunkSize.Load()
 }
 
+// GetETA renders GetETAState for the status block.
 func (c *buffered) GetETA() string {
-	c.Lock()
-	defer c.Unlock()
-	copiedRows, totalRows, pct := c.getCopyStats()
-	estimate, st := etaEstimate(copiedRows, totalRows, pct, c.rowsPerSecond.Load(), c.startTime)
-	switch st {
-	case status.ETADue:
-		return "DUE"
-	case status.ETAMeasuring:
-		return "TBD"
-	case status.ETAReady, status.ETANone:
-		// A ready estimate is formatted below; ETANone cannot occur during copy.
-	}
-	return estimate.String()
+	return c.GetETAState().String()
 }
 
 func (c *buffered) GetETAState() status.ETA {

pkg/copier/copier.go

@@ -65,9 +65,13 @@ type Copier interface {
 	GetThrottler() throttler.Throttler
 	StartTime() time.Time
 	GetProgress() string
-	// CopyProgress returns the same progress as GetProgress in numeric form,
-	// which the status block needs in order to lay the percentage and the
-	// row counts out as separate fields.
+	// CopyProgress returns the copier's own measure of the copy in numeric
+	// form: the chunker's Progress, which for the optimistic chunker is
+	// keyspace distance against the auto_increment max rather than rows. It
+	// is the measure the ETA is paced on. Callers reporting rows to a human
+	// or a wrapper should sum the chunker's per-table settled counts instead
+	// (status.CopyFromTables), which is what the runners do. GetProgress is
+	// this value rendered.
 	CopyProgress() status.CopyProgress
 	// ChunkSize returns the row count of the most recently claimed chunk, or
 	// 0 before the first one. This is the dynamic chunker's current sizing

pkg/copier/copiertest/stub.go

@@ -0,0 +1,29 @@
+// Package copiertest provides a Copier stub for tests of the runners that
+// report copier state, so every runner package shares one definition. It
+// lives beside the copier rather than in testutils because the copier's own
+// tests import testutils, which a stub there would turn into an import cycle.
+package copiertest
+
+import (
+	"github.com/block/spirit/pkg/copier"
+	"github.com/block/spirit/pkg/status"
+	"github.com/block/spirit/pkg/throttler"
+)
+
+// Stub answers the read-only status methods of copier.Copier from its
+// fields, and reports a throttler that never throttles. Every other method is
+// inherited from the embedded nil interface and panics if reached, which is
+// the point: a test that needs it is exercising more than status reporting.
+type Stub struct {
+	copier.Copier
+	ETA   status.ETA
+	Copy  status.CopyProgress
+	Chunk uint64
+}
+
+func (s Stub) GetETA() string                    { return s.ETA.String() }
+func (s Stub) GetETAState() status.ETA           { return s.ETA }
+func (s Stub) GetProgress() string               { return s.Copy.String() }
+func (s Stub) CopyProgress() status.CopyProgress { return s.Copy }
+func (s Stub) ChunkSize() uint64                 { return s.Chunk }
+func (s Stub) GetThrottler() throttler.Throttler { return &throttler.Noop{} }

pkg/datasync/progress_test.go

@@ -5,24 +5,12 @@ import (
 	"time"
 
 	"github.com/block/spirit/pkg/applier"
-	"github.com/block/spirit/pkg/copier"
+	"github.com/block/spirit/pkg/copier/copiertest"
 	"github.com/block/spirit/pkg/status"
 	"github.com/block/spirit/pkg/table"
 	"github.com/stretchr/testify/require"
 )
 
-type progressCopier struct{ copier.Copier }
-
-func (progressCopier) GetProgress() string { return "50%" }
-func (progressCopier) GetETA() string      { return "1m" }
-func (progressCopier) GetETAState() status.ETA {
-	return status.ETA{State: status.ETAReady, Duration: time.Minute}
-}
-func (progressCopier) CopyProgress() status.CopyProgress {
-	return status.CopyProgress{RowsCopied: 50, RowsTotal: 100}
-}
-func (progressCopier) ChunkSize() uint64 { return 25 }
-
 type progressApplier struct{ applier.Applier }
 
 func (progressApplier) Stats() applier.Stats { return applier.Stats{ActiveWorkers: 4} }
@@ -31,21 +19,44 @@ func TestSyncProgressAndLogFormat(t *testing.T) {
 	r, err := NewRunner(&Sync{})
 	require.NoError(t, err)
 	require.Empty(t, r.Progress().ETA)
-	r.copyChunker = table.NewMultiChunker(table.NewMockChunker("b", 100), table.NewMockChunker("a", 200))
-	r.copier = progressCopier{}
-	r.applier = progressApplier{}
+	b := table.NewMockChunker("b", 100)
+	a := table.NewMockChunker("a", 200)
+	b.Feedback(nil, 0, 30) // rows settled by the applier
+	a.Feedback(nil, 0, 40)
+	r.copyChunker = table.NewMultiChunker(b, a)
+	// With a chunker but no copier, the API and the log block agree: settled
+	// rows from the chunker, and an ETA that is not yet measured.
 	r.status.Set(status.CopyRows)
 	p := r.Progress()
+	require.Equal(t, status.CopyProgress{RowsCopied: 70, RowsTotal: 300}, p.Copy)
+	require.Equal(t, status.ETA{State: status.ETAMeasuring}, p.ETA)
+	require.Equal(t, "70/300 23.33% copyRows ETA TBD", p.Summary)
+	require.Contains(t, r.Status(), " 23.33%  70/300  chunk-size=0  eta=TBD")
+	r.copier = copiertest.Stub{
+		ETA: status.ETA{State: status.ETAReady, Duration: time.Minute},
+		// The copier's own measure, which neither Progress nor Status may report.
+		Copy:  status.CopyProgress{RowsCopied: 7, RowsTotal: 9},
+		Chunk: 25,
+	}
+	r.applier = progressApplier{}
+	r.status.Set(status.CopyRows)
+	p = r.Progress()
 	require.Equal(t, status.ETA{State: status.ETAReady, Duration: time.Minute}, p.ETA)
+	require.Equal(t, status.CopyProgress{RowsCopied: 70, RowsTotal: 300}, p.Copy) // Both counters summed across Tables.
+	require.Equal(t, "70/300 23.33% copyRows ETA 1m0s", p.Summary)
 	require.Len(t, p.Tables, 2)
 	require.Less(t, p.Tables[0].TableName, p.Tables[1].TableName)
 	block := r.Status()
 	for _, text := range []string{"copier-time=", "\n  copier", "\n  applier", "\n  binlog", "\n  ckpt"} {
 		require.Contains(t, block, text)
 	}
+	// The log block reports the same copy measure as the API, on the same tick.
+	require.Contains(t, block, " 23.33%  70/300  chunk-size=25  eta=1m0s")
+	require.NotContains(t, block, "7/9")
 	r.status.Set(status.ApplyChangeset)
 	require.Empty(t, r.Progress().ETA)
-	require.Empty(t, r.Progress().Checksum) // The continuous verifier has no finite initial-checksum phase.
+	require.Equal(t, status.CopyProgress{RowsCopied: 70, RowsTotal: 300}, r.Progress().Copy) // The copy reading outlives the copy phase.
+	require.Empty(t, r.Progress().Checksum)                                                  // The continuous verifier has no finite initial-checksum phase.
 	r.status.Set(status.RestoreSecondaryIndexes)
 	block = r.Status()
 	require.Contains(t, block, "state-time=")

pkg/datasync/runner.go

@@ -1599,16 +1599,29 @@ func (r *Runner) Progress() status.Progress {
 	repl := r.replClient
 	r.progMu.RUnlock()
 
+	tables := status.TablesFromChunker(chunker)
+	// The runner-wide copy is the sum of the per-table rows, so it reconciles
+	// with Tables and keeps its final reading once the copy has finished.
+	// Status derives its copier row the same way, so the API and the log
+	// block report one measure. The copier's own progress is not used for
+	// either: on an auto_increment key that measures keyspace distance, not
+	// rows.
+	copyProgress := status.CopyFromTables(tables)
+
 	var summary string
 	var eta status.ETA
 	switch state { //nolint:exhaustive // sync does not reach the cutover/checksum states
 	case status.CopyRows:
+		// The copy phase is entered only after the pipeline is built, so the
+		// copier is normally present; without one the estimate is not yet
+		// measured, the same reading Status gives.
+		eta = status.ETA{State: status.ETAMeasuring}
 		if cp != nil {
-			summary = fmt.Sprintf("%s copyRows ETA %s", cp.GetProgress(), cp.GetETA())
+			// One copier read, so the ETA in Summary and the ETA field
+			// describe the same instant.
 			eta = cp.GetETAState()
-		} else {
-			summary = "copyRows"
 		}
+		summary = fmt.Sprintf("%s copyRows ETA %s", copyProgress.String(), eta.String())
 	case status.ApplyChangeset:
 		if repl != nil {
 			summary = fmt.Sprintf("continuous sync position=%s pending-changes=%d", repl.Position(), repl.GetDeltaLen())
@@ -1619,14 +1632,13 @@ func (r *Runner) Progress() status.Progress {
 		summary = state.String()
 	}
 
-	tables := status.TablesFromChunker(chunker)
-
 	return status.Progress{
 		CurrentState: state,
 		Summary:      summary,
 		Resume:       r.resuming.Load(),
 		Tables:       tables,
 		ETA:          eta,
+		Copy:         copyProgress,
 		// Throttle is deliberately left zero: a sync copies through a Noop
 		// throttler, so there is nothing to report yet.
 	}
@@ -1640,6 +1652,7 @@ func (r *Runner) Status() string {
 
 	r.progMu.RLock()
 	cp := r.copier
+	chunker := r.copyChunker
 	repl := r.replClient
 	appl := r.applier
 	r.progMu.RUnlock()
@@ -1652,19 +1665,30 @@ func (r *Runner) Status() string {
 	switch state { //nolint:exhaustive // sync does not reach the cutover/checksum states
 	case status.CopyRows:
 		b := status.NewBlock("sync status: state=%s total-time=%s copier-time=%s", state.String(), elapsed, r.status.Elapsed().Round(time.Second))
-		// The copy pipeline is built asynchronously, so a status tick can land
-		// before there is a copier to report on.
-		if cp != nil {
-			progress := cp.CopyProgress()
+		// The chunker and the copier are published separately while the
+		// pipeline is built, and the copy phase is entered only once both
+		// exist, so these guards are defensive. The figures are settled rows
+		// from the chunker, the same measure Progress reports rather than the
+		// copier's own keyspace position; chunk-size and the ETA come from
+		// the copier and read as nothing claimed and nothing measured without
+		// one, as Progress does.
+		if chunker != nil {
+			progress := status.CopyFromTables(status.TablesFromChunker(chunker))
+			var chunkSize uint64
+			eta := status.ETA{State: status.ETAMeasuring}
+			if cp != nil {
+				chunkSize = cp.ChunkSize()
+				eta = cp.GetETAState()
+			}
 			// No throttled= here, unlike migrate and move: a sync copies
 			// through a Noop throttler, so the field would be a constant
 			// false.
 			b.Row("copier", "%6.2f%%  %d/%d  chunk-size=%d  eta=%s",
 				progress.Fraction()*100,
 				progress.RowsCopied,
 				progress.RowsTotal,
-				cp.ChunkSize(),
-				cp.GetETA(),
+				chunkSize,
+				eta.String(),
 			)
 		}
 		b.Row("applier", "%s", applier.StatusRow(appl))

pkg/migration/binlog_test.go

@@ -162,7 +162,7 @@ func TestE2EBinlogSubscribingCompositeKey(t *testing.T) {
 	require.NotNil(t, chunk)
 	require.Equal(t, "((`id1` < 1001)\n OR (`id1` = 1001 AND `id2` < 1))", chunk.String())
 	require.NoError(t, ccopier.CopyChunk(t.Context(), chunk))
-	require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1000/1200 83.33% copyRows ETA TBD", ETA: status.ETA{State: status.ETAMeasuring}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1000, RowsTotal: 1200, IsComplete: false}}}, m.Progress())
+	require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1000/1200 83.33% copyRows ETA TBD", ETA: status.ETA{State: status.ETAMeasuring}, Copy: status.CopyProgress{RowsCopied: 1000, RowsTotal: 1200}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1000, RowsTotal: 1200, IsComplete: false}}}, m.Progress())
 
 	// Now insert some data.
 	testutils.RunSQL(t, `insert into e2et1 (id1, id2) values (1002, 2)`)
@@ -177,7 +177,7 @@ func TestE2EBinlogSubscribingCompositeKey(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "((`id1` > 1001)\n OR (`id1` = 1001 AND `id2` >= 1))", chunk.String())
 	require.NoError(t, ccopier.CopyChunk(t.Context(), chunk))
-	require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1201/1200 100.08% copyRows ETA DUE", ETA: status.ETA{State: status.ETADue}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress())
+	require.Equal(t, status.Progress{CurrentState: status.CopyRows, Summary: "1201/1200 100.08% copyRows ETA DUE", ETA: status.ETA{State: status.ETADue}, Copy: status.CopyProgress{RowsCopied: 1201, RowsTotal: 1200}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress())
 
 	// Now insert some data.
 	// This should be picked up by the binlog subscription
@@ -207,7 +207,8 @@ func TestE2EBinlogSubscribingCompositeKey(t *testing.T) {
 	m.dbConfig = dbconn.NewDBConfig()
 	require.NoError(t, m.checksum(t.Context()))
 	require.Equal(t, "postChecksum", m.status.Get().String())
-	require.Equal(t, status.Progress{CurrentState: status.PostChecksum, Summary: "Applying Changeset Deltas=0", Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress())
+	// The copy reading outlives the copy phase.
+	require.Equal(t, status.Progress{CurrentState: status.PostChecksum, Summary: "Applying Changeset Deltas=0", Copy: status.CopyProgress{RowsCopied: 1201, RowsTotal: 1200}, Tables: []status.TableProgress{{TableName: "e2et1", RowsCopied: 1201, RowsTotal: 1200, IsComplete: true}}}, m.Progress())
 
 	// All done!
 	require.Equal(t, 0, m.db.Stats().InUse) // all connections are returned.