Coordinate incident recorder async persistence

internal/metrics/incident_recorder.go

@@ -118,6 +118,7 @@ type IncidentRecorder struct {
 	// Persistence
 	dataDir  string
 	filePath string
+	saveIOMu sync.Mutex
 
 	// Control
 	stopCh   chan struct{}
@@ -225,6 +226,7 @@ func (r *IncidentRecorder) Stop() {
 	r.mu.Lock()
 	if !r.running {
 		r.mu.Unlock()
+		r.waitForPendingSaves()
 		return
 	}
 	r.running = false
@@ -476,17 +478,8 @@ func (r *IncidentRecorder) completeWindowLocked(window *IncidentWindow) {
 		Int("data_points", len(window.DataPoints)).
 		Msg("completed incident recording")
 
-	// Save asynchronously
-	go func() {
-		if err := r.saveToDisk(); err != nil {
-			log.Warn().
-				Str("window_id", window.ID).
-				Str("resource_id", window.ResourceID).
-				Str("file_path", r.filePath).
-				Err(err).
-				Msg("Failed to save incident windows")
-		}
-	}()
+	// Save asynchronously.
+	r.requestAsyncSave()
 }
 
 // computeSummary computes statistics for a window
@@ -629,6 +622,8 @@ func (r *IncidentRecorder) saveToDisk() error {
 	if r.filePath == "" {
 		return nil
 	}
+	r.saveIOMu.Lock()
+	defer r.saveIOMu.Unlock()
 
 	data := struct {
 		CompletedWindows []*IncidentWindow `json:"completed_windows"`
@@ -683,6 +678,44 @@ func (r *IncidentRecorder) saveToDisk() error {
 	return os.Chmod(r.filePath, incidentRecorderFilePerm)
 }
 
+func (r *IncidentRecorder) requestAsyncSave() {
+	if r.filePath == "" {
+		return
+	}
+
+	r.saveMu.Lock()
+	if r.saveInProgress {
+		r.saveRequested = true
+		r.saveMu.Unlock()
+		return
+	}
+	r.saveInProgress = true
+	r.saveMu.Unlock()
+
+	go r.runAsyncSaves()
+}
+
+func (r *IncidentRecorder) runAsyncSaves() {
+	for {
+		if err := r.saveToDisk(); err != nil {
+			log.Warn().
+				Str("file_path", r.filePath).
+				Err(err).
+				Msg("Failed to save incident windows")
+		}
+
+		r.saveMu.Lock()
+		if !r.saveRequested {
+			r.saveInProgress = false
+			r.saveCond.Broadcast()
+			r.saveMu.Unlock()
+			return
+		}
+		r.saveRequested = false
+		r.saveMu.Unlock()
+	}
+}
+
 func (r *IncidentRecorder) snapshotCompletedWindows() []*IncidentWindow {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
@@ -704,6 +737,9 @@ func (r *IncidentRecorder) waitForPendingSaves() {
 		r.saveCond.Wait()
 	}
 	r.saveMu.Unlock()
+
+	r.saveIOMu.Lock()
+	r.saveIOMu.Unlock()
 }
 
 // loadFromDisk loads completed windows

internal/metrics/incident_recorder_coverage_test.go

@@ -51,6 +51,39 @@ func TestIncidentRecorderStartStopIdempotentGuards(t *testing.T) {
 	recorder.Stop()
 }
 
+func TestIncidentRecorderStopWaitsForPendingSavesWhenAlreadyStopped(t *testing.T) {
+	recorder := NewIncidentRecorder(IncidentRecorderConfig{
+		DataDir: t.TempDir(),
+	})
+
+	recorder.saveMu.Lock()
+	recorder.saveInProgress = true
+	recorder.saveMu.Unlock()
+
+	done := make(chan struct{})
+	go func() {
+		recorder.Stop()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		t.Fatal("expected Stop to wait for pending saves even when recorder is already stopped")
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	recorder.saveMu.Lock()
+	recorder.saveInProgress = false
+	recorder.saveCond.Broadcast()
+	recorder.saveMu.Unlock()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for Stop to return after pending saves were cleared")
+	}
+}
+
 func TestRecordSampleNoProviderNoop(t *testing.T) {
 	recorder := NewIncidentRecorder(IncidentRecorderConfig{})
 	recorder.recordSample()
@@ -291,11 +324,8 @@ func TestCompleteWindowAsyncSaveErrorPath(t *testing.T) {
 	recorder.activeWindows[window.ID] = window
 
 	recorder.completeWindowLocked(window)
+	recorder.waitForPendingSaves()
 
-	deadline := time.Now().Add(500 * time.Millisecond)
-	for !writer.hit.Load() && time.Now().Before(deadline) {
-		time.Sleep(10 * time.Millisecond)
-	}
 	if !writer.hit.Load() {
 		t.Fatal("expected async save error warning to be logged")
 	}