From 9af80adb9c4c2f22da3e492ff98ef064a8bc6381 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Tue, 8 Sep 2026 11:30:09 +0500 Subject: [PATCH] fix(recovery): bound overlapping incident retries to original plus two A new attempt hash after cooldown restarted the manager against the same stuck subjects. Count unfinished overlapping recoveries in durable history, keep cooldown and busy unhealthy, and refuse a fourth attempt without calling the restart. Signed-off-by: rldyourmnd Co-authored-by: Cursor --- internal/schedulerrecovery/controller.go | 10 +++ internal/schedulerrecovery/controller_test.go | 61 +++++++++++++++++++ .../progress_contract_test.go | 3 + internal/schedulerrecovery/recover.go | 56 +++++++++++++++++ internal/schedulerrecovery/recover_test.go | 6 ++ internal/schedulerrecovery/store.go | 9 +++ internal/schedulerrecovery/store_test.go | 4 ++ 7 files changed, 149 insertions(+) diff --git a/internal/schedulerrecovery/controller.go b/internal/schedulerrecovery/controller.go index 07839ef9..18cc61cb 100644 --- a/internal/schedulerrecovery/controller.go +++ b/internal/schedulerrecovery/controller.go @@ -74,6 +74,16 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error) } observation.HeartbeatAt = heartbeat.At decision := Evaluate(controller.Policy, observation) + if decision.Recover { + history, historyErr := controller.Attempts.History(ctx) + if historyErr != nil { + return decision, Result{}, fmt.Errorf("read recovery history: %w", historyErr) + } + if incidentRecoveryAttempts(history, decision.Stuck) >= maxIncidentRecoveryAttempts { + decision.Recover = false + decision.Reason = "recovery-retry-budget-exhausted" + } + } state := "healthy" if decision.Recover || len(decision.Stuck) > 0 { state = "unhealthy" diff --git a/internal/schedulerrecovery/controller_test.go b/internal/schedulerrecovery/controller_test.go index 77e063dd..18b419af 100644 --- a/internal/schedulerrecovery/controller_test.go +++ b/internal/schedulerrecovery/controller_test.go @@ -123,3 +123,64 @@ func TestControllerFinishesInterruptedRecoveryAfterRestartProgressed(t *testing. require.Empty(t, store.active) require.Equal(t, []string{"recovering", "recovered"}, []string{events.events[0].State, events.events[1].State}) } + +func TestControllerBoundsInfrastructureRetriesAcrossAttemptIDs(t *testing.T) { + t.Parallel() + at := time.Date(2026, 9, 8, 2, 35, 16, 0, time.UTC) + store := &memoryAttempts{} + executor := &faultExecutor{remaining: []string{"intent-a", "intent-b"}} + events := &eventRecorder{} + now := at + controller := Controller{ + Policy: Policy{MinimumStuckAge: time.Minute, MinimumUptime: time.Minute, Cooldown: time.Minute, HeartbeatStale: time.Minute}, + Observer: staticObserver{Observation{ + ObservedAt: at, ActiveIntents: 2, ManagerUptime: time.Hour, + StaleAssigned: []AssignedIntent{{ID: "intent-a", Age: time.Hour}, {ID: "intent-b", Age: time.Hour}}, + }}, + Heartbeat: staticHeartbeat{}, Attempts: store, Executor: executor, Events: events, + Now: func() time.Time { return now }, + } + for attempt := 0; attempt < maxIncidentRecoveryAttempts; attempt++ { + now = at.Add(time.Duration(attempt) * time.Minute) + observation := controller.Observer.(staticObserver).observation + observation.ObservedAt = now + controller.Observer = staticObserver{observation} + decision, result, err := controller.Tick(context.Background()) + require.Error(t, err) + require.True(t, decision.Recover) + require.False(t, result.Recovered) + require.Equal(t, attempt+1, executor.restarts) + } + now = at.Add(10 * time.Minute) + observation := controller.Observer.(staticObserver).observation + observation.ObservedAt = now + observation.StaleAssigned = []AssignedIntent{ + {ID: "intent-a", Age: time.Hour}, + {ID: "intent-b", Age: time.Hour}, + {ID: "intent-c", Age: time.Hour}, + } + controller.Observer = staticObserver{observation} + events.events = nil + decision, result, err := controller.Tick(context.Background()) + require.NoError(t, err) + require.False(t, decision.Recover) + require.Equal(t, "recovery-retry-budget-exhausted", decision.Reason) + require.False(t, result.Recovered) + require.Equal(t, maxIncidentRecoveryAttempts, executor.restarts) + require.Equal(t, "unhealthy", events.events[0].State) + require.Equal(t, []string{"intent-a", "intent-b", "intent-c"}, decision.Stuck) +} + +func TestIncidentRecoveryAttemptsResetAfterSuccessAndIgnoreSuppressed(t *testing.T) { + t.Parallel() + stuck := []string{"intent-a"} + history := []Result{ + {AttemptID: "one", Remaining: []string{"intent-a"}, Error: "incomplete"}, + {AttemptID: "two", Remaining: []string{"intent-a"}, Error: "incomplete"}, + {AttemptID: "dup", Remaining: []string{"intent-a"}, Suppressed: true}, + {AttemptID: "ok", Progressed: []string{"intent-a"}, Recovered: true}, + {AttemptID: "later", Remaining: []string{"intent-a", "intent-z"}, Error: "incomplete"}, + } + require.Equal(t, 1, incidentRecoveryAttempts(history, stuck)) + require.Equal(t, 0, incidentRecoveryAttempts(history, []string{"unrelated"})) +} diff --git a/internal/schedulerrecovery/progress_contract_test.go b/internal/schedulerrecovery/progress_contract_test.go index e86cb445..c6b8cfcb 100644 --- a/internal/schedulerrecovery/progress_contract_test.go +++ b/internal/schedulerrecovery/progress_contract_test.go @@ -67,6 +67,9 @@ func (s *evidenceStore) Finish(_ context.Context, r Result) error { s.results = append(s.results, r) return nil } +func (s *evidenceStore) History(context.Context) ([]Result, error) { + return append([]Result(nil), s.results...), nil +} type evidenceExecutor struct { progressed, remaining []string diff --git a/internal/schedulerrecovery/recover.go b/internal/schedulerrecovery/recover.go index aa72025a..f1b1b081 100644 --- a/internal/schedulerrecovery/recover.go +++ b/internal/schedulerrecovery/recover.go @@ -84,10 +84,59 @@ func (result *Result) UnmarshalJSON(data []byte) error { return nil } +const ( + originalRecoveryAttempt = 1 + maxInfrastructureRetries = 2 + maxIncidentRecoveryAttempts = originalRecoveryAttempt + maxInfrastructureRetries +) + type AttemptStore interface { Active(context.Context) ([]Attempt, error) Begin(context.Context, Attempt) (bool, error) Finish(context.Context, Result) error + History(context.Context) ([]Result, error) +} + +func subjectSetsOverlap(left []string, right map[string]struct{}) bool { + for _, id := range left { + if _, exists := right[id]; exists { + return true + } + } + return false +} + +func incidentSubjects(result Result) []string { + subjects := make([]string, 0, len(result.Progressed)+len(result.Remaining)) + subjects = append(subjects, result.Progressed...) + subjects = append(subjects, result.Remaining...) + return subjects +} + +// incidentRecoveryAttempts counts unfinished recoveries that overlap the +// current stuck set. A later successful overlapping recovery resets the count. +// The attempt ID, observation time and the lexicographic max subject may change +// without starting a new incident. +func incidentRecoveryAttempts(history []Result, stuck []string) int { + if len(stuck) == 0 { + return 0 + } + current := make(map[string]struct{}, len(stuck)) + for _, id := range stuck { + current[id] = struct{}{} + } + count := 0 + for _, result := range history { + if result.Suppressed || !subjectSetsOverlap(incidentSubjects(result), current) { + continue + } + if result.Recovered { + count = 0 + continue + } + count++ + } + return count } func resumeAcquired(ctx context.Context, attempt Attempt, store AttemptStore, executor Executor, now func() time.Time) (Result, error) { @@ -147,6 +196,13 @@ func Recover(ctx context.Context, observedAt time.Time, decision Decision, store if err := validateProgress(decision.Stuck, nil, decision.Stuck); err != nil { return Result{}, fmt.Errorf("invalid recovery identities: %w", err) } + history, err := store.History(ctx) + if err != nil { + return Result{}, fmt.Errorf("read recovery history: %w", err) + } + if incidentRecoveryAttempts(history, decision.Stuck) >= maxIncidentRecoveryAttempts { + return Result{}, fmt.Errorf("recovery refused: recovery-retry-budget-exhausted") + } attempt := NewAttempt(observedAt, decision.Stuck) acquired, err := store.Begin(ctx, attempt) if err != nil { diff --git a/internal/schedulerrecovery/recover_test.go b/internal/schedulerrecovery/recover_test.go index b915d649..36e14d66 100644 --- a/internal/schedulerrecovery/recover_test.go +++ b/internal/schedulerrecovery/recover_test.go @@ -52,6 +52,12 @@ func (store *memoryAttempts) Active(_ context.Context) ([]Attempt, error) { return attempts, nil } +func (store *memoryAttempts) History(_ context.Context) ([]Result, error) { + store.mu.Lock() + defer store.mu.Unlock() + return append([]Result(nil), store.finished...), nil +} + type faultExecutor struct { mu sync.Mutex checkpoints int diff --git a/internal/schedulerrecovery/store.go b/internal/schedulerrecovery/store.go index bc34053e..24d89140 100644 --- a/internal/schedulerrecovery/store.go +++ b/internal/schedulerrecovery/store.go @@ -105,6 +105,15 @@ func (store FileStore) Finish(_ context.Context, result Result) error { }) } +func (store FileStore) History(_ context.Context) ([]Result, error) { + var history []Result + err := store.locked(func(state *fileState) error { + history = slices.Clone(state.Finished) + return nil + }) + return history, err +} + func (store FileStore) locked(update func(*fileState) error) error { if store.Path == "" || store.LockPath == "" { return fmt.Errorf("state and lock paths are required") diff --git a/internal/schedulerrecovery/store_test.go b/internal/schedulerrecovery/store_test.go index 6794ab71..37faea04 100644 --- a/internal/schedulerrecovery/store_test.go +++ b/internal/schedulerrecovery/store_test.go @@ -39,6 +39,10 @@ func TestFileStoreSerializesDifferentConcurrentAttempts(t *testing.T) { 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})) + history, err := store.History(context.Background()) + require.NoError(t, err) + require.Len(t, history, 1) + require.Equal(t, active[0].Stuck, history[0].Remaining) loser := 0 if acquired[0] { loser = 1