From 08be51488687b9a2511c2efaecd66a24bebeee19 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Wed, 26 Aug 2026 20:18:56 +0500 Subject: [PATCH 1/2] fix(recovery): detect undispatched assigned intents --- internal/schedulerrecovery/evaluate.go | 24 +++++++++++++++++++-- internal/schedulerrecovery/evaluate_test.go | 14 ++++++++++++ internal/schedulerrecovery/observe.go | 22 ++++++++++++------- internal/schedulerrecovery/observe_test.go | 3 ++- 4 files changed, 52 insertions(+), 11 deletions(-) diff --git a/internal/schedulerrecovery/evaluate.go b/internal/schedulerrecovery/evaluate.go index 5bcfb11..e5d426c 100644 --- a/internal/schedulerrecovery/evaluate.go +++ b/internal/schedulerrecovery/evaluate.go @@ -24,11 +24,21 @@ type ProviderRetry struct { OverdueAge time.Duration `json:"overdue_age_nanoseconds"` } +// AssignedIntent identifies central queue ownership that never produced a +// corresponding GARM instance. Sibling scale sets can keep both the process +// heartbeat and provider retry journal moving while this exact dispatch path +// is no longer scheduled. +type AssignedIntent struct { + ID string `json:"id"` + Age time.Duration `json:"age_nanoseconds"` +} + type Observation struct { ObservedAt time.Time ActiveIntents int PendingCreates []PendingCreate OverdueRetries []ProviderRetry + StaleAssigned []AssignedIntent ManagerUptime time.Duration LastRecoveryAt time.Time HeartbeatAt time.Time @@ -48,8 +58,9 @@ func Evaluate(policy Policy, observation Observation) Decision { if observation.ActiveIntents == 0 { return Decision{Reason: "no-admitted-demand"} } - stuck := make([]string, 0, len(observation.PendingCreates)+len(observation.OverdueRetries)) + stuck := make([]string, 0, len(observation.PendingCreates)+len(observation.OverdueRetries)+len(observation.StaleAssigned)) overdueRetry := false + staleAssigned := false for _, pending := range observation.PendingCreates { if pending.CreateAttempt == 0 && pending.Age >= policy.MinimumStuckAge { stuck = append(stuck, pending.ID) @@ -61,13 +72,19 @@ func Evaluate(policy Policy, observation Observation) Decision { overdueRetry = true } } + for _, assigned := range observation.StaleAssigned { + if assigned.Age >= policy.MinimumStuckAge { + stuck = append(stuck, assigned.ID) + staleAssigned = true + } + } if len(stuck) == 0 { return Decision{Reason: "no-stale-undispatched-instance"} } // 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 { + if !overdueRetry && !staleAssigned && !observation.HeartbeatAt.IsZero() && observation.ObservedAt.Sub(observation.HeartbeatAt) < policy.HeartbeatStale { return Decision{Reason: "dispatcher-heartbeat-current", Stuck: stuck} } if observation.ManagerUptime < policy.MinimumUptime { @@ -80,5 +97,8 @@ func Evaluate(policy Policy, observation Observation) Decision { if overdueRetry { reason = "stale-provider-retry-past-next-allowed" } + if staleAssigned { + reason = "stale-assigned-intent-without-instance" + } return Decision{Recover: true, Reason: reason, Stuck: stuck} } diff --git a/internal/schedulerrecovery/evaluate_test.go b/internal/schedulerrecovery/evaluate_test.go index b538e41..ca15862 100644 --- a/internal/schedulerrecovery/evaluate_test.go +++ b/internal/schedulerrecovery/evaluate_test.go @@ -59,3 +59,17 @@ func TestEvaluateRecoversOverdueProviderRetryDespiteCurrentSiblingHeartbeat(t *t Stuck: []string{"scale-set:example:2:job:stuck"}, }, Evaluate(policy, observation)) } + +func TestEvaluateRecoversStaleAssignedIntentDespiteCurrentSiblingHeartbeat(t *testing.T) { + t.Parallel() + now := time.Date(2026, 8, 26, 13, 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: 8, ManagerUptime: time.Hour, + HeartbeatAt: now.Add(-10 * time.Second), + StaleAssigned: []AssignedIntent{{ID: "intent-skipped", Age: 2 * time.Minute}}, + } + require.Equal(t, Decision{ + Recover: true, Reason: "stale-assigned-intent-without-instance", Stuck: []string{"intent-skipped"}, + }, Evaluate(policy, observation)) +} diff --git a/internal/schedulerrecovery/observe.go b/internal/schedulerrecovery/observe.go index 0eaf0ea..304c124 100644 --- a/internal/schedulerrecovery/observe.go +++ b/internal/schedulerrecovery/observe.go @@ -17,13 +17,14 @@ type CommandObserver struct { } 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"` + ObservedAt time.Time `json:"observed_at"` + ActiveIntents int `json:"active_intents"` + PendingCreates []PendingCreate `json:"pending_creates"` + OverdueRetries []ProviderRetry `json:"overdue_provider_retries"` + StaleAssigned []AssignedIntent `json:"stale_assigned_intents"` + ManagerUptimeSeconds int64 `json:"manager_uptime_seconds"` + LastRecoveryAt time.Time `json:"last_recovery_at"` + RecoveryRunning bool `json:"recovery_running"` } func (observer CommandObserver) Validate() error { @@ -74,9 +75,14 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error return Observation{}, fmt.Errorf("scheduler observation contains an invalid overdue provider retry") } } + for _, assigned := range decoded.StaleAssigned { + if assigned.ID == "" || assigned.Age < 0 { + return Observation{}, fmt.Errorf("scheduler observation contains an invalid stale assigned intent") + } + } return Observation{ ObservedAt: decoded.ObservedAt, ActiveIntents: decoded.ActiveIntents, - PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries, + PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries, StaleAssigned: decoded.StaleAssigned, 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 1a8dd88..f66e868 100644 --- a/internal/schedulerrecovery/observe_test.go +++ b/internal/schedulerrecovery/observe_test.go @@ -17,6 +17,7 @@ func TestCommandObserverDecodesStrictSnapshot(t *testing.T) { 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, []AssignedIntent{{ID: "assigned-1", Age: 2 * time.Minute}}, observation.StaleAssigned) require.Equal(t, 2, observation.ActiveIntents) require.Equal(t, 10*time.Minute, observation.ManagerUptime) require.Equal(t, 2*time.Minute, observation.PendingCreates[0].Age) @@ -45,7 +46,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}],"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}`) + 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}],"stale_assigned_intents":[{"id":"assigned-1","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": From 1ef28e4c6aef72b321f073327812d97d8b900f8b Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Wed, 26 Aug 2026 21:04:06 +0500 Subject: [PATCH 2/2] fix(recovery): normalize result wire format --- internal/schedulerrecovery/recover.go | 57 ++++++++++++++++++++++++ internal/schedulerrecovery/store_test.go | 15 +++++++ 2 files changed, 72 insertions(+) diff --git a/internal/schedulerrecovery/recover.go b/internal/schedulerrecovery/recover.go index 7aca415..f5e8a3e 100644 --- a/internal/schedulerrecovery/recover.go +++ b/internal/schedulerrecovery/recover.go @@ -4,6 +4,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "encoding/json" "fmt" "slices" "strings" @@ -27,6 +28,62 @@ type Result struct { Error string } +type resultWire struct { + AttemptID string `json:"attempt_id"` + Suppressed bool `json:"suppressed"` + Checkpoint string `json:"checkpoint,omitempty"` + Progressed []string `json:"progressed,omitempty"` + Remaining []string `json:"remaining,omitempty"` + Recovered bool `json:"recovered"` + FinishedAt time.Time `json:"finished_at"` + Error string `json:"error,omitempty"` +} + +func (result Result) MarshalJSON() ([]byte, error) { + return json.Marshal(resultWire{ + AttemptID: result.AttemptID, Suppressed: result.Suppressed, Checkpoint: result.Checkpoint, + Progressed: result.Progressed, Remaining: result.Remaining, Recovered: result.Recovered, + FinishedAt: result.FinishedAt, Error: result.Error, + }) +} + +func (result *Result) UnmarshalJSON(data []byte) error { + var normalized resultWire + if err := json.Unmarshal(data, &normalized); err != nil { + return err + } + if normalized.AttemptID != "" || !normalized.FinishedAt.IsZero() { + *result = Result{ + AttemptID: normalized.AttemptID, Suppressed: normalized.Suppressed, Checkpoint: normalized.Checkpoint, + Progressed: normalized.Progressed, Remaining: normalized.Remaining, Recovered: normalized.Recovered, + FinishedAt: normalized.FinishedAt, Error: normalized.Error, + } + return nil + } + // v1 was deployed before an explicit JSON contract and therefore used Go + // field names. Decode it once; the next atomic state write normalizes every + // retained result without losing cooldown or recovery history. + var legacy struct { + AttemptID string + Suppressed bool + Checkpoint string + Progressed []string + Remaining []string + Recovered bool + FinishedAt time.Time + Error string + } + if err := json.Unmarshal(data, &legacy); err != nil { + return err + } + *result = Result{ + AttemptID: legacy.AttemptID, Suppressed: legacy.Suppressed, Checkpoint: legacy.Checkpoint, + Progressed: legacy.Progressed, Remaining: legacy.Remaining, Recovered: legacy.Recovered, + FinishedAt: legacy.FinishedAt, Error: legacy.Error, + } + return nil +} + type AttemptStore interface { Active(context.Context) ([]Attempt, error) Begin(context.Context, Attempt) (bool, error) diff --git a/internal/schedulerrecovery/store_test.go b/internal/schedulerrecovery/store_test.go index a3712fe..5f13a78 100644 --- a/internal/schedulerrecovery/store_test.go +++ b/internal/schedulerrecovery/store_test.go @@ -68,3 +68,18 @@ func TestFileStorePersistsMonotonicHeartbeat(t *testing.T) { require.Equal(t, Heartbeat{At: at.Add(time.Minute), Progress: "job-2"}, heartbeat) require.ErrorContains(t, reopened.RecordHeartbeat(context.Background(), Heartbeat{At: at.Add(-time.Second), Progress: "job-0"}), "moved backwards") } + +func TestFileStoreReadsLegacyResultAndNormalizesNextWrite(t *testing.T) { + t.Parallel() + directory := t.TempDir() + path := filepath.Join(directory, "state.json") + legacy := `{"schema_version":1,"heartbeat":{},"active":{},"finished":[{"AttemptID":"attempt-1","Recovered":false,"FinishedAt":"2026-08-26T15:38:36Z","Error":"incomplete"}]}` + require.NoError(t, os.WriteFile(path, []byte(legacy), 0o600)) + store := FileStore{Path: path, LockPath: filepath.Join(directory, "state.lock")} + _, err := store.ReadHeartbeat(context.Background()) + require.NoError(t, err) + data, err := os.ReadFile(path) + require.NoError(t, err) + require.Contains(t, string(data), `"finished_at": "2026-08-26T15:38:36Z"`) + require.NotContains(t, string(data), `"FinishedAt"`) +}