fix(operator): settle applies left behind their own settled operations

integration/operator_test.go

@@ -576,6 +576,78 @@ func TestOperator_StartResumesStoppedApplyWhoseDeploymentsNeverStarted(t *testin
 		"the serviced start request is completed")
 }
 
+// TestOperator_ProjectsApplyLeftBehindItsSettledOperations covers a parent apply
+// left behind its own children: a drive wrote its terminal operation row and
+// then stopped before projecting that outcome onto the parent, so every
+// operation is terminal while the apply itself is still running. No operation
+// claim arm matches a fully settled operation set, and until the parent settles
+// the one-active-apply guard refuses every new apply for that target. Once the
+// crashed drive's lease goes stale the operator must derive the parent from its
+// operations and release the target, leaving the operation rows untouched.
+func TestOperator_ProjectsApplyLeftBehindItsSettledOperations(t *testing.T) {
+	ctx := t.Context()
+	appDBName, appDSN := createTestDB(t, "operator_projection_")
+	ts := startTestServer(t, appDBName, appDSN)
+
+	now := time.Now()
+	applyID, err := ts.Storage.Applies().Create(ctx, &storage.Apply{
+		ApplyIdentifier: "apply-settled-ops-orphan",
+		Database:        appDBName,
+		DatabaseType:    "mysql",
+		Deployment:      appDBName,
+		Engine:          "spirit",
+		State:           state.Apply.Running,
+		Options:         []byte("{}"),
+		Environment:     "staging",
+		CreatedAt:       now,
+		UpdatedAt:       now,
+	})
+	require.NoError(t, err)
+
+	opID, err := ts.Storage.ApplyOperations().Insert(ctx, &storage.ApplyOperation{
+		ApplyID:    applyID,
+		Deployment: appDBName,
+		Target:     appDBName,
+		State:      state.ApplyOperation.Completed,
+	})
+	require.NoError(t, err)
+
+	// Age the apply's heartbeat past the lease staleness window so the crashed
+	// drive is no longer presumed live. Written directly because every storage
+	// write path refreshes updated_at, which is the heartbeat itself.
+	storageDB, err := sql.Open("mysql", schemabotDSN)
+	require.NoError(t, err, "open schemabot db")
+	require.NoError(t, storageDB.PingContext(ctx))
+	t.Cleanup(func() {
+		utils.CloseAndLog(storageDB)
+	})
+	_, err = storageDB.ExecContext(ctx,
+		`UPDATE applies SET updated_at = NOW() - INTERVAL ? SECOND WHERE id = ?`,
+		int64((2 * storage.ApplyLeaseStaleAfter).Seconds()), applyID)
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		apply, err := ts.Storage.Applies().Get(ctx, applyID)
+		if err != nil || apply == nil {
+			return false
+		}
+		return state.IsState(apply.State, state.Apply.Completed)
+	}, 30*time.Second, 200*time.Millisecond,
+		"operator should derive the parent apply from its already-settled operations")
+
+	apply, err := ts.Storage.Applies().Get(ctx, applyID)
+	require.NoError(t, err)
+	require.NotNil(t, apply)
+	assert.Equal(t, state.Apply.Completed, apply.State, "the parent settles to the state its operations derive")
+	require.NotNil(t, apply.CompletedAt, "a parent settled to a terminal state stamps completed_at")
+
+	op, err := ts.Storage.ApplyOperations().Get(ctx, opID)
+	require.NoError(t, err)
+	require.NotNil(t, op)
+	assert.Equal(t, state.ApplyOperation.Completed, op.State,
+		"the repair derives the parent from the operations and never rewrites them")
+}
+
 // TestOperator_StopReconciliationDrivesTaskStop covers the operation-claim stop
 // path for an apply whose operation row is still pending when a stop is
 // requested — the window before the operator claims and drives the operation.

pkg/api/handlers_test.go

@@ -501,6 +501,10 @@ func (s *capturingApplyStore) FindNextApplyForStopReconciliation(context.Context
 	return nil, nil
 }
 
+func (s *capturingApplyStore) FindNextApplyForOperationProjection(context.Context, string) (*storage.Apply, error) {
+	return nil, nil
+}
+
 func (s *capturingApplyStore) CheckLease(context.Context, storage.ApplyLease) error {
 	return nil
 }

pkg/api/operator.go

@@ -262,6 +262,15 @@ func (s *Service) recoverApplies(ctx context.Context, driverID int) {
 	if s.recoverApplyOperationCutover(ctx, driverID, owner) {
 		return
 	}
+	// Settle a parent left behind its own children before claiming new
+	// operation work: until it settles, the one-active-apply guard blocks
+	// every new apply for that target, so the repair unblocks a database
+	// rather than merely tidying a row. It matches nothing while any
+	// operation is still non-terminal or any driver is still heartbeating,
+	// so on a healthy plane this is a cheap no-op.
+	if s.recoverApplyOperationProjection(ctx, driverID, owner) {
+		return
+	}
 	s.recoverApplyOperation(ctx, driverID, owner)
 }
 
@@ -893,6 +902,88 @@ func (s *Service) recoverApplyPendingStop(ctx context.Context, driverID int, own
 	return true
 }
 
+// recoverApplyOperationProjection settles an apply whose operation rows have all
+// reached a terminal state while the apply itself is still non-terminal: a
+// parent left behind its own children because a drive stopped between writing
+// its terminal operation row and projecting that outcome onto the parent.
+//
+//	apply_operations              applies
+//	┌──────────────────┐          ┌──────────────────────────┐
+//	│ id=8  completed  │          │ id=7  running  ← stranded│
+//	│ id=9  completed  │  ──✗──▶  └──────────────────────────┘
+//	└──────────────────┘            the projection never ran
+//	  every child settled           the target stays blocked
+//
+// Nothing else recovers it. No operation-level claim arm matches a fully
+// terminal operation set, the parent's own state never moves again, and the
+// one-active-apply guard keeps refusing new applies for that target until it
+// does. Re-deriving the parent from its operations is the entire repair — no
+// engine work is driven, because the operations already carry their outcomes.
+//
+// Returns true when this driver spent its tick here, including on failure, so a
+// repair that could not complete is retried rather than falling through to
+// claim new work behind a target that is still blocked.
+func (s *Service) recoverApplyOperationProjection(ctx context.Context, driverID int, owner string) bool {
+	apply, err := s.storage.Applies().FindNextApplyForOperationProjection(ctx, owner)
+	if err != nil {
+		s.logger.Error("operator: failed to claim apply for operation projection",
+			"driver", driverID, "lease_owner", owner, "error", err)
+		metrics.RecordOperatorClaimFailure(ctx, "operation_projection_claim_error")
+		return true
+	}
+	if apply == nil {
+		s.logger.Debug("operator: no apply needs operation projection", "driver", driverID)
+		return false
+	}
+
+	lease := apply.Lease()
+	if !lease.Valid() {
+		s.logger.Error("operator: claimed apply for operation projection without a valid lease token; the apply stays non-terminal and its target stays blocked until a later tick reclaims it",
+			append(apply.LogAttrs(),
+				"driver", driverID, "lease_owner", owner)...)
+		metrics.RecordOperatorClaimFailure(ctx, "operation_projection_missing_lease_token")
+		return true
+	}
+	applyLeaseCtx := storage.WithApplyLease(ctx, lease)
+
+	s.logger.Info("operator: claimed an apply whose operations have all settled while it stayed non-terminal; deriving its state from them",
+		append(apply.LogAttrs(),
+			"driver", driverID, "lease_owner", owner)...)
+
+	result, err := s.updateApplyStateFromOperations(applyLeaseCtx, driverID, apply, allowLeaseScopedFailedReopen)
+	if err != nil {
+		s.logger.Error("operator: failed to derive apply state from its settled operations; the apply stays non-terminal and its target stays blocked until a later tick reclaims it",
+			append(apply.LogAttrs(),
+				"driver", driverID, "error", err)...)
+		return true
+	}
+	if !result.Swapped {
+		// The claim refreshed the heartbeat, so this apply is not reconsidered
+		// until the staleness window elapses again. Warn rather than debug: a
+		// terminal operation set that does not derive a terminal parent means the
+		// projection and the claim predicate disagree, and the target stays
+		// blocked in the meantime.
+		s.logger.Warn("operator: settled operations did not move the apply out of its non-terminal state; its target stays blocked until the state changes or an operator reconciles it",
+			append(apply.LogAttrs(),
+				"driver", driverID, "derived_state", result.DerivedState,
+				"operation_count", result.OperationCount)...)
+		return true
+	}
+	metrics.RecordOperatorOperationProjectionRepair(ctx, apply.Database, apply.Deployment, apply.Environment, result.DerivedState)
+
+	// No operation drive is left to publish on this apply's behalf, so this
+	// projection owes the single terminal summary if it won the terminal swap.
+	s.publishTerminalSummaryIfWon(applyLeaseCtx, driverID, apply, result)
+
+	if err := s.completePendingControlRequestsIfApplyResolved(applyLeaseCtx, driverID, apply.ID); err != nil {
+		s.logger.Error("operator: failed to complete pending control requests after deriving apply state from its settled operations",
+			append(apply.LogAttrs(),
+				"driver", driverID, "error", err)...)
+		return true
+	}
+	return true
+}
+
 // stopPendingOperationsForPendingStop terminalizes still-pending sibling
 // operations to stopped when the apply has a pending stop request, so the
 // rollout can settle instead of stranding running with siblings the claim gate

pkg/api/operator_test.go

@@ -1749,6 +1749,10 @@ func (s *operationClaimApplyStore) FindNextApplyForStopReconciliation(context.Co
 	return nil, nil
 }
 
+func (s *operationClaimApplyStore) FindNextApplyForOperationProjection(context.Context, string) (*storage.Apply, error) {
+	return nil, nil
+}
+
 func (s *operationClaimApplyStore) ClaimApplyByID(_ context.Context, _ int64, owner string) (*storage.Apply, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

pkg/metrics/README.md

@@ -101,7 +101,7 @@ available, such as `repository`, `github_app`, and `installation_id`.
 
 **status** (GitHub API): `success`, `error`, `unknown`
 
-**reason** (operator claim failures): `expire_retryable_error`, `missing_lease_token`, `operation_storage_error`, `missing_operation_lease_token`, `operation_set_list_error`, `operation_set_missing`, `operation_task_inspect_error`, `operation_cutover_storage_error`, `missing_operation_cutover_lease_token`, `operation_cutover_set_list_error`, `operation_cutover_set_invalid`, `operation_parent_load_error`, `operation_parent_missing`, `operation_parent_claim_error`, `operation_parent_not_claimable`, `operation_lease_release_error`, `missing_operation_deployment`, `stop_reconciliation_claim_error`, `stop_reconciliation_missing_lease_token`, `stranded_reaper_error`, `unknown`
+**reason** (operator claim failures): `expire_retryable_error`, `missing_lease_token`, `operation_storage_error`, `missing_operation_lease_token`, `operation_set_list_error`, `operation_set_missing`, `operation_task_inspect_error`, `operation_cutover_storage_error`, `missing_operation_cutover_lease_token`, `operation_cutover_set_list_error`, `operation_cutover_set_invalid`, `operation_parent_load_error`, `operation_parent_missing`, `operation_parent_claim_error`, `operation_parent_not_claimable`, `operation_lease_release_error`, `missing_operation_deployment`, `stop_reconciliation_claim_error`, `stop_reconciliation_missing_lease_token`, `operation_projection_claim_error`, `operation_projection_missing_lease_token`, `stranded_reaper_error`, `unknown`
 
 **reason** (operator resume failures): `missing_deployment`, `no_client`, `resume_error`, `lease_lost`, `retry_budget_exhausted`, `recovery_window_expired`
 

pkg/metrics/metrics.go

@@ -907,26 +907,28 @@ func RecordOperatorStrandedOperationReaped(ctx context.Context, database, deploy
 }
 
 var knownOperatorClaimFailureReasons = map[string]bool{
-	"expire_retryable_error":                  true,
-	"stranded_reaper_error":                   true,
-	"missing_lease_token":                     true,
-	"operation_storage_error":                 true,
-	"missing_operation_lease_token":           true,
-	"operation_set_list_error":                true,
-	"operation_set_missing":                   true,
-	"operation_task_inspect_error":            true,
-	"operation_cutover_storage_error":         true,
-	"missing_operation_cutover_lease_token":   true,
-	"operation_cutover_set_list_error":        true,
-	"operation_cutover_set_invalid":           true,
-	"operation_parent_load_error":             true,
-	"operation_parent_missing":                true,
-	"operation_parent_claim_error":            true,
-	"operation_parent_not_claimable":          true,
-	"operation_lease_release_error":           true,
-	"missing_operation_deployment":            true,
-	"stop_reconciliation_claim_error":         true,
-	"stop_reconciliation_missing_lease_token": true,
+	"expire_retryable_error":                   true,
+	"stranded_reaper_error":                    true,
+	"missing_lease_token":                      true,
+	"operation_storage_error":                  true,
+	"missing_operation_lease_token":            true,
+	"operation_set_list_error":                 true,
+	"operation_set_missing":                    true,
+	"operation_task_inspect_error":             true,
+	"operation_cutover_storage_error":          true,
+	"missing_operation_cutover_lease_token":    true,
+	"operation_cutover_set_list_error":         true,
+	"operation_cutover_set_invalid":            true,
+	"operation_parent_load_error":              true,
+	"operation_parent_missing":                 true,
+	"operation_parent_claim_error":             true,
+	"operation_parent_not_claimable":           true,
+	"operation_lease_release_error":            true,
+	"missing_operation_deployment":             true,
+	"stop_reconciliation_claim_error":          true,
+	"stop_reconciliation_missing_lease_token":  true,
+	"operation_projection_claim_error":         true,
+	"operation_projection_missing_lease_token": true,
 }
 
 // RecordOperatorClaimFailure increments the operator claim failure counter.
@@ -968,6 +970,26 @@ func RecordOperatorStuckPendingScanFailure(ctx context.Context) {
 	)
 }
 
+// RecordOperatorOperationProjectionRepair counts applies the operator settled by
+// re-deriving the parent from operation rows that had all already settled — a
+// parent left behind its own children because a drive stopped between writing
+// its terminal operation row and projecting the parent. Every increment is one
+// target that was blocked for the lease staleness window by the one-active-apply
+// guard, so a sustained rate means drives are dying mid-projection and the
+// expected operator action is to look for driver crashes, evictions, or
+// deploy-time terminations rather than at this repair path. derived_state names
+// the verdict the repair landed on, so a spike that is all failed reads
+// differently from one that is all completed.
+func RecordOperatorOperationProjectionRepair(ctx context.Context, database, deployment, environment, derivedState string) {
+	addOperatorCounter(ctx, "operation_projection_repairs_total",
+		"Total number of applies settled by re-deriving the parent from already-settled operation rows", "{apply}",
+		attribute.String("database", database),
+		DeploymentAttribute(deployment),
+		EnvironmentAttribute(environment),
+		attribute.String("derived_state", derivedState),
+	)
+}
+
 var knownOperatorTerminalSummaryFailureReasons = map[string]bool{
 	"reload_apply_error":           true,
 	"apply_missing":                true,

pkg/storage/mysqlstore/applies.go

@@ -1603,6 +1603,91 @@ func (s *applyStore) FindNextApplyForStopReconciliation(ctx context.Context, own
 	return apply, nil
 }
 
+// FindNextApplyForOperationProjection claims one apply whose operation rows have
+// all settled while the apply itself is still non-terminal and its heartbeat has
+// gone stale, so the operator can re-derive the parent from its operations. See
+// the interface doc for why a parent can be left behind its own children.
+func (s *applyStore) FindNextApplyForOperationProjection(ctx context.Context, owner string) (*storage.Apply, error) {
+	if owner == "" {
+		return nil, fmt.Errorf("operator owner is required to claim apply for operation projection: %w", storage.ErrApplyLeaseLost)
+	}
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return nil, fmt.Errorf("begin claim apply for operation projection transaction: %w", err)
+	}
+	defer rollbackTx(ctx, tx, "claim apply for operation projection")
+
+	// Parent eligibility is the recovery-claimable set: the same states
+	// ClaimApplyByID re-leases when a heartbeat goes stale, which is exactly the
+	// population a crashed operation drive can leave behind. pending is excluded
+	// because an apply whose operations all settled was necessarily driven, and
+	// the resumable stopped and failed_retryable states have their own resume
+	// paths.
+	parentStates := claimableApplyStates()
+	parentStatePlaceholders := placeholders(len(parentStates))
+
+	// Any operation not in a terminal state can still move the parent on its own,
+	// so the parent is not stranded and this path leaves it alone. That includes
+	// pending operations: a pending row under a non-terminal parent is queued
+	// work for the operation claim, not residue.
+	settledOpStates := terminalApplyStates()
+	settledOpStatePlaceholders := placeholders(len(settledOpStates))
+
+	// The staleness gate is what keeps this off live rollouts. A driver that is
+	// mid-projection still holds a fresh lease, so only a parent whose owner has
+	// stopped heartbeating for the full window is claimable here.
+	staleClaimCutoff := s.dialect.RelativeTime(TimestampPrecisionDefault, BeforeCurrentTime, LiteralIntervalAmount(uint64(storage.ApplyLeaseStaleAfter.Microseconds())), IntervalMicrosecond)
+
+	queryArgs := stringArgs(parentStates)
+	queryArgs = append(queryArgs, stringArgs(settledOpStates)...)
+
+	row := tx.QueryRowContext(ctx, fmt.Sprintf(`
+		SELECT %s
+		FROM applies a
+		WHERE a.state IN (%s)
+			AND a.updated_at < %s
+			AND EXISTS (
+				SELECT 1
+				FROM apply_operations ao
+				WHERE ao.apply_id = a.id
+			)
+			AND NOT EXISTS (
+				SELECT 1
+				FROM apply_operations unsettled
+				WHERE unsettled.apply_id = a.id
+					AND unsettled.state NOT IN (%s)
+			)
+		ORDER BY a.created_at
+		LIMIT 1
+		FOR UPDATE SKIP LOCKED
+	`, applyColumns, parentStatePlaceholders, staleClaimCutoff, settledOpStatePlaceholders), queryArgs...)
+
+	apply, err := scanApplyInto(row)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, nil // nothing to reconcile
+	}
+	if err != nil {
+		return nil, fmt.Errorf("query next apply for operation projection: %w", err)
+	}
+
+	// persistApplyClaim rotates the lease and refreshes the heartbeat. Refreshing
+	// it is what bounds re-claim churn: a parent the projection cannot settle is
+	// not reconsidered until the staleness window elapses again.
+	outcome, err := persistApplyClaim(ctx, s.db, s.locker, tx, apply, owner)
+	if err != nil {
+		return nil, err
+	}
+	if outcome != claimAcquired {
+		return nil, nil
+	}
+
+	if err := tx.Commit(); err != nil {
+		return nil, fmt.Errorf("commit claim apply %d (%s) for operation projection: %w", apply.ID, apply.ApplyIdentifier, err)
+	}
+
+	return apply, nil
+}
+
 // claimOutcome describes how persistApplyClaim resolved a claim attempt and
 // who owns committing the transaction.
 type claimOutcome int