From dbe49e52c40245c1711d8efdaa73b1c2452898b5 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 08:36:28 +0100 Subject: [PATCH] Join queued incident saves before test cleanup Every IncidentStore mutation queues a coalesced saveAsync goroutine that later runs MkdirAll on the data directory and writes its .tmp file. Nothing tracked those goroutines, so a t.TempDir-backed test could finish while one was still queued. The save then either wrote into the directory while TempDir's RemoveAll was deleting it (CI: "unlinkat ...: directory not empty" in TestIncidentQueryCanonicalOccurrencesAndNotes/memory, run 35932105100) or recreated the directory after cleanup and leaked it. The test's synchronous saveToDisk did not help: it cannot join a save that is already queued. Track queued saves with a WaitGroup and add an unexported flush(), mirroring KnowledgeStore (47cadcc188). The disk-backed incident tests now join through flush in place of polling, a 20ms sleep or waiting for a save count. The canonical-occurrences test reloads from RecordNote's own asynchronous save instead of a manual checkpoint. TestIncidentStoreFlushJoinsQueuedSave holds the save lock to prove flush waits for a queued save that has not started. With a 0-3ms delay injected before each save, the old code leaked a recreated data directory in 233 of 300 runs; with this change, none. Checkpoint timing, coalescing and the stored format are unchanged. --- .../v6/internal/subsystems/ai-runtime.md | 11 ++++++ internal/ai/memory/incident_query_test.go | 11 ++++-- internal/ai/memory/incidents.go | 17 ++++++++- .../memory/incidents_save_coalescing_test.go | 37 +++++++++++++++++++ internal/ai/memory/memory_coverage_test.go | 19 +++++----- 5 files changed, 79 insertions(+), 16 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 015c04f291..005efb5512 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -76,6 +76,17 @@ does it eliminate snapshot serialization. Proof: `internal/ai/memory/incidents_unchanged_test.go` covers unchanged evaluations, metadata, restart, resolution/recurrence, file loss and failed-write retry. +Coalesced asynchronous checkpoints are tracked until they finish. A queued save +still creates the data directory and writes its temp file after the mutation +returns, so a caller that removes or inspects that directory joins in-flight +saves through the store's unexported `flush` first. Tests backed by `t.TempDir` +register it as a cleanup; unjoined saves recreated the removed directory or +failed its cleanup with "directory not empty" on loaded CI runners. This is an +internal durability boundary. It does not change checkpoint timing, +coalescing or the stored format, and it is not a shutdown flush. +`TestIncidentStoreFlushJoinsQueuedSave` proves the join covers a queued save +that has not started. + Assistant owns composer registration and focus on every open, rather than only on component mount. Closing clears the registered input so later keyboard commands cannot target a detached composer. A handoff must leave Escape and diff --git a/internal/ai/memory/incident_query_test.go b/internal/ai/memory/incident_query_test.go index 0be4471ae5..de900c1420 100644 --- a/internal/ai/memory/incident_query_test.go +++ b/internal/ai/memory/incident_query_test.go @@ -23,6 +23,9 @@ func TestIncidentQueryCanonicalOccurrencesAndNotes(t *testing.T) { } config := IncidentStoreConfig{DataDir: t.TempDir()} memory := NewIncidentStore(config) + // Registered after TempDir, so it runs first: queued saves must + // finish writing before the data directory is removed. + t.Cleanup(memory.flush) memory.SetResourceTimelineStore(canonical) start := time.Now().UTC().Add(-time.Hour).Truncate(time.Second) second := start.Add(2 * time.Minute) @@ -64,8 +67,10 @@ func TestIncidentQueryCanonicalOccurrencesAndNotes(t *testing.T) { require.Equal(t, IncidentStatusResolved, noted.Incidents[0].Status) require.Equal(t, IncidentEventNote, noted.Incidents[0].Events[2].Type) require.Equal(t, "operator_note", noted.Incidents[0].Events[2].Source) - require.NoError(t, memory.saveToDisk()) + // Reload from what RecordNote's own asynchronous save wrote. + memory.flush() restored := NewIncidentStore(config) + t.Cleanup(restored.flush) restored.SetResourceTimelineStore(canonical) persisted, err := restored.QueryIncidents(IncidentQuery{AlertIdentifier: "repeat-alert", StartedAt: start, Limit: 1}) require.NoError(t, err) @@ -220,6 +225,7 @@ func TestIncidentQueryCanonicalAlertContextOwnsLegacyShell(t *testing.T) { func TestIncidentAlertContextRetainsOperatorNote(t *testing.T) { canonical := unifiedresources.NewMemoryStore() store := NewIncidentStore(IncidentStoreConfig{DataDir: t.TempDir()}) + t.Cleanup(store.flush) store.SetResourceTimelineStore(canonical) started := time.Now().UTC().Add(-time.Hour) change := unifiedresources.BuildAlertTimelineChange("resource-a", unifiedresources.ChangeAlertFired, started, "", unifiedresources.AlertTimelineChange{AlertIdentifier: "noted-alert"}) @@ -228,9 +234,6 @@ func TestIncidentAlertContextRetainsOperatorNote(t *testing.T) { require.NoError(t, err) require.Len(t, page.Incidents, 1) require.True(t, store.RecordNote("noted-alert", page.Incidents[0].ID, "Keep the old pool until its replacement is verified", "operator")) - // RecordNote persists asynchronously. Keep the temporary data directory - // alive until that write finishes so cleanup cannot race the save goroutine. - waitForCompletedSaves(t, store, 1) context := store.FormatForAlert("noted-alert", 10) require.Contains(t, context, "Note added by operator: Keep the old pool until its replacement is verified") require.Contains(t, context, "source=operator_note") diff --git a/internal/ai/memory/incidents.go b/internal/ai/memory/incidents.go index 64cba24fcd..d3d039b4f1 100644 --- a/internal/ai/memory/incidents.go +++ b/internal/ai/memory/incidents.go @@ -282,8 +282,9 @@ type IncidentStoreConfig struct { type IncidentStore struct { mu sync.RWMutex saveMu sync.Mutex - savePending atomic.Bool // a queued save will capture the latest state; further requests coalesce - savesCompleted atomic.Int64 // completed JSON replacements, observable by tests + savePending atomic.Bool // a queued save will capture the latest state; further requests coalesce + saveWG sync.WaitGroup // asynchronous saves still running or queued; joined by flush + savesCompleted atomic.Int64 // completed JSON replacements, observable by tests incidents []*incidentShell maxIncidents int maxEvents int @@ -1459,13 +1460,25 @@ func (s *IncidentStore) saveAsync() { if !s.savePending.CompareAndSwap(false, true) { return } + s.saveWG.Add(1) go func() { + defer s.saveWG.Done() if err := s.saveToDisk(); err != nil { log.Warn().Err(err).Msg("failed to save incident history") } }() } +// flush blocks until every asynchronous save queued by a mutation has +// finished. A queued save still creates the data directory and writes its +// temp file after the mutation returns, so callers that remove or inspect the +// data directory (including tests) join it first. Stop mutating the store +// before calling flush: WaitGroup forbids a new save being queued while a +// previous Wait is still returning. +func (s *IncidentStore) flush() { + s.saveWG.Wait() +} + func (s *IncidentStore) saveToDisk() error { s.saveMu.Lock() defer s.saveMu.Unlock() diff --git a/internal/ai/memory/incidents_save_coalescing_test.go b/internal/ai/memory/incidents_save_coalescing_test.go index c3010c3441..d126da3b26 100644 --- a/internal/ai/memory/incidents_save_coalescing_test.go +++ b/internal/ai/memory/incidents_save_coalescing_test.go @@ -1,6 +1,7 @@ package memory import ( + "os" "testing" "time" ) @@ -49,3 +50,39 @@ func TestIncidentStoreSaveCoalescing(t *testing.T) { t.Fatalf("persisted events = %d, want all 101 mutations captured", len(timeline.Events)) } } + +func TestIncidentStoreFlushJoinsQueuedSave(t *testing.T) { + dir := t.TempDir() + store := NewIncidentStore(IncidentStoreConfig{DataDir: dir}) + + // Hold the save lock so the save queued by the mutation cannot start. This + // is the window where TempDir cleanup used to race it: the mutation has + // returned, but the goroutine has not yet created its temp file. + store.saveMu.Lock() + store.RecordAnalysis("alert-flush", "analysis", nil) + + flushed := make(chan struct{}) + go func() { + store.flush() + close(flushed) + }() + select { + case <-flushed: + store.saveMu.Unlock() + t.Fatal("flush returned while a queued save had not run") + case <-time.After(50 * time.Millisecond): + } + + store.saveMu.Unlock() + select { + case <-flushed: + case <-time.After(5 * time.Second): + t.Fatal("flush did not return after the queued save ran") + } + if saves := store.savesCompleted.Load(); saves != 1 { + t.Fatalf("saves completed when flush returned = %d, want 1", saves) + } + if _, err := os.Stat(store.filePath + ".tmp"); !os.IsNotExist(err) { + t.Fatalf("temp file left behind after flush: %v", err) + } +} diff --git a/internal/ai/memory/memory_coverage_test.go b/internal/ai/memory/memory_coverage_test.go index 2e354e4f88..e96eb817af 100644 --- a/internal/ai/memory/memory_coverage_test.go +++ b/internal/ai/memory/memory_coverage_test.go @@ -580,16 +580,10 @@ func TestIncidentStore_SaveAsyncAndPersistence(t *testing.T) { } store.saveAsync() + store.flush() - deadline := time.Now().Add(500 * time.Millisecond) - for { - if _, err := os.Stat(store.filePath); err == nil { - break - } - if time.Now().After(deadline) { - t.Fatalf("expected saveAsync to create file") - } - time.Sleep(10 * time.Millisecond) + if _, err := os.Stat(store.filePath); err != nil { + t.Fatalf("expected saveAsync to create file: %v", err) } } @@ -608,7 +602,12 @@ func TestIncidentStore_SaveAsync_Error(t *testing.T) { filePath: filepath.Join(badDir, incidentFileName), } store.saveAsync() - time.Sleep(20 * time.Millisecond) + // An unjoined save could run MkdirAll after TempDir cleanup removed the + // blocking file, recreating the directory mid-removal. + store.flush() + if info, err := os.Stat(badDir); err != nil || info.IsDir() { + t.Fatalf("failed save must leave the blocking file in place: info=%v err=%v", info, err) + } } func TestIncidentStore_SaveToDisk_Scenarios(t *testing.T) {