From 88d44099aa3cd8404035ebdd78e8f4a6201a9582 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Wed, 7 Oct 2026 20:12:04 +0100 Subject: [PATCH] Stop tenant incident stores racing on shared ID counters Each tenant monitor owns its own IncidentStore, but incident and incident-event IDs came from package-level int64 counters bumped with ++ under nothing but the calling store's mutex. Two tenants recording alerts at once raced on them: go test -race failed TestLeavingMockModeClearsEveryRunningTenantMonitor on PR 2698 at incidents.go:1753. The remediation record counter had the same shape, and remediation logs are created per org. All six internal/ai/memory ID generators now mint through one atomic helper. The numeric suffix also stops wrapping at 1000: even with atomic counters, one generator minting more than a thousand IDs in a wall-clock second repeated an ID, and incident notes find their incident by ID. The prefix and timestamp layout is unchanged, and intToString lost its last callers. ids_test.go mints 2,400 IDs from every generator on 16 goroutines and records alerts into two tenant stores concurrently. Against the old files it reports DATA RACE at incidents.go:1753/1760 and remediation.go:355, and every generator mints duplicate or wrapped IDs. --- .../v6/internal/subsystems/ai-runtime.md | 19 +++ internal/ai/memory/changes.go | 14 -- internal/ai/memory/context.go | 9 +- internal/ai/memory/ids.go | 16 +++ internal/ai/memory/ids_test.go | 122 ++++++++++++++++++ internal/ai/memory/incidents.go | 10 +- internal/ai/memory/memory_coverage_test.go | 9 +- internal/ai/memory/remediation.go | 6 +- 8 files changed, 167 insertions(+), 38 deletions(-) create mode 100644 internal/ai/memory/ids.go create mode 100644 internal/ai/memory/ids_test.go diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 9ab54a9d35..540a33d923 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -128,6 +128,25 @@ on component mount. Closing clears the registered input so later keyboard commands cannot target a detached composer. A handoff must leave Escape and keyboard input in Assistant, not the underlying alert search. +### Process-wide memory ID sequences + +`internal/ai/memory` mints incident, incident-event, remediation-record and +context-memory IDs from package-level sequences that every store in the process +shares. Multi-tenant monitoring gives each tenant monitor its own +`IncidentStore`, and their alert checks call `RecordAlertFired` from separate +goroutines, while a store's mutex covers only its own state. Each sequence is +therefore an `atomic.Int64` minted through `nextSequencedID` in +`internal/ai/memory/ids.go`, never a plain package counter. The ID shape stays +`-`, but `n` no longer wraps at 1000. The old +`counter%1000` suffix repeated an ID whenever one generator minted more than a +thousand in a wall-clock second, and incident notes resolve an incident by ID +while timeline merges deduplicate events by ID. Within one process IDs are now +unique by construction. Across restarts only the timestamp separates them, so a +repeated local second (DST fall-back or a backward clock step) can still meet a +restarted sequence's suffix. Persisted IDs and the stored format are unchanged. +`internal/ai/memory/ids_test.go` mints from every generator concurrently and +records alerts into two tenant stores concurrently; run it under `-race`. + ### Canonical incident-history queries Incident context reads `IncidentStore.QueryIncidents`, whose page contains diff --git a/internal/ai/memory/changes.go b/internal/ai/memory/changes.go index 0771e9cd71..3b0a4e4bc0 100644 --- a/internal/ai/memory/changes.go +++ b/internal/ai/memory/changes.go @@ -131,17 +131,3 @@ func (d *ChangeDetector) loadFromDisk() error { d.trimChanges() return nil } - -// Helper functions - -func intToString(n int) string { - if n == 0 { - return "0" - } - var result string - for n > 0 { - result = string(rune('0'+n%10)) + result - n /= 10 - } - return result -} diff --git a/internal/ai/memory/context.go b/internal/ai/memory/context.go index 1b5547d882..9a42d03758 100644 --- a/internal/ai/memory/context.go +++ b/internal/ai/memory/context.go @@ -711,18 +711,15 @@ func (s *ContextStore) ForceSave() error { var memoryCounter, incidentMemCounter, patternMemCounter atomic.Int64 func generateMemoryID() string { - n := memoryCounter.Add(1) - return fmt.Sprintf("mem-%s-%d", time.Now().Format("20060102150405"), n%1000) + return nextSequencedID("mem-", &memoryCounter) } func generateIncidentMemoryID() string { - n := incidentMemCounter.Add(1) - return fmt.Sprintf("inc-mem-%s-%d", time.Now().Format("20060102150405"), n%1000) + return nextSequencedID("inc-mem-", &incidentMemCounter) } func generatePatternMemoryID() string { - n := patternMemCounter.Add(1) - return fmt.Sprintf("pat-mem-%s-%d", time.Now().Format("20060102150405"), n%1000) + return nextSequencedID("pat-mem-", &patternMemCounter) } func calculatePatternConfidence(occurrences int) float64 { diff --git a/internal/ai/memory/ids.go b/internal/ai/memory/ids.go new file mode 100644 index 0000000000..2be409a1da --- /dev/null +++ b/internal/ai/memory/ids.go @@ -0,0 +1,16 @@ +package memory + +import ( + "strconv" + "sync/atomic" + "time" +) + +// nextSequencedID mints prefix + local wall-clock second + "-" + the next value +// of seq. Each sequence is package-level and shared by every store in the +// process, including one store per tenant monitor, so it must stay atomic. +// The suffix never wraps, so one sequence never repeats an ID within a process +// however many IDs share a second. +func nextSequencedID(prefix string, seq *atomic.Int64) string { + return prefix + time.Now().Format("20060102150405") + "-" + strconv.FormatInt(seq.Add(1), 10) +} diff --git a/internal/ai/memory/ids_test.go b/internal/ai/memory/ids_test.go new file mode 100644 index 0000000000..530381625e --- /dev/null +++ b/internal/ai/memory/ids_test.go @@ -0,0 +1,122 @@ +package memory + +import ( + "fmt" + "regexp" + "strings" + "sync" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" +) + +// TestSequencedIDsStayUniqueUnderConcurrentMinting mints from every +// package-level ID sequence on many goroutines at once. Under -race it proves +// the sequences are synchronized. The uniqueness check proves the suffix does +// not wrap: more than a thousand IDs from one generator land in the same +// wall-clock second, which the old counter%1000 suffix turned into duplicates. +func TestSequencedIDsStayUniqueUnderConcurrentMinting(t *testing.T) { + generators := []struct { + prefix string + mint func() string + }{ + {"inc-", generateIncidentID}, + {"inc-evt-", generateIncidentEventID}, + {"rem-", generateRecordID}, + {"mem-", generateMemoryID}, + {"inc-mem-", generateIncidentMemoryID}, + {"pat-mem-", generatePatternMemoryID}, + } + const goroutines, perGoroutine = 16, 150 + + for _, gen := range generators { + t.Run(strings.TrimSuffix(gen.prefix, "-"), func(t *testing.T) { + shape := regexp.MustCompile("^" + regexp.QuoteMeta(gen.prefix) + `\d{14}-[1-9]\d*$`) + minted := make([][]string, goroutines) + start := make(chan struct{}) + var wg sync.WaitGroup + for g := range goroutines { + wg.Add(1) + go func() { + defer wg.Done() + <-start + batch := make([]string, perGoroutine) + for i := range batch { + batch[i] = gen.mint() + } + minted[g] = batch + }() + } + close(start) + wg.Wait() + + seen := make(map[string]bool, goroutines*perGoroutine) + for _, batch := range minted { + for _, id := range batch { + if !shape.MatchString(id) { + t.Fatalf("ID %q does not match %s", id, shape) + } + if seen[id] { + t.Fatalf("duplicate ID %q among %d concurrent mints", id, goroutines*perGoroutine) + } + seen[id] = true + } + } + }) + } +} + +// TestIncidentStoresMintIDsConcurrentlyAcrossTenants mirrors multi-tenant +// monitoring: each tenant monitor owns its own IncidentStore, and their alert +// checks call RecordAlertFired from separate goroutines. Each store's mutex +// covers only its own state, so the shared ID sequence must be synchronized on +// its own. Run under -race. +func TestIncidentStoresMintIDsConcurrentlyAcrossTenants(t *testing.T) { + const tenants, checkersPerTenant, alertsPerChecker = 2, 4, 50 + stores := make([]*IncidentStore, tenants) + for i := range stores { + stores[i] = NewIncidentStore(IncidentStoreConfig{MaxIncidents: checkersPerTenant * alertsPerChecker}) + } + + startedAt := time.Now().Add(-time.Minute) + start := make(chan struct{}) + var wg sync.WaitGroup + for tenant, store := range stores { + for checker := range checkersPerTenant { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for n := range alertsPerChecker { + store.RecordAlertFired(&alerts.Alert{ + ID: fmt.Sprintf("tenant-%d-checker-%d-alert-%d", tenant, checker, n), + Type: "cpu", + Level: alerts.AlertLevelWarning, + ResourceID: fmt.Sprintf("tenant-%d-vm-%d-%d", tenant, checker, n), + ResourceName: "vm", + StartTime: startedAt, + }) + } + }() + } + } + close(start) + wg.Wait() + + seen := make(map[string]bool, tenants*checkersPerTenant*alertsPerChecker) + for tenant, store := range stores { + store.mu.RLock() + incidents := len(store.incidents) + for _, shell := range store.incidents { + if seen[shell.ID] { + t.Errorf("tenant %d incident ID %q was already minted", tenant, shell.ID) + } + seen[shell.ID] = true + } + store.mu.RUnlock() + if incidents != checkersPerTenant*alertsPerChecker { + t.Errorf("tenant %d recorded %d incidents, want %d", tenant, incidents, checkersPerTenant*alertsPerChecker) + } + } +} diff --git a/internal/ai/memory/incidents.go b/internal/ai/memory/incidents.go index 8729c83cef..5c763b2d92 100644 --- a/internal/ai/memory/incidents.go +++ b/internal/ai/memory/incidents.go @@ -1747,18 +1747,14 @@ func cloneTime(value time.Time) *time.Time { return &cloned } -var incidentCounter int64 +var incidentCounter, incidentEventCounter atomic.Int64 func generateIncidentID() string { - incidentCounter++ - return "inc-" + time.Now().Format("20060102150405") + "-" + intToString(int(incidentCounter%1000)) + return nextSequencedID("inc-", &incidentCounter) } -var incidentEventCounter int64 - func generateIncidentEventID() string { - incidentEventCounter++ - return "inc-evt-" + time.Now().Format("20060102150405") + "-" + intToString(int(incidentEventCounter%1000)) + return nextSequencedID("inc-evt-", &incidentEventCounter) } func formatAlertSummary(alert *alerts.Alert) string { diff --git a/internal/ai/memory/memory_coverage_test.go b/internal/ai/memory/memory_coverage_test.go index 653f05796d..7ae826ea6a 100644 --- a/internal/ai/memory/memory_coverage_test.go +++ b/internal/ai/memory/memory_coverage_test.go @@ -11,18 +11,11 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/alerts" ) -func TestChangeDetector_DefaultsAndHelpers(t *testing.T) { +func TestChangeDetector_Defaults(t *testing.T) { detector := NewChangeDetector(ChangeDetectorConfig{}) if detector.maxChanges != 1000 { t.Fatalf("expected default maxChanges=1000, got %d", detector.maxChanges) } - - if got := intToString(0); got != "0" { - t.Errorf("intToString(0) = %q", got) - } - if got := intToString(42); got != "42" { - t.Errorf("intToString(42) = %q", got) - } } func TestNewChangeDetector_LoadsFromDisk(t *testing.T) { diff --git a/internal/ai/memory/remediation.go b/internal/ai/memory/remediation.go index da93f5f4f5..97d2c8ac81 100644 --- a/internal/ai/memory/remediation.go +++ b/internal/ai/memory/remediation.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "sync" + "sync/atomic" "time" "github.com/rcourtman/pulse-go-rewrite/internal/utils" @@ -349,11 +350,10 @@ func (r *RemediationLog) loadFromDisk() error { // Helper functions -var recordCounter int64 +var recordCounter atomic.Int64 func generateRecordID() string { - recordCounter++ - return "rem-" + time.Now().Format("20060102150405") + "-" + intToString(int(recordCounter%1000)) + return nextSequencedID("rem-", &recordCounter) } func truncateOutput(output string, maxLen int) string {