diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 6495996e2..90d8ad097 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -232,6 +232,45 @@ type RetentionConfig struct { // coverage rows in client_receptions / client_observers; 0 disables. Bounds // the table the opt-in coverage feature would otherwise grow without limit. ClientRxDays int `json:"clientRxDays"` + // The opt-in windows below cover the tables nothing else prunes (#329); + // 0 disables each. See TableRetention. + InactiveNodeDays int `json:"inactiveNodeDays"` + NodeChangeDays int `json:"nodeChangeDays"` + ObserverPurgeDays int `json:"observerPurgeDays"` +} + +// TableRetention holds the opt-in retention windows, in days, of the tables +// nothing else prunes (#329). 0 disables a window, so an instance that sets +// none of them keeps every row, as before. ping_triggers has no window and is +// kept forever: the all-time Ping Scores records join it with the history +// sidecar (#349, #241). +type TableRetention struct { + // InactiveNodeDays deletes inactive_nodes rows whose last advert is + // older, once the node has not come back. + InactiveNodeDays int + // NodeChangeDays deletes node_changes rows detected longer ago. + NodeChangeDays int + // ObserverPurgeDays hard-deletes observers that observerDays already + // soft-deleted, once nothing still references them (upstream#1886). + ObserverPurgeDays int +} + +// Enabled reports whether any window is set. +func (r TableRetention) Enabled() bool { + return r.InactiveNodeDays > 0 || r.NodeChangeDays > 0 || r.ObserverPurgeDays > 0 +} + +// TableRetention returns the configured opt-in table retention windows, +// with unset and negative values as 0 (disabled). +func (c *Config) TableRetention() TableRetention { + if c.Retention == nil { + return TableRetention{} + } + return TableRetention{ + InactiveNodeDays: max(c.Retention.InactiveNodeDays, 0), + NodeChangeDays: max(c.Retention.NodeChangeDays, 0), + ObserverPurgeDays: max(c.Retention.ObserverPurgeDays, 0), + } } // PacketDaysOrZero returns the configured retention.packetDays or 0 diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 7d99e88f6..d5dfb1c2a 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -268,6 +268,17 @@ func main() { } } + // Opt-in retention for the tables nothing else prunes (#329): after the + // observer soft-delete and the metrics and transmission prunes above, so + // the observer purge's reference guards see what they left. Every window + // unset (the default) = nothing deleted. + tableRetention := cfg.TableRetention() + if r := tableRetention; r.Enabled() { + log.Printf("[prune] table retention enabled (0 = off): inactiveNodeDays=%d nodeChangeDays=%d observerPurgeDays=%d", + r.InactiveNodeDays, r.NodeChangeDays, r.ObserverPurgeDays) + } + runTableRetention(store, tableRetention, "startup") + vacuumPages := cfg.IncrementalVacuumPages() store.RunIncrementalVacuum(vacuumPages) @@ -308,14 +319,18 @@ func main() { } }() - // Daily ticker for observer retention (every 24h, staggered 90s after startup) + // Daily ticker for observer retention (every 24h, staggered 90s after + // startup), followed by the opt-in table retention (#329), whose observer + // purge must follow the soft-delete. observerRetentionTicker := time.NewTicker(24 * time.Hour) go func() { time.Sleep(90 * time.Second) // stagger after metrics prune store.RemoveStaleObservers(observerDays) + runTableRetention(store, tableRetention, "daily") store.RunIncrementalVacuum(vacuumPages) for range observerRetentionTicker.C { store.RemoveStaleObservers(observerDays) + runTableRetention(store, tableRetention, "daily") store.RunIncrementalVacuum(vacuumPages) } }() diff --git a/cmd/ingestor/observer_purge_test.go b/cmd/ingestor/observer_purge_test.go new file mode 100644 index 000000000..3b00773e3 --- /dev/null +++ b/cmd/ingestor/observer_purge_test.go @@ -0,0 +1,322 @@ +package main + +import ( + "testing" + "time" +) + +// Tests for retention.observerPurgeDays (#329), ported from upstream#1886 +// (#34) and extended with the observer_neighbors / observer_neighbor_metrics +// rows that port did not cover. + +// staleObserver inserts an observer already soft-deleted (inactive = 1) with +// last_seen 90 days in the past, and returns its rowid. +func staleObserver(t *testing.T, s *Store, id string) int64 { + t.Helper() + if err := s.UpsertObserver(id, id, "LAX", nil); err != nil { + t.Fatal(err) + } + old := time.Now().UTC().AddDate(0, 0, -90).Format(time.RFC3339) + if _, err := s.db.Exec(`UPDATE observers SET last_seen = ?, inactive = 1 WHERE id = ?`, old, id); err != nil { + t.Fatal(err) + } + var rowid int64 + if err := s.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, id).Scan(&rowid); err != nil { + t.Fatal(err) + } + return rowid +} + +func observerExists(t *testing.T, s *Store, id string) bool { + t.Helper() + var n int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, id).Scan(&n); err != nil { + t.Fatal(err) + } + return n > 0 +} + +func TestPurgeStaleObserversDeletesUnreferencedRow(t *testing.T) { + store := newTestStore(t) + staleObserver(t, store, "obs-gone") + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 1 { + t.Errorf("purged=%d, want 1", purged) + } + if observerExists(t, store, "obs-gone") { + t.Error("obs-gone still present, want hard-deleted") + } +} + +// Regression: an observer whose rowid is still referenced by observations must +// survive the purge — deleting it orphans observations.observer_idx, which +// breaks the packets_v join and mis-attributes historical packets. +func TestPurgeStaleObserversKeepsObserverWithObservations(t *testing.T) { + store := newTestStore(t) + rowid := staleObserver(t, store, "obs-referenced") + + if _, err := store.db.Exec( + `INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (1, 'AA', 'h1', ?)`, + time.Now().UTC().Format(time.RFC3339), + ); err != nil { + t.Fatal(err) + } + if _, err := store.db.Exec( + `INSERT INTO observations (transmission_id, observer_idx, timestamp) VALUES (1, ?, 0)`, rowid, + ); err != nil { + t.Fatal(err) + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 0 { + t.Errorf("purged=%d, want 0 (row is referenced by observations)", purged) + } + if !observerExists(t, store, "obs-referenced") { + t.Error("obs-referenced was deleted, want kept") + } + + var orphans int + if err := store.db.QueryRow( + `SELECT COUNT(*) FROM observations WHERE observer_idx IS NOT NULL + AND observer_idx NOT IN (SELECT rowid FROM observers)`, + ).Scan(&orphans); err != nil { + t.Fatal(err) + } + if orphans != 0 { + t.Errorf("orphaned observations=%d, want 0", orphans) + } +} + +func TestPurgeStaleObserversKeepsObserverWithMetrics(t *testing.T) { + store := newTestStore(t) + staleObserver(t, store, "obs-metrics") + + if _, err := store.db.Exec( + `INSERT INTO observer_metrics (observer_id, timestamp) VALUES ('obs-metrics', ?)`, + time.Now().UTC().Format(time.RFC3339), + ); err != nil { + t.Fatal(err) + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 0 { + t.Errorf("purged=%d, want 0 (row is referenced by observer_metrics)", purged) + } + if !observerExists(t, store, "obs-metrics") { + t.Error("obs-metrics was deleted, want kept") + } +} + +func TestPurgeStaleObserversKeepsObserverWithDroppedPackets(t *testing.T) { + store := newTestStore(t) + staleObserver(t, store, "obs-dropped") + + if _, err := store.db.Exec( + `INSERT INTO dropped_packets (reason, observer_id) VALUES ('bad-sig', 'obs-dropped')`, + ); err != nil { + t.Fatal(err) + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 0 { + t.Errorf("purged=%d, want 0 (row is referenced by dropped_packets)", purged) + } + if !observerExists(t, store, "obs-dropped") { + t.Error("obs-dropped was deleted, want kept") + } +} + +// An observer old enough to purge but not yet soft-deleted must be left to +// RemoveStaleObservers — the hard purge only ever finalises an inactive row. +func TestPurgeStaleObserversKeepsActiveObserver(t *testing.T) { + store := newTestStore(t) + if err := store.UpsertObserver("obs-active", "Active", "LAX", nil); err != nil { + t.Fatal(err) + } + old := time.Now().UTC().AddDate(0, 0, -90).Format(time.RFC3339) + if _, err := store.db.Exec(`UPDATE observers SET last_seen = ? WHERE id = ?`, old, "obs-active"); err != nil { + t.Fatal(err) + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 0 { + t.Errorf("purged=%d, want 0 (inactive = 0)", purged) + } + if !observerExists(t, store, "obs-active") { + t.Error("obs-active was deleted, want kept") + } +} + +func TestPurgeStaleObserversKeepsObserverInsideWindow(t *testing.T) { + store := newTestStore(t) + if err := store.UpsertObserver("obs-recent", "Recent", "LAX", nil); err != nil { + t.Fatal(err) + } + recent := time.Now().UTC().AddDate(0, 0, -20).Format(time.RFC3339) + if _, err := store.db.Exec( + `UPDATE observers SET last_seen = ?, inactive = 1 WHERE id = ?`, recent, "obs-recent", + ); err != nil { + t.Fatal(err) + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 0 { + t.Errorf("purged=%d, want 0 (last_seen inside the 30-day window)", purged) + } + if !observerExists(t, store, "obs-recent") { + t.Error("obs-recent was deleted, want kept") + } +} + +func TestPurgeStaleObserversDisabled(t *testing.T) { + store := newTestStore(t) + staleObserver(t, store, "obs-kept") + + for _, days := range []int{0, -1} { + purged, err := store.PurgeStaleObservers(days) + if err != nil { + t.Fatal(err) + } + if purged != 0 { + t.Errorf("PurgeStaleObservers(%d) purged=%d, want 0 (disabled)", days, purged) + } + } + if !observerExists(t, store, "obs-kept") { + t.Error("obs-kept was deleted, want kept while purge is disabled") + } +} + +// observer_neighbors is a current-only snapshot that only the observer itself +// replaces, so a purged observer would otherwise serve its last neighbour +// list forever (GetObserverNeighbors does not join observers). The purge +// deletes it in the same transaction as the observer row. +func TestPurgeStaleObserversDeletesNeighborSnapshot(t *testing.T) { + store := newTestStore(t) + staleObserver(t, store, "obs-nbr") + staleObserver(t, store, "obs-nbr-kept") + if _, err := store.db.Exec( + `INSERT INTO observer_metrics (observer_id, timestamp) VALUES ('obs-nbr-kept', ?)`, + time.Now().UTC().Format(time.RFC3339), + ); err != nil { + t.Fatal(err) + } + for _, id := range []string{"obs-nbr", "obs-nbr-kept"} { + if _, err := store.db.Exec( + `INSERT INTO observer_neighbors (observer_id, neighbor_pubkey, status, reported_at) VALUES (?, 'aabb', 'ok', ?)`, + id, time.Now().UTC().AddDate(0, 0, -90).Format(time.RFC3339), + ); err != nil { + t.Fatal(err) + } + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 1 { + t.Errorf("purged=%d, want 1", purged) + } + if observerExists(t, store, "obs-nbr") { + t.Error("obs-nbr still present, want hard-deleted") + } + var n int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM observer_neighbors WHERE observer_id = 'obs-nbr'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Errorf("observer_neighbors rows of the purged observer = %d, want 0", n) + } + if err := store.db.QueryRow(`SELECT COUNT(*) FROM observer_neighbors WHERE observer_id = 'obs-nbr-kept'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Errorf("observer_neighbors rows of a kept observer = %d, want 1", n) + } +} + +// observer_neighbor_metrics is history that ages out by metricsDays, like +// observer_metrics, so it guards the observer the same way. +func TestPurgeStaleObserversKeepsObserverWithNeighborMetrics(t *testing.T) { + store := newTestStore(t) + staleObserver(t, store, "obs-nbr-metrics") + if _, err := store.db.Exec( + `INSERT INTO observer_neighbor_metrics (observer_id, neighbor_pubkey, timestamp, snr) VALUES ('obs-nbr-metrics', 'aabb', ?, 5)`, + time.Now().UTC().Format(time.RFC3339), + ); err != nil { + t.Fatal(err) + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 0 { + t.Errorf("purged=%d, want 0 (row is referenced by observer_neighbor_metrics)", purged) + } + if !observerExists(t, store, "obs-nbr-metrics") { + t.Error("obs-nbr-metrics was deleted, want kept") + } +} + +// More candidates than one batch: every one is purged, over several +// transactions. +func TestPurgeStaleObserversBatches(t *testing.T) { + setRetentionBatchRows(t, 2) + store := newTestStore(t) + for _, id := range []string{"obs-b1", "obs-b2", "obs-b3", "obs-b4", "obs-b5"} { + staleObserver(t, store, id) + } + ResetWriterStatsForTest() + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 5 { + t.Errorf("purged=%d, want 5", purged) + } + if got := countRows(t, store, "observers"); got != 0 { + t.Errorf("observers left = %d, want 0", got) + } + if got := store.WriterStatsSnapshot()["purge_observers"].Count; got != 3 { + t.Errorf("purge_observers transactions = %d, want 3 (batches of 2, 2, 1)", got) + } +} + +// SQLite lets a TEXT PRIMARY KEY hold NULL. Such a row is never a candidate, +// and it does not stop the purge of the others. +func TestPurgeStaleObserversSkipsNullID(t *testing.T) { + store := newTestStore(t) + staleObserver(t, store, "obs-real") + if _, err := store.db.Exec(`INSERT INTO observers (id, name, last_seen, inactive) VALUES (NULL, 'null-id', ?, 1)`, + time.Now().UTC().AddDate(0, 0, -90).Format(time.RFC3339)); err != nil { + t.Fatal(err) + } + + purged, err := store.PurgeStaleObservers(30) + if err != nil { + t.Fatal(err) + } + if purged != 1 || observerExists(t, store, "obs-real") { + t.Errorf("purged=%d, obs-real present=%v; want 1 and false", purged, observerExists(t, store, "obs-real")) + } +} diff --git a/cmd/ingestor/table_retention.go b/cmd/ingestor/table_retention.go new file mode 100644 index 000000000..ed295454b --- /dev/null +++ b/cmd/ingestor/table_retention.go @@ -0,0 +1,217 @@ +package main + +import ( + "database/sql" + "fmt" + "log" + "strings" + "time" +) + +// Opt-in retention for the tables nothing else prunes (#329): inactive_nodes, +// node_changes and soft-deleted observers. Each window is in days and 0 +// disables it, so an instance that sets none keeps every row, as before. Like +// pruneBatches, every delete runs in bounded WriterTx batches, so ingest waits +// for at most one batch. ping_triggers is deliberately left out and kept +// forever: the all-time Ping Scores records join it with the history sidecar +// (#349, #241), and an entry leaves them once its trigger is gone. + +// retentionBatchRows bounds the rows one table-retention transaction deletes. +// These rows are small and their indexes few, so a batch is a few +// milliseconds. A var so tests can exercise several batches. +var retentionBatchRows = 1000 + +// deleteInBatches runs del, a DELETE whose last parameter is the batch LIMIT, +// in WriterTx batches tagged tag until one deletes fewer rows than the limit. +// args are del's parameters before the LIMIT. On error the rows of the +// already-committed batches are returned alongside it. +func (s *Store) deleteInBatches(tag, del string, args ...any) (int64, error) { + params := append(append(make([]any, 0, len(args)+1), args...), retentionBatchRows) + var total int64 + for { + var n int64 + err := s.WriterTx(tag, func(tx *sql.Tx) error { + res, err := tx.Exec(del, params...) + if err != nil { + return err + } + n, _ = res.RowsAffected() + return nil + }) + if err != nil { + return total, fmt.Errorf("%s: %w", tag, err) + } + total += n + if n < int64(retentionBatchRows) { + return total, nil + } + } +} + +func retentionCutoff(days int) string { + return time.Now().UTC().AddDate(0, 0, -days).Format(time.RFC3339) +} + +// pruneInactiveNodesBatch deletes the next batch of inactive_nodes rows whose +// last advert is older than the cutoff (?1). It skips a node that came back: +// its old row stays next to the new nodes row (MoveStaleNodes does not delete +// it), still carries the node's confirmed default_scope evidence +// (UpdateNodeDefaultScope) and keeps the node out of New Nodes. It also skips +// the node of an observer seen since the cutoff, which MoveStaleNodes would +// not retire (#199; see staleNodesWhere). The walk is over +// idx_inactive_nodes_last_seen; the rows it steps over are bounded by the +// nodes and observers tables. +const pruneInactiveNodesBatch = `DELETE FROM inactive_nodes WHERE public_key IN ( + SELECT i.public_key FROM inactive_nodes i + WHERE i.last_seen < ?1 + AND NOT EXISTS (SELECT 1 FROM nodes n WHERE n.public_key = i.public_key) + AND lower(i.public_key) NOT IN ( + SELECT lower(id) FROM observers WHERE id IS NOT NULL AND last_seen >= ?1) + LIMIT ?2)` + +// PruneInactiveNodes deletes inactive_nodes rows not seen for days +// (retention.inactiveNodeDays, #329); days <= 0 disables it. A deleted node +// that adverts again later is a new node: no "resurrected" node_changes row, +// and it shows in New Nodes. +func (s *Store) PruneInactiveNodes(days int) (int64, error) { + if days <= 0 { + return 0, nil + } + n, err := s.deleteInBatches("prune_inactive_nodes", pruneInactiveNodesBatch, retentionCutoff(days)) + if n > 0 { + log.Printf("[prune] deleted %d inactive_nodes not seen in %d days", n, days) + } + return n, err +} + +// PruneNodeChanges deletes node_changes rows detected more than days ago +// (retention.nodeChangeDays, #329), oldest first over +// idx_node_changes_detected_at; days <= 0 disables it. +func (s *Store) PruneNodeChanges(days int) (int64, error) { + if days <= 0 { + return 0, nil + } + n, err := s.deleteInBatches("prune_node_changes", + `DELETE FROM node_changes WHERE id IN ( + SELECT id FROM node_changes WHERE detected_at < ? ORDER BY detected_at LIMIT ?)`, + retentionCutoff(days)) + if n > 0 { + log.Printf("[prune] deleted %d node_changes older than %d days", n, days) + } + return n, err +} + +// purgeObserverCandidates selects the next batch of observers to hard-delete: +// soft-deleted (inactive = 1) by RemoveStaleObservers, not seen since the +// cutoff (?1), and referenced by no row that outlives them by design. +// +// observations.observer_idx is a bare rowid with no foreign key, so deleting a +// still-referenced observer silently orphans history: packets_v stops +// resolving the observer and the packets are mis-attributed. observer_metrics, +// observer_neighbor_metrics and dropped_packets are keyed by observer id and +// age out by metricsDays. These four guards are what make the delete safe; +// each is an index seek per candidate (observers is O(100) rows), so a batch +// stays cheap. observer_neighbors is not a guard: it is a current-only +// snapshot that only the observer itself replaces, so it would keep a dead +// observer forever. PurgeStaleObservers deletes it with the observer instead. +const purgeObserverCandidates = `SELECT id FROM observers + WHERE inactive = 1 + AND last_seen < ?1 + AND id IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM observations o WHERE o.observer_idx = observers.rowid) + AND NOT EXISTS (SELECT 1 FROM observer_metrics m WHERE m.observer_id = observers.id) + AND NOT EXISTS (SELECT 1 FROM observer_neighbor_metrics nm WHERE nm.observer_id = observers.id) + AND NOT EXISTS (SELECT 1 FROM dropped_packets d WHERE d.observer_id = observers.id) + LIMIT ?2` + +// PurgeStaleObservers hard-deletes observers RemoveStaleObservers already +// soft-deleted, once they are older than purgeDays and nothing references them +// any more (retention.observerPurgeDays; upstream#1886, #329). It is the +// second stage of observer retention: the soft-delete hides the observer, this +// reclaims the row after its packets and metrics have aged out. purgeDays <= 0 +// disables it (the default). +// +// Each batch reads its candidates once and deletes exactly those ids, with +// their observer_neighbors rows, in the same transaction. +func (s *Store) PurgeStaleObservers(purgeDays int) (int64, error) { + if purgeDays <= 0 { + return 0, nil + } + cutoff := retentionCutoff(purgeDays) + var total int64 + for { + var n int + // Tagged for /api/perf writer-lock visibility (#1340). + err := s.WriterTx("purge_observers", func(tx *sql.Tx) error { + ids, err := queryStrings(tx, purgeObserverCandidates, cutoff, retentionBatchRows) + if err != nil { + return fmt.Errorf("select observers: %w", err) + } + n = len(ids) + if n == 0 { + return nil + } + in := "?" + strings.Repeat(",?", n-1) + args := make([]any, n) + for i, id := range ids { + args[i] = id + } + if _, err := tx.Exec(`DELETE FROM observer_neighbors WHERE observer_id IN (`+in+`)`, args...); err != nil { + return fmt.Errorf("purge observer_neighbors: %w", err) + } + if _, err := tx.Exec(`DELETE FROM observers WHERE id IN (`+in+`)`, args...); err != nil { + return fmt.Errorf("purge observers: %w", err) + } + return nil + }) + if err != nil { + return total, fmt.Errorf("purge stale observers: %w", err) + } + total += int64(n) + if n < retentionBatchRows { + break + } + } + if total > 0 { + log.Printf("[prune] purged %d inactive observer(s) with no remaining data (not seen in %d days)", total, purgeDays) + } + return total, nil +} + +// queryStrings returns the single string column of q's rows. +func queryStrings(tx *sql.Tx, q string, args ...any) ([]string, error) { + rows, err := tx.Query(q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var out []string + for rows.Next() { + var v string + if err := rows.Scan(&v); err != nil { + return nil, err + } + out = append(out, v) + } + return out, rows.Err() +} + +// runTableRetention runs one pass of the opt-in table retention (#329). when +// names the pass in the log ("startup", "daily"). The observer purge runs +// last, after the caller's RemoveStaleObservers, so a row crossing both +// thresholds is finalised in one pass. Each prune logs its own count; an +// error is logged and does not stop the others. +func runTableRetention(store *Store, r TableRetention, when string) { + for _, p := range []struct { + days int + prune func(int) (int64, error) + }{ + {r.InactiveNodeDays, store.PruneInactiveNodes}, + {r.NodeChangeDays, store.PruneNodeChanges}, + {r.ObserverPurgeDays, store.PurgeStaleObservers}, + } { + if _, err := p.prune(p.days); err != nil { + log.Printf("[prune] %s table retention error: %v", when, err) + } + } +} diff --git a/cmd/ingestor/table_retention_perf_test.go b/cmd/ingestor/table_retention_perf_test.go new file mode 100644 index 000000000..32198d0db --- /dev/null +++ b/cmd/ingestor/table_retention_perf_test.go @@ -0,0 +1,247 @@ +package main + +import ( + "bytes" + "database/sql" + "fmt" + "log" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// TestTableRetentionTiming measures the #329 table retention on a multi-GB +// synthetic DB next to today's daily prunes, and counts the [db-slow-writer] +// lines each emits (default 500 ms threshold) while a probe writer stands in +// for MQTT ingest. Fixture: 365 days of history; 3M transmissions (30k a day +// for the last 100 days) with four observations each; 20k inactive_nodes +// (2k of them nodes that came back), 500k node_changes, 50k ping_triggers +// (which no retention deletes; the test fails if one goes), 600 observers (300 soft-deleted, half of those still with metrics) with +// metrics and neighbour rows. The windows are 30 days, the operator's +// setting. Three passes, as in production order (transmissions and metrics +// first, then the table retention): the first run after enabling (335 days +// of backlog), the next day (one more day ages out) and a run with nothing to +// delete. Opt-in, as seeding takes several minutes and several GB of disk: +// +// CORESCOPE_RETENTION_PERF=1 go test -run TestTableRetentionTiming -v -timeout 60m . +// +// CORESCOPE_RETENTION_PERF_DIR puts the DB somewhere other than t.TempDir(). +func TestTableRetentionTiming(t *testing.T) { + if os.Getenv("CORESCOPE_RETENTION_PERF") == "" { + t.Skip("set CORESCOPE_RETENTION_PERF=1 to run the table retention timing on a multi-GB fixture") + } + dir := os.Getenv("CORESCOPE_RETENTION_PERF_DIR") + if dir == "" { + dir = t.TempDir() + } + path := filepath.Join(dir, "retention-perf.db") + os.Remove(path) + store, err := OpenStore(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { store.Close(); os.Remove(path) }) + store.WaitForAsyncMigrations() + seedTableRetentionPerf(t, store) + if fi, err := os.Stat(path); err == nil { + t.Logf("DB size %.2f GB", float64(fi.Size())/1e9) + } + + var logBuf syncBuffer + log.SetOutput(&logBuf) + t.Cleanup(func() { log.SetOutput(os.Stderr) }) + + type step struct { + label string + run func() error + } + today := func(days int) step { + return step{"today's prunes", func() error { + if _, err := store.PruneTransmissions(days, 0); err != nil { + return err + } + if _, err := store.PruneOldMetrics(days); err != nil { + return err + } + if _, err := store.PruneOldNeighborMetrics(days); err != nil { + return err + } + if _, err := store.PruneDroppedPackets(days); err != nil { + return err + } + _, err := store.RemoveStaleObservers(14) + return err + }} + } + tables := func(days int) step { + return step{"#329 table retention", func() error { + runTableRetention(store, TableRetention{InactiveNodeDays: days, NodeChangeDays: days, ObserverPurgeDays: days}, "perf") + if strings.Contains(logBuf.String(), "table retention error") { + return fmt.Errorf("%s", logBuf.String()) + } + return nil + }} + } + components := []string{"prune_packets", "prune_metrics", "prune_observers", + "prune_inactive_nodes", "prune_node_changes", "purge_observers"} + + for _, pass := range []struct { + name string + days int + }{{"first run (30)", 30}, {"next day (29)", 29}, {"nothing to delete (29)", 29}} { + for _, st := range []step{today(pass.days), tables(pass.days)} { + counts := map[string]int{} + for _, tbl := range []string{"transmissions", "inactive_nodes", "node_changes", "ping_triggers", "observers"} { + counts[tbl] = countRows(t, store, tbl) + } + ResetWriterStatsForTest() + logBuf.Reset() + stop, probeWaitMax := startIngestProbe(store) + start := time.Now() + if err := st.run(); err != nil { + t.Fatalf("%s %s: %v", pass.name, st.label, err) + } + elapsed := time.Since(start) + stop() + stats := store.WriterStatsSnapshot() + var parts []string + for _, c := range components { + if s, ok := stats[c]; ok { + parts = append(parts, fmt.Sprintf("%-21s %5d tx hold max %7.1fms p99 %7.1fms", c, s.Count, s.HoldMsMax, s.HoldMsP99)) + } + } + var deleted []string + for _, tbl := range []string{"transmissions", "inactive_nodes", "node_changes", "ping_triggers", "observers"} { + if d := counts[tbl] - countRows(t, store, tbl); d > 0 { + deleted = append(deleted, fmt.Sprintf("%s=%d", tbl, d)) + if tbl == "ping_triggers" { + t.Errorf("%s %s deleted %d ping_triggers; want none", pass.name, st.label, d) + } + } + } + slow := strings.Count(logBuf.String(), "[db-slow-writer]") + t.Logf("%-22s %-20s %9v [db-slow-writer] lines=%d ingest probe wait max %.1fms deleted[%s]\n %s", + pass.name, st.label, elapsed.Round(time.Millisecond), slow, *probeWaitMax, + strings.Join(deleted, " "), strings.Join(parts, "\n ")) + if slow > 0 { + for _, l := range strings.Split(logBuf.String(), "\n") { + if strings.Contains(l, "[db-slow-writer]") { + t.Logf(" %s", l) + } + } + } + } + } +} + +// startIngestProbe writes one small row through WriterTx every 5 ms, as MQTT +// ingest would, and records the longest wait for the writer lock. +func startIngestProbe(store *Store) (stop func(), waitMax *float64) { + var max float64 + done := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-done: + return + case <-time.After(5 * time.Millisecond): + } + start := time.Now() + _ = store.WriterTx("probe_ingest", func(tx *sql.Tx) error { + if w := float64(time.Since(start).Microseconds()) / 1000; w > max { + max = w + } + _, err := tx.Exec(`INSERT OR REPLACE INTO _migrations (name) VALUES ('perf-probe')`) + return err + }) + } + }() + return func() { close(done); wg.Wait() }, &max +} + +func seedTableRetentionPerf(t *testing.T, store *Store) { + t.Helper() + start := time.Now() + now := time.Now().UTC().Format(time.RFC3339) + exec := func(label, q string, args ...any) { + t.Helper() + s := time.Now() + if _, err := store.db.Exec(q, args...); err != nil { + t.Fatalf("seed %s: %v", label, err) + } + t.Logf("seeded %-20s in %v", label, time.Since(s).Round(time.Millisecond)) + } + // 3M transmissions over the last 100 days with realistic payload sizes. + exec("transmissions", ` + WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i + 1 < 3000000), + r(i, ts) AS (SELECT i, strftime('%Y-%m-%dT%H:%M:%SZ', ?1, '-' || (i * 86400 / 30000) || ' seconds') FROM n) + INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, last_seen) + SELECT hex(randomblob(110)), 'perf-' || i, ts, 1, CASE i % 3 WHEN 0 THEN 5 WHEN 1 THEN 4 ELSE 2 END, 1, + '{"type":"ADVERT","pubKey":"' || hex(randomblob(32)) || '","name":"node-' || (i % 3000) || '","lat":55.6,"lon":12.5,"flags":{"repeater":true},"pad":"' || hex(randomblob(90)) || '"}', + CAST(strftime('%s', ts) AS INTEGER) + FROM r`, now) + exec("observations", ` + WITH RECURSIVE k(j) AS (SELECT 0 UNION ALL SELECT j + 1 FROM k WHERE j + 1 < 4) + INSERT INTO observations (transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp) + SELECT t.id, 1 + (t.id * 7 + k.j) % 300, 'rx', 5.5, -100, 0, '["a1b2","c3d4","e5f6","0718"]', t.last_seen + k.j FROM transmissions t, k`) + // Nodes: 3000 active, 20k inactive over 365 days, 2k of them came back. + exec("nodes", ` + WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i + 1 < 3000) + INSERT INTO nodes (public_key, name, role, last_seen, first_seen, advert_count) + SELECT printf('%064x', i), 'node-' || i, 'repeater', ?1, ?1, 10 FROM n`, now) + exec("inactive_nodes", ` + WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i + 1 < 20000) + INSERT INTO inactive_nodes (public_key, name, role, last_seen, first_seen, advert_count) + SELECT printf('%064x', CASE WHEN i < 2000 THEN i ELSE 100000 + i END), 'gone-' || i, 'companion', + strftime('%Y-%m-%dT%H:%M:%SZ', ?1, '-' || (8 * 86400 + i * 357 * 86400 / 20000) || ' seconds'), ?1, 3 FROM n`, now) + exec("node_changes", ` + WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i + 1 < 500000) + INSERT INTO node_changes (public_key, change_type, old_value, new_value, detected_at) + SELECT printf('%064x', i % 23000), CASE i % 3 WHEN 0 THEN 'name' WHEN 1 THEN 'role' ELSE 'position' END, 'old-' || i, 'new-' || i, + strftime('%Y-%m-%dT%H:%M:%SZ', ?1, '-' || (i * 365 * 86400 / 500000) || ' seconds') FROM n`, now) + exec("ping_triggers", ` + WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i + 1 < 50000) + INSERT INTO ping_triggers (tx_id, hash, channel_hash, sender, first_seen) + SELECT 10000000 - i * 60, 'ping-' || i, '#ping', 'sender-' || (i % 500), + strftime('%Y-%m-%dT%H:%M:%SZ', ?1, '-' || (i * 365 * 86400 / 50000) || ' seconds') FROM n`, now) + // 600 observers, 1..300 referenced by observations; 301..600 soft-deleted + // 60-200 days ago, every second one with metrics still in the window. + exec("observers", ` + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i + 1 <= 600) + INSERT INTO observers (rowid, id, name, iata, last_seen, first_seen, packet_count, inactive) + SELECT i, printf('OBS%061d', i), 'obs-' || i, 'CPH', + CASE WHEN i <= 300 THEN ?1 ELSE strftime('%Y-%m-%dT%H:%M:%SZ', ?1, '-' || (60 + i % 140) || ' days') END, + ?1, 100, CASE WHEN i <= 300 THEN 0 ELSE 1 END FROM n`, now) + exec("observer_metrics", ` + WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i + 1 < 200000) + INSERT OR IGNORE INTO observer_metrics (observer_id, timestamp, noise_floor) + SELECT printf('OBS%061d', CASE WHEN i % 4 = 0 THEN 301 + 2 * (i % 150) ELSE 1 + i % 300 END), + strftime('%Y-%m-%dT%H:%M:%SZ', ?1, '-' || (i * 60 * 86400 / 200000) || ' seconds'), -110 FROM n`, now) + exec("observer_neighbors", ` + WITH RECURSIVE n(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM n WHERE i + 1 < 12000) + INSERT OR IGNORE INTO observer_neighbors (observer_id, neighbor_pubkey, status, reported_at) + SELECT printf('OBS%061d', 1 + i % 600), printf('%064x', i), 'ok', ?1 FROM n`, now) + if _, err := store.db.Exec(`ANALYZE`); err != nil { + t.Fatal(err) + } + t.Logf("seeded fixture in %v", time.Since(start).Round(time.Second)) +} + +type syncBuffer struct { + mu sync.Mutex + b bytes.Buffer +} + +func (s *syncBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.b.Write(p) +} +func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() } +func (s *syncBuffer) Reset() { s.mu.Lock(); defer s.mu.Unlock(); s.b.Reset() } diff --git a/cmd/ingestor/table_retention_test.go b/cmd/ingestor/table_retention_test.go new file mode 100644 index 000000000..ffb0865d0 --- /dev/null +++ b/cmd/ingestor/table_retention_test.go @@ -0,0 +1,392 @@ +package main + +// Tests for the opt-in retention of the tables nothing else prunes (#329): +// retention.inactiveNodeDays and nodeChangeDays here, and observerPurgeDays in +// observer_purge_test.go. ping_triggers is deliberately not among them; see +// TestTableRetentionLeavesPingTriggers. + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// daysAgo is an RFC3339 timestamp n days before now, the format every +// retention column is written in. +func daysAgo(n int) string { + return time.Now().UTC().AddDate(0, 0, -n).Format(time.RFC3339) +} + +// setRetentionBatchRows lowers retentionBatchRows for one test, so a handful +// of rows spans several batches. +func setRetentionBatchRows(t *testing.T, n int) { + t.Helper() + old := retentionBatchRows + retentionBatchRows = n + t.Cleanup(func() { retentionBatchRows = old }) +} + +func mustExec(t *testing.T, s *Store, query string, args ...any) { + t.Helper() + if _, err := s.db.Exec(query, args...); err != nil { + t.Fatalf("%s: %v", query, err) + } +} + +func rowExists(t *testing.T, s *Store, query string, args ...any) bool { + t.Helper() + var n int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM (`+query+`)`, args...).Scan(&n); err != nil { + t.Fatalf("%s: %v", query, err) + } + return n > 0 +} + +// ─── config ──────────────────────────────────────────────────────────────── + +func TestTableRetentionConfig(t *testing.T) { + cases := []struct { + name string + json string + want TableRetention + }{ + {"no retention block", `{}`, TableRetention{}}, + {"unset", `{"retention":{"nodeDays":7,"packetDays":30}}`, TableRetention{}}, + {"set", `{"retention":{"inactiveNodeDays":30,"nodeChangeDays":31,"observerPurgeDays":33}}`, + TableRetention{InactiveNodeDays: 30, NodeChangeDays: 31, ObserverPurgeDays: 33}}, + {"negative disables", `{"retention":{"inactiveNodeDays":-1,"nodeChangeDays":-1,"observerPurgeDays":-1}}`, TableRetention{}}, + // An earlier draft of #329 had this knob; ping_triggers is never pruned. + {"pingTriggerDays ignored", `{"retention":{"pingTriggerDays":30}}`, TableRetention{}}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + var cfg Config + if err := json.Unmarshal([]byte(c.json), &cfg); err != nil { + t.Fatal(err) + } + got := cfg.TableRetention() + if got != c.want { + t.Errorf("TableRetention() = %+v, want %+v", got, c.want) + } + if got.Enabled() != (c.want != TableRetention{}) { + t.Errorf("Enabled() = %v for %+v", got.Enabled(), got) + } + }) + } +} + +// config.example.json documents every window, set to 0 so that copying it as +// a live config changes nothing. +func TestConfigExampleDocumentsTableRetention(t *testing.T) { + data, err := os.ReadFile(filepath.Join("..", "..", "config.example.json")) + if err != nil { + t.Fatal(err) + } + var cfg Config + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatal(err) + } + if got := cfg.TableRetention(); got.Enabled() { + t.Errorf("config.example.json enables table retention: %+v, want every window 0", got) + } + var raw struct { + Retention map[string]any `json:"retention"` + } + if err := json.Unmarshal(data, &raw); err != nil { + t.Fatal(err) + } + doc, _ := raw.Retention["_comment_tableRetention"].(string) + for _, k := range []string{"inactiveNodeDays", "nodeChangeDays", "observerPurgeDays"} { + if v, ok := raw.Retention[k]; !ok || v != float64(0) { + t.Errorf("retention.%s = %v (present %v), want 0", k, v, ok) + } + if !strings.Contains(doc, k+":") { + t.Errorf("retention._comment_tableRetention does not document %s", k) + } + } + // ping_triggers is kept forever (operator decision on #329), so neither + // the example nor the user guide offers a knob for it. + guide, err := os.ReadFile(filepath.Join("..", "..", "docs", "user-guide", "configuration.md")) + if err != nil { + t.Fatal(err) + } + for name, text := range map[string]string{"config.example.json": string(data), "configuration.md": string(guide)} { + if strings.Contains(text, "pingTriggerDays") { + t.Errorf("%s documents pingTriggerDays; ping_triggers is never pruned", name) + } + } +} + +// ─── inactive_nodes ──────────────────────────────────────────────────────── + +func insertInactiveNode(t *testing.T, s *Store, pubkey string, ageDays int) { + t.Helper() + mustExec(t, s, `INSERT INTO inactive_nodes (public_key, name, last_seen, first_seen) VALUES (?, ?, ?, ?)`, + pubkey, pubkey, daysAgo(ageDays), daysAgo(ageDays+10)) +} + +func inactiveNodeExists(t *testing.T, s *Store, pubkey string) bool { + t.Helper() + return rowExists(t, s, `SELECT 1 FROM inactive_nodes WHERE public_key = ?`, pubkey) +} + +func TestPruneInactiveNodesDeletesOldKeepsRecent(t *testing.T) { + store := newTestStore(t) + insertInactiveNode(t, store, "aa01", 90) + insertInactiveNode(t, store, "aa02", 20) + + n, err := store.PruneInactiveNodes(30) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Errorf("deleted=%d, want 1", n) + } + if inactiveNodeExists(t, store, "aa01") { + t.Error("aa01 (90 days) still present, want deleted") + } + if !inactiveNodeExists(t, store, "aa02") { + t.Error("aa02 (20 days) deleted, want kept") + } +} + +// A node that came back has a nodes row next to its old inactive_nodes row. +// That row still carries the node's confirmed default_scope evidence and +// keeps the node out of New Nodes, so it stays while the node is active. +func TestPruneInactiveNodesKeepsReturnedNode(t *testing.T) { + store := newTestStore(t) + insertInactiveNode(t, store, "aa03", 90) + if err := store.UpsertNode("aa03", "Back", "repeater", nil, nil, daysAgo(0)); err != nil { + t.Fatal(err) + } + + if _, err := store.PruneInactiveNodes(30); err != nil { + t.Fatal(err) + } + if !inactiveNodeExists(t, store, "aa03") { + t.Error("inactive row of a node that came back was deleted, want kept") + } +} + +// MoveStaleNodes keeps the node of an observer that is still uploading (#199); +// the purge does not delete what MoveStaleNodes would keep. +func TestPruneInactiveNodesKeepsActiveObserverNode(t *testing.T) { + store := newTestStore(t) + insertInactiveNode(t, store, "aa04", 90) + if err := store.UpsertObserver("AA04", "Obs", "LAX", nil); err != nil { + t.Fatal(err) + } + + if _, err := store.PruneInactiveNodes(30); err != nil { + t.Fatal(err) + } + if !inactiveNodeExists(t, store, "aa04") { + t.Error("inactive row of an active observer was deleted, want kept") + } +} + +func TestPruneInactiveNodesBatches(t *testing.T) { + setRetentionBatchRows(t, 2) + store := newTestStore(t) + for _, pk := range []string{"ab01", "ab02", "ab03", "ab04", "ab05"} { + insertInactiveNode(t, store, pk, 90) + } + insertInactiveNode(t, store, "ab06", 1) + ResetWriterStatsForTest() + + n, err := store.PruneInactiveNodes(30) + if err != nil { + t.Fatal(err) + } + if n != 5 || countRows(t, store, "inactive_nodes") != 1 { + t.Errorf("deleted=%d left=%d, want 5 and 1", n, countRows(t, store, "inactive_nodes")) + } + if got := store.WriterStatsSnapshot()["prune_inactive_nodes"].Count; got != 3 { + t.Errorf("prune_inactive_nodes transactions = %d, want 3", got) + } +} + +// ─── node_changes ────────────────────────────────────────────────────────── + +func insertNodeChange(t *testing.T, s *Store, pubkey string, ageDays int) { + t.Helper() + mustExec(t, s, `INSERT INTO node_changes (public_key, change_type, old_value, new_value, detected_at) VALUES (?, 'name', 'a', 'b', ?)`, + pubkey, daysAgo(ageDays)) +} + +func TestPruneNodeChangesDeletesOldKeepsRecent(t *testing.T) { + store := newTestStore(t) + insertNodeChange(t, store, "old", 90) + insertNodeChange(t, store, "new", 20) + + n, err := store.PruneNodeChanges(30) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Errorf("deleted=%d, want 1", n) + } + if rowExists(t, store, `SELECT 1 FROM node_changes WHERE public_key = 'old'`) { + t.Error("90-day change still present, want deleted") + } + if !rowExists(t, store, `SELECT 1 FROM node_changes WHERE public_key = 'new'`) { + t.Error("20-day change deleted, want kept") + } +} + +func TestPruneNodeChangesBatches(t *testing.T) { + setRetentionBatchRows(t, 2) + store := newTestStore(t) + for i := 0; i < 4; i++ { + insertNodeChange(t, store, "old", 90+i) + } + insertNodeChange(t, store, "new", 1) + ResetWriterStatsForTest() + + n, err := store.PruneNodeChanges(30) + if err != nil { + t.Fatal(err) + } + if n != 4 || countRows(t, store, "node_changes") != 1 { + t.Errorf("deleted=%d left=%d, want 4 and 1", n, countRows(t, store, "node_changes")) + } + if got := store.WriterStatsSnapshot()["prune_node_changes"].Count; got != 3 { + t.Errorf("prune_node_changes transactions = %d, want 3 (batches of 2, 2, 0)", got) + } +} + +// ─── ping_triggers: never pruned ─────────────────────────────────────────── + +func insertPingTrigger(t *testing.T, s *Store, txID int64, ageDays int) { + t.Helper() + mustExec(t, s, `INSERT INTO ping_triggers (tx_id, hash, channel_hash, sender, first_seen) VALUES (?, ?, '#ping', 'Alice', ?)`, + txID, fmt.Sprintf("hash%d", txID), daysAgo(ageDays)) +} + +// pingTriggerRows dumps every ping_triggers row, so a test can tell that none +// was deleted or rewritten. +func pingTriggerRows(t *testing.T, s *Store) []string { + t.Helper() + rows, err := s.db.Query(`SELECT tx_id || '|' || hash || '|' || IFNULL(channel_hash, '') || '|' || + IFNULL(sender, '') || '|' || first_seen FROM ping_triggers ORDER BY tx_id`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var r string + if err := rows.Scan(&r); err != nil { + t.Fatal(err) + } + out = append(out, r) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + return out +} + +// ping_triggers keeps every row (operator decision on #329): the all-time Ping +// Scores records join it with the history sidecar (#349, #241). With every +// other retention knob set, and the transmission prune run first, no trigger +// is deleted or changed, however old it is or whether its transmission is +// gone. pingTriggerDays, a knob in an earlier draft of #329, is ignored. +func TestTableRetentionLeavesPingTriggers(t *testing.T) { + var cfg Config + if err := json.Unmarshal([]byte(`{"retention":{"packetDays":30,"inactiveNodeDays":30,`+ + `"nodeChangeDays":30,"observerPurgeDays":30,"pingTriggerDays":30}}`), &cfg); err != nil { + t.Fatal(err) + } + store := newTestStore(t) + insertChanTx(t, store, "pingpruned03", "Carol: ping", "#ping") + mustExec(t, store, `UPDATE transmissions SET first_seen = ? WHERE hash = 'pingpruned03'`, daysAgo(90)) + mustExec(t, store, `UPDATE ping_triggers SET first_seen = ? WHERE hash = 'pingpruned03'`, daysAgo(90)) + insertPingTrigger(t, store, 101, 90) + insertPingTrigger(t, store, 102, 1) + seedTableRetentionRows(t, store) + before := pingTriggerRows(t, store) + if len(before) != 3 { + t.Fatalf("seeded %d ping_triggers, want 3", len(before)) + } + + if n, err := store.PruneOldPackets(cfg.PacketDaysOrZero()); err != nil || n != 1 { + t.Fatalf("PruneOldPackets = %d, %v; want the ping's transmission pruned", n, err) + } + runTableRetention(store, cfg.TableRetention(), "test") + + for _, table := range []string{"inactive_nodes", "node_changes", "observers"} { + if got := countRows(t, store, table); got != 1 { + t.Errorf("%s rows = %d, want 1 (its knob is set)", table, got) + } + } + after := pingTriggerRows(t, store) + if strings.Join(after, "\n") != strings.Join(before, "\n") { + t.Errorf("ping_triggers changed by retention:\nbefore %q\nafter %q", before, after) + } +} + +// ─── wiring ──────────────────────────────────────────────────────────────── + +// seedTableRetentionRows gives each of the three tables one 90-day-old row +// and one 1-day-old row. +func seedTableRetentionRows(t *testing.T, s *Store) { + t.Helper() + insertInactiveNode(t, s, "cc01", 90) + insertInactiveNode(t, s, "cc02", 1) + insertNodeChange(t, s, "cc01", 90) + insertNodeChange(t, s, "cc02", 1) + staleObserver(t, s, "obs-cc01") + if err := s.UpsertObserver("obs-cc02", "obs-cc02", "LAX", nil); err != nil { + t.Fatal(err) + } + mustExec(t, s, `UPDATE observers SET inactive = 1, last_seen = ? WHERE id = 'obs-cc02'`, daysAgo(1)) +} + +// Every knob unset: nothing is deleted from any of the three tables. +func TestRunTableRetentionUnsetChangesNothing(t *testing.T) { + store := newTestStore(t) + seedTableRetentionRows(t, store) + + runTableRetention(store, TableRetention{}, "test") + + for _, table := range []string{"inactive_nodes", "node_changes", "observers"} { + if got := countRows(t, store, table); got != 2 { + t.Errorf("%s rows = %d, want 2 (all knobs unset)", table, got) + } + } +} + +// Each knob prunes its own table and only that one. +func TestRunTableRetentionEachKnob(t *testing.T) { + cases := []struct { + table string + r TableRetention + }{ + {"inactive_nodes", TableRetention{InactiveNodeDays: 30}}, + {"node_changes", TableRetention{NodeChangeDays: 30}}, + {"observers", TableRetention{ObserverPurgeDays: 30}}, + } + for _, c := range cases { + t.Run(c.table, func(t *testing.T) { + store := newTestStore(t) + seedTableRetentionRows(t, store) + + runTableRetention(store, c.r, "test") + + for _, table := range []string{"inactive_nodes", "node_changes", "observers"} { + want := 2 + if table == c.table { + want = 1 + } + if got := countRows(t, store, table); got != want { + t.Errorf("%s rows = %d, want %d", table, got, want) + } + } + }) + } +} diff --git a/cmd/server/ping_score_history_retention_329_test.go b/cmd/server/ping_score_history_retention_329_test.go new file mode 100644 index 000000000..6140729af --- /dev/null +++ b/cmd/server/ping_score_history_retention_329_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "testing" + "time" +) + +// #329: the ingestor's table retention never deletes ping_triggers rows, so a +// Ping Scores history entry outlives the packetDays prune of its transmission. +// This pins the server side of that on the shared DB: an entry already marked +// data_pruned keeps rendering (the #329 acceptance criterion). + +// A ping settled 40 days ago loses its transmission and observations the way +// the ingestor's packetDays prune removes them (the ping_triggers row stays). +// Its history entry is marked data_pruned and keeps rendering in the all-time +// records, cycle after cycle. +func TestPingScoresDataPrunedEntryRendersWhileTriggerKept(t *testing.T) { + fx := setupEngineFixture(t, pingScoreHistoryEngineConfig{SettleDebounce: time.Minute, DeepSweepBatchSize: 100, RetentionDuration: 30 * 24 * time.Hour}) + firstSeen := fx.clock.Now().Add(-40 * 24 * time.Hour).UTC().Format(time.RFC3339) + txID := setupSettledPingWithGoneData(t, fx, "keeptrigger329", firstSeen) + if _, err := fx.srv.db.conn.Exec(`DELETE FROM transmissions WHERE id = ?`, txID); err != nil { + t.Fatal(err) + } + fx.clock.Advance(2 * time.Minute) + if _, err := fx.engine.Cycle(); err != nil { + t.Fatal(err) + } + if e, _ := fx.engine.index.Get(txID); !e.DataPruned { + t.Fatalf("sanity check failed: entry not data_pruned after its transmission was pruned: %+v", e) + } + fx.clock.Advance(time.Minute) + if _, err := fx.engine.Cycle(); err != nil { + t.Fatal(err) + } + + snap, err := fx.srv.buildPingScoresSnapshotFromHistory(mustFetchTriggers(t, fx), fx.engine.index.Entries(), fx.clock.Now()) + if err != nil { + t.Fatal(err) + } + if snap.MostHopsPing == nil || snap.MostHopsPing.Hash != "keeptrigger329" { + t.Errorf("data_pruned entry not rendered in the all-time records: %+v", snap.MostHopsPing) + } +} diff --git a/cmd/server/readonly_invariant_test.go b/cmd/server/readonly_invariant_test.go index 6e8b3e25e..03c27bf48 100644 --- a/cmd/server/readonly_invariant_test.go +++ b/cmd/server/readonly_invariant_test.go @@ -92,6 +92,10 @@ func TestServerDBHasNoWriteMethods(t *testing.T) { "PruneOldPackets", "PruneOldMetrics", "RemoveStaleObservers", + // #329 opt-in table retention: ingestor-only. + "PruneInactiveNodes", + "PruneNodeChanges", + "PurgeStaleObservers", // #738 / one-click geo-prune: the DELETE must live on the // ingestor's *Store. The server's HTTP handler now enqueues a // marker file (see internal/prunequeue); it does not write. diff --git a/config.example.json b/config.example.json index a89aff342..28cf27d93 100644 --- a/config.example.json +++ b/config.example.json @@ -15,6 +15,10 @@ "packetDays": 30, "channelDays": 0, "clientRxDays": 30, + "inactiveNodeDays": 0, + "nodeChangeDays": 0, + "observerPurgeDays": 0, + "_comment_tableRetention": "Opt-in (ingestor, #329) retention for the tables nothing else prunes; each is in days, 0 or omitted = keep forever (the default). Applied at ingestor startup and daily with the observer retention pass, in batches. inactiveNodeDays: inactive_nodes rows whose last advert is older than N days are deleted, unless the node has come back or is an observer still uploading; a deleted node that adverts again later counts as new (no 'resurrected' change, listed in New Nodes). Set it above nodeDays. nodeChangeDays: node_changes rows (Tools > Node Changes) detected more than N days ago are deleted. observerPurgeDays: observers already marked inactive by observerDays AND not seen in N days are hard-deleted with their observer_neighbors snapshot, but only when no observation, observer_metrics, observer_neighbor_metrics or dropped_packets row still references them. Set it above observerDays, packetDays (and channelDays) and metricsDays; below those an observer waits until that data has aged out. Example: an operator honouring a 30-day retention period sets all three to 30. ping_triggers (Ping Scores triggers, with sender names) has no window and is kept forever, because the all-time Ping Scores records need it.", "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). packetDays: transmissions older than N days are deleted (0 = disabled). channelDays: channel messages (GRP_TXT) and their observations are kept until they are N days old instead of packetDays; only takes effect when packetDays is set and channelDays is larger (0 = channel messages follow packetDays, default) (#296). The server's in-memory window (packetStore.retentionHours) is separate; channel history is read from the database. clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes." }, "db": { diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 6717499c4..9e9126068 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -98,6 +98,9 @@ How long (in hours) before a node is marked degraded or silent: | `retention.nodeDays` | `7` | Nodes not seen in N days move to inactive | | `retention.packetDays` | `30` | Packets older than N days are deleted daily | | `retention.channelDays` | `0` | Channel messages (GRP_TXT) and their observations are kept until they are N days old instead of `packetDays`. Takes effect only when `packetDays` is set and `channelDays` is larger; `0` = channel messages follow `packetDays` | +| `retention.inactiveNodeDays` | `0` | Rows in `inactive_nodes` whose last advert is older than N days are deleted, unless the node has come back or is an observer still uploading. `0` = keep forever | +| `retention.nodeChangeDays` | `0` | Node Changes history (`node_changes`) older than N days is deleted. `0` = keep forever | +| `retention.observerPurgeDays` | `0` | Observers already marked inactive by `observerDays` and not seen in N days are deleted, once no packet, metric or dropped-packet row references them. `0` = keep forever | `retention.channelDays` lets an instance keep a short `packetDays` to bound the database while keeping chat history longer. Channel messages are a small share @@ -113,6 +116,25 @@ database (`/api/channels`, `/api/channels/{hash}/messages`), so messages kept by `channelDays` stay visible there even when they are older than the in-memory window. +The three opt-in settings (`inactiveNodeDays`, `nodeChangeDays`, +`observerPurgeDays`) cover the tables nothing else prunes, so an instance can +honour a fixed retention period for node and observer data. The ingestor +applies them at startup and then daily, in small batches, and logs each count, +for example `[prune] deleted 12 node_changes older than 30 days`. Leaving them +unset keeps today's behaviour. + +- A node deleted by `inactiveNodeDays` that adverts again later counts as a new + node: it shows in New Nodes, and Node Changes records no return. Set it above + `nodeDays`. +- `observerPurgeDays` deletes an observer only after its packets, metrics and + dropped packets have aged out, so set it above `observerDays`, `packetDays` + (and `channelDays`) and `metricsDays`; below those, an observer waits until + that data is gone. The observer's current neighbour list is deleted with it. +- Ping Scores triggers (`ping_triggers`, which hold sender names) have no + setting and are never deleted. The all-time Ping Scores records need them: + a ping stays in those records after `packetDays` has removed its packet, but + only while its trigger exists. + > **Note:** Lowering retention does **not** immediately shrink the database file. > SQLite marks deleted pages as free but does not return them to the filesystem > unless [incremental auto-vacuum](database.md) is enabled. New databases created