fix(alerts): bound alert event log snapshot volume

A permanently-offline derived identity flapped a docker-host alert and wrote 4,748 resolved/refired pairs in three days; each lifecycle event stores a full alert-state snapshot averaging ~170 KB, so events.db reached 1.2 GB and filled the host root filesystem (#2113 downstream evidence). The 90-day age retention is not a volume bound: one alert can still write unbounded payload inside the window.

The hourly prune now also enforces a total stored snapshot-payload cap and removes the oldest events first when it is exceeded, raising the same retention revision age pruning uses so the history projection rebuilds from the remaining log. Normal installs stay far below the cap and are unaffected; the bound only triggers on pathological amplification. Focused proof in eventlog_snapshot_test.go; the alerts subsystem contract records the bound.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-09-20 22:06:17 +01:00
parent 9a99f9948f
commit 1c8fee0119
3 changed files with 157 additions and 3 deletions
@@ -1420,7 +1420,12 @@ evidence. Failed asynchronous batches do not advance the successful-write
counter, and `Flush` must report the failure or timeout rather than falsely
claiming the batch landed. An unavailable or failed lifecycle store degrades
history reads to the recovery model without hiding live active-alert truth.
Events retain for 90 days and prune hourly. Fired and
Events retain for 90 days and prune hourly. Retention alone does not bound disk
use: one flapping alert can write thousands of full alert-state snapshots inside
the window, so the hourly prune also enforces a total stored snapshot-payload
cap and removes the oldest events first when it is exceeded. A volume prune
raises the same retention revision as age pruning, so the history projection
rebuilds from the remaining log rather than reusing a stale fold. Fired and
refired lifecycle events come only from the reducer core's explicit activation
events: canonical lifecycle reactivation maps `EventRefired` separately, while
shared metric activation records one fired event when its pending incident
+91 -2
View File
@@ -103,6 +103,14 @@ const (
defaultRetention = 90 * 24 * time.Hour
pruneInterval = time.Hour
eventLogFileName = "events.db"
// maxRetainedSnapshotBytes bounds the total lifecycle-snapshot payload the
// log keeps. Retention bounds how long an event lives; a single flapping
// alert can still write hundreds of megabytes of full state snapshots
// inside that window and fill the host disk. When the stored payload
// exceeds this cap the oldest events are removed first, exactly as age
// pruning removes them, so the history projection rebuilds from what
// remains.
maxRetainedSnapshotBytes = 256 << 20
)
// Store is the SQLite-backed event log. A nil *Store is valid: Append and
@@ -612,10 +620,91 @@ func (s *Store) forgetFailedDeliveryEpisodes(events []Event) {
}
func (s *Store) pruneOld() {
if s.retention <= 0 {
if s.retention > 0 {
s.pruneEventsBefore(time.Now().Add(-s.retention))
}
s.pruneSnapshotVolume(maxRetainedSnapshotBytes)
}
// pruneSnapshotVolume removes the oldest events until the total stored snapshot
// payload is within maxBytes. Age retention alone cannot bound disk use: one
// flapping alert can write thousands of full state snapshots inside the
// retention window. Deleting the oldest rows raises the retention revision so
// the history projection rebuilds from the remaining log.
func (s *Store) pruneSnapshotVolume(maxBytes int64) {
if s == nil || maxBytes <= 0 {
return
}
s.pruneEventsBefore(time.Now().Add(-s.retention))
var total int64
if err := s.db.QueryRow(`SELECT COALESCE(SUM(length(snapshot)), 0) FROM alert_events`).Scan(&total); err != nil {
log.Error().Err(err).Msg("alert event log volume check failed")
return
}
if total <= maxBytes {
return
}
// Read only ids and payload lengths, oldest first, until enough bytes are
// scheduled for removal. The cursor must be closed before deleting because
// the store serializes writes on a single connection.
rows, err := s.db.Query(`SELECT id, length(snapshot) FROM alert_events ORDER BY occurred_at ASC, id ASC`)
if err != nil {
log.Error().Err(err).Msg("alert event log volume scan failed")
return
}
ids := make([]int64, 0)
for total > maxBytes && rows.Next() {
var id, size int64
if scanErr := rows.Scan(&id, &size); scanErr != nil {
rows.Close()
log.Error().Err(scanErr).Msg("alert event log volume scan failed")
return
}
ids = append(ids, id)
total -= size
}
scanErr := rows.Err()
rows.Close()
if scanErr != nil {
log.Error().Err(scanErr).Msg("alert event log volume scan failed")
return
}
if len(ids) == 0 {
return
}
s.deleteEventsByID(ids)
}
func (s *Store) deleteEventsByID(ids []int64) {
tx, err := s.db.Begin()
if err != nil {
log.Error().Err(err).Msg("alert event log volume prune failed")
return
}
defer tx.Rollback()
for start := 0; start < len(ids); start += 500 {
end := start + 500
if end > len(ids) {
end = len(ids)
}
placeholders := make([]string, 0, end-start)
args := make([]any, 0, end-start)
for _, id := range ids[start:end] {
placeholders = append(placeholders, "?")
args = append(args, id)
}
if _, err = tx.Exec("DELETE FROM alert_events WHERE id IN ("+strings.Join(placeholders, ",")+")", args...); err != nil {
log.Error().Err(err).Msg("alert event log volume prune failed")
return
}
}
if _, err = tx.Exec(`INSERT INTO alert_store_meta (key, value) VALUES ('retention_revision', '1')
ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1`); err != nil {
log.Error().Err(err).Msg("alert event log volume prune failed")
return
}
if err = tx.Commit(); err != nil {
log.Error().Err(err).Msg("alert event log volume prune failed")
}
}
func (s *Store) pruneEventsBefore(cutoff time.Time) {
@@ -1,8 +1,10 @@
package eventlog
import (
"bytes"
"database/sql"
"encoding/json"
"fmt"
"path/filepath"
"sync"
"testing"
@@ -312,3 +314,61 @@ func TestConcurrentUnchangedDeliveryDecisionAdmitsOneEpisode(t *testing.T) {
t.Fatalf("concurrent reevaluations wrote %d episodes, want one", len(events))
}
}
func TestSnapshotVolumeBoundPrunesOldestAndAdvancesRetention(t *testing.T) {
store := newTestStore(t)
base := time.Now().Add(-time.Hour).UTC()
const eventCount = 8
payload := bytes.Repeat([]byte("x"), 1024)
for i := 0; i < eventCount; i++ {
snapshot, err := json.Marshal(map[string]any{"id": fmt.Sprintf("a%d", i), "blob": string(payload)})
if err != nil {
t.Fatal(err)
}
if err := store.AppendDurable(Event{
OccurredAt: base.Add(time.Duration(i) * time.Minute),
Type: TypeFired,
AlertID: fmt.Sprintf("a%d", i),
Snapshot: snapshot,
}); err != nil {
t.Fatal(err)
}
}
before, err := store.ReplayBoundary()
if err != nil {
t.Fatal(err)
}
// Keep roughly three events' payload; the older rows must be removed.
store.pruneSnapshotVolume(3 * 1200)
after, err := store.ReplayBoundary()
if err != nil {
t.Fatal(err)
}
if after.RetentionRevision <= before.RetentionRevision {
t.Fatalf("volume prune did not advance retention revision: before=%+v after=%+v", before, after)
}
if after.LastID != before.LastID {
t.Fatalf("volume prune changed the newest id: before=%+v after=%+v", before, after)
}
remaining, err := store.Query(Filter{})
if err != nil {
t.Fatal(err)
}
if len(remaining) == 0 || len(remaining) >= eventCount {
t.Fatalf("volume prune left %d of %d events", len(remaining), eventCount)
}
for _, event := range remaining {
if event.AlertID == "a0" {
t.Fatalf("oldest event survived volume prune: %+v", remaining)
}
}
// A second prune already within the cap must not change the boundary.
store.pruneSnapshotVolume(3 * 1200)
again, err := store.ReplayBoundary()
if err != nil {
t.Fatal(err)
}
if again != after {
t.Fatalf("prune within the cap changed the boundary: %+v -> %+v", after, again)
}
}