From 7ebf4e7e09fb50f434a18aa441ebacaaca75a4a1 Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 7 Oct 2026 10:39:44 +0000 Subject: [PATCH 1/5] feat(ingestor): opt-in retention for inactive_nodes, node_changes, ping_triggers and observers (#329) Four new retention knobs, each in days, 0/unset = keep forever (today's behaviour): inactiveNodeDays, nodeChangeDays, pingTriggerDays and observerPurgeDays. The ingestor applies them at startup and daily after the observer soft-delete, in bounded WriterTx batches, and logs the counts as [prune] lines. - inactive_nodes: by last_seen, skipping nodes that came back (their old row carries confirmed default_scope evidence and keeps them out of New Nodes) and observers still uploading (#199). - node_changes: by detected_at, over idx_node_changes_detected_at. - ping_triggers: by their own first_seen, not with their transmission. The server's Ping Scores history drops an entry once its trigger is gone, and data_pruned entries are exactly those whose transmission was pruned. No index covers first_seen, so the table is walked by tx_id ranges, starting below every key (ids can be 0 or negative). - observers: port of upstream#1886 (#34), plus the observer_neighbor_metrics guard and deletion of the current-only observer_neighbors snapshot. PruneOrphanRouteMaskChanges now uses the shared key-range walk. Co-authored-by: efiten Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/config.go | 41 +++ cmd/ingestor/main.go | 17 +- cmd/ingestor/maintenance.go | 34 +- cmd/ingestor/observer_purge_test.go | 322 +++++++++++++++++++ cmd/ingestor/table_retention.go | 282 +++++++++++++++++ cmd/ingestor/table_retention_test.go | 430 ++++++++++++++++++++++++++ cmd/server/readonly_invariant_test.go | 5 + config.example.json | 5 + docs/user-guide/configuration.md | 23 ++ 9 files changed, 1128 insertions(+), 31 deletions(-) create mode 100644 cmd/ingestor/observer_purge_test.go create mode 100644 cmd/ingestor/table_retention.go create mode 100644 cmd/ingestor/table_retention_test.go diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 6495996e2..0a70698cd 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -232,6 +232,47 @@ 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"` + PingTriggerDays int `json:"pingTriggerDays"` + 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. +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 + // PingTriggerDays deletes ping_triggers rows first seen longer ago. + PingTriggerDays 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.PingTriggerDays > 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), + PingTriggerDays: max(c.Retention.PingTriggerDays, 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 911226e21..945ababec 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -261,6 +261,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 pingTriggerDays=%d observerPurgeDays=%d", + r.InactiveNodeDays, r.NodeChangeDays, r.PingTriggerDays, r.ObserverPurgeDays) + } + runTableRetention(store, tableRetention, "startup") + vacuumPages := cfg.IncrementalVacuumPages() store.RunIncrementalVacuum(vacuumPages) @@ -301,14 +312,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/maintenance.go b/cmd/ingestor/maintenance.go index 9029ae3a7..2fea20815 100644 --- a/cmd/ingestor/maintenance.go +++ b/cmd/ingestor/maintenance.go @@ -278,36 +278,10 @@ func (s *Store) PruneOrphanRouteMaskChanges() (int64, error) { if batch <= 0 { batch = 1000 } - var total, after int64 - for { - var examined, deleted int64 - err := s.WriterTx("prune_route_mask_changes", func(tx *sql.Tx) error { - var last sql.NullInt64 - if err := tx.QueryRow(`SELECT MAX(id), COUNT(*) FROM ( - SELECT id FROM route_mask_changes WHERE id > ? ORDER BY id LIMIT ?)`, - after, batch).Scan(&last, &examined); err != nil { - return fmt.Errorf("scan route_mask_changes: %w", err) - } - if examined == 0 { - return nil - } - res, err := tx.Exec(`DELETE FROM route_mask_changes WHERE id > ? AND id <= ? - AND NOT EXISTS (SELECT 1 FROM transmissions t WHERE t.id = route_mask_changes.transmission_id)`, - after, last.Int64) - if err != nil { - return fmt.Errorf("prune orphan route_mask_changes: %w", err) - } - deleted, _ = res.RowsAffected() - after = last.Int64 - return nil - }) - if err != nil { - return total, err - } - total += deleted - if examined < int64(batch) { - break - } + total, err := s.deleteInKeyRanges("prune_route_mask_changes", "route_mask_changes", "id", + `NOT EXISTS (SELECT 1 FROM transmissions t WHERE t.id = route_mask_changes.transmission_id)`, batch) + if err != nil { + return total, err } if total > 0 { log.Printf("[prune] deleted %d route_mask_changes rows of deleted transmissions", total) 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..fa6f4bd3a --- /dev/null +++ b/cmd/ingestor/table_retention.go @@ -0,0 +1,282 @@ +package main + +import ( + "database/sql" + "fmt" + "log" + "math" + "strings" + "time" +) + +// Opt-in retention for the tables nothing else prunes (#329): inactive_nodes, +// node_changes, ping_triggers 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. + +// retentionBatchRows bounds the rows one table-retention transaction deletes +// (or, for ping_triggers, examines). 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 + } + } +} + +// deleteInKeyRanges walks table in ascending order of its integer key, batch +// keys per WriterTx (tagged tag), and deletes the rows of each range that +// match cond (args are cond's parameters). Each transaction's work is bounded +// by batch however sparse the matches are, which suits a cond no index +// serves. Returns the rows deleted; on error, those of the committed ranges. +func (s *Store) deleteInKeyRanges(tag, table, key, cond string, batch int, args ...any) (int64, error) { + scan := fmt.Sprintf(`SELECT MAX(%[2]s), COUNT(*) FROM (SELECT %[2]s FROM %[1]s WHERE %[2]s > ? ORDER BY %[2]s LIMIT ?)`, table, key) + del := fmt.Sprintf(`DELETE FROM %s WHERE %s > ? AND %[2]s <= ? AND (%s)`, table, key, cond) + var total int64 + after := int64(math.MinInt64) // below every key: ids can be 0 or negative + for { + var examined, deleted int64 + err := s.WriterTx(tag, func(tx *sql.Tx) error { + var last sql.NullInt64 + if err := tx.QueryRow(scan, after, batch).Scan(&last, &examined); err != nil { + return fmt.Errorf("scan %s: %w", table, err) + } + if examined == 0 { + return nil + } + res, err := tx.Exec(del, append([]any{after, last.Int64}, args...)...) + if err != nil { + return fmt.Errorf("prune %s: %w", table, err) + } + deleted, _ = res.RowsAffected() + after = last.Int64 + return nil + }) + if err != nil { + return total, err + } + total += deleted + if examined < int64(batch) { + 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 +} + +// PrunePingTriggers deletes ping_triggers rows first seen more than days ago +// (retention.pingTriggerDays, #329); days <= 0 disables it. +// +// The window is its own and does not follow the transmission prune. The +// server's Ping Scores history keeps an entry only while its trigger row +// exists (planPingScoreHistoryReconcile deletes the rest), and an entry is +// marked data_pruned exactly when its transmission is gone. Deleting the +// triggers of pruned transmissions would therefore drop every data_pruned +// entry from the all-time records at once; with its own window an entry +// renders until its trigger is pingTriggerDays old, and its sender name then +// leaves both the trigger and the history entry. +// +// No index covers first_seen, so the table is walked by tx_id +// (deleteInKeyRanges): one primary-key range per transaction. +func (s *Store) PrunePingTriggers(days int) (int64, error) { + if days <= 0 { + return 0, nil + } + n, err := s.deleteInKeyRanges("prune_ping_triggers", "ping_triggers", "tx_id", + "first_seen < ?", retentionBatchRows, retentionCutoff(days)) + if n > 0 { + log.Printf("[prune] deleted %d ping_triggers 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.PingTriggerDays, store.PrunePingTriggers}, + {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_test.go b/cmd/ingestor/table_retention_test.go new file mode 100644 index 000000000..2398307e3 --- /dev/null +++ b/cmd/ingestor/table_retention_test.go @@ -0,0 +1,430 @@ +package main + +// Tests for the opt-in retention of the tables nothing else prunes (#329): +// retention.inactiveNodeDays, nodeChangeDays and pingTriggerDays here, and +// observerPurgeDays in observer_purge_test.go. + +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,"pingTriggerDays":32,"observerPurgeDays":33}}`, + TableRetention{InactiveNodeDays: 30, NodeChangeDays: 31, PingTriggerDays: 32, ObserverPurgeDays: 33}}, + {"negative disables", `{"retention":{"inactiveNodeDays":-1,"nodeChangeDays":-1,"pingTriggerDays":-1,"observerPurgeDays":-1}}`, 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", "pingTriggerDays", "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) + } + } +} + +// ─── 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 ───────────────────────────────────────────────────────── + +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)) +} + +func pingTriggerExists(t *testing.T, s *Store, txID int64) bool { + t.Helper() + return rowExists(t, s, `SELECT 1 FROM ping_triggers WHERE tx_id = ?`, txID) +} + +func TestPrunePingTriggersDeletesOldKeepsRecent(t *testing.T) { + store := newTestStore(t) + insertPingTrigger(t, store, 1, 90) + insertPingTrigger(t, store, 2, 20) + + n, err := store.PrunePingTriggers(30) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Errorf("deleted=%d, want 1", n) + } + if pingTriggerExists(t, store, 1) { + t.Error("90-day trigger still present, want deleted") + } + if !pingTriggerExists(t, store, 2) { + t.Error("20-day trigger deleted, want kept") + } +} + +// The Ping Scores history keeps an entry only while its ping_triggers row +// exists (planPingScoreHistoryReconcile drops the rest), and an entry is +// marked data_pruned exactly when its transmission is gone. So the trigger +// of a pruned transmission must survive until pingTriggerDays: following the +// transmission prune would remove every data_pruned entry at once. +func TestPrunePingTriggersKeepsTriggerOfPrunedTransmission(t *testing.T) { + store := newTestStore(t) + insertChanTx(t, store, "pingpruned01", "Alice: ping", "#ping") + var txID int64 + if err := store.db.QueryRow(`SELECT tx_id FROM ping_triggers`).Scan(&txID); err != nil { + t.Fatal(err) + } + mustExec(t, store, `UPDATE ping_triggers SET first_seen = ? WHERE tx_id = ?`, daysAgo(20), txID) + mustExec(t, store, `UPDATE transmissions SET first_seen = ? WHERE id = ?`, daysAgo(20), txID) + if n, err := store.PruneOldPackets(10); err != nil || n != 1 { + t.Fatalf("PruneOldPackets(10) = %d, %v; want the transmission pruned", n, err) + } + + if _, err := store.PrunePingTriggers(30); err != nil { + t.Fatal(err) + } + if !pingTriggerExists(t, store, txID) { + t.Error("trigger of a pruned transmission, inside pingTriggerDays, was deleted; want kept (data_pruned entries need it)") + } +} + +// pingTriggerDays unset: the transmission prune leaves ping_triggers alone, as +// before #329. +func TestPruneOldPacketsLeavesPingTriggers(t *testing.T) { + store := newTestStore(t) + insertChanTx(t, store, "pingpruned02", "Bob: ping", "#ping") + mustExec(t, store, `UPDATE transmissions SET first_seen = ?`, daysAgo(90)) + mustExec(t, store, `UPDATE ping_triggers SET first_seen = ?`, daysAgo(90)) + if n, err := store.PruneOldPackets(30); err != nil || n != 1 { + t.Fatalf("PruneOldPackets(30) = %d, %v; want the transmission pruned", n, err) + } + runTableRetention(store, TableRetention{}, "test") + if got := countPingTriggers(t, store); got != 1 { + t.Errorf("ping_triggers = %d, want 1", got) + } +} + +// Triggers are walked by tx_id, a batch of ids per transaction, because no +// index covers first_seen. Old and recent rows interleave across batches. +func TestPrunePingTriggersBatches(t *testing.T) { + setRetentionBatchRows(t, 2) + store := newTestStore(t) + ages := []int{90, 1, 90, 90, 1, 90, 90} + for i, age := range ages { + insertPingTrigger(t, store, int64(i+1), age) + } + ResetWriterStatsForTest() + + n, err := store.PrunePingTriggers(30) + if err != nil { + t.Fatal(err) + } + if n != 5 { + t.Errorf("deleted=%d, want 5", n) + } + for _, id := range []int64{2, 5} { + if !pingTriggerExists(t, store, id) { + t.Errorf("recent trigger %d deleted, want kept", id) + } + } + if got := store.WriterStatsSnapshot()["prune_ping_triggers"].Count; got != 4 { + t.Errorf("prune_ping_triggers transactions = %d, want 4 (7 ids in ranges of 2)", got) + } +} + +// ─── wiring ──────────────────────────────────────────────────────────────── + +// seedTableRetentionRows gives each of the four 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) + insertPingTrigger(t, s, 1, 90) + insertPingTrigger(t, s, 2, 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 four tables. +func TestRunTableRetentionUnsetChangesNothing(t *testing.T) { + store := newTestStore(t) + seedTableRetentionRows(t, store) + + runTableRetention(store, TableRetention{}, "test") + + for _, table := range []string{"inactive_nodes", "node_changes", "ping_triggers", "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}}, + {"ping_triggers", TableRetention{PingTriggerDays: 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", "ping_triggers", "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) + } + } + }) + } +} + +// tx_id is a transmissions id, and those can be 0 or negative (the CI E2E +// fixture seeds both). The walk starts below every key. +func TestPrunePingTriggersNonPositiveTxID(t *testing.T) { + store := newTestStore(t) + insertPingTrigger(t, store, -1000000, 90) + insertPingTrigger(t, store, 0, 90) + insertPingTrigger(t, store, 1, 90) + + n, err := store.PrunePingTriggers(30) + if err != nil { + t.Fatal(err) + } + if n != 3 || countPingTriggers(t, store) != 0 { + t.Errorf("deleted=%d left=%d, want 3 and 0", n, countPingTriggers(t, store)) + } +} diff --git a/cmd/server/readonly_invariant_test.go b/cmd/server/readonly_invariant_test.go index 6e8b3e25e..ca2819db6 100644 --- a/cmd/server/readonly_invariant_test.go +++ b/cmd/server/readonly_invariant_test.go @@ -92,6 +92,11 @@ func TestServerDBHasNoWriteMethods(t *testing.T) { "PruneOldPackets", "PruneOldMetrics", "RemoveStaleObservers", + // #329 opt-in table retention: ingestor-only. + "PruneInactiveNodes", + "PruneNodeChanges", + "PrunePingTriggers", + "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..4b298c97a 100644 --- a/config.example.json +++ b/config.example.json @@ -15,6 +15,11 @@ "packetDays": 30, "channelDays": 0, "clientRxDays": 30, + "inactiveNodeDays": 0, + "nodeChangeDays": 0, + "pingTriggerDays": 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. pingTriggerDays: ping_triggers rows (Ping Scores, holding sender names) first seen more than N days ago are deleted, by their own age and not with their transmission, so Ping Scores history entries already marked data_pruned keep rendering until then; the server then drops the entry, sender name included, from its history store, so N bounds the all-time Ping Scores records. 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 four to 30.", "_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..ef3ac449f 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -98,6 +98,10 @@ 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.pingTriggerDays` | `0` | Ping Scores triggers (`ping_triggers`, which hold sender names) first seen more than N days ago are 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 +117,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 four opt-in settings (`inactiveNodeDays`, `nodeChangeDays`, +`pingTriggerDays`, `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`. +- `pingTriggerDays` counts from when the ping was first seen, not from when its + packet was pruned. Ping Scores keeps a ping in its all-time records after + `packetDays` has removed the packet, and drops it, with its sender name, once + the trigger is deleted. So `pingTriggerDays` also bounds those records. +- `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. + > **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 From f6c2fd466a74c7580cfadd7922338a562b570ef3 Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 7 Oct 2026 10:39:44 +0000 Subject: [PATCH 2/5] test(server): data_pruned Ping Scores entries render while their trigger is kept (#329) Pins the server half of why retention.pingTriggerDays has its own age: a data_pruned entry keeps rendering after its transmission is pruned, and leaves the history store (sender name included) once its ping_triggers row is deleted. Co-Authored-By: Claude Opus 5.5 --- .../ping_score_history_retention_329_test.go | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 cmd/server/ping_score_history_retention_329_test.go 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..b55936e26 --- /dev/null +++ b/cmd/server/ping_score_history_retention_329_test.go @@ -0,0 +1,83 @@ +package main + +import ( + "testing" + "time" +) + +// #329: the ingestor's retention.pingTriggerDays deletes ping_triggers rows by +// their own age, not when their transmission is pruned. These tests pin the +// two halves of why, on the server side of the shared DB. + +// dataPrunedFixture settles one ping 40 days old, removes its transmission +// and observations the way the ingestor's packetDays prune does (the +// ping_triggers row stays), and runs the cycle that marks it data_pruned. +func dataPrunedFixture(t *testing.T, hash string) (*engineFixture, int64) { + t.Helper() + 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, hash, 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) + } + return fx, txID +} + +// A data_pruned entry whose trigger row is still there keeps rendering in the +// all-time records, cycle after cycle, with its transmission gone. +func TestPingScoresDataPrunedEntryRendersWhileTriggerKept(t *testing.T) { + fx, _ := dataPrunedFixture(t, "keeptrigger329") + 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) + } +} + +// Once its trigger row is deleted, the entry leaves the history store too +// (and with it the sender name). This is why ping_triggers cannot follow the +// transmission prune: every data_pruned entry has a pruned transmission. +func TestPingScoresDataPrunedEntryDroppedWithItsTrigger(t *testing.T) { + fx, txID := dataPrunedFixture(t, "droptrigger329") + if _, err := fx.srv.db.conn.Exec(`DELETE FROM ping_triggers WHERE tx_id = ?`, txID); err != nil { + t.Fatal(err) + } + fx.clock.Advance(time.Minute) + if _, err := fx.engine.Cycle(); err != nil { + t.Fatal(err) + } + + if _, ok := fx.engine.index.Get(txID); ok { + t.Error("history entry still indexed after its trigger was deleted") + } + stored, err := fx.store.LoadAll() + if err != nil { + t.Fatal(err) + } + for _, e := range stored { + if e.TxID == txID { + t.Errorf("history entry still stored after its trigger was deleted: %+v", e) + } + } + 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 == "droptrigger329" { + t.Error("entry of a deleted trigger still rendered") + } +} From 398af2b64dc41884d440251fba3cfe00bf246942 Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 7 Oct 2026 10:39:44 +0000 Subject: [PATCH 3/5] test(ingestor): time the #329 table retention on a multi-GB fixture Opt-in (CORESCOPE_RETENTION_PERF=1). Seeds ~5 GB of synthetic history and reports writer-lock hold, [db-slow-writer] lines and an ingest probe's lock wait for today's prunes and for the new table retention, on the first run, the next day and a run with nothing to delete. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/table_retention_perf_test.go | 244 ++++++++++++++++++++++ 1 file changed, 244 insertions(+) create mode 100644 cmd/ingestor/table_retention_perf_test.go diff --git a/cmd/ingestor/table_retention_perf_test.go b/cmd/ingestor/table_retention_perf_test.go new file mode 100644 index 000000000..cbca1d9e7 --- /dev/null +++ b/cmd/ingestor/table_retention_perf_test.go @@ -0,0 +1,244 @@ +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, +// 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, PingTriggerDays: 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", "prune_ping_triggers", "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)) + } + } + 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() } From 5fdce2af4cfb16dafabce26b5c9afa661de37a39 Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 7 Oct 2026 12:05:44 +0000 Subject: [PATCH 4/5] test(ingestor): lock ping_triggers out of the table retention (#329) Operator decision on #329: ping_triggers keeps every row, because the all-time Ping Scores records join it with the history sidecar (#349, #241). TestTableRetentionLeavesPingTriggers sets every other knob (and a stale pingTriggerDays), runs the transmission prune first, and requires every ping_triggers row unchanged. The config test requires pingTriggerDays to be ignored, and the docs test requires neither config.example.json nor the user guide to offer it. Red on 259684f3: the stale knob deletes the old triggers, TableRetention reads it, and both docs still list it. The tests of the removed knob go. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/table_retention_test.go | 174 +++++++++++---------------- 1 file changed, 68 insertions(+), 106 deletions(-) diff --git a/cmd/ingestor/table_retention_test.go b/cmd/ingestor/table_retention_test.go index 2398307e3..ffb0865d0 100644 --- a/cmd/ingestor/table_retention_test.go +++ b/cmd/ingestor/table_retention_test.go @@ -1,8 +1,9 @@ package main // Tests for the opt-in retention of the tables nothing else prunes (#329): -// retention.inactiveNodeDays, nodeChangeDays and pingTriggerDays here, and -// observerPurgeDays in observer_purge_test.go. +// retention.inactiveNodeDays and nodeChangeDays here, and observerPurgeDays in +// observer_purge_test.go. ping_triggers is deliberately not among them; see +// TestTableRetentionLeavesPingTriggers. import ( "encoding/json" @@ -55,9 +56,11 @@ func TestTableRetentionConfig(t *testing.T) { }{ {"no retention block", `{}`, TableRetention{}}, {"unset", `{"retention":{"nodeDays":7,"packetDays":30}}`, TableRetention{}}, - {"set", `{"retention":{"inactiveNodeDays":30,"nodeChangeDays":31,"pingTriggerDays":32,"observerPurgeDays":33}}`, - TableRetention{InactiveNodeDays: 30, NodeChangeDays: 31, PingTriggerDays: 32, ObserverPurgeDays: 33}}, - {"negative disables", `{"retention":{"inactiveNodeDays":-1,"nodeChangeDays":-1,"pingTriggerDays":-1,"observerPurgeDays":-1}}`, 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) { @@ -97,7 +100,7 @@ func TestConfigExampleDocumentsTableRetention(t *testing.T) { t.Fatal(err) } doc, _ := raw.Retention["_comment_tableRetention"].(string) - for _, k := range []string{"inactiveNodeDays", "nodeChangeDays", "pingTriggerDays", "observerPurgeDays"} { + 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) } @@ -105,6 +108,17 @@ func TestConfigExampleDocumentsTableRetention(t *testing.T) { 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 ──────────────────────────────────────────────────────── @@ -245,7 +259,7 @@ func TestPruneNodeChangesBatches(t *testing.T) { } } -// ─── ping_triggers ───────────────────────────────────────────────────────── +// ─── ping_triggers: never pruned ─────────────────────────────────────────── func insertPingTrigger(t *testing.T, s *Store, txID int64, ageDays int) { t.Helper() @@ -253,104 +267,72 @@ func insertPingTrigger(t *testing.T, s *Store, txID int64, ageDays int) { txID, fmt.Sprintf("hash%d", txID), daysAgo(ageDays)) } -func pingTriggerExists(t *testing.T, s *Store, txID int64) bool { +// 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() - return rowExists(t, s, `SELECT 1 FROM ping_triggers WHERE tx_id = ?`, txID) -} - -func TestPrunePingTriggersDeletesOldKeepsRecent(t *testing.T) { - store := newTestStore(t) - insertPingTrigger(t, store, 1, 90) - insertPingTrigger(t, store, 2, 20) - - n, err := store.PrunePingTriggers(30) + 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) } - if n != 1 { - t.Errorf("deleted=%d, want 1", n) - } - if pingTriggerExists(t, store, 1) { - t.Error("90-day trigger still present, want deleted") - } - if !pingTriggerExists(t, store, 2) { - t.Error("20-day trigger deleted, want kept") + 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) } -} - -// The Ping Scores history keeps an entry only while its ping_triggers row -// exists (planPingScoreHistoryReconcile drops the rest), and an entry is -// marked data_pruned exactly when its transmission is gone. So the trigger -// of a pruned transmission must survive until pingTriggerDays: following the -// transmission prune would remove every data_pruned entry at once. -func TestPrunePingTriggersKeepsTriggerOfPrunedTransmission(t *testing.T) { - store := newTestStore(t) - insertChanTx(t, store, "pingpruned01", "Alice: ping", "#ping") - var txID int64 - if err := store.db.QueryRow(`SELECT tx_id FROM ping_triggers`).Scan(&txID); err != nil { + if err := rows.Err(); err != nil { t.Fatal(err) } - mustExec(t, store, `UPDATE ping_triggers SET first_seen = ? WHERE tx_id = ?`, daysAgo(20), txID) - mustExec(t, store, `UPDATE transmissions SET first_seen = ? WHERE id = ?`, daysAgo(20), txID) - if n, err := store.PruneOldPackets(10); err != nil || n != 1 { - t.Fatalf("PruneOldPackets(10) = %d, %v; want the transmission pruned", n, err) - } + return out +} - if _, err := store.PrunePingTriggers(30); err != nil { +// 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) } - if !pingTriggerExists(t, store, txID) { - t.Error("trigger of a pruned transmission, inside pingTriggerDays, was deleted; want kept (data_pruned entries need it)") - } -} - -// pingTriggerDays unset: the transmission prune leaves ping_triggers alone, as -// before #329. -func TestPruneOldPacketsLeavesPingTriggers(t *testing.T) { store := newTestStore(t) - insertChanTx(t, store, "pingpruned02", "Bob: ping", "#ping") - mustExec(t, store, `UPDATE transmissions SET first_seen = ?`, daysAgo(90)) - mustExec(t, store, `UPDATE ping_triggers SET first_seen = ?`, daysAgo(90)) - if n, err := store.PruneOldPackets(30); err != nil || n != 1 { - t.Fatalf("PruneOldPackets(30) = %d, %v; want the transmission pruned", n, err) - } - runTableRetention(store, TableRetention{}, "test") - if got := countPingTriggers(t, store); got != 1 { - t.Errorf("ping_triggers = %d, want 1", got) + 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)) } -} -// Triggers are walked by tx_id, a batch of ids per transaction, because no -// index covers first_seen. Old and recent rows interleave across batches. -func TestPrunePingTriggersBatches(t *testing.T) { - setRetentionBatchRows(t, 2) - store := newTestStore(t) - ages := []int{90, 1, 90, 90, 1, 90, 90} - for i, age := range ages { - insertPingTrigger(t, store, int64(i+1), age) + if n, err := store.PruneOldPackets(cfg.PacketDaysOrZero()); err != nil || n != 1 { + t.Fatalf("PruneOldPackets = %d, %v; want the ping's transmission pruned", n, err) } - ResetWriterStatsForTest() + runTableRetention(store, cfg.TableRetention(), "test") - n, err := store.PrunePingTriggers(30) - if err != nil { - t.Fatal(err) - } - if n != 5 { - t.Errorf("deleted=%d, want 5", n) - } - for _, id := range []int64{2, 5} { - if !pingTriggerExists(t, store, id) { - t.Errorf("recent trigger %d deleted, want kept", id) + 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) } } - if got := store.WriterStatsSnapshot()["prune_ping_triggers"].Count; got != 4 { - t.Errorf("prune_ping_triggers transactions = %d, want 4 (7 ids in ranges of 2)", 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 four tables one 90-day-old row +// 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() @@ -358,8 +340,6 @@ func seedTableRetentionRows(t *testing.T, s *Store) { insertInactiveNode(t, s, "cc02", 1) insertNodeChange(t, s, "cc01", 90) insertNodeChange(t, s, "cc02", 1) - insertPingTrigger(t, s, 1, 90) - insertPingTrigger(t, s, 2, 1) staleObserver(t, s, "obs-cc01") if err := s.UpsertObserver("obs-cc02", "obs-cc02", "LAX", nil); err != nil { t.Fatal(err) @@ -367,14 +347,14 @@ func seedTableRetentionRows(t *testing.T, s *Store) { 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 four tables. +// 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", "ping_triggers", "observers"} { + 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) } @@ -389,7 +369,6 @@ func TestRunTableRetentionEachKnob(t *testing.T) { }{ {"inactive_nodes", TableRetention{InactiveNodeDays: 30}}, {"node_changes", TableRetention{NodeChangeDays: 30}}, - {"ping_triggers", TableRetention{PingTriggerDays: 30}}, {"observers", TableRetention{ObserverPurgeDays: 30}}, } for _, c := range cases { @@ -399,7 +378,7 @@ func TestRunTableRetentionEachKnob(t *testing.T) { runTableRetention(store, c.r, "test") - for _, table := range []string{"inactive_nodes", "node_changes", "ping_triggers", "observers"} { + for _, table := range []string{"inactive_nodes", "node_changes", "observers"} { want := 2 if table == c.table { want = 1 @@ -411,20 +390,3 @@ func TestRunTableRetentionEachKnob(t *testing.T) { }) } } - -// tx_id is a transmissions id, and those can be 0 or negative (the CI E2E -// fixture seeds both). The walk starts below every key. -func TestPrunePingTriggersNonPositiveTxID(t *testing.T) { - store := newTestStore(t) - insertPingTrigger(t, store, -1000000, 90) - insertPingTrigger(t, store, 0, 90) - insertPingTrigger(t, store, 1, 90) - - n, err := store.PrunePingTriggers(30) - if err != nil { - t.Fatal(err) - } - if n != 3 || countPingTriggers(t, store) != 0 { - t.Errorf("deleted=%d left=%d, want 3 and 0", n, countPingTriggers(t, store)) - } -} From 7228e08314531761573e1143c0c53d939e91aa77 Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 7 Oct 2026 12:05:44 +0000 Subject: [PATCH 5/5] fix(ingestor): drop the pingTriggerDays knob; ping_triggers is kept (#329) Scope change on #329: ping_triggers must not be pruned (Ping Scores all-time, #349/#241). Remove the knob end to end: - RetentionConfig/TableRetention lose PingTriggerDays; a config that still sets it is ignored. The startup log lists the three knobs. - PrunePingTriggers and its wiring in runTableRetention go, and with them deleteInKeyRanges, which only existed to share the key-range walk with it: PruneOrphanRouteMaskChanges is back to master's code. - The server read-only invariant no longer names PrunePingTriggers. Of the two server tests, the data_pruned-renders one stays (the #329 acceptance criterion); the one pinning the knob's cost goes. - The perf test no longer prunes ping_triggers and fails if retention deletes one. - config.example.json and the user guide drop the knob and say that ping_triggers is kept forever. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/config.go | 10 +-- cmd/ingestor/main.go | 4 +- cmd/ingestor/maintenance.go | 34 +++++++- cmd/ingestor/table_retention.go | 83 ++----------------- cmd/ingestor/table_retention_perf_test.go | 11 ++- .../ping_score_history_retention_329_test.go | 60 +++----------- cmd/server/readonly_invariant_test.go | 1 - config.example.json | 3 +- docs/user-guide/configuration.md | 21 +++-- 9 files changed, 73 insertions(+), 154 deletions(-) diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 0a70698cd..90d8ad097 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -236,21 +236,20 @@ type RetentionConfig struct { // 0 disables each. See TableRetention. InactiveNodeDays int `json:"inactiveNodeDays"` NodeChangeDays int `json:"nodeChangeDays"` - PingTriggerDays int `json:"pingTriggerDays"` 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. +// 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 - // PingTriggerDays deletes ping_triggers rows first seen longer ago. - PingTriggerDays int // ObserverPurgeDays hard-deletes observers that observerDays already // soft-deleted, once nothing still references them (upstream#1886). ObserverPurgeDays int @@ -258,7 +257,7 @@ type TableRetention struct { // Enabled reports whether any window is set. func (r TableRetention) Enabled() bool { - return r.InactiveNodeDays > 0 || r.NodeChangeDays > 0 || r.PingTriggerDays > 0 || r.ObserverPurgeDays > 0 + return r.InactiveNodeDays > 0 || r.NodeChangeDays > 0 || r.ObserverPurgeDays > 0 } // TableRetention returns the configured opt-in table retention windows, @@ -270,7 +269,6 @@ func (c *Config) TableRetention() TableRetention { return TableRetention{ InactiveNodeDays: max(c.Retention.InactiveNodeDays, 0), NodeChangeDays: max(c.Retention.NodeChangeDays, 0), - PingTriggerDays: max(c.Retention.PingTriggerDays, 0), ObserverPurgeDays: max(c.Retention.ObserverPurgeDays, 0), } } diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 612340c10..d5dfb1c2a 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -274,8 +274,8 @@ func main() { // 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 pingTriggerDays=%d observerPurgeDays=%d", - r.InactiveNodeDays, r.NodeChangeDays, r.PingTriggerDays, r.ObserverPurgeDays) + log.Printf("[prune] table retention enabled (0 = off): inactiveNodeDays=%d nodeChangeDays=%d observerPurgeDays=%d", + r.InactiveNodeDays, r.NodeChangeDays, r.ObserverPurgeDays) } runTableRetention(store, tableRetention, "startup") diff --git a/cmd/ingestor/maintenance.go b/cmd/ingestor/maintenance.go index 2fea20815..9029ae3a7 100644 --- a/cmd/ingestor/maintenance.go +++ b/cmd/ingestor/maintenance.go @@ -278,10 +278,36 @@ func (s *Store) PruneOrphanRouteMaskChanges() (int64, error) { if batch <= 0 { batch = 1000 } - total, err := s.deleteInKeyRanges("prune_route_mask_changes", "route_mask_changes", "id", - `NOT EXISTS (SELECT 1 FROM transmissions t WHERE t.id = route_mask_changes.transmission_id)`, batch) - if err != nil { - return total, err + var total, after int64 + for { + var examined, deleted int64 + err := s.WriterTx("prune_route_mask_changes", func(tx *sql.Tx) error { + var last sql.NullInt64 + if err := tx.QueryRow(`SELECT MAX(id), COUNT(*) FROM ( + SELECT id FROM route_mask_changes WHERE id > ? ORDER BY id LIMIT ?)`, + after, batch).Scan(&last, &examined); err != nil { + return fmt.Errorf("scan route_mask_changes: %w", err) + } + if examined == 0 { + return nil + } + res, err := tx.Exec(`DELETE FROM route_mask_changes WHERE id > ? AND id <= ? + AND NOT EXISTS (SELECT 1 FROM transmissions t WHERE t.id = route_mask_changes.transmission_id)`, + after, last.Int64) + if err != nil { + return fmt.Errorf("prune orphan route_mask_changes: %w", err) + } + deleted, _ = res.RowsAffected() + after = last.Int64 + return nil + }) + if err != nil { + return total, err + } + total += deleted + if examined < int64(batch) { + break + } } if total > 0 { log.Printf("[prune] deleted %d route_mask_changes rows of deleted transmissions", total) diff --git a/cmd/ingestor/table_retention.go b/cmd/ingestor/table_retention.go index fa6f4bd3a..ed295454b 100644 --- a/cmd/ingestor/table_retention.go +++ b/cmd/ingestor/table_retention.go @@ -4,21 +4,21 @@ import ( "database/sql" "fmt" "log" - "math" "strings" "time" ) // Opt-in retention for the tables nothing else prunes (#329): inactive_nodes, -// node_changes, ping_triggers 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. +// 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 -// (or, for ping_triggers, examines). These rows are small and their indexes -// few, so a batch is a few milliseconds. A var so tests can exercise several -// batches. +// 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, @@ -48,44 +48,6 @@ func (s *Store) deleteInBatches(tag, del string, args ...any) (int64, error) { } } -// deleteInKeyRanges walks table in ascending order of its integer key, batch -// keys per WriterTx (tagged tag), and deletes the rows of each range that -// match cond (args are cond's parameters). Each transaction's work is bounded -// by batch however sparse the matches are, which suits a cond no index -// serves. Returns the rows deleted; on error, those of the committed ranges. -func (s *Store) deleteInKeyRanges(tag, table, key, cond string, batch int, args ...any) (int64, error) { - scan := fmt.Sprintf(`SELECT MAX(%[2]s), COUNT(*) FROM (SELECT %[2]s FROM %[1]s WHERE %[2]s > ? ORDER BY %[2]s LIMIT ?)`, table, key) - del := fmt.Sprintf(`DELETE FROM %s WHERE %s > ? AND %[2]s <= ? AND (%s)`, table, key, cond) - var total int64 - after := int64(math.MinInt64) // below every key: ids can be 0 or negative - for { - var examined, deleted int64 - err := s.WriterTx(tag, func(tx *sql.Tx) error { - var last sql.NullInt64 - if err := tx.QueryRow(scan, after, batch).Scan(&last, &examined); err != nil { - return fmt.Errorf("scan %s: %w", table, err) - } - if examined == 0 { - return nil - } - res, err := tx.Exec(del, append([]any{after, last.Int64}, args...)...) - if err != nil { - return fmt.Errorf("prune %s: %w", table, err) - } - deleted, _ = res.RowsAffected() - after = last.Int64 - return nil - }) - if err != nil { - return total, err - } - total += deleted - if examined < int64(batch) { - return total, nil - } - } -} - func retentionCutoff(days int) string { return time.Now().UTC().AddDate(0, 0, -days).Format(time.RFC3339) } @@ -139,32 +101,6 @@ func (s *Store) PruneNodeChanges(days int) (int64, error) { return n, err } -// PrunePingTriggers deletes ping_triggers rows first seen more than days ago -// (retention.pingTriggerDays, #329); days <= 0 disables it. -// -// The window is its own and does not follow the transmission prune. The -// server's Ping Scores history keeps an entry only while its trigger row -// exists (planPingScoreHistoryReconcile deletes the rest), and an entry is -// marked data_pruned exactly when its transmission is gone. Deleting the -// triggers of pruned transmissions would therefore drop every data_pruned -// entry from the all-time records at once; with its own window an entry -// renders until its trigger is pingTriggerDays old, and its sender name then -// leaves both the trigger and the history entry. -// -// No index covers first_seen, so the table is walked by tx_id -// (deleteInKeyRanges): one primary-key range per transaction. -func (s *Store) PrunePingTriggers(days int) (int64, error) { - if days <= 0 { - return 0, nil - } - n, err := s.deleteInKeyRanges("prune_ping_triggers", "ping_triggers", "tx_id", - "first_seen < ?", retentionBatchRows, retentionCutoff(days)) - if n > 0 { - log.Printf("[prune] deleted %d ping_triggers 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. @@ -272,7 +208,6 @@ func runTableRetention(store *Store, r TableRetention, when string) { }{ {r.InactiveNodeDays, store.PruneInactiveNodes}, {r.NodeChangeDays, store.PruneNodeChanges}, - {r.PingTriggerDays, store.PrunePingTriggers}, {r.ObserverPurgeDays, store.PurgeStaleObservers}, } { if _, err := p.prune(p.days); err != nil { diff --git a/cmd/ingestor/table_retention_perf_test.go b/cmd/ingestor/table_retention_perf_test.go index cbca1d9e7..32198d0db 100644 --- a/cmd/ingestor/table_retention_perf_test.go +++ b/cmd/ingestor/table_retention_perf_test.go @@ -18,8 +18,8 @@ import ( // 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, -// 600 observers (300 soft-deleted, half of those still with metrics) with +// (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 @@ -78,7 +78,7 @@ func TestTableRetentionTiming(t *testing.T) { } tables := func(days int) step { return step{"#329 table retention", func() error { - runTableRetention(store, TableRetention{InactiveNodeDays: days, NodeChangeDays: days, PingTriggerDays: days, ObserverPurgeDays: days}, "perf") + runTableRetention(store, TableRetention{InactiveNodeDays: days, NodeChangeDays: days, ObserverPurgeDays: days}, "perf") if strings.Contains(logBuf.String(), "table retention error") { return fmt.Errorf("%s", logBuf.String()) } @@ -86,7 +86,7 @@ func TestTableRetentionTiming(t *testing.T) { }} } components := []string{"prune_packets", "prune_metrics", "prune_observers", - "prune_inactive_nodes", "prune_node_changes", "prune_ping_triggers", "purge_observers"} + "prune_inactive_nodes", "prune_node_changes", "purge_observers"} for _, pass := range []struct { name string @@ -117,6 +117,9 @@ func TestTableRetentionTiming(t *testing.T) { 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]") diff --git a/cmd/server/ping_score_history_retention_329_test.go b/cmd/server/ping_score_history_retention_329_test.go index b55936e26..6140729af 100644 --- a/cmd/server/ping_score_history_retention_329_test.go +++ b/cmd/server/ping_score_history_retention_329_test.go @@ -5,18 +5,19 @@ import ( "time" ) -// #329: the ingestor's retention.pingTriggerDays deletes ping_triggers rows by -// their own age, not when their transmission is pruned. These tests pin the -// two halves of why, on the server side of the shared DB. +// #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). -// dataPrunedFixture settles one ping 40 days old, removes its transmission -// and observations the way the ingestor's packetDays prune does (the -// ping_triggers row stays), and runs the cycle that marks it data_pruned. -func dataPrunedFixture(t *testing.T, hash string) (*engineFixture, int64) { - t.Helper() +// 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, hash, firstSeen) + txID := setupSettledPingWithGoneData(t, fx, "keeptrigger329", firstSeen) if _, err := fx.srv.db.conn.Exec(`DELETE FROM transmissions WHERE id = ?`, txID); err != nil { t.Fatal(err) } @@ -27,13 +28,6 @@ func dataPrunedFixture(t *testing.T, hash string) (*engineFixture, int64) { 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) } - return fx, txID -} - -// A data_pruned entry whose trigger row is still there keeps rendering in the -// all-time records, cycle after cycle, with its transmission gone. -func TestPingScoresDataPrunedEntryRendersWhileTriggerKept(t *testing.T) { - fx, _ := dataPrunedFixture(t, "keeptrigger329") fx.clock.Advance(time.Minute) if _, err := fx.engine.Cycle(); err != nil { t.Fatal(err) @@ -47,37 +41,3 @@ func TestPingScoresDataPrunedEntryRendersWhileTriggerKept(t *testing.T) { t.Errorf("data_pruned entry not rendered in the all-time records: %+v", snap.MostHopsPing) } } - -// Once its trigger row is deleted, the entry leaves the history store too -// (and with it the sender name). This is why ping_triggers cannot follow the -// transmission prune: every data_pruned entry has a pruned transmission. -func TestPingScoresDataPrunedEntryDroppedWithItsTrigger(t *testing.T) { - fx, txID := dataPrunedFixture(t, "droptrigger329") - if _, err := fx.srv.db.conn.Exec(`DELETE FROM ping_triggers WHERE tx_id = ?`, txID); err != nil { - t.Fatal(err) - } - fx.clock.Advance(time.Minute) - if _, err := fx.engine.Cycle(); err != nil { - t.Fatal(err) - } - - if _, ok := fx.engine.index.Get(txID); ok { - t.Error("history entry still indexed after its trigger was deleted") - } - stored, err := fx.store.LoadAll() - if err != nil { - t.Fatal(err) - } - for _, e := range stored { - if e.TxID == txID { - t.Errorf("history entry still stored after its trigger was deleted: %+v", e) - } - } - 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 == "droptrigger329" { - t.Error("entry of a deleted trigger still rendered") - } -} diff --git a/cmd/server/readonly_invariant_test.go b/cmd/server/readonly_invariant_test.go index ca2819db6..03c27bf48 100644 --- a/cmd/server/readonly_invariant_test.go +++ b/cmd/server/readonly_invariant_test.go @@ -95,7 +95,6 @@ func TestServerDBHasNoWriteMethods(t *testing.T) { // #329 opt-in table retention: ingestor-only. "PruneInactiveNodes", "PruneNodeChanges", - "PrunePingTriggers", "PurgeStaleObservers", // #738 / one-click geo-prune: the DELETE must live on the // ingestor's *Store. The server's HTTP handler now enqueues a diff --git a/config.example.json b/config.example.json index 4b298c97a..28cf27d93 100644 --- a/config.example.json +++ b/config.example.json @@ -17,9 +17,8 @@ "clientRxDays": 30, "inactiveNodeDays": 0, "nodeChangeDays": 0, - "pingTriggerDays": 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. pingTriggerDays: ping_triggers rows (Ping Scores, holding sender names) first seen more than N days ago are deleted, by their own age and not with their transmission, so Ping Scores history entries already marked data_pruned keep rendering until then; the server then drops the entry, sender name included, from its history store, so N bounds the all-time Ping Scores records. 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 four to 30.", + "_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 ef3ac449f..9e9126068 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -100,7 +100,6 @@ How long (in hours) before a node is marked degraded or silent: | `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.pingTriggerDays` | `0` | Ping Scores triggers (`ping_triggers`, which hold sender names) first seen more than N days ago are 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 @@ -117,24 +116,24 @@ 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 four opt-in settings (`inactiveNodeDays`, `nodeChangeDays`, -`pingTriggerDays`, `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. +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`. -- `pingTriggerDays` counts from when the ping was first seen, not from when its - packet was pruned. Ping Scores keeps a ping in its all-time records after - `packetDays` has removed the packet, and drops it, with its sender name, once - the trigger is deleted. So `pingTriggerDays` also bounds those records. - `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