diff --git a/integration/operator_test.go b/integration/operator_test.go index dc9dcf57..2afe1b19 100644 --- a/integration/operator_test.go +++ b/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. diff --git a/pkg/api/handlers_test.go b/pkg/api/handlers_test.go index a745003e..f88b2e21 100644 --- a/pkg/api/handlers_test.go +++ b/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 } diff --git a/pkg/api/operator.go b/pkg/api/operator.go index d2ab9485..49b6c33e 100644 --- a/pkg/api/operator.go +++ b/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 diff --git a/pkg/api/operator_test.go b/pkg/api/operator_test.go index 92e74121..5d25cfd5 100644 --- a/pkg/api/operator_test.go +++ b/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() diff --git a/pkg/metrics/README.md b/pkg/metrics/README.md index ada44dea..65a20ffd 100644 --- a/pkg/metrics/README.md +++ b/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` diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 5d906b7e..d10952c2 100644 --- a/pkg/metrics/metrics.go +++ b/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, diff --git a/pkg/storage/mysqlstore/applies.go b/pkg/storage/mysqlstore/applies.go index 54ffc0ef..c07bbdd7 100644 --- a/pkg/storage/mysqlstore/applies.go +++ b/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 diff --git a/pkg/storage/mysqlstore/applies_test.go b/pkg/storage/mysqlstore/applies_test.go index a11ae851..dc603b1a 100644 --- a/pkg/storage/mysqlstore/applies_test.go +++ b/pkg/storage/mysqlstore/applies_test.go @@ -4439,6 +4439,138 @@ func TestApplyStore_FindNextApplyForStopReconciliation_SkipsApplyWithoutPendingS assert.Nil(t, claimed, "an apply without a pending stop is not a reconciliation candidate") } +// stageOperationProjectionOrphan builds an apply whose operations are all in the +// given states while the apply itself stays in parentState, then ages the apply's +// heartbeat by staleBy so the projection claim's staleness gate can be exercised +// in either direction. +func stageOperationProjectionOrphan(t *testing.T, store *Storage, identifier, parentState string, staleBy time.Duration, opStates ...string) *storage.Apply { + t.Helper() + ctx := t.Context() + + lock := createTestLock(t, store, "testdb", "mysql", "staging") + apply := createTestApplyWithStateAndEnv(t, store, lock, identifier, 1, parentState, "staging") + for i, opState := range opStates { + _, err := store.ApplyOperations().Insert(ctx, &storage.ApplyOperation{ + ApplyID: apply.ID, + Deployment: fmt.Sprintf("region-%d", i), + State: opState, + }) + require.NoError(t, err) + } + _, err := testDB.ExecContext(ctx, + `UPDATE applies SET updated_at = NOW() - INTERVAL ? SECOND WHERE id = ?`, + int64(staleBy.Seconds()), apply.ID) + require.NoError(t, err) + return apply +} + +// TestApplyStore_FindNextApplyForOperationProjection_ClaimsSettledOperationOrphan +// verifies the repair trigger for a parent left behind its own children: every +// operation has reached a terminal state, the apply itself is still running, and +// its heartbeat has gone stale, so no drive is coming back to project the +// outcome. The apply is claimable here so the operator can derive its state from +// the operations and release the target. +func TestApplyStore_FindNextApplyForOperationProjection_ClaimsSettledOperationOrphan(t *testing.T) { + clearTables(t) + store := New(testDB) + + apply := stageOperationProjectionOrphan(t, store, "apply_op_projection", state.Apply.Running, + 2*storage.ApplyLeaseStaleAfter, state.ApplyOperation.Completed, state.ApplyOperation.Completed) + + claimed, err := store.Applies().FindNextApplyForOperationProjection(t.Context(), "test-operator") + require.NoError(t, err) + require.NotNil(t, claimed, "an apply whose operations have all settled must be claimable for projection") + assert.Equal(t, apply.ID, claimed.ID) + assert.Equal(t, "test-operator", claimed.LeaseOwner, "the claim rotates the lease owner") + assert.Equal(t, state.Apply.Running, claimed.State, "the claim refreshes the lease without changing apply state") +} + +// TestApplyStore_FindNextApplyForOperationProjection_SkipsLiveAndUnsettledApplies +// verifies the two gates that keep this repair off applies that are not +// stranded. A fresh heartbeat means a driver is still on the apply and may be +// mid-projection. Any non-terminal operation — pending queued work included — +// means an operation claim arm can still move the parent on its own. +func TestApplyStore_FindNextApplyForOperationProjection_SkipsLiveAndUnsettledApplies(t *testing.T) { + cases := []struct { + name string + staleBy time.Duration + opStates []string + reason string + }{ + { + name: "fresh heartbeat", + staleBy: 0, + opStates: []string{state.ApplyOperation.Completed, state.ApplyOperation.Completed}, + reason: "a driver still heartbeating the apply may be mid-projection", + }, + { + name: "pending operation left", + staleBy: 2 * storage.ApplyLeaseStaleAfter, + opStates: []string{state.ApplyOperation.Completed, state.ApplyOperation.Pending}, + reason: "a pending operation is queued work for the operation claim, not residue", + }, + { + name: "running operation left", + staleBy: 2 * storage.ApplyLeaseStaleAfter, + opStates: []string{state.ApplyOperation.Completed, state.ApplyOperation.Running}, + reason: "a running operation is recovered by the operation claim, which projects the parent itself", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + clearTables(t) + store := New(testDB) + + stageOperationProjectionOrphan(t, store, "apply_op_projection_skip", state.Apply.Running, tc.staleBy, tc.opStates...) + + claimed, err := store.Applies().FindNextApplyForOperationProjection(t.Context(), "test-operator") + require.NoError(t, err) + assert.Nil(t, claimed, tc.reason) + }) + } +} + +// TestApplyStore_FindNextApplyForOperationProjection_SkipsTerminalAndChildlessApplies +// verifies the repair is scoped to parents that are genuinely behind their +// children. A terminal apply has already recorded its verdict, and an apply with +// no operation rows has nothing to derive a state from. +func TestApplyStore_FindNextApplyForOperationProjection_SkipsTerminalAndChildlessApplies(t *testing.T) { + t.Run("terminal apply", func(t *testing.T) { + clearTables(t) + store := New(testDB) + + stageOperationProjectionOrphan(t, store, "apply_op_projection_terminal", state.Apply.Completed, + 2*storage.ApplyLeaseStaleAfter, state.ApplyOperation.Completed) + + claimed, err := store.Applies().FindNextApplyForOperationProjection(t.Context(), "test-operator") + require.NoError(t, err) + assert.Nil(t, claimed, "a terminal apply has already recorded its verdict") + }) + + t.Run("apply with no operations", func(t *testing.T) { + clearTables(t) + store := New(testDB) + + stageOperationProjectionOrphan(t, store, "apply_op_projection_childless", state.Apply.Running, + 2*storage.ApplyLeaseStaleAfter) + + claimed, err := store.Applies().FindNextApplyForOperationProjection(t.Context(), "test-operator") + require.NoError(t, err) + assert.Nil(t, claimed, "an apply with no operation rows has nothing to project from") + }) +} + +// TestApplyStore_FindNextApplyForOperationProjection_RequiresOwner verifies the +// claim fails closed without an owner: an unowned claim would rotate a lease +// nobody holds, leaving the apply writable by any driver. +func TestApplyStore_FindNextApplyForOperationProjection_RequiresOwner(t *testing.T) { + clearTables(t) + store := New(testDB) + + _, err := store.Applies().FindNextApplyForOperationProjection(t.Context(), "") + require.ErrorIs(t, err, storage.ErrApplyLeaseLost) +} + // SetRevertSkipped records skip-revert on the apply and the timestamp round-trips // through Get, so progress can show that revert was skipped without an // engine-specific side table. It is a targeted write that leaves other fields diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 0b022af1..3d1393e8 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -453,6 +453,22 @@ type ApplyStore interface { // by a peer. FindNextApplyForStopReconciliation(ctx context.Context, owner string) (*Apply, error) + // FindNextApplyForOperationProjection atomically claims one apply whose + // operation rows have all settled while the apply itself is still + // non-terminal and its heartbeat has gone stale, rotating the lease onto it + // like ClaimApplyByID. It is the repair path for a parent left behind its own + // children: an operation drive writes its terminal operation row and then + // projects the parent, so a crash between those two writes leaves nothing to + // finish the projection — every operation is terminal, so no operation-level + // claim arm matches, and the one-active-apply guard keeps that target blocked + // until the parent settles. Requiring a stale heartbeat is what keeps this off + // live rollouts: a driver mid-projection still holds a fresh lease. Applies + // with no operation rows are excluded (there is nothing to project from), as + // are pending, stopped, and failed_retryable parents, which have their own + // resume paths. Returns nil when no such apply exists or it is locked by a + // peer. + FindNextApplyForOperationProjection(ctx context.Context, owner string) (*Apply, error) + // Heartbeat updates the apply's updated_at timestamp to maintain the lease. // Should be called every 10 seconds while working on an apply. // If not called for > 1 minute, another driver can claim the apply.