From b14cb376bc0140907e47e4bcd19b6841416fd556 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Thu, 23 Jul 2026 21:56:38 +0100 Subject: [PATCH] Prove resolved alert lock discipline Route canonical cooldown refires through one resolved-state critical section and document the m.mu to resolvedMutex invariant. Exercise production refire, recovery, broadcast, alias repair, cleanup, and shutdown paths under the race detector while preserving cooldown history and start-time semantics. --- .../v6/internal/subsystems/alerts.md | 10 + .../v6/internal/subsystems/registry.json | 1 + internal/alerts/active_lifecycle.go | 24 ++ internal/alerts/canonical_lifecycle.go | 40 +-- .../alerts/migration_characterization_test.go | 2 - .../alerts/resolved_lock_discipline_test.go | 311 +++++++++++++++--- internal/alerts/test_helpers_test.go | 6 +- 7 files changed, 316 insertions(+), 78 deletions(-) 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