mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-09-15 01:06:08 +00:00
feat(retention): add observerPurgeDays hard-delete for long-inactive observers (#1886)
## 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) <noreply@anthropic.com>
This commit is contained in:
efiten
+16
-1
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
+2
-1
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user