From b3c587eadd3516931112c913736a2e984fcf4016 Mon Sep 17 00:00:00 2001 From: Danil Silantyev Date: Mon, 7 Sep 2026 04:14:17 +0500 Subject: [PATCH] fix(recovery): require exact progress and preserve blocked stalled work Separate restart eligibility from stalled identity observation. Reject incomplete progress partitions, prevent interrupted attempts from replaying restarts using stale authorization, and retain blocked incidents as unhealthy. Add focused regressions and a coordinated rollout/evidence contract. --- docs/runbooks/recovery-evidence-contract.md | 53 ++++++ internal/schedulerrecovery/command.go | 7 + internal/schedulerrecovery/controller.go | 6 +- internal/schedulerrecovery/evaluate.go | 21 +- internal/schedulerrecovery/observe.go | 11 ++ .../schedulerrecovery/progress_contract.go | 40 ++++ .../progress_contract_test.go | 180 ++++++++++++++++++ internal/schedulerrecovery/recover.go | 40 +++- 8 files changed, 343 insertions(+), 15 deletions(-) create mode 100644 docs/runbooks/recovery-evidence-contract.md create mode 100644 internal/schedulerrecovery/progress_contract.go create mode 100644 internal/schedulerrecovery/progress_contract_test.go diff --git a/docs/runbooks/recovery-evidence-contract.md b/docs/runbooks/recovery-evidence-contract.md new file mode 100644 index 00000000..24451aba --- /dev/null +++ b/docs/runbooks/recovery-evidence-contract.md @@ -0,0 +1,53 @@ +# Scheduler recovery evidence contract + +## Implemented boundary + +A stalled identity remains observable even while restarting the dispatcher is +unsafe. The observation command may return `recovery_blockers`, an array of +bounded reason codes. `Evaluate` preserves `Decision.Stuck` and refuses recovery; +the controller emits `unhealthy`, not `healthy`, for this condition. The collector +must evaluate actual worker/claim occupancy independently from stalled-job +observation. Do not remove worker protection to make the queue appear responsive. + +Progress commands must return exactly one JSON object whose `progressed` and +`remaining` arrays form a complete disjoint partition of the attempt's expected +identities. Missing identities, unrelated work, empty objects, duplicate IDs and +concatenated JSON values are rejected. The recovery engine checks this contract +for every executor, not only the command adapter. Invalid attempts are refused +before acquiring their durable state. + +Interrupted attempts may verify progress but must not replay a manager restart +from old authorization. Incomplete or unknown progress finishes the attempt as +failed; a subsequent current observation must pass the normal policy and +cooldown. A syntactically valid partition still needs real per-identity evidence +from the deployment adapter. A vanished row is not a forward transition. + +## Coordinated rollout + +Pause only the recovery timer during the binary/adapter/config replacement; +do not stop running build workers. Back up the installed recovery artifacts and +state using the deployment's normal maintenance path. Install the new binary +before the adapter that emits `recovery_blockers`: older binaries reject the +unknown field rather than interpreting it as permission to restart. Verify the +exact installed source/build/config hashes and exercise observation without an +actual restart before re-enabling the timer. + +The adapter must bind checkpoints and progress to `GHA_SCHEDULER_RECOVERY_ATTEMPT` +and the exact `GHA_SCHEDULER_RECOVERY_STUCK` set; preserve original snapshots and +recheck eligibility immediately before a restart. Do not hold provider-journal +locks across systemctl shutdown. An atomic admission fence and scale-set-local +repair are separate work; a last-moment read alone does not eliminate that race. + +## Verification + +Run `go test -race ./internal/schedulerrecovery` and the repository's full checks +on its pinned Go toolchain. The focused offline review exercised the changed +production files with Go 1.23.2, an unchanged type-only extraction of `Heartbeat`, +and the new standard-library regression tests. It did not execute FileStore, +the full dependency graph, systemd, Incus, or real GitHub job delivery. + +The regressions cover exact partitions, unknown resume progress, fresh-decision +requirements, blocked-but-visible incidents, and strict single-value JSON. +Runtime acceptance additionally requires an affected real job to advance, no +unrelated running job lost, and matched pre-start/total-workflow measurements. +Neither a green PR nor fewer notifications is fleet throughput acceptance. diff --git a/internal/schedulerrecovery/command.go b/internal/schedulerrecovery/command.go index c5b309a9..4b9b676d 100644 --- a/internal/schedulerrecovery/command.go +++ b/internal/schedulerrecovery/command.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "os/exec" "path/filepath" "strings" @@ -78,6 +79,12 @@ func (executor CommandExecutor) AwaitProgress(ctx context.Context, attempt Attem if err := decoder.Decode(&progress); err != nil { return nil, nil, fmt.Errorf("decode progress output: %w", err) } + if err := decoder.Decode(new(any)); err != io.EOF { + return nil, nil, fmt.Errorf("progress output must contain exactly one JSON value") + } + if err := validateProgress(attempt.Stuck, progress.Progressed, progress.Remaining); err != nil { + return nil, nil, err + } return progress.Progressed, progress.Remaining, nil } diff --git a/internal/schedulerrecovery/controller.go b/internal/schedulerrecovery/controller.go index fb55f9c6..56939589 100644 --- a/internal/schedulerrecovery/controller.go +++ b/internal/schedulerrecovery/controller.go @@ -3,6 +3,7 @@ package schedulerrecovery import ( "context" "fmt" + "strings" "time" ) @@ -75,7 +76,7 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error) observation.HeartbeatAt = heartbeat.At decision := Evaluate(controller.Policy, observation) state := "healthy" - if decision.Recover { + if decision.Recover || strings.HasPrefix(decision.Reason, "recovery-blocked:") { state = "unhealthy" } if err := controller.emit(ctx, Event{At: observation.ObservedAt, State: state, Reason: decision.Reason, Stuck: decision.Stuck}); err != nil { @@ -84,6 +85,9 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error) if !decision.Recover { return decision, Result{}, nil } + if err := validateProgress(decision.Stuck, nil, decision.Stuck); err != nil { + return decision, Result{}, fmt.Errorf("invalid recovery identities: %w", err) + } attempt := NewAttempt(observation.ObservedAt, decision.Stuck) acquired, err := controller.Attempts.Begin(ctx, attempt) if err != nil { diff --git a/internal/schedulerrecovery/evaluate.go b/internal/schedulerrecovery/evaluate.go index f335bdeb..5f823406 100644 --- a/internal/schedulerrecovery/evaluate.go +++ b/internal/schedulerrecovery/evaluate.go @@ -1,6 +1,10 @@ package schedulerrecovery -import "time" +import ( + "slices" + "strings" + "time" +) type Policy struct { MinimumStuckAge time.Duration @@ -40,10 +44,12 @@ type Observation struct { OverdueRetries []ProviderRetry StaleAssigned []AssignedIntent CapacityBackpressure bool - ManagerUptime time.Duration - LastRecoveryAt time.Time - HeartbeatAt time.Time - RecoveryRunning bool + // RecoveryBlockers preserve stalled identities while a manager-wide restart is unsafe. + RecoveryBlockers []string + ManagerUptime time.Duration + LastRecoveryAt time.Time + HeartbeatAt time.Time + RecoveryRunning bool } type Decision struct { @@ -82,6 +88,11 @@ func Evaluate(policy Policy, observation Observation) Decision { if len(stuck) == 0 { return Decision{Reason: "no-stale-undispatched-instance"} } + if len(observation.RecoveryBlockers) != 0 { + blockers := slices.Clone(observation.RecoveryBlockers) + slices.Sort(blockers) + return Decision{Reason: "recovery-blocked:" + strings.Join(slices.Compact(blockers), ","), Stuck: stuck} + } // A process-wide heartbeat proves only that some dispatcher work advanced. // It cannot clear an exact retry that is already overdue: production has // shown one scale set parked while sibling classes kept the heartbeat fresh. diff --git a/internal/schedulerrecovery/observe.go b/internal/schedulerrecovery/observe.go index f31707f0..3ef6e3e1 100644 --- a/internal/schedulerrecovery/observe.go +++ b/internal/schedulerrecovery/observe.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "io" "os/exec" "path/filepath" "strings" @@ -26,6 +27,7 @@ type observationOutput struct { ManagerUptimeSeconds int64 `json:"manager_uptime_seconds"` LastRecoveryAt time.Time `json:"last_recovery_at"` RecoveryRunning bool `json:"recovery_running"` + RecoveryBlockers []string `json:"recovery_blockers"` } func (observer CommandObserver) Validate() error { @@ -63,6 +65,14 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error if err := decoder.Decode(&decoded); err != nil { return Observation{}, fmt.Errorf("decode scheduler observation: %w", err) } + if err := decoder.Decode(new(any)); err != io.EOF { + return Observation{}, fmt.Errorf("scheduler observation must contain exactly one JSON value") + } + for _, blocker := range decoded.RecoveryBlockers { + if strings.TrimSpace(blocker) == "" || len(blocker) > 128 || strings.ContainsAny(blocker, "\r\n\x00,") { + return Observation{}, fmt.Errorf("scheduler observation contains an invalid recovery blocker") + } + } if decoded.ObservedAt.IsZero() || decoded.ActiveIntents < 0 || decoded.ManagerUptimeSeconds < 0 { return Observation{}, fmt.Errorf("scheduler observation contains invalid values") } @@ -85,6 +95,7 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error ObservedAt: decoded.ObservedAt, ActiveIntents: decoded.ActiveIntents, PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries, StaleAssigned: decoded.StaleAssigned, CapacityBackpressure: decoded.CapacityBackpressure, + RecoveryBlockers: decoded.RecoveryBlockers, ManagerUptime: time.Duration(decoded.ManagerUptimeSeconds) * time.Second, LastRecoveryAt: decoded.LastRecoveryAt, RecoveryRunning: decoded.RecoveryRunning, }, nil diff --git a/internal/schedulerrecovery/progress_contract.go b/internal/schedulerrecovery/progress_contract.go new file mode 100644 index 00000000..84c2022c --- /dev/null +++ b/internal/schedulerrecovery/progress_contract.go @@ -0,0 +1,40 @@ +package schedulerrecovery + +import ( + "fmt" + "strings" +) + +// validateProgress requires a complete, disjoint partition of the original +// identities. Empty output, omitted identities, duplicates and unrelated jobs +// are not proof of recovery. Adapters must separately prove each transition. +func validateProgress(expected, progressed, remaining []string) error { + if len(expected) == 0 { + return fmt.Errorf("recovery progress requires expected identities") + } + want := make(map[string]bool, len(expected)) + for _, id := range expected { + if strings.TrimSpace(id) == "" || strings.ContainsAny(id, ",\x00\r\n") { + return fmt.Errorf("invalid recovery identity") + } + if _, duplicate := want[id]; duplicate { + return fmt.Errorf("duplicate expected recovery identity") + } + want[id] = false + } + for _, group := range [][]string{progressed, remaining} { + for _, id := range group { + seen, exists := want[id] + if !exists || seen { + return fmt.Errorf("progress contains an unexpected or repeated identity") + } + want[id] = true + } + } + for _, seen := range want { + if !seen { + return fmt.Errorf("progress omits an expected recovery identity") + } + } + return nil +} diff --git a/internal/schedulerrecovery/progress_contract_test.go b/internal/schedulerrecovery/progress_contract_test.go new file mode 100644 index 00000000..e86cb445 --- /dev/null +++ b/internal/schedulerrecovery/progress_contract_test.go @@ -0,0 +1,180 @@ +package schedulerrecovery + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestProgressPartitionRejectsMissingForeignAndRepeatedIdentities(t *testing.T) { + cases := []struct { + name string + expected, progressed, remaining []string + valid bool + }{ + {"complete", []string{"a", "b"}, []string{"b", "a"}, nil, true}, + {"partial", []string{"a", "b"}, []string{"a"}, []string{"b"}, true}, + {"none", []string{"a"}, nil, []string{"a"}, true}, + {"empty-output", []string{"a"}, nil, nil, false}, + {"foreign", []string{"a"}, []string{"b"}, nil, false}, + {"omitted", []string{"a", "b"}, []string{"a"}, nil, false}, + {"overlap", []string{"a"}, []string{"a"}, []string{"a"}, false}, + {"duplicate-progress", []string{"a"}, []string{"a", "a"}, nil, false}, + {"duplicate-expected", []string{"a", "a"}, []string{"a"}, nil, false}, + {"empty-expected", nil, nil, nil, false}, + {"delimiter", []string{"a,b"}, nil, []string{"a,b"}, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if err := validateProgress(tc.expected, tc.progressed, tc.remaining); (err == nil) != tc.valid { + t.Fatalf("valid=%v, got %v", tc.valid, err) + } + }) + } +} + +func TestRecoveryBlockersPreserveExactStalledIdentities(t *testing.T) { + now := time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC) + policy := Policy{MinimumStuckAge: time.Minute, MinimumUptime: time.Minute, Cooldown: time.Minute, HeartbeatStale: time.Minute} + observation := Observation{ObservedAt: now, ActiveIntents: 1, ManagerUptime: time.Hour, + StaleAssigned: []AssignedIntent{{ID: "a", Age: time.Hour}}, + RecoveryBlockers: []string{"provider-leases-active", "provider-journal-unknown"}} + decision := Evaluate(policy, observation) + if decision.Recover || len(decision.Stuck) != 1 || decision.Stuck[0] != "a" || !strings.HasPrefix(decision.Reason, "recovery-blocked:") { + t.Fatalf("blocked work must remain visible: %+v", decision) + } + if observation.RecoveryBlockers[0] != "provider-leases-active" { + t.Fatal("evaluation mutated caller-owned observation") + } + observation.RecoveryBlockers = nil + if !Evaluate(policy, observation).Recover { + t.Fatal("eligible unchanged intent did not recover after blocker cleared") + } +} + +type evidenceStore struct { + results []Result + begins int +} + +func (s *evidenceStore) Active(context.Context) ([]Attempt, error) { return nil, nil } +func (s *evidenceStore) Begin(context.Context, Attempt) (bool, error) { s.begins++; return true, nil } +func (s *evidenceStore) Finish(_ context.Context, r Result) error { + s.results = append(s.results, r) + return nil +} + +type evidenceExecutor struct { + progressed, remaining []string + err error + restarts int +} + +func (e *evidenceExecutor) Checkpoint(context.Context, Attempt) (string, error) { + return "checkpoint", nil +} +func (e *evidenceExecutor) RestartDispatcher(context.Context, Attempt) error { + e.restarts++ + return nil +} +func (e *evidenceExecutor) AwaitProgress(context.Context, Attempt) ([]string, []string, error) { + return e.progressed, e.remaining, e.err +} + +func TestRecoveryCannotSucceedWithEmptyExecutorProof(t *testing.T) { + store, executor := &evidenceStore{}, &evidenceExecutor{} + r, err := Recover(context.Background(), time.Now(), Decision{Recover: true, Stuck: []string{"a"}}, store, executor, time.Now) + if err == nil || r.Recovered || len(r.Remaining) != 1 || len(store.results) != 1 { + t.Fatalf("malformed progress accepted: result=%+v err=%v", r, err) + } +} + +func TestResumeDoesNotRestartOnUnknownProgress(t *testing.T) { + for _, proofErr := range []error{nil, errors.New("observer unavailable")} { + store, executor := &evidenceStore{}, &evidenceExecutor{err: proofErr} + r, err := resumeAcquired(context.Background(), Attempt{ID: "attempt", Stuck: []string{"a"}}, store, executor, time.Now) + if err == nil || r.Recovered || executor.restarts != 0 || len(store.results) != 1 { + t.Fatalf("unknown progress caused destructive retry: result=%+v err=%v restarts=%d", r, err, executor.restarts) + } + } +} + +func TestRecoveryRejectsAmbiguousAttemptBeforeStoreMutation(t *testing.T) { + store := &evidenceStore{} + _, err := Recover(context.Background(), time.Now(), Decision{Recover: true, Stuck: []string{"a", "a"}}, store, &evidenceExecutor{}, time.Now) + if err == nil || store.begins != 0 { + t.Fatalf("invalid attempt was acquired: %v", err) + } +} + +func outputCommand(t *testing.T, body string) []string { + t.Helper() + path := filepath.Join(t.TempDir(), "output.sh") + if err := os.WriteFile(path, []byte("#!/bin/sh\ncat <<'PROOF'\n"+body+"\nPROOF\n"), 0o700); err != nil { + t.Fatal(err) + } + return []string{path} +} + +func TestCommandProgressRejectsTrailingJSONAndEmptyObjects(t *testing.T) { + for _, body := range []string{"{}", `{"progressed":["a"],"remaining":[]} {}`, `{"progressed":["b"],"remaining":[]}`} { + argv := outputCommand(t, body) + executor := CommandExecutor{Config: CommandConfig{Checkpoint: argv, Restart: argv, Progress: argv, Timeout: time.Second}} + if _, _, err := executor.AwaitProgress(context.Background(), Attempt{Stuck: []string{"a"}}); err == nil { + t.Fatalf("accepted invalid proof %s", body) + } + } +} + +func TestCommandObservationPreservesRecoveryBlockers(t *testing.T) { + body := `{"observed_at":"2026-09-07T00:00:00Z","active_intents":1,"manager_uptime_seconds":600,"stale_assigned_intents":[{"id":"a","age_nanoseconds":600000000000}],"recovery_blockers":["provider-journal-unknown"]}` + observer := CommandObserver{Argv: outputCommand(t, body), Timeout: time.Second} + observation, err := observer.Observe(context.Background()) + if err != nil || len(observation.RecoveryBlockers) != 1 { + t.Fatalf("lost blocker: %+v %v", observation, err) + } + observer.Argv = outputCommand(t, body+` {}`) + if _, err := observer.Observe(context.Background()); err == nil { + t.Fatal("accepted concatenated JSON") + } +} + +func TestResumeRequiresFreshDecisionForRemainingWork(t *testing.T) { + store, executor := &evidenceStore{}, &evidenceExecutor{remaining: []string{"a"}} + r, err := resumeAcquired(context.Background(), Attempt{ID: "attempt", Stuck: []string{"a"}}, store, executor, time.Now) + if err == nil || r.Recovered || executor.restarts != 0 || len(store.results) != 1 { + t.Fatalf("stored attempt bypassed fresh eligibility: result=%+v err=%v", r, err) + } +} + +type proofObserver struct{ observation Observation } + +func (o proofObserver) Observe(context.Context) (Observation, error) { return o.observation, nil } + +type proofHeartbeat struct{} + +func (proofHeartbeat) ReadHeartbeat(context.Context) (Heartbeat, error) { return Heartbeat{}, nil } + +type proofEvents struct{ events []Event } + +func (e *proofEvents) Emit(_ context.Context, event Event) error { + e.events = append(e.events, event) + return nil +} + +func TestBlockedControllerIsUnhealthyWithoutRestart(t *testing.T) { + at := time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC) + store, executor, events := &evidenceStore{}, &evidenceExecutor{}, &proofEvents{} + controller := Controller{Policy: Policy{MinimumStuckAge: time.Minute}, + Observer: proofObserver{Observation{ObservedAt: at, ActiveIntents: 1, ManagerUptime: time.Hour, + StaleAssigned: []AssignedIntent{{ID: "a", Age: time.Hour}}, RecoveryBlockers: []string{"provider-leases-active"}}}, + Heartbeat: proofHeartbeat{}, Attempts: store, Executor: executor, Events: events, Now: func() time.Time { return at }} + decision, _, err := controller.Tick(context.Background()) + if err != nil || decision.Recover || store.begins != 0 || executor.restarts != 0 || len(events.events) != 1 || events.events[0].State != "unhealthy" { + t.Fatalf("blocked work was hidden or restarted: decision=%+v events=%+v err=%v", decision, events.events, err) + } +} diff --git a/internal/schedulerrecovery/recover.go b/internal/schedulerrecovery/recover.go index f5e8a3e4..40d5b2ef 100644 --- a/internal/schedulerrecovery/recover.go +++ b/internal/schedulerrecovery/recover.go @@ -92,7 +92,19 @@ type AttemptStore interface { func resumeAcquired(ctx context.Context, attempt Attempt, store AttemptStore, executor Executor, now func() time.Time) (Result, error) { progressed, remaining, progressErr := executor.AwaitProgress(ctx, attempt) - if progressErr == nil && len(remaining) == 0 { + if progressErr == nil { + progressErr = validateProgress(attempt.Stuck, progressed, remaining) + } + if progressErr != nil { + // Unknown or malformed progress never authorizes another manager restart. + result := Result{AttemptID: attempt.ID, Remaining: slices.Clone(attempt.Stuck), + FinishedAt: now().UTC(), Error: "verify resumed recovery progress: " + progressErr.Error()} + if err := store.Finish(ctx, result); err != nil { + return result, fmt.Errorf("finish resumed recovery attempt: %w", err) + } + return result, fmt.Errorf("verify resumed recovery progress: %w", progressErr) + } + if len(remaining) == 0 { result := Result{ AttemptID: attempt.ID, Progressed: slices.Clone(progressed), Recovered: true, FinishedAt: now().UTC(), @@ -102,13 +114,16 @@ func resumeAcquired(ctx context.Context, attempt Attempt, store AttemptStore, ex } return result, nil } - // The previous process may have died before or during the manager restart. - // Re-running the checkpoint-first sequence is idempotent; restricting it to - // the still-stuck identities avoids replaying work already proven progressed. - if len(remaining) > 0 { - attempt.Stuck = slices.Clone(remaining) + // A stored attempt is not fresh authorization to restart the dispatcher. + // Finish it as incomplete and let the next observation re-evaluate current + // work, blockers, startup grace and cooldown instead of replaying a restart. + result := Result{AttemptID: attempt.ID, Progressed: slices.Clone(progressed), + Remaining: slices.Clone(remaining), FinishedAt: now().UTC(), + Error: "resumed recovery incomplete: fresh evaluation required"} + if err := store.Finish(ctx, result); err != nil { + return result, fmt.Errorf("finish resumed recovery attempt: %w", err) } - return recoverAcquired(ctx, attempt, store, executor, now) + return result, fmt.Errorf("resumed recovery incomplete: %d identities remain", len(remaining)) } type Executor interface { @@ -129,6 +144,9 @@ func Recover(ctx context.Context, observedAt time.Time, decision Decision, store if !decision.Recover { return Result{}, fmt.Errorf("recovery refused: %s", decision.Reason) } + if err := validateProgress(decision.Stuck, nil, decision.Stuck); err != nil { + return Result{}, fmt.Errorf("invalid recovery identities: %w", err) + } attempt := NewAttempt(observedAt, decision.Stuck) acquired, err := store.Begin(ctx, attempt) if err != nil { @@ -161,11 +179,15 @@ func recoverAcquired(ctx context.Context, attempt Attempt, store AttemptStore, e return finish(fmt.Errorf("restart dispatcher: %w", err)) } progressed, remaining, err := executor.AwaitProgress(ctx, attempt) - result.Progressed = slices.Clone(progressed) - result.Remaining = slices.Clone(remaining) + if err == nil { + err = validateProgress(attempt.Stuck, progressed, remaining) + } if err != nil { + result.Remaining = slices.Clone(attempt.Stuck) return finish(fmt.Errorf("verify dispatcher progress: %w", err)) } + result.Progressed = slices.Clone(progressed) + result.Remaining = slices.Clone(remaining) result.Recovered = len(remaining) == 0 if !result.Recovered { return finish(fmt.Errorf("recovery incomplete: %d stuck instances remain", len(remaining)))