Integrate reviewed bounded alert checkpoint worker

Change-source: pulse-maintainer

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,

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.

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)
+	}
+}

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)
+	}
+}

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 {