diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index ba53f8f52..f1cbeb1c6 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -298,6 +298,16 @@ unbounded history store. `recentlyResolved` must prune expired entries and cap the newest retained entries on insert as well as during cleanup, so monitor sync and websocket state snapshots remain bounded; durable resolved-alert history belongs in the alert history store, not in this live transition cache. +`recentlyResolved` and `resolvedAlias` have one lock owner: +`Manager.resolvedMutex`. Every access holds that mutex, including alias-repair +lookups, which require the write lock. When an operation needs both manager +state and resolved state, the only permitted nested order is +`Manager.mu` then `Manager.resolvedMutex`; no path may acquire `Manager.mu` +while holding `Manager.resolvedMutex`. Resolved critical sections are limited +to map access and must not dispatch, persist history, invoke callbacks, or +perform notification work. Canonical lifecycle and stateful cooldown refires +consume resolved state through the shared lock-order-aware helper, preserve the +original alert `StartTime`, and keep the five-minute refire/history semantics. The browser thresholds surface is also platform-shaped: Proxmox, Docker, Kubernetes, TrueNAS, vSphere, PBS, PMG, and Systems. It must use the shared FilterBar chip and "+ Filter" pattern for resource filtering, and alert tables diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index e85d7c558..5a78ed218 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -2319,6 +2319,7 @@ "internal/alerts/history_test.go", "internal/alerts/intent_policy_test.go", "internal/alerts/operational_contract_test.go", + "internal/alerts/resolved_lock_discipline_test.go", "internal/alerts/synology_test.go", "internal/alerts/threshold_resolution_shared_test.go", "internal/alerts/update_alerts_test.go", diff --git a/internal/alerts/active_lifecycle.go b/internal/alerts/active_lifecycle.go index e63b551f8..9cb1efdda 100644 --- a/internal/alerts/active_lifecycle.go +++ b/internal/alerts/active_lifecycle.go @@ -59,6 +59,30 @@ func (m *Manager) pruneRecentlyResolvedUnlocked(now time.Time) { } } +// consumeRecentlyResolvedForRefireWithPrimaryLock consumes a resolved alert +// that is still inside the refire window. The caller must hold m.mu. +// +// Lock order is always m.mu -> resolvedMutex. This helper performs only +// resolved-map access while resolvedMutex is held; history, dispatch, and +// notification work remain the caller's responsibility. +func (m *Manager) consumeRecentlyResolvedForRefireWithPrimaryLock(storageKey string, now time.Time) (time.Time, time.Time, bool) { + m.resolvedMutex.Lock() + defer m.resolvedMutex.Unlock() + + resolved, ok := m.getResolvedAlertNoLock(storageKey) + if !ok || resolved == nil || resolved.Alert == nil { + return time.Time{}, time.Time{}, false + } + if !resolved.ResolvedTime.After(now.Add(-recentlyResolvedRetention)) { + return time.Time{}, time.Time{}, false + } + + startTime := resolved.Alert.StartTime + resolvedAt := resolved.ResolvedTime + m.removeResolvedAlertUnlocked(storageKey) + return startTime, resolvedAt, true +} + // addRecentlyResolvedWithPrimaryLock records a resolved alert while preserving the caller's // ownership of m.mu. Callers must hold m.mu before invoking this helper. func (m *Manager) addRecentlyResolvedWithPrimaryLock(resolved *ResolvedAlert) { diff --git a/internal/alerts/canonical_lifecycle.go b/internal/alerts/canonical_lifecycle.go index ed22b9485..c97d829ac 100644 --- a/internal/alerts/canonical_lifecycle.go +++ b/internal/alerts/canonical_lifecycle.go @@ -445,23 +445,11 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert return result, true } - // The resolved maps are resolvedMutex-guarded; take it for the - // lookup/removal only and keep history/dispatch outside the lock. - var reactivatedAt time.Time - reactivated := false - m.resolvedMutex.Lock() - if resolved, ok := m.getResolvedAlertNoLock(storageKey); ok && resolved != nil && resolved.Alert != nil { - if resolved.ResolvedTime.After(time.Now().Add(-5 * time.Minute)) { - if !resolved.Alert.StartTime.IsZero() { - alert.StartTime = resolved.Alert.StartTime - } - m.removeResolvedAlertUnlocked(storageKey) - reactivated = true - reactivatedAt = resolved.ResolvedTime - } - } - m.resolvedMutex.Unlock() + reactivatedStart, reactivatedAt, reactivated := m.consumeRecentlyResolvedForRefireWithPrimaryLock(storageKey, time.Now()) if reactivated { + if !reactivatedStart.IsZero() { + alert.StartTime = reactivatedStart + } if params.AddToHistory { m.historyManager.UpdateAlertLastSeenForAlert(alert, alert.LastSeen) } @@ -618,23 +606,11 @@ func (m *Manager) evaluateCanonicalStatefulAlert(params canonicalStatefulAlertPa } if existing == nil { - // The resolved maps are resolvedMutex-guarded; take it for the - // lookup/removal only and keep history/dispatch outside the lock. - var reactivatedAt time.Time - reactivated := false - m.resolvedMutex.Lock() - if resolved, ok := m.getResolvedAlertNoLock(storageKey); ok && resolved != nil && resolved.Alert != nil { - if resolved.ResolvedTime.After(time.Now().Add(-5 * time.Minute)) { - if !resolved.Alert.StartTime.IsZero() { - alert.StartTime = resolved.Alert.StartTime - } - m.removeResolvedAlertUnlocked(storageKey) - reactivated = true - reactivatedAt = resolved.ResolvedTime - } - } - m.resolvedMutex.Unlock() + reactivatedStart, reactivatedAt, reactivated := m.consumeRecentlyResolvedForRefireWithPrimaryLock(storageKey, time.Now()) if reactivated { + if !reactivatedStart.IsZero() { + alert.StartTime = reactivatedStart + } if params.AddToHistory { m.historyManager.UpdateAlertLastSeenForAlert(alert, alert.LastSeen) } diff --git a/internal/alerts/migration_characterization_test.go b/internal/alerts/migration_characterization_test.go index 4317ee901..d2034310f 100644 --- a/internal/alerts/migration_characterization_test.go +++ b/internal/alerts/migration_characterization_test.go @@ -463,9 +463,7 @@ func TestAlertCharacterizationReevaluatesAlertsWhenConfigChanges(t *testing.T) { assertAlertMissing(t, m, alertID) - m.resolvedMutex.RLock() _, wasResolved := testLookupResolvedAlert(t, m, alertID) - m.resolvedMutex.RUnlock() if !wasResolved { t.Fatalf("expected %q in recently resolved after config change", alertID) } diff --git a/internal/alerts/resolved_lock_discipline_test.go b/internal/alerts/resolved_lock_discipline_test.go index 9d95a5b0f..a862028c3 100644 --- a/internal/alerts/resolved_lock_discipline_test.go +++ b/internal/alerts/resolved_lock_discipline_test.go @@ -2,81 +2,308 @@ package alerts import ( "fmt" + "runtime" "sync" "testing" "time" + + alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rs/zerolog" ) // Regression for issue #1590. The canonical alert evaluation paths used to // read and mutate recentlyResolved/resolvedAlias while holding only m.mu, // while the broadcast and recovery paths guarded them with resolvedMutex, so // the two lock domains did not exclude each other and Go's runtime aborted -// the process with a concurrent map access fault on flapping resources. Run -// with -race: this exercises the cooldown-reactivation lookup pattern against -// the broadcaster and the recovery writer concurrently. -func TestRecentlyResolvedConcurrentAccessIsRaceFree(t *testing.T) { - m := NewManagerWithDataDir(t.TempDir()) +// the process with a concurrent map access fault on flapping resources. +// +// Run with -race. Unlike the original reproducer, this proof drives both +// production canonical evaluators through fire -> recovery -> cooldown refire +// while public broadcast/lookup, alias repair, cleanup, and shutdown paths run +// concurrently. +func TestRecentlyResolvedProductionPathsAreRaceAndDeadlockFree(t *testing.T) { + originalLogLevel := zerolog.GlobalLevel() + zerolog.SetGlobalLevel(zerolog.Disabled) + t.Cleanup(func() { + zerolog.SetGlobalLevel(originalLogLevel) + }) - mk := func(id string) *ResolvedAlert { - return &ResolvedAlert{ - Alert: &Alert{ID: id, ResourceID: id, Type: "connectivity"}, - ResolvedTime: time.Now(), + m := NewManagerWithDataDir(t.TempDir()) + t.Cleanup(m.Stop) + + lifecycleResourceID := "agent:issue-1590-lifecycle" + lifecycleSpec, err := buildCanonicalConnectivitySpec( + lifecycleResourceID, + "Issue 1590 lifecycle", + unifiedresources.ResourceTypeAgent, + AlertLevelWarning, + 1, + false, + ) + if err != nil { + t.Fatal(err) + } + lifecycleTracking := make(map[string]int) + lifecycleParams := canonicalLifecycleAlertParams{ + Spec: lifecycleSpec, + Tracking: lifecycleTracking, + TrackingKey: lifecycleResourceID, + AlertID: "host-offline-issue-1590-lifecycle", + AlertType: "connectivity", + ResourceID: lifecycleResourceID, + ResourceName: "Issue 1590 lifecycle", + Instance: "issue-1590", + Message: "lifecycle test alert", + Metadata: map[string]interface{}{"resourceType": string(unifiedresources.ResourceTypeAgent)}, + AddToHistory: true, + } + lifecycleStateID := canonicalTrackingKeyForSpec(lifecycleSpec, lifecycleParams.AlertID) + + statefulResourceID := "storage:issue-1590-stateful" + statefulSpec, err := buildCanonicalHealthAssessmentSpec( + "issue-1590-health", + statefulResourceID, + "Issue 1590 stateful", + unifiedresources.ResourceTypeStorage, + "pool-health", + []string{"degraded"}, + false, + ) + if err != nil { + t.Fatal(err) + } + statefulParams := canonicalStatefulAlertParams{ + Spec: statefulSpec, + AlertID: "issue-1590-stateful-health", + AlertType: "storage-health", + ResourceID: statefulResourceID, + ResourceName: "Issue 1590 stateful", + Instance: "issue-1590", + Message: "stateful test alert", + Metadata: map[string]interface{}{"resourceType": string(unifiedresources.ResourceTypeStorage)}, + AddToHistory: true, + } + statefulStateID := canonicalTrackingKeyForSpec(statefulSpec, statefulParams.AlertID) + + fireLifecycle := func(observedAt time.Time) alertspecs.AlertState { + lifecycleParams.Evidence = alertspecs.AlertEvidence{ + ObservedAt: observedAt, + Connectivity: &alertspecs.ConnectivityEvidence{ + Signal: "status", + Connected: false, + }, } + result, _ := m.evaluateCanonicalLifecycleAlert(lifecycleParams) + return result.State.State + } + recoverLifecycle := func(observedAt time.Time) alertspecs.AlertState { + lifecycleParams.Evidence = alertspecs.AlertEvidence{ + ObservedAt: observedAt, + Connectivity: &alertspecs.ConnectivityEvidence{ + Signal: "status", + Connected: true, + }, + } + result, _ := m.evaluateCanonicalLifecycleAlert(lifecycleParams) + return result.State.State + } + fireStateful := func(observedAt time.Time) alertspecs.AlertState { + statefulParams.Evidence = alertspecs.AlertEvidence{ + ObservedAt: observedAt, + HealthAssessment: &alertspecs.HealthAssessmentEvidence{ + Signal: "pool-health", + Severity: alertspecs.AlertSeverityWarning, + Codes: []string{"degraded"}, + }, + } + result, _ := m.evaluateCanonicalStatefulAlert(statefulParams) + return result.State.State + } + recoverStateful := func(observedAt time.Time) alertspecs.AlertState { + statefulParams.Evidence = alertspecs.AlertEvidence{ + ObservedAt: observedAt, + HealthAssessment: &alertspecs.HealthAssessmentEvidence{Signal: "pool-health"}, + } + result, _ := m.evaluateCanonicalStatefulAlert(statefulParams) + return result.State.State } - stop := make(chan struct{}) + base := time.Now().Add(-time.Minute) + if got := fireLifecycle(base); got != alertspecs.AlertStateFiring { + t.Fatalf("initial lifecycle state = %q, want firing", got) + } + lifecycleStart := testRequireActiveAlert(t, m, lifecycleStateID).StartTime + if got := fireStateful(base); got != alertspecs.AlertStateFiring { + t.Fatalf("initial stateful state = %q, want firing", got) + } + statefulStart := testRequireActiveAlert(t, m, statefulStateID).StartTime + if got := recoverLifecycle(base.Add(time.Millisecond)); got != alertspecs.AlertStateClear { + t.Fatalf("initial lifecycle recovery state = %q, want clear", got) + } + if got := recoverStateful(base.Add(time.Millisecond)); got != alertspecs.AlertStateClear { + t.Fatalf("initial stateful recovery state = %q, want clear", got) + } + + // Force the canonical-alias fallback path. GetResolvedAlert repairs this + // alias under the resolved write lock while broadcasts iterate the map. + aliasStateID := buildCanonicalStateID("agent:issue-1590-alias", "issue-1590-alias") + m.resolvedMutex.Lock() + m.recentlyResolved["issue-1590-legacy-storage-key"] = &ResolvedAlert{ + Alert: &Alert{ + ID: "issue-1590-legacy-alert", + ResourceID: "agent:issue-1590-alias", + CanonicalSpecID: "issue-1590-alias", + CanonicalState: aliasStateID, + StartTime: base, + }, + ResolvedTime: time.Now(), + } + m.resolvedMutex.Unlock() + + start := make(chan struct{}) + errs := make(chan error, 4) var wg sync.WaitGroup - wg.Add(3) + wg.Add(4) - // Recovery path: writes resolved alerts under resolvedMutex. + // Canonical lifecycle recovery and cooldown refire path. go func() { defer wg.Done() - for i := 0; ; i++ { - select { - case <-stop: + <-start + for i := 0; i < 100; i++ { + observedAt := base.Add(time.Duration(2*i+2) * time.Millisecond) + if got := fireLifecycle(observedAt); got != alertspecs.AlertStateFiring { + errs <- fmt.Errorf("lifecycle iteration %d fire state = %q", i, got) return - default: } - m.addRecentlyResolvedUnlocked(mk(fmt.Sprintf("res-%d", i%8))) + if got := recoverLifecycle(observedAt.Add(time.Microsecond)); got != alertspecs.AlertStateClear { + errs <- fmt.Errorf("lifecycle iteration %d recovery state = %q", i, got) + return + } + runtime.Gosched() } }() - // Broadcast path: iterates the resolved map every poll cycle. + // Canonical stateful recovery and cooldown refire path. go func() { defer wg.Done() - for { - select { - case <-stop: + <-start + for i := 0; i < 100; i++ { + observedAt := base.Add(time.Duration(2*i+2) * time.Millisecond) + if got := fireStateful(observedAt); got != alertspecs.AlertStateFiring { + errs <- fmt.Errorf("stateful iteration %d fire state = %q", i, got) return - default: } + if got := recoverStateful(observedAt.Add(time.Microsecond)); got != alertspecs.AlertStateClear { + errs <- fmt.Errorf("stateful iteration %d recovery state = %q", i, got) + return + } + runtime.Gosched() + } + }() + + // Production broadcast and point-lookup paths. + go func() { + defer wg.Done() + <-start + for i := 0; i < 800; i++ { m.GetRecentlyResolved() - m.GetResolvedAlert("res-1") + m.GetResolvedAlert(aliasStateID) + m.GetResolvedAlert(lifecycleStateID) + m.GetResolvedAlert(statefulStateID) + runtime.Gosched() } }() - // Canonical eval path: cooldown lookup and removal while holding m.mu, - // with the resolved maps guarded by the subordinate resolvedMutex. + // Cleanup is the other production path that holds m.mu before acquiring + // resolvedMutex. go func() { defer wg.Done() - for i := 0; ; i++ { - select { - case <-stop: - return - default: - } - key := fmt.Sprintf("res-%d", i%8) - m.mu.Lock() - m.resolvedMutex.Lock() - if resolved, ok := m.getResolvedAlertNoLock(key); ok && resolved != nil { - m.removeResolvedAlertUnlocked(key) - } - m.resolvedMutex.Unlock() - m.mu.Unlock() + <-start + for i := 0; i < 100; i++ { + m.Cleanup(time.Hour) + runtime.Gosched() } }() - time.Sleep(500 * time.Millisecond) - close(stop) - wg.Wait() + close(start) + waitForResolvedConcurrencyGroup(t, &wg) + close(errs) + for err := range errs { + t.Error(err) + } + if t.Failed() { + return + } + + finalObservedAt := base.Add(2 * time.Second) + if got := fireLifecycle(finalObservedAt); got != alertspecs.AlertStateFiring { + t.Fatalf("final lifecycle state = %q, want firing", got) + } + if got := fireStateful(finalObservedAt); got != alertspecs.AlertStateFiring { + t.Fatalf("final stateful state = %q, want firing", got) + } + if got := testRequireActiveAlert(t, m, lifecycleStateID).StartTime; !got.Equal(lifecycleStart) { + t.Fatalf("lifecycle refire start = %v, want original %v", got, lifecycleStart) + } + if got := testRequireActiveAlert(t, m, statefulStateID).StartTime; !got.Equal(statefulStart) { + t.Fatalf("stateful refire start = %v, want original %v", got, statefulStart) + } + if got := len(m.historyManager.GetAllHistory(1000)); got != 2 { + t.Fatalf("history entries after repeated cooldown refires = %d, want 2", got) + } + if got := m.GetResolvedAlert(lifecycleStateID); got != nil { + t.Fatal("lifecycle resolved entry survived cooldown refire") + } + if got := m.GetResolvedAlert(statefulStateID); got != nil { + t.Fatal("stateful resolved entry survived cooldown refire") + } + if got := m.GetResolvedAlert(aliasStateID); got == nil || got.Alert == nil { + t.Fatal("canonical alias lookup did not survive concurrent repair and broadcasts") + } + + // Shutdown must not introduce a reverse-order wait against concurrent + // resolved reads or cleanup. + var shutdownWG sync.WaitGroup + shutdownWG.Add(6) + for i := 0; i < 2; i++ { + go func() { + defer shutdownWG.Done() + for j := 0; j < 100; j++ { + m.GetRecentlyResolved() + m.GetResolvedAlert(aliasStateID) + } + }() + } + for i := 0; i < 2; i++ { + go func() { + defer shutdownWG.Done() + for j := 0; j < 30; j++ { + m.Cleanup(time.Hour) + } + }() + } + for i := 0; i < 2; i++ { + go func() { + defer shutdownWG.Done() + m.Stop() + }() + } + waitForResolvedConcurrencyGroup(t, &shutdownWG) +} + +func waitForResolvedConcurrencyGroup(t *testing.T, wg *sync.WaitGroup) { + t.Helper() + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(30 * time.Second): + t.Fatal("resolved alert concurrency paths deadlocked") + } } diff --git a/internal/alerts/test_helpers_test.go b/internal/alerts/test_helpers_test.go index 19e3b31ab..3ebd75ee8 100644 --- a/internal/alerts/test_helpers_test.go +++ b/internal/alerts/test_helpers_test.go @@ -164,8 +164,10 @@ func testHasActiveAlert(t testing.TB, m *Manager, alertID string) bool { func testLookupResolvedAlert(t testing.TB, m *Manager, alertID string) (*ResolvedAlert, bool) { t.Helper() - m.resolvedMutex.RLock() - defer m.resolvedMutex.RUnlock() + // getResolvedAlertNoLock can repair a missing canonical alias, so lookups + // require the write lock even when the resolved alert itself is read-only. + m.resolvedMutex.Lock() + defer m.resolvedMutex.Unlock() if resolved, exists := m.getResolvedAlertNoLock(alertID); exists { return resolved, true