From 5dea312f2b17323a25e616b81324855b4ca939da Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Thu, 1 Oct 2026 00:28:53 +0100 Subject: [PATCH] Bound asynchronous alert checkpoints during bursts Coalesce full active-alert snapshots into one worker and one requested follow-up instead of queueing a goroutine per mutation. Preserve immediate lifecycle durability, fresh snapshots, failure recovery and the final shutdown save; reproduce the old burst amplification and validate the new admission and persistence boundaries. Change-source: pulse-maintainer --- .../v6/internal/subsystems/alerts.md | 28 +++ internal/alerts/active_persistence.go | 78 +++++-- internal/alerts/alerts_test.go | 79 +++++++ internal/alerts/async_checkpoint_test.go | 214 ++++++++++++++++++ internal/alerts/manager.go | 5 + 5 files changed, 385 insertions(+), 19 deletions(-) create mode 100644 internal/alerts/async_checkpoint_test.go diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 8eea5b7709..b9fecc921f 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -3044,6 +3044,34 @@ starting database workers, including restart loading and Unix hardening. The active mirror's unchanged fast path alone does not cover this companion file. This is a narrow write contributor, not aggregate installed-write acceptance. +### Coalesced asynchronous active-alert checkpoints + +`active_persistence.go` admits one asynchronous checkpoint worker per Manager, +not one goroutine and full serialization per mutation. Admission is protected +by the shutdown lock. A worker starts immediately, takes fresh state through +`SaveActiveAlerts`, and retains one follow-up request for mutations arriving +during its current pass. Returning to idle and accepting another request share +the same lock, so requests cannot be stranded at the idle boundary. The worker +never holds the shutdown lock while acquiring persistence or manager-state +locks; callers may request a checkpoint while holding the manager-state lock. + +This coalesces derived snapshots, not lifecycle events or notification work. +Synchronous lifecycle durability, checkpoint revision/failure-epoch checks, +recovery-mirror replacement, intent timing and periodic saves are unchanged. +An error or recovered panic releases admission after any requested follow-up; +there is no automatic busy-loop retry. Once shutdown closes admission, the +worker finishes only its in-flight pass; `Stop` joins it and writes the final +current snapshot synchronously instead of draining redundant queued snapshots. + +`TestAsyncActiveCheckpointBurstIsBounded` in `internal/alerts/alerts_test.go` +blocks the real checkpoint writer, bounds a 512-request burst to one worker, +then verifies current acknowledgement, incident age, metadata and pending intent +through actual JSON persistence and a fresh Manager. The scheduler regressions +in `internal/alerts/async_checkpoint_test.go` pin one follow-up for 10,000 +in-flight requests, idle restart, error/panic recovery, retry after a real file +failure, and shutdown admission/draining. These checks establish bounded +checkpoint work, not the installed CPU/write rate or any reporter's recovery. + ### Warning-level destination preference persistence Email API decoding and encoding, plus webhook edit/create and list presentation, diff --git a/internal/alerts/active_persistence.go b/internal/alerts/active_persistence.go index 6ced54c0d4..d5c0c122a1 100644 --- a/internal/alerts/active_persistence.go +++ b/internal/alerts/active_persistence.go @@ -20,31 +20,71 @@ import ( ) func (m *Manager) saveActiveAlertsAsync(context string) { - m.stopMu.RLock() - if m.stopping { - m.stopMu.RUnlock() - return + if m.queueActiveAlertsSave(context) { + go m.runActiveAlertsSaveWorker(m.SaveActiveAlerts) } - m.workerWG.Add(1) - m.stopMu.RUnlock() +} - go func() { - defer m.workerWG.Done() - defer func() { - if r := recover(); r != nil { +// queueActiveAlertsSave reserves at most one worker and one follow-up snapshot. +// Callers may hold m.mu; never wait for a checkpoint under stopMu. +func (m *Manager) queueActiveAlertsSave(context string) bool { + m.stopMu.Lock() + defer m.stopMu.Unlock() + if m.stopping { + return false + } + m.activeSavePending = true + m.activeSaveContext = context + if m.activeSaveRunning { + return false + } + m.activeSaveRunning = true + // Add under the shutdown admission lock, before Stop can begin waiting. + m.workerWG.Add(1) + return true +} + +// Each pass takes a fresh snapshot. Requests arriving during a pass mark +// one follow-up, not another goroutine/full snapshot per mutation. No timer +// delays alert persistence, and lifecycle events remain independently durable. +func (m *Manager) runActiveAlertsSaveWorker(save func() error) { + defer m.workerWG.Done() + for { + m.stopMu.Lock() + if m.stopping || !m.activeSavePending { + m.activeSaveRunning = false + m.activeSavePending = false + m.activeSaveContext = "" + m.stopMu.Unlock() + // Stop drains this in-flight worker and owns the final synchronous + // checkpoint; do not drain redundant queued snapshots at shutdown. + return + } + context := m.activeSaveContext + m.activeSavePending = false + m.activeSaveContext = "" + m.stopMu.Unlock() + + // Recover per pass so a failed checkpoint cannot strand later requests + // or leave worker admission permanently occupied. Retry only on another + // request (or the existing periodic/shutdown save), never in a busy loop. + func() { + defer func() { + if r := recover(); r != nil { + log.Error(). + Interface("panic", r). + Str("context", context). + Msg("Panic in SaveActiveAlerts goroutine") + } + }() + if err := save(); err != nil { log.Error(). - Interface("panic", r). + Err(err). Str("context", context). - Msg("Panic in SaveActiveAlerts goroutine") + Msg("Failed to save active alerts") } }() - if err := m.SaveActiveAlerts(); err != nil { - log.Error(). - Err(err). - Str("context", context). - Msg("Failed to save active alerts") - } - }() + } } // SaveActiveAlerts persists active alerts to disk. diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index 46bc008b69..fb16bccb97 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -1,6 +1,7 @@ package alerts import ( + "bytes" "encoding/json" "fmt" "math" @@ -8,6 +9,7 @@ import ( "path/filepath" "reflect" "runtime" + "runtime/pprof" "strings" "sync" "testing" @@ -21560,3 +21562,80 @@ func TestIntentCheckpointRetriesFailedWrite(t *testing.T) { t.Fatalf("retry wrote %q", got) } } + +// A slow checkpoint must not turn an alert burst into one blocked worker per +// mutation. Exercise the real asynchronous entry point and real JSON writer. +func TestAsyncActiveCheckpointBurstIsBounded(t *testing.T) { + dir := t.TempDir() + id, alert := testNewCanonicalAlert("checkpoint-host", "metric-threshold:cpu", "host", "cpu") + now := time.Now().UTC() + alert.Level, alert.StartTime, alert.LastSeen = AlertLevelWarning, now.Add(-time.Hour), now + m := &Manager{ + alertsDir: filepath.Join(dir, "alerts"), + activeAlerts: map[string]*Alert{id: alert}, + intentPending: make(map[string]IntentPendingState), + } + m.saveMu.Lock() + locked := true + defer func() { + if locked { + m.saveMu.Unlock() + } + m.workerWG.Wait() + }() + const requests = 512 + for range requests { + m.saveActiveAlertsAsync("burst checkpoint") + } + var stacks bytes.Buffer + if err := pprof.Lookup("goroutine").WriteTo(&stacks, 2); err != nil { + t.Fatal(err) + } + workers := strings.Count(stacks.String(), "created by github.com/rcourtman/pulse-go-rewrite/internal/alerts.(*Manager).saveActiveAlertsAsync") + t.Logf("blocked asynchronous checkpoint workers: %d for %d requests", workers, requests) + if workers != 1 { + t.Errorf("burst started %d checkpoint workers, want one", workers) + } + + // State changed while the first checkpoint was blocked must reach both + // restart files, not the snapshot that existed at request admission. + m.mu.Lock() + alert.Acknowledged, alert.AckUser, alert.AckTime = true, "checkpoint-operator", &now + alert.LastSeen = now.Add(time.Second) + alert.Metadata["checkpoint"] = "latest" + m.intentPending["pending-cpu"] = IntentPendingState{ + TrackingKey: "pending-cpu", ResourceID: "checkpoint-host", Signal: "cpu", + FirstMatchedAt: now.Add(-time.Minute), LastObservedAt: now, ElapsedNanos: int64(time.Minute), + } + m.mu.Unlock() + m.saveActiveAlertsAsync("latest state") + m.saveMu.Unlock() + locked = false + m.workerWG.Wait() + + data, err := os.ReadFile(filepath.Join(m.alertsDir, "active-alerts.json")) + if err != nil { + t.Fatal(err) + } + var persisted []*Alert + if err := json.Unmarshal(data, &persisted); err != nil { + t.Fatal(err) + } + if len(persisted) != 1 || !persisted[0].Acknowledged || persisted[0].AckUser != "checkpoint-operator" || + !persisted[0].StartTime.Equal(alert.StartTime) || !persisted[0].LastSeen.Equal(alert.LastSeen) || + persisted[0].Metadata["checkpoint"] != "latest" { + t.Fatalf("checkpoint lost latest incident state: %s", data) + } + restarted := NewManagerWithDataDir(dir) + t.Cleanup(restarted.Stop) + restored := testRequireActiveAlert(t, restarted, id) + if !restored.Acknowledged || restored.AckUser != "checkpoint-operator" || !restored.StartTime.Equal(alert.StartTime) { + t.Fatalf("restart lost acknowledgement or incident age: %+v", restored) + } + restarted.mu.RLock() + pending := restarted.intentPending["pending-cpu"] + restarted.mu.RUnlock() + if pending.ElapsedNanos != int64(time.Minute) || pending.ResourceID != "checkpoint-host" { + t.Fatalf("restart lost latest pending intent: %+v", pending) + } +} diff --git a/internal/alerts/async_checkpoint_test.go b/internal/alerts/async_checkpoint_test.go new file mode 100644 index 0000000000..1f8d55abe8 --- /dev/null +++ b/internal/alerts/async_checkpoint_test.go @@ -0,0 +1,214 @@ +package alerts + +import ( + "errors" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func awaitCheckpointSignal(t *testing.T, signal <-chan struct{}) { + t.Helper() + select { + case <-signal: + case <-time.After(5 * time.Second): + t.Fatal("checkpoint worker did not reach its barrier") + } +} + +func TestAsyncActiveCheckpointCoalescesInFlightRequests(t *testing.T) { + m := &Manager{} + started, release := make(chan struct{}), make(chan struct{}) + var calls atomic.Int32 + if !m.queueActiveAlertsSave("initial") { + t.Fatal("first request did not start a worker") + } + go m.runActiveAlertsSaveWorker(func() error { + if calls.Add(1) == 1 { + close(started) + <-release + } + return nil + }) + // Always release the worker before test cleanup, even on an assertion failure. + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }); m.workerWG.Wait() }) + awaitCheckpointSignal(t, started) + const requests = 10_000 + for range requests { + if m.queueActiveAlertsSave("during checkpoint") { + t.Fatal("request during a write admitted another worker") + } + } + releaseOnce.Do(func() { close(release) }) + m.workerWG.Wait() + if got := calls.Load(); got != 2 { + t.Fatalf("%d in-flight requests executed %d checkpoints, want exactly two", requests, got) + } + t.Logf("%d requests during one blocked write produced one follow-up checkpoint", requests) + if !m.queueActiveAlertsSave("after idle") { + t.Fatal("worker remained stuck after becoming idle") + } + go m.runActiveAlertsSaveWorker(func() error { calls.Add(1); return nil }) + m.workerWG.Wait() + if got := calls.Load(); got != 3 { + t.Fatalf("new request after idle executed %d total checkpoints, want three", got) + } +} + +func TestAsyncActiveCheckpointFailureDoesNotStrandRequests(t *testing.T) { + for _, failure := range []string{"error", "panic"} { + t.Run(failure, func(t *testing.T) { + m := &Manager{} + started, release := make(chan struct{}), make(chan struct{}) + var calls atomic.Int32 + if !m.queueActiveAlertsSave("initial failure") { + t.Fatal("first request did not start a worker") + } + go m.runActiveAlertsSaveWorker(func() error { + if calls.Add(1) == 1 { + close(started) + <-release + if failure == "panic" { + panic("test checkpoint panic") + } + return errors.New("test checkpoint failure") + } + return nil + }) + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }); m.workerWG.Wait() }) + awaitCheckpointSignal(t, started) + if m.queueActiveAlertsSave("after failure") { + t.Fatal("request during a failing write admitted another worker") + } + releaseOnce.Do(func() { close(release) }) + m.workerWG.Wait() + if got := calls.Load(); got != 2 { + t.Fatalf("failure stranded the follow-up request: %d checkpoints", got) + } + if !m.queueActiveAlertsSave("later retry") { + t.Fatal("failed worker did not release admission") + } + go m.runActiveAlertsSaveWorker(func() error { calls.Add(1); return nil }) + m.workerWG.Wait() + if got := calls.Load(); got != 3 { + t.Fatalf("later retry did not execute: %d checkpoints", got) + } + }) + } +} + +func TestAsyncActiveCheckpointStopOwnsFinalState(t *testing.T) { + m := &Manager{ + alertsDir: t.TempDir(), activeAlerts: make(map[string]*Alert), + escalationStop: make(chan struct{}), cleanupStop: make(chan struct{}), + } + started, release, stopped := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var calls atomic.Int32 + if !m.queueActiveAlertsSave("initial") { + t.Fatal("first request did not start a worker") + } + go m.runActiveAlertsSaveWorker(func() error { + calls.Add(1) + close(started) + <-release + return nil + }) + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }); m.Stop() }) + awaitCheckpointSignal(t, started) + if m.queueActiveAlertsSave("superseded at shutdown") { + t.Fatal("follow-up request admitted another worker") + } + go func() { m.Stop(); close(stopped) }() + awaitCheckpointSignal(t, m.escalationStop) + if m.queueActiveAlertsSave("too late") { + t.Fatal("shutdown admitted a new worker") + } + select { + case <-stopped: + t.Fatal("Stop returned before the in-flight checkpoint finished") + default: + } + m.mu.Lock() + m.activeAlerts["shutdown-final"] = &Alert{ID: "shutdown-final", Acknowledged: true, AckUser: "final-operator"} + m.mu.Unlock() + releaseOnce.Do(func() { close(release) }) + awaitCheckpointSignal(t, stopped) + if got := calls.Load(); got != 1 { + t.Fatalf("shutdown drained %d redundant async checkpoints, want one in-flight", got) + } + data, err := os.ReadFile(filepath.Join(m.alertsDir, "active-alerts.json")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(data), "final-operator") { + t.Fatalf("Stop did not checkpoint the latest state: %s", data) + } +} + +func TestAsyncActiveCheckpointConcurrentRequestsAndStop(t *testing.T) { + m := NewManagerWithDataDir(t.TempDir()) + m.saveMu.Lock() + var unlockOnce sync.Once + t.Cleanup(func() { unlockOnce.Do(m.saveMu.Unlock); m.Stop() }) + m.saveActiveAlertsAsync("initial blocked checkpoint") + start := make(chan struct{}) + entered := make(chan struct{}, 16) + var requesters sync.WaitGroup + for range 16 { + requesters.Add(1) + go func() { + defer requesters.Done() + <-start + m.saveActiveAlertsAsync("before shutdown") + entered <- struct{}{} + for range 256 { + m.saveActiveAlertsAsync("concurrent request") + } + }() + } + close(start) + for range 16 { + <-entered + } + stopped := make(chan struct{}) + go func() { m.Stop(); close(stopped) }() + awaitCheckpointSignal(t, m.escalationStop) + requesters.Wait() + unlockOnce.Do(m.saveMu.Unlock) + awaitCheckpointSignal(t, stopped) + m.stopMu.RLock() + defer m.stopMu.RUnlock() + if m.activeSaveRunning { + t.Fatal("checkpoint worker survived shutdown") + } +} + +func TestAsyncActiveCheckpointWriteFailureCanRetry(t *testing.T) { + dir := filepath.Join(t.TempDir(), "alerts") + if err := os.WriteFile(dir, []byte("block checkpoint directory creation"), 0600); err != nil { + t.Fatal(err) + } + m := &Manager{alertsDir: dir, activeAlerts: make(map[string]*Alert)} + m.saveActiveAlertsAsync("unavailable directory") + m.workerWG.Wait() + data, err := os.ReadFile(dir) + if err != nil || string(data) != "block checkpoint directory creation" { + t.Fatalf("failed save changed the source: %q, %v", data, err) + } + if err := os.Remove(dir); err != nil { + t.Fatal(err) + } + m.saveActiveAlertsAsync("directory restored") + m.workerWG.Wait() + data, err = os.ReadFile(filepath.Join(dir, "active-alerts.json")) + if err != nil || string(data) != "[]" { + t.Fatalf("retry after a failed checkpoint did not persist: %q, %v", data, err) + } +} diff --git a/internal/alerts/manager.go b/internal/alerts/manager.go index 747351f36e..79aa6dc464 100644 --- a/internal/alerts/manager.go +++ b/internal/alerts/manager.go @@ -145,6 +145,11 @@ type Manager struct { stopMu sync.RWMutex stopping bool workerWG sync.WaitGroup + // stopMu protects checkpoint worker admission and its coalesced request. + // The worker releases stopMu before taking saveMu or the manager state lock. + activeSaveRunning bool + activeSavePending bool + activeSaveContext string } type ackRecord struct {