mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-07-09 16:00:59 +00:00
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
This commit is contained in:
rcourtman
parent
d6aed650b1
commit
028e8c8df2
2 changed files with 389 additions and 8 deletions
273
internal/unifiedresources/retention_test.go
Normal file
273
internal/unifiedresources/retention_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
@ -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 {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue