From d821d9a390422179b6a5689cfdd33a77dd42b56e Mon Sep 17 00:00:00 2001 From: efiten Date: Wed, 2 Sep 2026 14:22:30 +0200 Subject: [PATCH] feat(retention): add observerPurgeDays hard-delete for long-inactive observers (#1886) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem `RemoveStaleObservers` only soft-deletes — it sets `inactive = 1` and the row stays forever. On a long-running deployment those rows just accumulate: on a two-year-old instance roughly 25% of the `observers` table was rows nobody can ever see again. There is currently no way to reclaim them. ## Fix A second retention stage. `PurgeStaleObservers` hard-deletes rows that are: - already `inactive = 1` (so the soft-delete stage owns the decision of *when* an observer goes stale), **and** - older than `retention.observerPurgeDays`, **and** - referenced by nothing. New config field `retention.observerPurgeDays`, default `0` = disabled. Existing deployments are unaffected until they opt in. Set it above both `observerDays` and `packetDays` — below those the reference guards keep every candidate row anyway. ## Why the reference guards are the point `observations.observer_idx` is a bare rowid with no foreign key. Deleting a still-referenced observer silently orphans history — `packets_v` stops resolving the observer and those packets get mis-attributed. Nothing errors; the data just quietly goes wrong. So the statement guards on all three referencing tables: ```sql 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 dropped_packets d WHERE d.observer_id = observers.id) ``` This is correctness, not defensive padding — it was found the hard way, by orphaning 280 observation rows during a manual purge that skipped one of these checks. Each guard has its own test. ## Performance Each `NOT EXISTS` is an index seek per candidate row (`idx_observations_observer_idx`, `idx_dropped_observer`, the `observer_metrics` PK), and `observers` is O(100). It runs on the existing daily retention tick alongside `RemoveStaleObservers`, never on the ingest path. ## Tests Eight tests in `cmd/ingestor/observer_purge_test.go`, written before the implementation: - deletes an unreferenced stale row - keeps a row referenced by `observations` — and asserts zero orphans afterwards - keeps a row referenced by `observer_metrics` - keeps a row referenced by `dropped_packets` - keeps a row that is old enough but still `inactive = 0` - keeps a row inside the retention window - no-ops when disabled (`0` and `-1`) - config accessor table test ## Invariant Writes stay in `cmd/ingestor` per #1283. `cmd/server/readonly_invariant_test.go` now also forbids `PurgeStaleObservers` as a method on the server's `*DB`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5 (1M context) --- cmd/ingestor/config.go | 17 +- cmd/ingestor/db.go | 36 +++++ cmd/ingestor/main.go | 10 ++ cmd/ingestor/observer_purge_test.go | 222 ++++++++++++++++++++++++++ cmd/server/readonly_invariant_test.go | 1 + config.example.json | 3 +- 6 files changed, 287 insertions(+), 2 deletions(-) create mode 100644 cmd/ingestor/observer_purge_test.go diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 5074b976..095f6304 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -153,7 +153,13 @@ func (c *Config) ClientRxCoverageEnabled() bool { type RetentionConfig struct { NodeDays int `json:"nodeDays"` ObserverDays int `json:"observerDays"` - MetricsDays int `json:"metricsDays"` + // ObserverPurgeDays hard-deletes observers that observerDays already + // soft-deleted, once they are older than this and no observation, + // metric or dropped-packet row still references them; 0 disables. + // Only meaningful above observerDays and packetDays — below those the + // reference guards keep every candidate row anyway. + ObserverPurgeDays int `json:"observerPurgeDays"` + MetricsDays int `json:"metricsDays"` // PacketDays is the retention window for transmissions (#1283). // Ownership moved from cmd/server to cmd/ingestor; 0 disables. PacketDays int `json:"packetDays"` @@ -172,6 +178,15 @@ func (c *Config) PacketDaysOrZero() int { return 0 } +// ObserverPurgeDaysOrZero returns the configured retention.observerPurgeDays +// or 0 (disabled) if not set. +func (c *Config) ObserverPurgeDaysOrZero() int { + if c.Retention != nil && c.Retention.ObserverPurgeDays > 0 { + return c.Retention.ObserverPurgeDays + } + return 0 +} + // ClientRxDaysOrZero returns the configured retention.clientRxDays or 0 // (disabled) if not set. func (c *Config) ClientRxDaysOrZero() int { diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index bd67c8c1..0df01fac 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -1673,6 +1673,42 @@ func (s *Store) RemoveStaleObservers(observerDays int) (int64, error) { return removed, nil } +// PurgeStaleObservers hard-deletes rows RemoveStaleObservers already soft-deleted, +// once they are older than purgeDays and nothing references them any more. It is +// the second stage of observer retention: the soft-delete hides the observer, +// this reclaims the row after its packets have aged out via packetDays. +// +// 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. The three NOT EXISTS guards +// are what make the delete safe; they are correctness, not defensive padding. +// Each is an index seek per candidate row (observers is O(100)), so the whole +// statement stays cheap enough for the daily retention tick. +// +// purgeDays <= 0 disables the purge (the default). +func (s *Store) PurgeStaleObservers(purgeDays int) (int64, error) { + if purgeDays <= 0 { + return 0, nil // disabled + } + cutoff := time.Now().UTC().AddDate(0, 0, -purgeDays).Format(time.RFC3339) + // Tagged for /api/perf writer-lock visibility (#1340). + result, err := s.instrumentedExec("purge_observers", ` + DELETE FROM observers + WHERE inactive = 1 + AND last_seen < ? + 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 dropped_packets d WHERE d.observer_id = observers.id)`, cutoff) + if err != nil { + return 0, fmt.Errorf("purge stale observers: %w", err) + } + purged, _ := result.RowsAffected() + if purged > 0 { + log.Printf("Purged %d inactive observer(s) with no remaining data (not seen in %d days)", purged, purgeDays) + } + return purged, nil +} + // DroppedPacket holds data for a packet rejected during ingest. type DroppedPacket struct { Hash string diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 963be045..bc26cc69 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -252,6 +252,14 @@ func main() { observerDays := cfg.ObserverDaysOrDefault() store.RemoveStaleObservers(observerDays) + // Observer purge: second stage, hard-deletes long-inactive observers whose + // packets have already aged out. Always runs after the soft-delete so a row + // crossing both thresholds is finalised in a single pass. 0 = disabled. + observerPurgeDays := cfg.ObserverPurgeDaysOrZero() + if _, err := store.PurgeStaleObservers(observerPurgeDays); err != nil { + log.Printf("[prune] error: %v", err) + } + // Metrics retention: prune old metrics on startup metricsDays := cfg.MetricsRetentionDays() store.PruneOldMetrics(metricsDays) @@ -310,9 +318,11 @@ func main() { go func() { time.Sleep(90 * time.Second) // stagger after metrics prune store.RemoveStaleObservers(observerDays) + store.PurgeStaleObservers(observerPurgeDays) store.RunIncrementalVacuum(vacuumPages) for range observerRetentionTicker.C { store.RemoveStaleObservers(observerDays) + store.PurgeStaleObservers(observerPurgeDays) store.RunIncrementalVacuum(vacuumPages) } }() diff --git a/cmd/ingestor/observer_purge_test.go b/cmd/ingestor/observer_purge_test.go new file mode 100644 index 00000000..8a3f6a6a --- /dev/null +++ b/cmd/ingestor/observer_purge_test.go @@ -0,0 +1,222 @@ +package main + +import ( + "testing" + "time" +) + +// staleObserver inserts an observer already soft-deleted (inactive = 1) with +// last_seen ageDays 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") + } +} + +func TestObserverPurgeDaysOrZero(t *testing.T) { + tests := []struct { + name string + cfg *Config + want int + }{ + {"nil retention", &Config{}, 0}, + {"unset", &Config{Retention: &RetentionConfig{}}, 0}, + {"configured", &Config{Retention: &RetentionConfig{ObserverPurgeDays: 30}}, 30}, + {"negative disables", &Config{Retention: &RetentionConfig{ObserverPurgeDays: -1}}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.cfg.ObserverPurgeDaysOrZero(); got != tt.want { + t.Errorf("ObserverPurgeDaysOrZero() = %d, want %d", got, tt.want) + } + }) + } +} diff --git a/cmd/server/readonly_invariant_test.go b/cmd/server/readonly_invariant_test.go index ab1fadef..f97baa48 100644 --- a/cmd/server/readonly_invariant_test.go +++ b/cmd/server/readonly_invariant_test.go @@ -98,6 +98,7 @@ func TestServerDBHasNoWriteMethods(t *testing.T) { "PruneOldPackets", "PruneOldMetrics", "RemoveStaleObservers", + "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 ac8d971c..2b7e13b2 100644 --- a/config.example.json +++ b/config.example.json @@ -10,9 +10,10 @@ "retention": { "nodeDays": 7, "observerDays": 14, + "observerPurgeDays": 0, "packetDays": 30, "clientRxDays": 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). 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." + "_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). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). 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": { "vacuumOnStartup": false,