From 5549459c53ef7f63f737db028eb36bc66fc759aa Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Mon, 7 Sep 2026 08:01:52 +0500 Subject: [PATCH] fix(recovery): preserve exact evidence and serialize attempt admission --- CHANGELOG.md | 6 + docs/runbooks/recovery-evidence-contract.md | 59 ++++++ internal/schedulerrecovery/command.go | 7 + internal/schedulerrecovery/controller.go | 7 +- internal/schedulerrecovery/controller_test.go | 33 +++- internal/schedulerrecovery/evaluate.go | 21 +- internal/schedulerrecovery/observe.go | 11 ++ .../schedulerrecovery/progress_contract.go | 40 ++++ .../progress_contract_test.go | 180 ++++++++++++++++++ internal/schedulerrecovery/recover.go | 44 ++++- internal/schedulerrecovery/recover_test.go | 16 +- internal/schedulerrecovery/store.go | 4 +- internal/schedulerrecovery/store_test.go | 38 ++++ 13 files changed, 445 insertions(+), 21 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/CHANGELOG.md b/CHANGELOG.md index db002a5d..f018d49e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## Unreleased +- Recovery requires exact complete progress evidence, preserves unresolved + identities on checkpoint/restart failures, and never replays interrupted + restart authorization. Blocked/suppressed stalled work remains unhealthy; + FileStore admits only one unfinished attempt across concurrent decisions. + Scoped action ownership and admission fencing remain separate rollout work. + - Publish background CI failures as unassigned repository-local issues with exact run/attempt/job identity. Classify `actions/ci-feedback/feedback.py` in the network-bootstrap inventory. The in-repo `workflow_run` caller includes diff --git a/docs/runbooks/recovery-evidence-contract.md b/docs/runbooks/recovery-evidence-contract.md new file mode 100644 index 00000000..8c2d02ab --- /dev/null +++ b/docs/runbooks/recovery-evidence-contract.md @@ -0,0 +1,59 @@ +# 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. +Startup grace, cooldown, current coarse heartbeat and duplicate-attempt +suppression also retain unhealthy state when exact stalled subjects remain. + +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. +Checkpoint or restart failure retains every original subject as unresolved. +FileStore atomically admits only one unfinished attempt, including when distinct +decisions race. This does not itself serialize all concurrent Tick/resume actors; +the production action-owner/fencing protocol remains part of scoped recovery. + +## 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 integration review now runs the complete package +on Go 1.26.7, including real FileStore races with distinct attempt IDs. Earlier +isolated Go 1.23.2 results are historical, not the integration boundary. No +systemd/Incus mutation or real-job recovery is proved by these source tests. + +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..07839ef9 100644 --- a/internal/schedulerrecovery/controller.go +++ b/internal/schedulerrecovery/controller.go @@ -75,7 +75,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 || len(decision.Stuck) > 0 { state = "unhealthy" } if err := controller.emit(ctx, Event{At: observation.ObservedAt, State: state, Reason: decision.Reason, Stuck: decision.Stuck}); err != nil { @@ -84,6 +84,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 { @@ -91,7 +94,7 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error) } if !acquired { result := Result{AttemptID: attempt.ID, Suppressed: true} - if err := controller.emit(ctx, Event{At: controller.Now().UTC(), State: "healthy", Reason: "duplicate-recovery-suppressed", AttemptID: attempt.ID, Stuck: attempt.Stuck}); err != nil { + if err := controller.emit(ctx, Event{At: controller.Now().UTC(), State: "unhealthy", Reason: "duplicate-recovery-suppressed", AttemptID: attempt.ID, Stuck: attempt.Stuck}); err != nil { return decision, result, err } return decision, result, nil diff --git a/internal/schedulerrecovery/controller_test.go b/internal/schedulerrecovery/controller_test.go index a792ae6f..77e063dd 100644 --- a/internal/schedulerrecovery/controller_test.go +++ b/internal/schedulerrecovery/controller_test.go @@ -57,7 +57,7 @@ func TestControllerRecoversFaultInjectedStoppedDispatcher(t *testing.T) { require.Equal(t, 1, executor.restarts) } -func TestControllerReportsHealthyCurrentHeartbeat(t *testing.T) { +func TestControllerKeepsStalledWorkUnhealthyDuringCurrentHeartbeat(t *testing.T) { t.Parallel() at := time.Date(2026, 8, 24, 10, 0, 0, 0, time.UTC) events := &eventRecorder{} @@ -71,7 +71,36 @@ func TestControllerReportsHealthyCurrentHeartbeat(t *testing.T) { require.NoError(t, err) require.False(t, decision.Recover) require.Equal(t, "dispatcher-heartbeat-current", decision.Reason) - require.Equal(t, "healthy", events.events[0].State) + require.Equal(t, "unhealthy", events.events[0].State) + require.Equal(t, []string{"instance-1"}, events.events[0].Stuck) +} + +func TestControllerSuppressionDoesNotClearIncident(t *testing.T) { + at := time.Now().UTC() + for _, reason := range []string{"manager-startup-grace", "recovery-cooldown"} { + t.Run(reason, func(t *testing.T) { + observation := Observation{ObservedAt: at, ActiveIntents: 1, ManagerUptime: time.Hour, + StaleAssigned: []AssignedIntent{{ID: "intent-a", Age: time.Hour}}} + if reason == "manager-startup-grace" { + observation.ManagerUptime = time.Second + } else { + observation.LastRecoveryAt = at.Add(-time.Second) + } + events, executor := &eventRecorder{}, &faultExecutor{} + controller := Controller{ + Policy: Policy{MinimumStuckAge: time.Minute, MinimumUptime: time.Minute, Cooldown: time.Minute, HeartbeatStale: time.Minute}, + Observer: staticObserver{observation}, Heartbeat: staticHeartbeat{}, Attempts: &memoryAttempts{}, + Executor: executor, Events: events, Now: func() time.Time { return at }, + } + decision, _, err := controller.Tick(context.Background()) + require.NoError(t, err) + require.False(t, decision.Recover) + require.Equal(t, reason, decision.Reason) + require.Zero(t, executor.restarts) + require.Equal(t, "unhealthy", events.events[0].State) + require.Equal(t, []string{"intent-a"}, events.events[0].Stuck) + }) + } } func TestControllerFinishesInterruptedRecoveryAfterRestartProgressed(t *testing.T) { 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..aa72025a 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 { @@ -141,7 +159,9 @@ func Recover(ctx context.Context, observedAt time.Time, decision Decision, store } func recoverAcquired(ctx context.Context, attempt Attempt, store AttemptStore, executor Executor, now func() time.Time) (Result, error) { - result := Result{AttemptID: attempt.ID} + // Until a complete proof is validated every original subject is unresolved, + // including when checkpointing or the restart command fails first. + result := Result{AttemptID: attempt.ID, Remaining: slices.Clone(attempt.Stuck)} finish := func(operationErr error) (Result, error) { result.FinishedAt = now().UTC() if operationErr != nil { @@ -161,11 +181,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))) diff --git a/internal/schedulerrecovery/recover_test.go b/internal/schedulerrecovery/recover_test.go index b343132f..b915d649 100644 --- a/internal/schedulerrecovery/recover_test.go +++ b/internal/schedulerrecovery/recover_test.go @@ -59,6 +59,7 @@ type faultExecutor struct { progressed []string remaining []string checkpoint error + restart error } func (executor *faultExecutor) Checkpoint(_ context.Context, _ Attempt) (string, error) { @@ -72,7 +73,7 @@ func (executor *faultExecutor) RestartDispatcher(_ context.Context, _ Attempt) e executor.mu.Lock() defer executor.mu.Unlock() executor.restarts++ - return nil + return executor.restart } func (executor *faultExecutor) AwaitProgress(_ context.Context, _ Attempt) ([]string, []string, error) { @@ -117,4 +118,17 @@ func TestRecoverNeverRestartsWithoutCheckpoint(t *testing.T) { require.Empty(t, result.Checkpoint) require.Len(t, store.finished, 1) require.False(t, result.Recovered) + require.Equal(t, []string{"instance-1"}, result.Remaining) +} + +func TestRestartFailureRetainsOriginalSubjects(t *testing.T) { + store := &memoryAttempts{} + executor := &faultExecutor{restart: errors.New("restart timeout")} + at := time.Now().UTC() + result, err := Recover(context.Background(), at, + Decision{Recover: true, Stuck: []string{"instance-1", "instance-2"}}, store, executor, func() time.Time { return at }) + require.ErrorContains(t, err, "restart dispatcher") + require.Equal(t, []string{"instance-1", "instance-2"}, result.Remaining) + require.Equal(t, result.Remaining, store.finished[0].Remaining) + require.False(t, result.Recovered) } diff --git a/internal/schedulerrecovery/store.go b/internal/schedulerrecovery/store.go index f783b00e..bc34053e 100644 --- a/internal/schedulerrecovery/store.go +++ b/internal/schedulerrecovery/store.go @@ -59,7 +59,9 @@ func (store FileStore) ReadHeartbeat(_ context.Context) (Heartbeat, error) { func (store FileStore) Begin(_ context.Context, attempt Attempt) (bool, error) { acquired := false err := store.locked(func(state *fileState) error { - if _, exists := state.Active[attempt.ID]; exists { + // Distinct decisions may race between Active and Begin. Persist only one + // unfinished attempt, not merely one copy of each attempt ID. + if len(state.Active) != 0 { return nil } for _, result := range state.Finished { diff --git a/internal/schedulerrecovery/store_test.go b/internal/schedulerrecovery/store_test.go index 5f13a783..6794ab71 100644 --- a/internal/schedulerrecovery/store_test.go +++ b/internal/schedulerrecovery/store_test.go @@ -4,12 +4,50 @@ import ( "context" "os" "path/filepath" + "sync" "testing" "time" "github.com/stretchr/testify/require" ) +func TestFileStoreSerializesDifferentConcurrentAttempts(t *testing.T) { + t.Parallel() + directory := t.TempDir() + store := FileStore{Path: filepath.Join(directory, "state.json"), LockPath: filepath.Join(directory, "state.lock")} + at := time.Now().UTC() + attempts := []Attempt{NewAttempt(at, []string{"instance-a"}), NewAttempt(at, []string{"instance-b"})} + acquired := make([]bool, len(attempts)) + errs := make([]error, len(attempts)) + var workers sync.WaitGroup + start := make(chan struct{}) + for index := range attempts { + workers.Add(1) + go func() { + defer workers.Done() + <-start + independent := FileStore{Path: store.Path, LockPath: store.LockPath} + acquired[index], errs[index] = independent.Begin(context.Background(), attempts[index]) + }() + } + close(start) + workers.Wait() + require.NoError(t, errs[0]) + require.NoError(t, errs[1]) + require.NotEqual(t, acquired[0], acquired[1], "exactly one different attempt may be active") + active, err := store.Active(context.Background()) + require.NoError(t, err) + require.Len(t, active, 1) + require.NoError(t, store.Finish(context.Background(), Result{AttemptID: active[0].ID, FinishedAt: at, Remaining: active[0].Stuck})) + loser := 0 + if acquired[0] { + loser = 1 + } + ok, err := store.Begin(context.Background(), attempts[loser]) + require.NoError(t, err) + require.True(t, ok, "finishing the active attempt must release admission") +} + func TestFileStorePersistsAttemptAndSuppressesReplay(t *testing.T) { t.Parallel() directory := t.TempDir()