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

pkg/datasync/progress_test.go

@@ -37,6 +37,8 @@ func TestSyncProgressAndLogFormat(t *testing.T) {
 	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: 50, RowsTotal: 100}, p.Copy)
+	require.Equal(t, "50/100 50.00% copyRows ETA 1m", p.Summary)
 	require.Len(t, p.Tables, 2)
 	require.Less(t, p.Tables[0].TableName, p.Tables[1].TableName)
 	block := r.Status()
@@ -45,6 +47,7 @@ func TestSyncProgressAndLogFormat(t *testing.T) {
 	}
 	r.status.Set(status.ApplyChangeset)
 	require.Empty(t, r.Progress().ETA)
+	require.Empty(t, r.Progress().Copy)
 	require.Empty(t, r.Progress().Checksum) // The continuous verifier has no finite initial-checksum phase.
 	r.status.Set(status.RestoreSecondaryIndexes)
 	block = r.Status()

pkg/datasync/runner.go

@@ -1601,10 +1601,12 @@ func (r *Runner) Progress() status.Progress {
 
 	var summary string
 	var eta status.ETA
+	var copyProgress status.CopyProgress
 	switch state { //nolint:exhaustive // sync does not reach the cutover/checksum states
 	case status.CopyRows:
 		if cp != nil {
-			summary = fmt.Sprintf("%s copyRows ETA %s", cp.GetProgress(), cp.GetETA())
+			copyProgress = cp.CopyProgress()
+			summary = fmt.Sprintf("%v copyRows ETA %s", copyProgress, cp.GetETA())
 			eta = cp.GetETAState()
 		} else {
 			summary = "copyRows"
@@ -1627,6 +1629,7 @@ func (r *Runner) Progress() status.Progress {
 		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.
 	}

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

pkg/migration/progress_test.go

@@ -3,7 +3,9 @@ package migration
 import (
 	"sync/atomic"
 	"testing"
+	"time"
 
+	"github.com/block/spirit/pkg/copier"
 	"github.com/block/spirit/pkg/status"
 	"github.com/block/spirit/pkg/testutils"
 	"github.com/block/spirit/pkg/throttler"
@@ -15,6 +17,35 @@ import (
 // reads nothing but the throttler, so the fields under test can be exercised
 // without a live migration.
 
+type progressCopier struct{ copier.Copier }
+
+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}
+}
+
+// TestProgressReportsCopyDuringCopyRows pins that the row-copy counts are a
+// structured field alongside the ETA, populated only while copying, and that
+// Summary renders from the same reading so the two never disagree.
+func TestProgressReportsCopyDuringCopyRows(t *testing.T) {
+	r := &Runner{copier: progressCopier{}}
+	require.Empty(t, r.Progress().Copy)
+
+	r.status.Set(status.CopyRows)
+	p := r.Progress()
+	require.Equal(t, status.CopyProgress{RowsCopied: 50, RowsTotal: 100}, p.Copy)
+	require.Equal(t, status.ETA{State: status.ETAReady, Duration: time.Minute}, p.ETA)
+	require.Equal(t, "50/100 50.00% copyRows ETA 1m", p.Summary)
+
+	r.status.Set(status.WaitingOnSentinelTable)
+	p = r.Progress()
+	require.Empty(t, p.Copy)
+	require.Empty(t, p.ETA)
+}
+
 func TestProgressReportsResume(t *testing.T) {
 	// Resume exists so a wrapper can tell a recovering run from one that is
 	// starting over — a resumed run walks the whole state machine again, so

pkg/migration/runner.go

@@ -1454,11 +1454,13 @@ func (r *Runner) Progress() status.Progress {
 	state := r.status.Get()
 	var summary string
 	var eta status.ETA
+	var copyProgress status.CopyProgress
 	var checksum status.ChecksumProgress
 	switch state { //nolint: exhaustive
 	case status.CopyRows:
+		copyProgress = r.copier.CopyProgress()
 		summary = fmt.Sprintf("%v %s ETA %v",
-			r.copier.GetProgress(),
+			copyProgress,
 			state.String(),
 			r.copier.GetETA(),
 		)
@@ -1485,6 +1487,7 @@ func (r *Runner) Progress() status.Progress {
 		Resume:       r.usedResumeFromCheckpoint.Load(),
 		Throttle:     r.throttleStatus(state),
 		ETA:          eta,
+		Copy:         copyProgress,
 		Checksum:     checksum,
 		Tables:       tables,
 	}

pkg/move/progress_test.go

@@ -22,6 +22,9 @@ 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}
+}
 
 type progressChecker struct{ checksum.Checker }
 
@@ -49,18 +52,22 @@ func TestMoveProgress(t *testing.T) {
 	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: 50, RowsTotal: 100}, p.Copy)
+	require.Equal(t, "50/100 50.00% copyRows ETA 1m", p.Summary)
 	r.checker = progressChecker{}
 	r.status.Set(status.Checksum)
 	p = r.Progress()
 	require.Equal(t, status.ChecksumProgress{RowsChecked: 25, RowsTotal: 100}, p.Checksum)
 	require.Equal(t, "Checksum Progress="+p.Checksum.String(), p.Summary)
 	require.Empty(t, p.ETA)
+	require.Empty(t, p.Copy)
 	r.usedResumeFromCheckpoint.Store(true)
 	r.status.Set(status.WaitingOnSentinelTable)
 	p = r.Progress()
 	require.True(t, p.Resume)
 	require.Equal(t, "Waiting on Sentinel Table", p.Summary) // No logging or target access.
 	require.Empty(t, p.ETA)
+	require.Empty(t, p.Copy)
 	require.Empty(t, p.Checksum)
 }
 

pkg/move/runner.go

@@ -1983,11 +1983,13 @@ func (r *Runner) Progress() status.Progress {
 	state := r.status.Get()
 	var summary string
 	var eta status.ETA
+	var copyProgress status.CopyProgress
 	var checksum status.ChecksumProgress
 	switch state { //nolint: exhaustive
 	case status.CopyRows:
+		copyProgress = r.copier.CopyProgress()
 		summary = fmt.Sprintf("%v %s ETA %v",
-			r.copier.GetProgress(),
+			copyProgress,
 			state.String(),
 			r.copier.GetETA(),
 		)
@@ -2012,6 +2014,7 @@ func (r *Runner) Progress() status.Progress {
 		Summary:      summary,
 		Resume:       r.usedResumeFromCheckpoint.Load(),
 		ETA:          eta,
+		Copy:         copyProgress,
 		Checksum:     checksum,
 		Tables:       tables,
 		// Throttle is deliberately zero: move currently uses a Noop throttler.

pkg/status/progress.go

@@ -109,6 +109,12 @@ type Progress struct {
 	// ETA is the structured remaining row-copy estimate and its availability.
 	ETA ETA
 
+	// Copy is the structured progress of the row copy across every table in the
+	// run, populated while CurrentState is CopyRows and zero otherwise. It is the
+	// structured form of the copy progress embedded in Summary. RowsTotal is an
+	// estimate from table statistics, so RowsCopied can exceed it.
+	Copy CopyProgress
+
 	// Checksum is the structured progress of the post-copy checksum phase,
 	// populated while CurrentState is Checksum and zero otherwise. It is the
 	// structured form of the checksum progress embedded in Summary.