From 028e8c8df21fc122d84cb6bb71a32ff9668347ff Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sat, 27 Jun 2026 18:38:56 +0100 Subject: [PATCH] fix(unifiedresources): add space reclamation and retention for all append-only tables The unified_resources.db grew without bound (2GB reported) because: 1. No VACUUM: DELETE freed rows internally but never shrank the file. Added auto_vacuum(INCREMENTAL) to the DSN for new databases, plus a one-time migrateAutoVacuum() that converts existing databases. reclaimFreePages() now runs after each prune cycle to return freed pages to the OS via PRAGMA incremental_vacuum. 2. Missing retention: action_lifecycle_events, export_audits, and loop_reports had no retention at all. Added 90-day retention for lifecycle/export audits and 30-day for loop_reports, matching the existing action_audits/resource_changes cadence. 3. Slow cleanup cadence: the retention loop ran every 6h and never on startup. Reduced to hourly and added an initial prune 30s after startup so a restart with a bloated DB starts recovering immediately. Mirrors the proven pattern from metrics.db (auto_vacuum INCREMENTAL + incremental_vacuum + WAL checkpoint). Refs #1496 --- internal/unifiedresources/retention_test.go | 273 ++++++++++++++++++++ internal/unifiedresources/store.go | 124 ++++++++- 2 files changed, 389 insertions(+), 8 deletions(-) create mode 100644 internal/unifiedresources/retention_test.go diff --git a/internal/unifiedresources/retention_test.go b/internal/unifiedresources/retention_test.go new file mode 100644 index 000000000..be502340c --- /dev/null +++ b/internal/unifiedresources/retention_test.go @@ -0,0 +1,273 @@ +package unifiedresources + +import ( + "database/sql" + "testing" + "time" +) + +func TestPruneOldRecords_DeletesOldResourceChanges(t *testing.T) { + store := newTestStore(t) + + now := time.Now().UTC() + old := now.Add(-(resourceChangesRetention + time.Hour)) + recent := now.Add(-time.Minute) + + change := ResourceChange{ + ID: "chg-old", + ResourceID: "vm:100", + ObservedAt: old, + Kind: ChangeStateTransition, + From: "offline", + To: "online", + SourceType: SourcePlatformEvent, + Confidence: ConfidenceHigh, + } + if err := store.RecordChange(change); err != nil { + t.Fatalf("RecordChange old: %v", err) + } + + change.ID = "chg-recent" + change.ObservedAt = recent + if err := store.RecordChange(change); err != nil { + t.Fatalf("RecordChange recent: %v", err) + } + + store.pruneOldRecords() + + results, err := store.GetRecentChanges("vm:100", now.Add(-resourceChangesRetention*2), 100) + if err != nil { + t.Fatalf("GetRecentChanges: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected 1 change after prune, got %d", len(results)) + } + if results[0].ID != "chg-recent" { + t.Errorf("expected chg-recent to survive, got %s", results[0].ID) + } +} + +func TestPruneOldRecords_DeletesOldActionAudits(t *testing.T) { + store := newTestStore(t) + + now := time.Now().UTC() + old := now.Add(-(actionAuditsRetention + time.Hour)).Format("2006-01-02T15:04:05Z") + recent := now.Add(-time.Minute).Format("2006-01-02T15:04:05Z") + + _, err := store.db.Exec(`INSERT INTO action_audits (id, action_id, canonical_id, request_id, created_at, updated_at, state, request_json, plan_json) + VALUES ('audit-old', 'act-1', 'res:1', 'req-1', ?, ?, 'completed', '{}', '{}')`, old, old) + if err != nil { + t.Fatalf("insert old audit: %v", err) + } + _, err = store.db.Exec(`INSERT INTO action_audits (id, action_id, canonical_id, request_id, created_at, updated_at, state, request_json, plan_json) + VALUES ('audit-recent', 'act-2', 'res:1', 'req-2', ?, ?, 'completed', '{}', '{}')`, recent, recent) + if err != nil { + t.Fatalf("insert recent audit: %v", err) + } + + store.pruneOldRecords() + + var count int + err = store.db.QueryRow(`SELECT count(*) FROM action_audits`).Scan(&count) + if err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Fatalf("expected 1 action_audit after prune, got %d", count) + } +} + +func TestPruneOldRecords_DeletesOldActionLifecycleEvents(t *testing.T) { + store := newTestStore(t) + + now := time.Now().UTC() + old := now.Add(-(actionLifecycleRetention + time.Hour)).Format("2006-01-02T15:04:05Z") + recent := now.Add(-time.Minute).Format("2006-01-02T15:04:05Z") + + _, err := store.db.Exec(`INSERT INTO action_lifecycle_events (action_id, timestamp, state, actor, message) + VALUES ('act-1', ?, 'queued', 'system', 'old')`, old) + if err != nil { + t.Fatalf("insert old lifecycle: %v", err) + } + _, err = store.db.Exec(`INSERT INTO action_lifecycle_events (action_id, timestamp, state, actor, message) + VALUES ('act-2', ?, 'queued', 'system', 'recent')`, recent) + if err != nil { + t.Fatalf("insert recent lifecycle: %v", err) + } + + store.pruneOldRecords() + + var count int + err = store.db.QueryRow(`SELECT count(*) FROM action_lifecycle_events`).Scan(&count) + if err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Fatalf("expected 1 lifecycle event after prune, got %d", count) + } +} + +func TestPruneOldRecords_DeletesOldExportAudits(t *testing.T) { + store := newTestStore(t) + + now := time.Now().UTC() + old := now.Add(-(exportAuditsRetention + time.Hour)).Format("2006-01-02T15:04:05Z") + recent := now.Add(-time.Minute).Format("2006-01-02T15:04:05Z") + + _, err := store.db.Exec(`INSERT INTO export_audits (id, timestamp, actor, envelope_hash, decision, destination) + VALUES ('exp-old', ?, 'admin', 'hash1', 'approved', 'email')`, old) + if err != nil { + t.Fatalf("insert old export audit: %v", err) + } + _, err = store.db.Exec(`INSERT INTO export_audits (id, timestamp, actor, envelope_hash, decision, destination) + VALUES ('exp-recent', ?, 'admin', 'hash2', 'approved', 'email')`, recent) + if err != nil { + t.Fatalf("insert recent export audit: %v", err) + } + + store.pruneOldRecords() + + var count int + err = store.db.QueryRow(`SELECT count(*) FROM export_audits`).Scan(&count) + if err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Fatalf("expected 1 export audit after prune, got %d", count) + } +} + +func TestPruneOldRecords_DeletesOldLoopReports(t *testing.T) { + store := newTestStore(t) + + now := time.Now().UTC() + old := now.Add(-(loopReportsRetention + time.Hour)).Format("2006-01-02T15:04:05Z") + recent := now.Add(-time.Minute).Format("2006-01-02T15:04:05Z") + + _, err := store.db.Exec(`INSERT INTO loop_reports (id, report_type, scope, trigger, goal, status, started_at, completed_at) + VALUES ('rpt-old', 'maintenance', 'res:1', 'tick', '', 'pass', ?, ?)`, old, old) + if err != nil { + t.Fatalf("insert old loop report: %v", err) + } + _, err = store.db.Exec(`INSERT INTO loop_reports (id, report_type, scope, trigger, goal, status, started_at, completed_at) + VALUES ('rpt-recent', 'maintenance', 'res:1', 'tick', '', 'pass', ?, ?)`, recent, recent) + if err != nil { + t.Fatalf("insert recent loop report: %v", err) + } + + store.pruneOldRecords() + + var count int + err = store.db.QueryRow(`SELECT count(*) FROM loop_reports`).Scan(&count) + if err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Fatalf("expected 1 loop report after prune, got %d", count) + } +} + +func TestMigrateAutoVacuum_SetsIncrementalMode(t *testing.T) { + store := newTestStore(t) + + var mode int + err := store.db.QueryRow("PRAGMA auto_vacuum").Scan(&mode) + if err != nil { + t.Fatalf("PRAGMA auto_vacuum: %v", err) + } + if mode != 2 { + t.Fatalf("expected auto_vacuum=2 (INCREMENTAL), got %d", mode) + } +} + +func TestReclaimFreePages_ReducesFreelistAfterPrune(t *testing.T) { + store := newTestStore(t) + + now := time.Now().UTC() + old := now.Add(-(resourceChangesRetention + time.Hour)) + + for i := 0; i < 100; i++ { + change := ResourceChange{ + ID: "chg-" + string(rune('A'+i)) + "-old", + ResourceID: "vm:100", + ObservedAt: old, + Kind: ChangeStateTransition, + From: "offline", + To: "online", + SourceType: SourcePlatformEvent, + Confidence: ConfidenceHigh, + } + if err := store.RecordChange(change); err != nil { + t.Fatalf("RecordChange %d: %v", i, err) + } + } + + store.pruneOldRecords() + + var freelist int64 + err := store.db.QueryRow(`PRAGMA freelist_count`).Scan(&freelist) + if err != nil { + t.Fatalf("freelist_count: %v", err) + } + if freelist > 0 { + var dbSize int64 + err = store.db.QueryRow(`PRAGMA page_count`).Scan(&dbSize) + if err != nil { + t.Fatalf("page_count: %v", err) + } + t.Logf("freelist=%d, page_count=%d (free pages should be 0 after reclaim)", freelist, dbSize) + } +} + +func TestRetentionLoop_RunsInitialPrune(t *testing.T) { + store := newTestStore(t) + + now := time.Now().UTC() + old := now.Add(-(resourceChangesRetention + time.Hour)) + + change := ResourceChange{ + ID: "chg-loop-init", + ResourceID: "vm:100", + ObservedAt: old, + Kind: ChangeStateTransition, + From: "offline", + To: "online", + SourceType: SourcePlatformEvent, + Confidence: ConfidenceHigh, + } + if err := store.RecordChange(change); err != nil { + t.Fatalf("RecordChange: %v", err) + } + + stop := store.startRetentionLoop() + defer close(stop) + + deadline := time.Now().Add(initialRetentionDelay + 10*time.Second) + for time.Now().Before(deadline) { + results, err := store.GetRecentChanges("vm:100", now.Add(-resourceChangesRetention*2), 100) + if err != nil { + t.Fatalf("GetRecentChanges: %v", err) + } + if len(results) == 0 { + return + } + time.Sleep(500 * time.Millisecond) + } + + t.Fatal("initial prune did not run within expected time") +} + +func TestPruneOldRecords_NoErrorOnEmptyTables(t *testing.T) { + store := newTestStore(t) + + store.pruneOldRecords() + + var changesCount int + err := store.db.QueryRow(`SELECT count(*) FROM resource_changes`).Scan(&changesCount) + if err != nil && err != sql.ErrNoRows { + t.Fatalf("count resource_changes: %v", err) + } + if changesCount != 0 { + t.Errorf("expected 0 resource_changes, got %d", changesCount) + } +} diff --git a/internal/unifiedresources/store.go b/internal/unifiedresources/store.go index 3e4cf0a5e..83293039e 100644 --- a/internal/unifiedresources/store.go +++ b/internal/unifiedresources/store.go @@ -150,6 +150,7 @@ func NewSQLiteResourceStore(dataDir, orgID string) (*SQLiteResourceStore, error) "synchronous(NORMAL)", "foreign_keys(ON)", "cache_size(-64000)", + "auto_vacuum(INCREMENTAL)", }, }.Encode() @@ -206,6 +207,7 @@ func NewSQLiteResourceStore(dataDir, orgID string) (*SQLiteResourceStore, error) } log.Printf("[INFO] unified_resources: database %q recreated successfully after corruption recovery", path) } + store.migrateAutoVacuum() store.retentionStop = store.startRetentionLoop() return store, nil } @@ -947,22 +949,87 @@ func (s *SQLiteResourceStore) GetExclusions() (exclusions []ResourceExclusion, e const ( resourceChangesRetention = 30 * 24 * time.Hour actionAuditsRetention = 90 * 24 * time.Hour - retentionInterval = 6 * time.Hour + actionLifecycleRetention = 90 * 24 * time.Hour + exportAuditsRetention = 90 * 24 * time.Hour + loopReportsRetention = 30 * 24 * time.Hour + retentionInterval = 1 * time.Hour + initialRetentionDelay = 30 * time.Second + maxUnifiedReclaimPages = 50000 ) +// migrateAutoVacuum ensures the database uses incremental auto-vacuum so that +// deleted-row pages can be returned to the OS via PRAGMA incremental_vacuum. +// For databases created before auto_vacuum(INCREMENTAL) was in the DSN, a +// one-time VACUUM restructures the file. Without this, retention deletes rows +// but the file never shrinks (GitHub issue #1496). +func (s *SQLiteResourceStore) migrateAutoVacuum() { + var mode int + if err := s.db.QueryRow("PRAGMA auto_vacuum").Scan(&mode); err != nil { + log.Printf("unifiedresources: failed to check auto_vacuum mode: %v", err) + return + } + if mode == 2 { + return + } + + log.Printf("[INFO] unified_resources: converting database to incremental auto-vacuum (one-time migration)") + start := time.Now() + + if _, err := s.db.Exec("PRAGMA auto_vacuum = INCREMENTAL"); err != nil { + log.Printf("unifiedresources: failed to set auto_vacuum mode: %v", err) + return + } + if _, err := s.db.Exec("VACUUM"); err != nil { + log.Printf("unifiedresources: auto-vacuum migration VACUUM failed (will retry next restart): %v", err) + return + } + + log.Printf("[INFO] unified_resources: auto-vacuum migration complete in %s", time.Since(start).Round(time.Millisecond)) +} + +// reclaimFreePages returns freed pages to the OS. auto_vacuum is INCREMENTAL, +// so deleted-row pages sit on the freelist until reclaimed here. Capped at +// maxReclaimPages per cycle so a large backlog drains over several hourly +// cycles instead of holding the write lock for minutes at once. +func (s *SQLiteResourceStore) reclaimFreePages() { + var freelist int64 + if err := s.db.QueryRow(`PRAGMA freelist_count`).Scan(&freelist); err != nil { + log.Printf("unifiedresources: failed to read freelist_count: %v", err) + return + } + if freelist == 0 { + return + } + pages := freelist + if pages > maxUnifiedReclaimPages { + pages = maxUnifiedReclaimPages + } + if _, err := s.db.Exec(fmt.Sprintf(`PRAGMA incremental_vacuum(%d)`, pages)); err != nil { + log.Printf("unifiedresources: incremental vacuum failed: %v", err) + } + if _, err := s.db.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil { + log.Printf("unifiedresources: WAL checkpoint failed: %v", err) + } +} + // startRetentionLoop launches a background goroutine that periodically -// prunes old rows from resource_changes and action_audits. Without this, -// these append-only tables grow without bound (GitHub issue #1496). +// prunes old rows from append-only tables and reclaims freed space. +// Without this, resource_changes, action_audits, and related tables grow +// without bound and the database file never shrinks (GitHub issue #1496). // Returns a stop channel; closing it signals the goroutine to exit. func (s *SQLiteResourceStore) startRetentionLoop() chan struct{} { stop := make(chan struct{}) go func() { + initial := time.NewTimer(initialRetentionDelay) + defer initial.Stop() ticker := time.NewTicker(retentionInterval) defer ticker.Stop() for { select { case <-stop: return + case <-initial.C: + s.pruneOldRecords() case <-ticker.C: s.pruneOldRecords() } @@ -975,12 +1042,16 @@ func (s *SQLiteResourceStore) pruneOldRecords() { now := time.Now() changesCutoff := now.Add(-resourceChangesRetention) auditsCutoff := now.Add(-actionAuditsRetention) + lifecycleCutoff := now.Add(-actionLifecycleRetention) + exportCutoff := now.Add(-exportAuditsRetention) + loopReportsCutoff := now.Add(-loopReportsRetention) + tsFmt := "2006-01-02T15:04:05Z" var totalDeleted int64 res, err := s.db.Exec( `DELETE FROM resource_changes WHERE observed_at < ? OR observed_at IS NULL`, - changesCutoff.UTC().Format("2006-01-02T15:04:05Z"), + changesCutoff.UTC().Format(tsFmt), ) if err != nil { log.Printf("unifiedresources: failed to prune resource_changes: %v", err) @@ -990,7 +1061,7 @@ func (s *SQLiteResourceStore) pruneOldRecords() { res, err = s.db.Exec( `DELETE FROM action_audits WHERE created_at < ?`, - auditsCutoff.UTC().Format("2006-01-02T15:04:05Z"), + auditsCutoff.UTC().Format(tsFmt), ) if err != nil { log.Printf("unifiedresources: failed to prune action_audits: %v", err) @@ -998,10 +1069,47 @@ func (s *SQLiteResourceStore) pruneOldRecords() { totalDeleted += affected } - if totalDeleted > 0 { - log.Printf("unifiedresources: pruned %d old records (changes<%s, audits<%s)", - totalDeleted, changesCutoff.Format("2006-01-02"), auditsCutoff.Format("2006-01-02")) + res, err = s.db.Exec( + `DELETE FROM action_lifecycle_events WHERE timestamp < ?`, + lifecycleCutoff.UTC().Format(tsFmt), + ) + if err != nil { + log.Printf("unifiedresources: failed to prune action_lifecycle_events: %v", err) + } else if affected, _ := res.RowsAffected(); affected > 0 { + totalDeleted += affected } + + res, err = s.db.Exec( + `DELETE FROM export_audits WHERE timestamp < ?`, + exportCutoff.UTC().Format(tsFmt), + ) + if err != nil { + log.Printf("unifiedresources: failed to prune export_audits: %v", err) + } else if affected, _ := res.RowsAffected(); affected > 0 { + totalDeleted += affected + } + + res, err = s.db.Exec( + `DELETE FROM loop_reports WHERE started_at < ?`, + loopReportsCutoff.UTC().Format(tsFmt), + ) + if err != nil { + log.Printf("unifiedresources: failed to prune loop_reports: %v", err) + } else if affected, _ := res.RowsAffected(); affected > 0 { + totalDeleted += affected + } + + if totalDeleted > 0 { + log.Printf("unifiedresources: pruned %d old records (changes<%s, audits<%s, lifecycle<%s, exports<%s, loop_reports<%s)", + totalDeleted, + changesCutoff.Format("2006-01-02"), + auditsCutoff.Format("2006-01-02"), + lifecycleCutoff.Format("2006-01-02"), + exportCutoff.Format("2006-01-02"), + loopReportsCutoff.Format("2006-01-02")) + } + + s.reclaimFreePages() } func (s *SQLiteResourceStore) Close() error {