diff --git a/internal/schedulerrecovery/evaluate.go b/internal/schedulerrecovery/evaluate.go index 6c160de..5bcfb11 100644 --- a/internal/schedulerrecovery/evaluate.go +++ b/internal/schedulerrecovery/evaluate.go @@ -15,10 +15,20 @@ type PendingCreate struct { CreateAttempt int `json:"create_attempt"` } +// ProviderRetry identifies a non-terminal provider-create retry that has not +// advanced or cleared after its own next_allowed_at deadline. Unlike the +// dispatcher heartbeat, this is scoped to one exact create path: sibling +// scale sets may continue making progress while this retry is parked. +type ProviderRetry struct { + ID string `json:"id"` + OverdueAge time.Duration `json:"overdue_age_nanoseconds"` +} + type Observation struct { ObservedAt time.Time ActiveIntents int PendingCreates []PendingCreate + OverdueRetries []ProviderRetry ManagerUptime time.Duration LastRecoveryAt time.Time HeartbeatAt time.Time @@ -38,16 +48,26 @@ func Evaluate(policy Policy, observation Observation) Decision { if observation.ActiveIntents == 0 { return Decision{Reason: "no-admitted-demand"} } - stuck := make([]string, 0, len(observation.PendingCreates)) + stuck := make([]string, 0, len(observation.PendingCreates)+len(observation.OverdueRetries)) + overdueRetry := false for _, pending := range observation.PendingCreates { if pending.CreateAttempt == 0 && pending.Age >= policy.MinimumStuckAge { stuck = append(stuck, pending.ID) } } + for _, retry := range observation.OverdueRetries { + if retry.OverdueAge >= policy.MinimumStuckAge { + stuck = append(stuck, retry.ID) + overdueRetry = true + } + } if len(stuck) == 0 { return Decision{Reason: "no-stale-undispatched-instance"} } - if !observation.HeartbeatAt.IsZero() && observation.ObservedAt.Sub(observation.HeartbeatAt) < policy.HeartbeatStale { + // 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. + if !overdueRetry && !observation.HeartbeatAt.IsZero() && observation.ObservedAt.Sub(observation.HeartbeatAt) < policy.HeartbeatStale { return Decision{Reason: "dispatcher-heartbeat-current", Stuck: stuck} } if observation.ManagerUptime < policy.MinimumUptime { @@ -56,5 +76,9 @@ func Evaluate(policy Policy, observation Observation) Decision { if !observation.LastRecoveryAt.IsZero() && observation.ObservedAt.Sub(observation.LastRecoveryAt) < policy.Cooldown { return Decision{Reason: "recovery-cooldown", Stuck: stuck} } - return Decision{Recover: true, Reason: "stale-pending-create-attempt-zero", Stuck: stuck} + reason := "stale-pending-create-attempt-zero" + if overdueRetry { + reason = "stale-provider-retry-past-next-allowed" + } + return Decision{Recover: true, Reason: reason, Stuck: stuck} } diff --git a/internal/schedulerrecovery/evaluate_test.go b/internal/schedulerrecovery/evaluate_test.go index d20e836..b538e41 100644 --- a/internal/schedulerrecovery/evaluate_test.go +++ b/internal/schedulerrecovery/evaluate_test.go @@ -44,3 +44,18 @@ func TestEvaluatePreventsDuplicateRecovery(t *testing.T) { observation.RecoveryRunning = true require.Equal(t, "recovery-already-running", Evaluate(policy, observation).Reason) } + +func TestEvaluateRecoversOverdueProviderRetryDespiteCurrentSiblingHeartbeat(t *testing.T) { + t.Parallel() + now := time.Date(2026, 8, 26, 12, 0, 0, 0, time.UTC) + policy := Policy{MinimumStuckAge: 90 * time.Second, MinimumUptime: 2 * time.Minute, Cooldown: 10 * time.Minute, HeartbeatStale: time.Minute} + observation := Observation{ + ObservedAt: now, ActiveIntents: 9, ManagerUptime: time.Hour, + HeartbeatAt: now.Add(-10 * time.Second), + OverdueRetries: []ProviderRetry{{ID: "scale-set:example:2:job:stuck", OverdueAge: 2 * time.Minute}}, + } + require.Equal(t, Decision{ + Recover: true, Reason: "stale-provider-retry-past-next-allowed", + Stuck: []string{"scale-set:example:2:job:stuck"}, + }, Evaluate(policy, observation)) +} diff --git a/internal/schedulerrecovery/observe.go b/internal/schedulerrecovery/observe.go index 0c87a19..0eaf0ea 100644 --- a/internal/schedulerrecovery/observe.go +++ b/internal/schedulerrecovery/observe.go @@ -20,6 +20,7 @@ type observationOutput struct { ObservedAt time.Time `json:"observed_at"` ActiveIntents int `json:"active_intents"` PendingCreates []PendingCreate `json:"pending_creates"` + OverdueRetries []ProviderRetry `json:"overdue_provider_retries"` ManagerUptimeSeconds int64 `json:"manager_uptime_seconds"` LastRecoveryAt time.Time `json:"last_recovery_at"` RecoveryRunning bool `json:"recovery_running"` @@ -68,9 +69,15 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error return Observation{}, fmt.Errorf("scheduler observation contains an invalid pending create") } } + for _, retry := range decoded.OverdueRetries { + if retry.ID == "" || retry.OverdueAge < 0 { + return Observation{}, fmt.Errorf("scheduler observation contains an invalid overdue provider retry") + } + } return Observation{ ObservedAt: decoded.ObservedAt, ActiveIntents: decoded.ActiveIntents, - PendingCreates: decoded.PendingCreates, ManagerUptime: time.Duration(decoded.ManagerUptimeSeconds) * time.Second, + PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries, + ManagerUptime: time.Duration(decoded.ManagerUptimeSeconds) * time.Second, LastRecoveryAt: decoded.LastRecoveryAt, RecoveryRunning: decoded.RecoveryRunning, }, nil } diff --git a/internal/schedulerrecovery/observe_test.go b/internal/schedulerrecovery/observe_test.go index 30acfbc..1a8dd88 100644 --- a/internal/schedulerrecovery/observe_test.go +++ b/internal/schedulerrecovery/observe_test.go @@ -16,6 +16,7 @@ func TestCommandObserverDecodesStrictSnapshot(t *testing.T) { require.NoError(t, err) observation, err := (CommandObserver{Argv: []string{command, "-test.run=TestSchedulerRecoveryObserverHelper", "--", "valid"}, Timeout: 10 * time.Second}).Observe(context.Background()) require.NoError(t, err) + require.Equal(t, []ProviderRetry{{ID: "retry-1", OverdueAge: 2 * time.Minute}}, observation.OverdueRetries) require.Equal(t, 2, observation.ActiveIntents) require.Equal(t, 10*time.Minute, observation.ManagerUptime) require.Equal(t, 2*time.Minute, observation.PendingCreates[0].Age) @@ -44,7 +45,7 @@ func TestSchedulerRecoveryObserverHelper(t *testing.T) { } switch os.Args[separator+1] { case "valid": - fmt.Print(`{"observed_at":"2026-08-24T10:00:00Z","active_intents":2,"pending_creates":[{"id":"instance-1","age_nanoseconds":120000000000,"create_attempt":0}],"manager_uptime_seconds":600,"last_recovery_at":"0001-01-01T00:00:00Z","recovery_running":false}`) + fmt.Print(`{"observed_at":"2026-08-24T10:00:00Z","active_intents":2,"pending_creates":[{"id":"instance-1","age_nanoseconds":120000000000,"create_attempt":0}],"overdue_provider_retries":[{"id":"retry-1","overdue_age_nanoseconds":120000000000}],"manager_uptime_seconds":600,"last_recovery_at":"0001-01-01T00:00:00Z","recovery_running":false}`) case "unknown": fmt.Print(`{"observed_at":"2026-08-24T10:00:00Z","active_intents":1,"manager_uptime_seconds":1,"unexpected":true}`) case "invalid":