feat(observability): report the post-copy verify pass as its own engine state

pkg/engine/engine.go

@@ -470,8 +470,14 @@ type ShardProgress struct {
 type State string
 
 const (
-	StatePending           State = "pending"
-	StateRunning           State = "running"
+	StatePending State = "pending"
+	StateRunning State = "running"
+	// StateChecksumming is the post-copy verification pass: the engine is
+	// comparing the copied data against the source before cutover. On a large
+	// table it runs for hours, so it is reported as its own state rather than
+	// folded into Running, which would leave the operator watching a finished
+	// copy bar with nothing to indicate a verification is under way.
+	StateChecksumming      State = "checksumming"
 	StateWaitingForDeploy  State = "waiting_for_deploy"
 	StateWaitingForCutover State = "waiting_for_cutover"
 	StateCuttingOver       State = "cutting_over"

pkg/engine/spirit/spirit.go

@@ -822,19 +822,57 @@ func spiritPostCopyPhase(s status.State) bool {
 // progressState resolves the state reported for a progress poll. The tracked
 // state is authoritative for terminal outcomes: they are recorded before the
 // runner is closed, so a runner observed mid-teardown never changes the
-// recorded outcome. Spirit's status only refines a non-terminal state, e.g.
-// surfacing the sentinel wait for a deferred cutover. A stopped tracked state
-// with a volume restart in flight reports running because the schema change
-// is restarting with new settings.
+// recorded outcome. Spirit's phase only refines a non-terminal state. A stopped
+// tracked state with a volume restart in flight reports running because the
+// schema change is restarting with new settings.
 func progressState(rm *runningSchemaChange, spiritState status.State) engine.State {
-	state := rm.state
-	if state == engine.StateStopped && rm.volumeRestartInProgress {
-		state = engine.StateRunning
+	tracked := rm.state
+	if tracked == engine.StateStopped && rm.volumeRestartInProgress {
+		tracked = engine.StateRunning
 	}
-	if !state.IsTerminal() && spiritState == status.WaitingOnSentinelTable && rm.deferCutover {
-		state = engine.StateWaitingForCutover
+	if tracked.IsTerminal() {
+		return tracked
+	}
+	if refined, ok := spiritPhaseState(spiritState, rm.deferCutover); ok {
+		return refined
+	}
+	return tracked
+}
+
+// spiritPhaseState maps a Spirit phase to the engine state it should be
+// reported as, and reports whether the phase refines the tracked state at all.
+// Phases with no operator-visible distinction of their own deliberately flatten
+// into the tracked state; they are listed rather than left to the default so
+// that adding vocabulary for one of them is a decision made here, not an
+// omission discovered on a long apply.
+func spiritPhaseState(spiritState status.State, deferCutover bool) (engine.State, bool) {
+	switch spiritState {
+	case status.Checksum:
+		// Verifying the copied data against the source. Hours on a large table,
+		// with its own row counters, so it gets its own state.
+		return engine.StateChecksumming, true
+
+	case status.WaitingOnSentinelTable:
+		// Spirit parks on the sentinel table only because SchemaBot asked it to
+		// hold the cutover. Without that request the wait is Spirit's own brief
+		// internal step and says nothing to the operator.
+		if deferCutover {
+			return engine.StateWaitingForCutover, true
+		}
+		return "", false
+
+	case status.Initial, status.CopyRows, status.ApplyChangeset,
+		status.RestoreSecondaryIndexes, status.AnalyzeTable, status.PostChecksum,
+		status.CutOver, status.Close, status.ErrCleanup:
+		// Flattened into the tracked state. The changeset applies and the analyze
+		// are short; cutover, close and cleanup are the tracked state's own
+		// business. The secondary-index rebuild is the one that can run for hours
+		// like the checksum does, and is the next candidate for its own state.
+		return "", false
+
+	default:
+		return "", false
 	}
-	return state
 }
 
 // fetchCurrentSchema retrieves table schemas from the database, filtering out

pkg/engine/spirit/spirit_test.go

@@ -91,6 +91,30 @@ func TestProgressState(t *testing.T) {
 			spiritState: status.WaitingOnSentinelTable,
 			want:        engine.StateRunning,
 		},
+		{
+			name: "verify pass surfaces as checksumming",
+			rm: &runningSchemaChange{
+				state: engine.StateRunning,
+			},
+			spiritState: status.Checksum,
+			want:        engine.StateChecksumming,
+		},
+		{
+			name: "verify pass does not reopen a terminal outcome",
+			rm: &runningSchemaChange{
+				state: engine.StateCancelled,
+			},
+			spiritState: status.Checksum,
+			want:        engine.StateCancelled,
+		},
+		{
+			name: "the changeset apply after the verify stays running",
+			rm: &runningSchemaChange{
+				state: engine.StateRunning,
+			},
+			spiritState: status.PostChecksum,
+			want:        engine.StateRunning,
+		},
 		{
 			name: "volume restart reports stopped state as running",
 			rm: &runningSchemaChange{

pkg/tern/state_converters.go

@@ -31,6 +31,8 @@ func engineStateToStorage(es engine.State) string {
 		return state.Task.Pending
 	case engine.StateRunning:
 		return state.Task.Running
+	case engine.StateChecksumming:
+		return state.Task.Checksumming
 	case engine.StateWaitingForDeploy:
 		return state.Task.WaitingForDeploy
 	case engine.StateWaitingForCutover:

pkg/tern/state_converters_test.go

@@ -44,6 +44,14 @@ func TestPostCopyPhaseStateConversions(t *testing.T) {
 	}
 }
 
+// The engine's verify pass converts to the checksumming task state, so the
+// phase the proto boundary already carries is populated from Spirit rather than
+// flattening into a generic running task.
+func TestChecksummingStateConversions(t *testing.T) {
+	assert.Equal(t, state.Task.Checksumming, engineStateToStorage(engine.StateChecksumming))
+	assert.False(t, engine.StateChecksumming.IsTerminal(), "the verify pass is in-flight, not terminal")
+}
+
 // A null namespace value in the proto map (e.g. JSON `{"default": null}`)
 // converts to an empty namespace rather than dereferencing a nil pointer.
 func TestProtoToSchemaFiles_NilNamespaceValue(t *testing.T) {