From 2494930f04d21e920fda62c7f1dd774ee2f9f12a Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:43:32 +0100 Subject: [PATCH] alerts: answer history reads from the shared fold without full copies On a long-running install the alert history projection was the largest part of the Go heap: 36 of 78 MB live on the dogfood install, one merged alert per occurrence for the event log's 90 days (#1803). Every read then deep-copied the whole projection before applying its limit, so the attention summary's 500-row read copied all 1,895 occurrences, and every Since read (the 30-day alert-quality window, date-filtered history) built a fresh projection and re-decoded every snapshot in its window. Each occurrence now keeps one small mark per folded event (time, id and the LastSeen it contributed) and the index of its latest acknowledged snapshot. A Since window starts at the first event the log's own occurred_at >= since text comparison admits, takes LastSeen and acknowledgement from there and orders occurrences by their first in-window event, so it equals a walk of only the window. Overlay, sort and limit run on lightweight rows under the projection lock, and only returned rows are copied, once, by canonicalizeAlertHistoryForOutput. On a log shaped like the dogfood install's (about 1,900 occurrences, 40k events), a 30-day read went from 68.9 ms and 19.8 MB allocated to 5.2 ms and 2.8 MB, a 500-row read from 12.7 ms and 8.5 MB to 3.6 ms and 2.0 MB, and a full read from 20.8 ms and 12.9 MB to 13.4 ms and 6.9 MB. A randomized test holds every window and limit equal to an independent walk of only the window, across acknowledgements on both sides of the boundary, uneven fractional seconds, same-instant events, a history clear and live overlays. Negative controls: ignoring the window, a fresh fold per Since read and copying every row each fail their test. Related to #1803 --- .../v6/internal/subsystems/alerts.md | 13 + internal/alerts/history_projection.go | 312 +++++++++++++----- .../alerts/history_projection_parity_test.go | 263 +++++++++++++++ 3 files changed, 497 insertions(+), 91 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index ebea6b5277..8388d6fad9 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -1028,6 +1028,19 @@ Live active-alert overlays and returned objects are independent clones. Explicit Since queries preserve their existing event-window semantics without evicting the normal full-history fold. Repeated attention requests must not replay all retained snapshots independently or accumulate behind the history connection pool. +Since queries are answered from that same fold (#1803). Each occurrence keeps one +mark per folded event (time, durable ID and the LastSeen it contributed) and the +index of its latest acknowledged snapshot. A window starts at the occurrence's +first event that the log's own `occurred_at >= since` text comparison admits, +takes LastSeen and acknowledgement from that point on, and orders occurrences by +their first in-window event as a walk of only the window would. A read selects +and limits rows before copying, so only returned rows are cloned. +`TestWindowedHistoryMatchesAWalkOfOnlyTheWindow` holds this equal to an +independent walk of only the window, and +`TestLimitedHistoryReadCopiesOnlyReturnedRows`, +`TestWindowedHistoryReusesTheFullHistoryFold` and +`TestHistoryReadResultsDoNotAliasTheSharedFold` pin the copy, reuse and +isolation rules. PBS node-status availability is distinct from connectivity. While a connected diff --git a/internal/alerts/history_projection.go b/internal/alerts/history_projection.go index 77dc8d2a75..1e83218a7a 100644 --- a/internal/alerts/history_projection.go +++ b/internal/alerts/history_projection.go @@ -17,19 +17,102 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog" ) +// historyEventMark is what one folded lifecycle event contributed to its +// occurrence. The marks let a Since window be answered from the full fold +// instead of decoding every snapshot in the window again (#1803). +type historyEventMark struct { + at time.Time + id int64 + lastSeen time.Time // the snapshot's LastSeen, or the resolution time if later +} + // historyOccurrence accumulates one alert occurrence from its lifecycle // events, oldest to newest. type historyOccurrence struct { - alert Alert - firstEvent time.Time - lastEvent time.Time - resolved bool + alert Alert + events []historyEventMark + // lastAcked indexes the latest event whose snapshot was acknowledged, or + // is -1. A window starting after it never saw the acknowledgement. + lastAcked int + // The latest snapshot's own acknowledgement fields, before merging carried + // an earlier acknowledgement forward onto them. + rawAcknowledged bool + rawAckTime *time.Time + rawAckUser string } func historyOccurrenceKey(alertID string, snapshot *Alert) string { return alertID + "\x00" + snapshot.StartTime.UTC().Format(time.RFC3339Nano) } +// windowStart returns the index of the first event the log's Since filter +// (occurred_at >= since, compared as RFC3339Nano text) admits, or +// len(events) when none is. Events are folded in the log's (occurred_at, +// id) text order, so the comparison is monotonic. +func (occ *historyOccurrence) windowStart(sinceKey string) int { + return sort.Search(len(occ.events), func(i int) bool { + return occ.events[i].at.UTC().Format(time.RFC3339Nano) >= sinceKey + }) +} + +// windowAck is the acknowledgement that folding events[start:] alone would +// produce: an acknowledgement is carried forward, so only one inside the +// window survives into it. +func (occ *historyOccurrence) windowAck(start int) (bool, *time.Time, string) { + if occ.lastAcked >= start { + return occ.alert.Acknowledged, occ.alert.AckTime, occ.alert.AckUser + } + return occ.rawAcknowledged, occ.rawAckTime, occ.rawAckUser +} + +// windowLastSeen is the LastSeen that folding events[start:] alone would +// produce: the latest contribution inside the window. +func (occ *historyOccurrence) windowLastSeen(start int) time.Time { + if start == 0 { + return occ.alert.LastSeen + } + var latest time.Time + for _, mark := range occ.events[start:] { + if mark.lastSeen.After(latest) { + latest = mark.lastSeen + } + } + return latest +} + +// historyRow is one candidate history entry. Rows reference the shared fold +// and are copied only when selected for output. +type historyRow struct { + key string + occ *historyOccurrence // nil for an active alert the log never saw fire + start int + firstEvent time.Time + overlay *Alert // the live active alert shown instead of the folded state +} + +func (row *historyRow) ack() (bool, *time.Time, string) { + if row.overlay != nil { + return row.overlay.Acknowledged, row.overlay.AckTime, row.overlay.AckUser + } + return row.occ.windowAck(row.start) +} + +// materialize returns the row's entry. It shares the fold's maps and pointers; +// projectedHistory deep-copies every returned entry through +// canonicalizeAlertHistoryForOutput before releasing the projection lock, so +// that is the only copy a read makes. +func (row *historyRow) materialize() Alert { + if row.overlay != nil { + return *row.overlay + } + alert := row.occ.alert + if row.start > 0 { + alert.LastSeen = row.occ.windowLastSeen(row.start) + alert.Acknowledged, alert.AckTime, alert.AckUser = row.occ.windowAck(row.start) + } + return alert +} + // AlertHistoryFromEvents projects the alert history list from the event // log. It reports ok=false when no event log is enabled, letting callers // fall back to the JSON-backed history manager. @@ -38,73 +121,19 @@ func (m *Manager) AlertHistoryFromEvents(since time.Time, limit int) ([]Alert, b if store == nil || !m.eventHistoryAuthoritative.Load() { return nil, false } - - occurrences, order, err := m.historyOccurrences(store, since) + results, err := m.projectedHistory(store, since, limit) if err != nil { return nil, false } - - // Overlay the live active alerts: an occurrence still firing shows its - // current state (fresh LastSeen and value), exactly as the JSON - // history's continuously updated entries do. Active alerts the log - // never saw fire (log enabled mid-flight) are appended. - m.mu.RLock() - active := make([]*Alert, 0, len(m.activeAlerts)) - for storageKey, alert := range m.activeAlerts { - if alert == nil { - continue - } - exported := cloneAlertForOutput(alert) - if exported == nil { - continue - } - if exported.ID == "" { - exported.ID = effectiveAlertID(alert, storageKey) - } - active = append(active, exported) - } - m.mu.RUnlock() - for _, alert := range active { - key := historyOccurrenceKey(alert.ID, alert) - if occ, exists := occurrences[key]; exists { - ack := occ.alert.Acknowledged - ackTime, ackUser := occ.alert.AckTime, occ.alert.AckUser - occ.alert = *alert - if ack && !occ.alert.Acknowledged { - occ.alert.Acknowledged = true - occ.alert.AckTime = ackTime - occ.alert.AckUser = ackUser - } - occ.resolved = false - } else if since.IsZero() || alert.StartTime.After(since) { - occurrences[key] = &historyOccurrence{ - alert: *alert, - firstEvent: alert.StartTime, - lastEvent: alert.LastSeen, - } - order = append(order, key) - } - } - - sort.SliceStable(order, func(i, j int) bool { - return occurrences[order[i]].firstEvent.Before(occurrences[order[j]].firstEvent) - }) - - results := make([]Alert, 0, len(order)) - for i := len(order) - 1; i >= 0; i-- { - if limit > 0 && len(results) >= limit { - break - } - results = append(results, occurrences[order[i]].alert) - } - return m.applyCurrentNodeDisplayNames(canonicalizeAlertHistoryForOutput(results)), true + return m.applyCurrentNodeDisplayNames(results), true } // historyProjection is a disposable fold of the durable log, never a second // source of truth. Its cursor is bounded by a committed event ID. Old imports, // retention and store replacement rebuild it using the original chronological -// fold. A moving Since window keeps its original event-filter semantics and -// does not evict the full-history fold used by attention and normal polling. +// fold. A Since window is answered from this full-history fold with the same +// result as walking only the window's events (windowStart, windowAck and +// windowLastSeen), so windowed reads neither rebuild nor evict it. type historyProjection struct { store *eventlog.Store boundary eventlog.ReplayBoundary @@ -119,7 +148,7 @@ func newHistoryProjection(store *eventlog.Store) *historyProjection { return &historyProjection{store: store, occurrences: make(map[string]*historyOccurrence)} } -func (p *historyProjection) fold(since time.Time, throughID int64) error { +func (p *historyProjection) fold(throughID int64) error { if throughID == 0 || throughID == p.boundary.LastID { return nil } @@ -130,7 +159,7 @@ func (p *historyProjection) fold(since time.Time, throughID int64) error { eventlog.TypeSnoozed, eventlog.TypeUnsnoozed, eventlog.TypeEscalated, eventlog.TypeHistoryImported, eventlog.TypeHistoryCleared, }, - Since: since, AfterID: p.boundary.LastID, ThroughID: throughID, + AfterID: p.boundary.LastID, ThroughID: throughID, }, func(event eventlog.Event) error { // Match the log's (occurred_at, id) ordering. Equal timestamps are // safe because all appended IDs exceed the preceding boundary. @@ -156,72 +185,173 @@ func (p *historyProjection) fold(since time.Time, throughID int64) error { key := historyOccurrenceKey(event.AlertID, &snapshot) occ, exists := p.occurrences[key] if !exists { - occ = &historyOccurrence{alert: snapshot, firstEvent: event.OccurredAt} + occ = &historyOccurrence{alert: snapshot, lastAcked: -1} p.occurrences[key] = occ p.order = append(p.order, key) } else { occ.alert = mergeHistoryAlertSnapshots(occ.alert, snapshot) } - occ.lastEvent = event.OccurredAt + mark := historyEventMark{at: event.OccurredAt, id: event.ID, lastSeen: snapshot.LastSeen} if event.Type == eventlog.TypeResolved { - occ.resolved = true // The JSON history's resolve path stamps the entry's LastSeen // with the resolution time so the row reflects the true // duration; mirror that. if event.OccurredAt.After(occ.alert.LastSeen) { occ.alert.LastSeen = event.OccurredAt } - } else { - occ.resolved = false + if event.OccurredAt.After(mark.lastSeen) { + mark.lastSeen = event.OccurredAt + } } + occ.events = append(occ.events, mark) + if snapshot.Acknowledged { + occ.lastAcked = len(occ.events) - 1 + } + occ.rawAcknowledged, occ.rawAckTime, occ.rawAckUser = snapshot.Acknowledged, snapshot.AckTime, snapshot.AckUser return nil }) } -func (m *Manager) historyOccurrences(store *eventlog.Store, since time.Time) (map[string]*historyOccurrence, []string, error) { - m.historyProjectionMu.Lock() - defer m.historyProjectionMu.Unlock() +// currentHistoryProjection folds the log through its current boundary and +// returns the shared projection. The caller must hold historyProjectionMu. +func (m *Manager) currentHistoryProjection(store *eventlog.Store) (*historyProjection, error) { for { boundary, err := store.ReplayBoundary() if err != nil { m.historyProjection = nil - return nil, nil, err + return nil, err } p := m.historyProjection - if !since.IsZero() || p == nil || p.store != store || + if p == nil || p.store != store || p.boundary.RetentionRevision != boundary.RetentionRevision || p.boundary.LastID > boundary.LastID { p = newHistoryProjection(store) } - err = p.fold(since, boundary.LastID) + err = p.fold(boundary.LastID) if errors.Is(err, errHistoryEventBeforeCursor) { p = newHistoryProjection(store) - err = p.fold(since, boundary.LastID) + err = p.fold(boundary.LastID) } if err != nil { m.historyProjection = nil - return nil, nil, err + return nil, err } after, err := store.ReplayBoundary() if err != nil { m.historyProjection = nil - return nil, nil, err + return nil, err } if after.RetentionRevision != boundary.RetentionRevision { m.historyProjection = nil continue } p.boundary = boundary - if since.IsZero() { - m.historyProjection = p - } - // The live-state overlay and callers may mutate their result. Never - // let those changes contaminate the durable fold or another reader. - occurrences := make(map[string]*historyOccurrence, len(p.occurrences)) - for key, occurrence := range p.occurrences { - copy := *occurrence - copy.alert = *occurrence.alert.Clone() - occurrences[key] = © - } - return occurrences, append([]string(nil), p.order...), nil + m.historyProjection = p + return p, nil } } + +// projectedHistory selects the history list, newest first, under the +// projection lock, and copies only the entries it returns. Callers may mutate +// the result without touching the shared fold or another reader's result. +func (m *Manager) projectedHistory(store *eventlog.Store, since time.Time, limit int) ([]Alert, error) { + m.historyProjectionMu.Lock() + defer m.historyProjectionMu.Unlock() + p, err := m.currentHistoryProjection(store) + if err != nil { + return nil, err + } + + sinceKey := "" + if !since.IsZero() { + sinceKey = since.UTC().Format(time.RFC3339Nano) + } + rows := make([]historyRow, 0, len(p.order)) + for _, key := range p.order { + occ := p.occurrences[key] + start := 0 + if sinceKey != "" { + if start = occ.windowStart(sinceKey); start == len(occ.events) { + continue + } + } + rows = append(rows, historyRow{key: key, occ: occ, start: start, firstEvent: occ.events[start].at}) + } + if sinceKey != "" { + // Walking only the window appends each occurrence at its first event + // inside the window, in the log's (occurred_at, id) order. + keys := make([]string, len(rows)) + for i := range rows { + keys[i] = rows[i].firstEvent.UTC().Format(time.RFC3339Nano) + } + order := make([]int, len(rows)) + for i := range order { + order[i] = i + } + sort.SliceStable(order, func(a, b int) bool { + ra, rb := order[a], order[b] + if keys[ra] != keys[rb] { + return keys[ra] < keys[rb] + } + return rows[ra].occ.events[rows[ra].start].id < rows[rb].occ.events[rows[rb].start].id + }) + sorted := make([]historyRow, len(rows)) + for i, row := range order { + sorted[i] = rows[row] + } + rows = sorted + } + + // Overlay the live active alerts: an occurrence still firing shows its + // current state (fresh LastSeen and value), exactly as the JSON + // history's continuously updated entries do. Active alerts the log + // never saw fire (log enabled mid-flight) are appended. + m.mu.RLock() + active := make([]*Alert, 0, len(m.activeAlerts)) + for storageKey, alert := range m.activeAlerts { + if alert == nil { + continue + } + exported := cloneAlertForOutput(alert) + if exported == nil { + continue + } + if exported.ID == "" { + exported.ID = effectiveAlertID(alert, storageKey) + } + active = append(active, exported) + } + m.mu.RUnlock() + index := make(map[string]int, len(rows)) + for i := range rows { + index[rows[i].key] = i + } + for _, alert := range active { + key := historyOccurrenceKey(alert.ID, alert) + if i, exists := index[key]; exists { + acknowledged, ackTime, ackUser := rows[i].ack() + overlay := *alert + if acknowledged && !overlay.Acknowledged { + overlay.Acknowledged, overlay.AckTime, overlay.AckUser = true, ackTime, ackUser + } + rows[i].overlay = &overlay + } else if since.IsZero() || alert.StartTime.After(since) { + overlay := *alert + index[key] = len(rows) + rows = append(rows, historyRow{key: key, firstEvent: alert.StartTime, overlay: &overlay}) + } + } + + sort.SliceStable(rows, func(i, j int) bool { + return rows[i].firstEvent.Before(rows[j].firstEvent) + }) + + size := len(rows) + if limit > 0 && limit < size { + size = limit + } + results := make([]Alert, 0, size) + for i := len(rows) - 1; i >= 0 && len(results) < size; i-- { + results = append(results, rows[i].materialize()) + } + return canonicalizeAlertHistoryForOutput(results), nil +} diff --git a/internal/alerts/history_projection_parity_test.go b/internal/alerts/history_projection_parity_test.go index 039b75d0b2..b028865299 100644 --- a/internal/alerts/history_projection_parity_test.go +++ b/internal/alerts/history_projection_parity_test.go @@ -10,7 +10,10 @@ package alerts import ( "encoding/json" "fmt" + "math/rand" "reflect" + "runtime" + "sort" "sync" "testing" "time" @@ -291,3 +294,263 @@ func TestHistoryProjectionWalksEveryLifecycleEventBeyondQueryCap(t *testing.T) { t.Fatalf("limited projection order = %q ... %q", limited[0].ID, limited[24].ID) } } + +// Windowed and limited history reads are answered from the one shared +// full-history fold (#1803). Before, every read deep-copied the whole fold and +// every Since read re-decoded each snapshot in its window. These tests hold +// the shared answer equal to an independent walk of only the window's events, +// which is the original algorithm. + +// referenceWindowedHistory is the original read path: a fresh fold of only +// the events the log's Since filter admits, the live overlay, then the sort. +func referenceWindowedHistory(t *testing.T, m *Manager, since time.Time, limit int) []Alert { + t.Helper() + type occurrence struct { + alert Alert + firstEvent time.Time + } + occurrences := map[string]*occurrence{} + var order []string + err := m.eventLogStore().WalkOldest(eventlog.Filter{ + Types: []string{ + eventlog.TypeFired, eventlog.TypeRefired, eventlog.TypeResolved, + eventlog.TypeAcknowledged, eventlog.TypeUnacknowledged, + eventlog.TypeSnoozed, eventlog.TypeUnsnoozed, eventlog.TypeEscalated, + eventlog.TypeHistoryImported, eventlog.TypeHistoryCleared, + }, + Since: since, + }, func(event eventlog.Event) error { + if event.Type == eventlog.TypeHistoryCleared { + occurrences = map[string]*occurrence{} + order = order[:0] + return nil + } + var snapshot Alert + if len(event.Snapshot) == 0 || json.Unmarshal(event.Snapshot, &snapshot) != nil { + return nil + } + key := historyOccurrenceKey(event.AlertID, &snapshot) + occ, exists := occurrences[key] + if !exists { + occ = &occurrence{alert: snapshot, firstEvent: event.OccurredAt} + occurrences[key] = occ + order = append(order, key) + } else { + occ.alert = mergeHistoryAlertSnapshots(occ.alert, snapshot) + } + if event.Type == eventlog.TypeResolved && event.OccurredAt.After(occ.alert.LastSeen) { + occ.alert.LastSeen = event.OccurredAt + } + return nil + }) + if err != nil { + t.Fatalf("reference walk: %v", err) + } + m.mu.RLock() + var active []*Alert + for storageKey, alert := range m.activeAlerts { + exported := cloneAlertForOutput(alert) + if exported.ID == "" { + exported.ID = effectiveAlertID(alert, storageKey) + } + active = append(active, exported) + } + m.mu.RUnlock() + for _, alert := range active { + key := historyOccurrenceKey(alert.ID, alert) + if occ, exists := occurrences[key]; exists { + ack, ackTime, ackUser := occ.alert.Acknowledged, occ.alert.AckTime, occ.alert.AckUser + occ.alert = *alert + if ack && !occ.alert.Acknowledged { + occ.alert.Acknowledged, occ.alert.AckTime, occ.alert.AckUser = true, ackTime, ackUser + } + } else if since.IsZero() || alert.StartTime.After(since) { + occurrences[key] = &occurrence{alert: *alert, firstEvent: alert.StartTime} + order = append(order, key) + } + } + sort.SliceStable(order, func(i, j int) bool { + return occurrences[order[i]].firstEvent.Before(occurrences[order[j]].firstEvent) + }) + results := []Alert{} + for i := len(order) - 1; i >= 0; i-- { + if limit > 0 && len(results) >= limit { + break + } + results = append(results, occurrences[order[i]].alert) + } + return m.applyCurrentNodeDisplayNames(canonicalizeAlertHistoryForOutput(results)) +} + +// writeRandomHistory appends a lifecycle history with acknowledgements on +// both sides of window boundaries, out-of-order LastSeen values, uneven +// fractional seconds (so RFC3339Nano text order and time order can differ), +// same-instant events and a history clear. It returns every event time. +func writeRandomHistory(t *testing.T, m *Manager, seed int64) []time.Time { + t.Helper() + rng := rand.New(rand.NewSource(seed)) + store := m.eventLogStore() + base := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) + kinds := []string{ + eventlog.TypeFired, eventlog.TypeRefired, eventlog.TypeAcknowledged, + eventlog.TypeUnacknowledged, eventlog.TypeSnoozed, eventlog.TypeEscalated, eventlog.TypeResolved, + } + fractions := []time.Duration{0, 500 * time.Millisecond, 50 * time.Millisecond, 123456789} + var times []time.Time + at := base + for i := 0; i < 120; i++ { + if rng.Intn(4) != 0 { + at = at.Add(time.Duration(rng.Intn(3)) * time.Second) + } + occurred := at.Add(fractions[rng.Intn(len(fractions))]) + times = append(times, occurred) + if i == 60 { + if err := store.AppendDurable(eventlog.Event{Type: eventlog.TypeHistoryCleared, AlertID: "history", OccurredAt: occurred}); err != nil { + t.Fatal(err) + } + continue + } + id := fmt.Sprintf("alert-%d", rng.Intn(5)) + snapshot := Alert{ + ID: id, + ResourceID: "vm-" + id, + StartTime: base.Add(time.Duration(rng.Intn(3)) * time.Hour), + LastSeen: occurred.Add(time.Duration(rng.Intn(240)-120) * time.Second), + Value: float64(i), + } + kind := kinds[rng.Intn(len(kinds))] + if kind == eventlog.TypeAcknowledged || (kind != eventlog.TypeUnacknowledged && rng.Intn(5) == 0) { + ackTime := occurred + snapshot.Acknowledged, snapshot.AckTime, snapshot.AckUser = true, &ackTime, fmt.Sprintf("user-%d", i) + } + payload, err := json.Marshal(snapshot) + if err != nil { + t.Fatal(err) + } + if err := store.AppendDurable(eventlog.Event{Type: kind, AlertID: id, OccurredAt: occurred, Snapshot: payload}); err != nil { + t.Fatal(err) + } + } + return times +} + +func TestWindowedHistoryMatchesAWalkOfOnlyTheWindow(t *testing.T) { + for seed := int64(1); seed <= 25; seed++ { + t.Run(fmt.Sprintf("seed-%d", seed), func(t *testing.T) { + m := newHistoryParityManager(t) + times := writeRandomHistory(t, m, seed) + // Live overlay: one alert the log knows (with a fresh value) and + // one it never saw fire. + m.mu.Lock() + m.activeAlerts["live-known"] = &Alert{ID: "alert-1", ResourceID: "vm-alert-1", + StartTime: time.Date(2026, 9, 1, 1, 0, 0, 0, time.UTC), LastSeen: times[len(times)-1], Value: 999} + m.activeAlerts["live-new"] = &Alert{ID: "alert-new", ResourceID: "vm-new", + StartTime: times[len(times)/2], LastSeen: times[len(times)-1], Value: 7} + m.mu.Unlock() + + windows := []time.Time{{}, times[0].Add(-time.Hour), times[len(times)-1].Add(time.Hour)} + for i := 0; i < len(times); i += 7 { + windows = append(windows, times[i], times[i].Add(time.Nanosecond), times[i].Add(-time.Nanosecond)) + } + for _, since := range windows { + for _, limit := range []int{0, 1, 3} { + got, ok := m.AlertHistoryFromEvents(since, limit) + if !ok { + t.Fatal("history unavailable") + } + want := referenceWindowedHistory(t, m, since, limit) + if !reflect.DeepEqual(got, want) { + t.Fatalf("since=%s limit=%d: shared fold differs from a walk of only the window\n got: %+v\nwant: %+v", + since.Format(time.RFC3339Nano), limit, got, want) + } + } + } + }) + } +} + +func TestWindowedHistoryReusesTheFullHistoryFold(t *testing.T) { + m := newHistoryParityManager(t) + times := writeRandomHistory(t, m, 7) + if _, ok := m.AlertHistoryFromEvents(time.Time{}, 0); !ok { + t.Fatal("history unavailable") + } + m.historyProjectionMu.Lock() + full := m.historyProjection + m.historyProjectionMu.Unlock() + if full == nil { + t.Fatal("a full-history read did not keep its fold") + } + for _, since := range []time.Time{times[10], times[90], times[len(times)-1]} { + if _, ok := m.AlertHistoryFromEvents(since, 0); !ok { + t.Fatal("history unavailable") + } + m.historyProjectionMu.Lock() + current := m.historyProjection + m.historyProjectionMu.Unlock() + if current != full { + t.Fatalf("a Since read at %s replaced the shared fold instead of reusing it", since) + } + } +} + +func TestHistoryReadResultsDoNotAliasTheSharedFold(t *testing.T) { + m := newHistoryParityManager(t) + times := writeRandomHistory(t, m, 11) + m.mu.Lock() + m.activeAlerts["live-known"] = &Alert{ID: "alert-2", ResourceID: "vm-alert-2", + StartTime: time.Date(2026, 9, 1, 2, 0, 0, 0, time.UTC), LastSeen: times[len(times)-1]} + m.mu.Unlock() + for _, since := range []time.Time{{}, times[30], times[100]} { + first, _ := m.AlertHistoryFromEvents(since, 0) + before := referenceWindowedHistory(t, m, since, 0) + for i := range first { + first[i].Value = -1 + first[i].LastSeen = time.Time{} + if first[i].AckTime != nil { + *first[i].AckTime = time.Time{} + } + } + if again, _ := m.AlertHistoryFromEvents(since, 0); !reflect.DeepEqual(again, before) { + t.Fatalf("since=%s: mutating a result changed the next read", since) + } + } +} + +// A limited read copies only the rows it returns, not the whole fold. +func TestLimitedHistoryReadCopiesOnlyReturnedRows(t *testing.T) { + m := newHistoryParityManager(t) + store := m.eventLogStore() + at := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) + metadata := map[string]interface{}{} + for i := 0; i < 40; i++ { + metadata[fmt.Sprintf("key-%d", i)] = fmt.Sprintf("value-%d-with-some-detail", i) + } + for i := 0; i < 2000; i++ { + occurred := at.Add(time.Duration(i) * time.Minute) + snapshot, err := json.Marshal(Alert{ID: fmt.Sprintf("alert-%d", i), ResourceID: "vm", StartTime: occurred, + LastSeen: occurred, Message: "disk usage above threshold on a long resource name", Metadata: metadata}) + if err != nil { + t.Fatal(err) + } + if err := store.AppendDurable(eventlog.Event{Type: eventlog.TypeFired, AlertID: fmt.Sprintf("alert-%d", i), + OccurredAt: occurred, Snapshot: snapshot}); err != nil { + t.Fatal(err) + } + } + allocated := func(limit int) uint64 { + var before, after runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&before) + if got, ok := m.AlertHistoryFromEvents(time.Time{}, limit); !ok || (limit > 0 && len(got) != limit) { + t.Fatalf("history read with limit %d failed", limit) + } + runtime.ReadMemStats(&after) + return after.TotalAlloc - before.TotalAlloc + } + allocated(0) // build the shared fold + everything, ten := allocated(0), allocated(10) + if ten*5 > everything { + t.Fatalf("a 10-row read allocated %d bytes against %d for all 2000 rows; it must not copy the whole fold", ten, everything) + } +}