diff --git a/internal/metrics/incident_recorder.go b/internal/metrics/incident_recorder.go index cb4f607c9..69abc8c57 100644 --- a/internal/metrics/incident_recorder.go +++ b/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 diff --git a/internal/metrics/incident_recorder_coverage_test.go b/internal/metrics/incident_recorder_coverage_test.go index e79ce08a8..f365531ae 100644 --- a/internal/metrics/incident_recorder_coverage_test.go +++ b/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") }