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) + } +}