fix(github): surface a mirrored control rejection on every path an operator reads

- Mirror settled control requests from the stop/cancel progress reads. Those
  paths reconcile a terminal remote and end the drive, so a rejection the data
  plane settled after the last regular poll would otherwise never be recorded.
- Append the rejection notice to the terminal summary comment, not only to the
  progress comment the summary supersedes.
- Omit the attribution clause when the stored request names no operator, so a
  proxied command is not credited to SchemaBot's internal forwarding caller.
- Rewrite the remote apply identifier out of the mirrored reason on an
  operation-scoped drive, where the parent apply carries no external id.
- Name an unrecognized settled status in a warning instead of dropping it
  silently.
- Document the rejection counter and the release operation in the metrics
  reference, and count release as a known control operation.

Author: aparajon

pkg/metrics/README.md

@@ -39,6 +39,7 @@ available, such as `repository`, `github_app`, and `installation_id`.
 | `schemabot.github.rate_limit.remaining` | Gauge | environment, operation, resource, repository, github_app, installation_id | GitHub primary rate limit requests remaining for the observed API resource |
 | `schemabot.github.rate_limit.used` | Gauge | environment, operation, resource, repository, github_app, installation_id | GitHub primary rate limit requests used for the observed API resource |
 | `schemabot.control_operations_total` | Counter | operation, database, environment, status | Control operations (cutover, stop, start, etc.) |
+| `schemabot.remote_control_requests.rejected_total` | Counter | operation, engine, database, deployment, environment | Control requests a remote data plane accepted and its own driver then failed, mirrored back so the operator learns the command never took effect — see [Control Operations](#control-operations) |
 | `schemabot.lock_operations_total` | Counter | operation, database, environment, status | Lock acquire/release operations |
 | `schemabot.direct_write_authorization.total` | Counter | operation, database, environment, status, reason | Per-database direct-write (CLI/API) authorization decisions at the handler layer |
 | `schemabot.operator.resumed_total` | Counter | database, environment, previous_state | Applies resumed by the operator |
@@ -95,10 +96,12 @@ available, such as `repository`, `github_app`, and `installation_id`.
 
 **status** (status checks): `success`, `error`, `skipped`, `stale`, `noop`, `blocked` (operation outcome, not GitHub Check Run conclusion)
 
-**operation** (control): `cutover`, `stop`, `start`, `volume`, `revert`, `skip_revert`, `rollback_plan`
+**operation** (control): `cutover`, `stop`, `start`, `volume`, `revert`, `skip_revert`, `release`, `rollback_plan`
 
 **status** (control): `success`, `error`, `rejected`
 
+**engine** (control): the engine that rejected the command, as recorded on the apply (for example `spirit`, `planetscale`)
+
 **operation** (locks): `acquire`, `release`
 
 **status** (locks): `success`, `conflict`, `not_found`, `not_owned`, `error`
@@ -290,8 +293,18 @@ Operation values:
 | `volume` | Change the engine's throttling or volume setting for an apply. |
 | `revert` | Request that the engine revert an apply. |
 | `skip_revert` | Skip the post-deploy revert window for an apply. |
+| `release` | Release a rollout paused after a failure so the remaining work proceeds. |
 | `rollback_plan` | Generate a rollback plan from a previous completed apply. |
 
+`schemabot.remote_control_requests.rejected_total` counts the same operations
+from the other side: a control request the data plane acknowledged and its own
+driver then failed. Acceptance only means the request was queued, so without
+this mirror the operator is told the command succeeded and never learns the
+effect did not land. A non-zero rate names the operation and `engine` that is
+refusing operator commands — chart it by operation to see which control surface
+is unsupported or broken on that engine, and read the apply log entry recorded
+alongside it for the engine's own reason.
+
 ### Lock Operations
 
 `schemabot.lock_operations_total` tracks database-level lock acquisition and

pkg/metrics/metrics.go

@@ -740,11 +740,12 @@ var knownControlOperations = map[string]bool{
 	"volume":        true,
 	"revert":        true,
 	"skip_revert":   true,
+	"release":       true,
 	"rollback_plan": true,
 }
 
 // RecordControlOperation increments the control operations counter.
-// Operation should be one of: cutover, stop, start, volume, revert, skip_revert, rollback_plan.
+// Operation should be one of: cutover, stop, start, volume, revert, skip_revert, release, rollback_plan.
 // Status should be "success" or "error".
 func RecordControlOperation(ctx context.Context, operation, database, deployment, environment, status string) {
 	if !knownControlOperations[operation] {

pkg/tern/grpc_client.go

@@ -896,10 +896,7 @@ func (c *GRPCClient) processPendingStopControlRequest(ctx context.Context, apply
 		}
 	}
 
-	progress, err := c.client.Progress(ctx, &ternv1.ProgressRequest{
-		ApplyId:     remoteID,
-		Environment: apply.Environment,
-	})
+	progress, err := c.controlPathProgress(ctx, apply, remoteID)
 	if err != nil {
 		return true, fmt.Errorf("sync remote gRPC stop for apply %s remote %s: %w", apply.ApplyIdentifier, remoteID, err)
 	}
@@ -1014,7 +1011,7 @@ func (c *GRPCClient) processPendingCancelControlRequest(ctx context.Context, app
 			logRemoteControlResend(ctx, logger, apply, controlReq, now)
 		}
 	}
-	progress, err := c.client.Progress(ctx, &ternv1.ProgressRequest{ApplyId: remoteID, Environment: apply.Environment})
+	progress, err := c.controlPathProgress(ctx, apply, remoteID)
 	if err != nil {
 		return true, fmt.Errorf("sync remote gRPC cancel for apply %s remote %s: %w", apply.ApplyIdentifier, remoteID, err)
 	}
@@ -1050,6 +1047,23 @@ func (c *GRPCClient) processPendingCancelControlRequest(ctx context.Context, app
 	return false, nil
 }
 
+// controlPathProgress polls remote progress on behalf of a control-request path
+// and mirrors any control rejections the response carries. A cancel or stop that
+// reconciles a terminal remote ends the drive, so the regular poll loop never
+// runs again: a rejection the data plane settled after the last regular poll
+// reaches the operator only if it is mirrored from here.
+func (c *GRPCClient) controlPathProgress(ctx context.Context, apply *storage.Apply, remoteID string) (*ternv1.ProgressResponse, error) {
+	progress, err := c.client.Progress(ctx, &ternv1.ProgressRequest{
+		ApplyId:     remoteID,
+		Environment: apply.Environment,
+	})
+	if err != nil {
+		return nil, err
+	}
+	c.mirrorRemoteControlRejections(ctx, apply, remoteID, progress.SettledControlRequests)
+	return progress, nil
+}
+
 func (c *GRPCClient) processPendingCancelOrStopControlRequest(ctx context.Context, apply *storage.Apply, scope applyTaskScope) (bool, error) {
 	if handled, err := c.processPendingCancelControlRequest(ctx, apply, scope); handled || err != nil {
 		return handled, err
@@ -1063,10 +1077,7 @@ func (c *GRPCClient) completeRemoteStopFromTerminalProgress(ctx context.Context,
 	// tracks its remote apply on the operation, not on the parent apply's
 	// ExternalID.
 	remoteID := scope.remoteApplyID(apply)
-	progress, err := c.client.Progress(ctx, &ternv1.ProgressRequest{
-		ApplyId:     remoteID,
-		Environment: apply.Environment,
-	})
+	progress, err := c.controlPathProgress(ctx, apply, remoteID)
 	if err != nil {
 		logger.WarnContext(ctx, "remote gRPC stop error could not be reconciled from progress",
 			append(apply.MutableLogAttrs(),
@@ -1134,10 +1145,7 @@ func (c *GRPCClient) completeRemoteCancelFromTerminalProgress(ctx context.Contex
 	// tracks its remote apply on the operation, not on the parent apply's
 	// ExternalID.
 	remoteID := scope.remoteApplyID(apply)
-	progress, err := c.client.Progress(ctx, &ternv1.ProgressRequest{
-		ApplyId:     remoteID,
-		Environment: apply.Environment,
-	})
+	progress, err := c.controlPathProgress(ctx, apply, remoteID)
 	if err != nil {
 		logger.WarnContext(ctx, "remote gRPC cancel error could not be reconciled from progress",
 			append(apply.MutableLogAttrs(),
@@ -1669,7 +1677,12 @@ func mirrorRemoteVolume(logger *slog.Logger, apply *storage.Apply, remoteVolume
 // operator retries the operation, so a mirror that fails is retried on the next
 // tick rather than aborting the drive, and a mirror that succeeds is surfaced
 // exactly once (RecordRemoteFailure reports whether the stored row changed).
-func (c *GRPCClient) mirrorRemoteControlRejections(ctx context.Context, apply *storage.Apply, settled []*ternv1.SettledControlRequest) {
+//
+// remoteID is the remote identifier this drive addressed. The data plane's
+// message names its own apply, which is meaningless to the operator reading the
+// PR — and on an operation-scoped drive the parent apply's ExternalID is empty,
+// so the remote identifier is only redactable when it is passed in here.
+func (c *GRPCClient) mirrorRemoteControlRejections(ctx context.Context, apply *storage.Apply, remoteID string, settled []*ternv1.SettledControlRequest) {
 	if c.storage == nil || apply == nil || len(settled) == 0 {
 		return
 	}
@@ -1695,13 +1708,21 @@ func (c *GRPCClient) mirrorRemoteControlRejections(ctx context.Context, apply *s
 			continue
 		}
 		if entry.Status != string(storage.ControlRequestFailed) {
+			// Only a failure needs mirroring, and completion is handled above. A
+			// newer data plane reporting some other terminal status would drop the
+			// request here, so name it rather than skipping silently.
+			logger.Warn("data plane reported a settled control request in an unrecognized status; it will not reach the operator",
+				append(apply.MutableLogAttrs(),
+					"operation", entry.Operation,
+					"status", entry.Status,
+					"settled_at", entry.SettledAt)...)
 			continue
 		}
 		message := remoteControlRejectionMessage(entry)
 		changed, err := controlStore.RecordRemoteFailure(ctx, &storage.ApplyControlRequest{
 			ApplyID:      apply.ID,
 			Operation:    operation,
-			ErrorMessage: apply.OperatorFacingMessage(message),
+			ErrorMessage: apply.OperatorFacingMessage(message, remoteID),
 			RequestedBy:  entry.RequestedBy,
 		})
 		if err != nil {
@@ -3895,7 +3916,7 @@ func (c *GRPCClient) pollForCompletion(ctx context.Context, apply *storage.Apply
 				logger.Info("mirrored remote volume level onto control-plane apply options",
 					append(apply.MutableLogAttrs(), "volume", resp.Volume)...)
 			}
-			c.mirrorRemoteControlRejections(ctx, apply, resp.SettledControlRequests)
+			c.mirrorRemoteControlRejections(ctx, apply, remoteID, resp.SettledControlRequests)
 
 			terminal := isTerminalProtoState(resp.State)
 			if terminal {

pkg/tern/grpc_client_test.go

@@ -570,6 +570,7 @@ type capturingTernServer struct {
 	progressTables    []*ternv1.TableProgress
 	progressVolume    int32
 	progressError     string
+	progressSettled   []*ternv1.SettledControlRequest
 	progressErr       error
 	startErr          error
 	cancelErr         error
@@ -676,6 +677,7 @@ func (s *capturingTernServer) Progress(_ context.Context, req *ternv1.ProgressRe
 	tables := s.progressTables
 	volume := s.progressVolume
 	errorMessage := s.progressError
+	settled := s.progressSettled
 	err := s.progressErr
 	s.mu.Unlock()
 	if err != nil {
@@ -684,7 +686,13 @@ func (s *capturingTernServer) Progress(_ context.Context, req *ternv1.ProgressRe
 	if !psSet {
 		ps = ternv1.State_STATE_COMPLETED
 	}
-	return &ternv1.ProgressResponse{State: ps, Tables: tables, Volume: volume, ErrorMessage: errorMessage}, nil
+	return &ternv1.ProgressResponse{
+		State:                  ps,
+		Tables:                 tables,
+		Volume:                 volume,
+		ErrorMessage:           errorMessage,
+		SettledControlRequests: settled,
+	}, nil
 }
 func (s *capturingTernServer) Logs(context.Context, *ternv1.LogsRequest) (*ternv1.LogsResponse, error) {
 	return &ternv1.LogsResponse{}, nil
@@ -2242,6 +2250,77 @@ func TestGRPCClient_ProcessPendingCancelControlRequestCompletesWholeApply(t *tes
 	assert.Nil(t, cancelReq)
 }
 
+// A cancel that reconciles a terminal remote ends the drive, so the regular
+// poll loop never runs again. A control command the data plane settled after
+// the last regular poll — here a volume change its engine refused — reaches the
+// operator only if the cancel path's own progress read mirrors it; otherwise the
+// operator is left believing a command they were told was accepted took effect.
+func TestGRPCClient_CancelPathMirrorsSettledControlRejections(t *testing.T) {
+	server := &capturingTernServer{
+		progressState:    ternv1.State_STATE_CANCELLED,
+		progressStateSet: true,
+		progressTables: []*ternv1.TableProgress{{
+			Namespace: "default",
+			TableName: "users",
+			Status:    state.Task.Cancelled,
+		}},
+		progressSettled: []*ternv1.SettledControlRequest{{
+			Operation:    string(storage.ControlOperationVolume),
+			Status:       string(storage.ControlRequestFailed),
+			ErrorMessage: "throttle endpoint returned 404",
+			RequestedBy:  "cli:alice",
+		}},
+	}
+	client, cleanup := testCapturingGRPCClient(t, server)
+	defer cleanup()
+
+	apply := &storage.Apply{
+		ID:              7,
+		ApplyIdentifier: "apply-grpc-cancel-mirror",
+		ExternalID:      "remote-grpc-cancel-mirror",
+		PlanID:          99,
+		Database:        "testdb",
+		DatabaseType:    storage.DatabaseTypeMySQL,
+		Environment:     "staging",
+		State:           state.Apply.Running,
+	}
+	task := &storage.Task{
+		ID:             11,
+		TaskIdentifier: "task-users",
+		ApplyID:        apply.ID,
+		Namespace:      "default",
+		TableName:      "users",
+		State:          state.Task.Running,
+	}
+	controlRequests := &testControlRequestStore{requests: []*storage.ApplyControlRequest{{
+		ApplyID:     apply.ID,
+		Operation:   storage.ControlOperationCancel,
+		Status:      storage.ControlRequestPending,
+		RequestedBy: "cli:alice",
+	}}}
+	storedApply := *apply
+	logs := &mockApplyLogStore{}
+	client.storage = &mockStorage{
+		applies:         &mockApplyStore{apply: &storedApply},
+		tasks:           &mockTaskStore{tasks: []*storage.Task{task}},
+		logs:            logs,
+		controlRequests: controlRequests,
+	}
+
+	handled, err := client.processPendingCancelControlRequest(t.Context(), apply, wholeApplyTaskScope())
+	require.NoError(t, err)
+	require.True(t, handled)
+
+	rejected, err := controlRequests.GetByOperation(t.Context(), apply.ID, storage.ControlOperationVolume)
+	require.NoError(t, err)
+	require.NotNil(t, rejected, "the rejection the data plane settled must survive the drive that ends here")
+	assert.Equal(t, storage.ControlRequestFailed, rejected.Status)
+	assert.Contains(t, rejected.ErrorMessage, "throttle endpoint returned 404")
+	assert.Equal(t, "cli:alice", rejected.RequestedBy)
+	assert.True(t, hasLogMessageContaining(logs.logs, "Volume was accepted but not applied"),
+		"the operator must find the rejection on the schema change's log")
+}
+
 func TestGRPCClient_ProcessPendingCancelOperationLeavesApplyCancelPending(t *testing.T) {
 	// A multi-deployment apply has one durable cancel request shared by every
 	// deployment. One operation reaching a terminal remote state must not

pkg/tern/grpc_control_rejection_integration_test.go

@@ -115,7 +115,7 @@ func TestGRPCClient_MirrorsRemoteControlRejection(t *testing.T) {
 		Status:    string(storage.ControlRequestCompleted),
 	}}
 
-	client.mirrorRemoteControlRejections(ctx, apply, rejection)
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, rejection)
 
 	stored, err := stor.ControlRequests().GetByOperation(ctx, apply.ID, storage.ControlOperationVolume)
 	require.NoError(t, err)
@@ -136,7 +136,7 @@ func TestGRPCClient_MirrorsRemoteControlRejection(t *testing.T) {
 	// The data plane reports the same rejection on every poll until the
 	// operator retries the operation; mirroring it again must not append a
 	// second entry.
-	client.mirrorRemoteControlRejections(ctx, apply, rejection)
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, rejection)
 	logs, err = stor.ApplyLogs().GetByApply(ctx, apply.ID)
 	require.NoError(t, err)
 	assert.Equal(t, 1, countLogMessages(logs, "Volume was accepted but not applied"),
@@ -182,7 +182,7 @@ func TestGRPCClient_MirrorLeavesAReissuedCommandPending(t *testing.T) {
 
 	// The data plane has not seen the new request yet, so it keeps reporting the
 	// old one it settled.
-	client.mirrorRemoteControlRejections(ctx, apply, []*ternv1.SettledControlRequest{{
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, []*ternv1.SettledControlRequest{{
 		Operation:    string(storage.ControlOperationCutover),
 		Status:       string(storage.ControlRequestFailed),
 		ErrorMessage: "cutover was not applied because apply is recovering",
@@ -235,8 +235,8 @@ func TestGRPCClient_MirrorReattributesARejectionToTheOperatorWhoReissuedIt(t *te
 		}}
 	}
 
-	client.mirrorRemoteControlRejections(ctx, apply, rejection("cli:alice"))
-	client.mirrorRemoteControlRejections(ctx, apply, rejection("cli:bob"))
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, rejection("cli:alice"))
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, rejection("cli:bob"))
 
 	stored, err := stor.ControlRequests().GetByOperation(ctx, apply.ID, storage.ControlOperationVolume)
 	require.NoError(t, err)
@@ -272,7 +272,7 @@ func TestGRPCClient_MirrorRetiresARejectionTheDataPlaneLaterCompleted(t *testing
 	defer cleanup()
 	client.storage = stor
 
-	client.mirrorRemoteControlRejections(ctx, apply, []*ternv1.SettledControlRequest{{
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, []*ternv1.SettledControlRequest{{
 		Operation:    string(storage.ControlOperationVolume),
 		Status:       string(storage.ControlRequestFailed),
 		ErrorMessage: "throttle endpoint returned 404",
@@ -281,7 +281,7 @@ func TestGRPCClient_MirrorRetiresARejectionTheDataPlaneLaterCompleted(t *testing
 	requireControlRequestStatus(t, stor, apply.ID, storage.ControlOperationVolume, storage.ControlRequestFailed)
 
 	// The operator re-issues volume; this time the data plane applies it.
-	client.mirrorRemoteControlRejections(ctx, apply, []*ternv1.SettledControlRequest{{
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, []*ternv1.SettledControlRequest{{
 		Operation:   string(storage.ControlOperationVolume),
 		Status:      string(storage.ControlRequestCompleted),
 		RequestedBy: "cli:alice",
@@ -359,7 +359,7 @@ func TestGRPCClient_MirrorKeepsTheOperatorNameOnALocallyQueuedRequest(t *testing
 	failLocallyQueuedControlRequest(t, dsn, stor, apply.ID, storage.ControlOperationCutover,
 		"octocat", "cutover request was not applied because apply is failed")
 
-	client.mirrorRemoteControlRejections(ctx, apply, []*ternv1.SettledControlRequest{{
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, []*ternv1.SettledControlRequest{{
 		Operation:    string(storage.ControlOperationCutover),
 		Status:       string(storage.ControlRequestFailed),
 		ErrorMessage: "deploy request is not in a cutover-ready state",
@@ -403,7 +403,7 @@ func TestGRPCClient_MirrorReattributesToANamedOperator(t *testing.T) {
 	failLocallyQueuedControlRequest(t, dsn, stor, apply.ID, storage.ControlOperationCutover,
 		"octocat", "cutover request was not applied because apply is failed")
 
-	client.mirrorRemoteControlRejections(ctx, apply, []*ternv1.SettledControlRequest{{
+	client.mirrorRemoteControlRejections(ctx, apply, apply.ExternalID, []*ternv1.SettledControlRequest{{
 		Operation:    string(storage.ControlOperationCutover),
 		Status:       string(storage.ControlRequestFailed),
 		ErrorMessage: "deploy request is not in a cutover-ready state",

pkg/tern/local_client_test.go

@@ -651,6 +651,36 @@ func (s *testControlRequestStore) FailPending(_ context.Context, applyID int64,
 	return nil
 }
 
+// RecordRemoteFailure mirrors the store's contract: a pending row is a live
+// request this plane has not forwarded yet, so a settled report describes a
+// superseded attempt and is ignored; anything else takes the remote reason.
+func (s *testControlRequestStore) RecordRemoteFailure(_ context.Context, req *storage.ApplyControlRequest) (bool, error) {
+	for _, existing := range s.requests {
+		if existing.ApplyID != req.ApplyID || existing.Operation != req.Operation {
+			continue
+		}
+		if existing.Status == storage.ControlRequestPending {
+			return false, nil
+		}
+		if existing.Status == storage.ControlRequestFailed &&
+			existing.ErrorMessage == req.ErrorMessage &&
+			existing.RequestedBy == req.RequestedBy {
+			return false, nil
+		}
+		existing.Status = storage.ControlRequestFailed
+		existing.ErrorMessage = req.ErrorMessage
+		if req.RequestedBy != "" {
+			existing.RequestedBy = req.RequestedBy
+		}
+		return true, nil
+	}
+	stored := cloneTestControlRequest(req)
+	stored.ID = int64(len(s.requests) + 1)
+	stored.Status = storage.ControlRequestFailed
+	s.requests = append(s.requests, stored)
+	return true, nil
+}
+
 func cloneTestControlRequest(req *storage.ApplyControlRequest) *storage.ApplyControlRequest {
 	if req == nil {
 		return nil