port(upstream#1886): opt-in observerPurgeDays hard-delete for long-inactive observers

cmd/ingestor/config.go

@@ -144,7 +144,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"`
@@ -163,6 +169,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 {

cmd/ingestor/db.go

@@ -1839,6 +1839,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

cmd/ingestor/main.go

@@ -264,6 +264,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)
@@ -323,9 +331,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)
 		}
 	}()

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)
+			}
+		})
+	}
+}

cmd/server/readonly_invariant_test.go

@@ -92,6 +92,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.

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,