diff --git a/README.md b/README.md index 53e1fd7..ef3026d 100644 --- a/README.md +++ b/README.md @@ -204,6 +204,15 @@ nodes: delete_after: 720h # delete a node entirely after this long unseen (default: 30 days, same default as packets.retention) clock_drift_threshold: 5m # |device clock - server clock| above which clockOutOfSync=true for a repeater/room server (default: 5m) +# Optional observer age-out (disabled by default; set e.g. 720h to opt in). +# Enable only when one Beacon ingest process owns the database: presence-cache +# preparation is local to that process, including both of its MQTT workers. +# Deletes at most 1000 observers per background.cleanup interval, only when +# last_seen/last_status_at are old and no observations, telemetry or ownership remain. +# Broker/location/scope metadata cascades; a returning observer receives a new ID. +observers: + delete_after: 0s # omitted or nonpositive disables deletion + # Redis caching layer (optional). # Caches read-heavy, slow-changing responses to reduce PostgreSQL load. # Connection details (address, password, database) are set via environment diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index fd92c27..ef46a38 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -29,6 +29,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/presence" "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" + "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" "github.com/joho/godotenv" ) @@ -255,11 +256,19 @@ func main() { go broker1.Start(ctx) go broker2.Start(ctx) - scheduler := background.New([]background.Task{ + tasks := []background.Task{ background.ViewRefreshTask(store, resolved.ViewRefreshInterval), background.CleanupTask(store, resolved.TelemetryRetention, resolved.PacketRetention, resolved.NodeDeleteAfter, resolved.CleanupInterval), background.ReconfirmTask(store, resolved.RouteRetention, resolved.RouteGrace, int64(resolved.RouteMinObservations), resolved.ReconfirmInterval), - }) + } + if resolved.ObserverDeleteAfter > 0 { + var onDelete func(context.Context, uuid.UUID) + if cr, ok := reader.(*cache.CachedReader); ok { + onDelete = cr.InvalidateObserver + } + tasks = append(tasks, background.ObserverCleanupTask(coalescer, resolved.ObserverDeleteAfter, resolved.CleanupInterval, onDelete)) + } + scheduler := background.New(tasks) go scheduler.Start(ctx) // ── HTTP server ────────────────────────────────────────────────────────── diff --git a/config.yaml.example b/config.yaml.example index b039cad..cc91a39 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -103,6 +103,13 @@ websocket: # delete_after: 720h # default: 30 days, same default as packets.retention # clock_drift_threshold: 5m # default: 5m +# Optional observer age-out. Omitted or nonpositive leaves observers intact. +# Enable only with one Beacon ingest process for this database. +# At most 1000 old observers are deleted per background.cleanup interval; +# retained observations, telemetry and ownership records prevent deletion. +#observers: +# delete_after: 720h # opt in to 30 days unseen + # CORS configuration. # Controls which origins, methods and headers are allowed for cross-origin requests. # Defaults to allowing all origins with GET/HEAD/OPTIONS if omitted — appropriate diff --git a/db/observer_retention_integration_test.go b/db/observer_retention_integration_test.go new file mode 100644 index 0000000..8672ea4 --- /dev/null +++ b/db/observer_retention_integration_test.go @@ -0,0 +1,147 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "bytes" + "context" + "os" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/presence" + "github.com/jackc/pgx/v5" +) + +func TestObserverRetentionPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN for the PostgreSQL regression test") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + exec := func(query string, args ...any) { + t.Helper() + if _, err := tx.Exec(ctx, query, args...); err != nil { + t.Fatal(err) + } + } + for _, table := range []string{"observers", "packet_observations", "observer_telemetry", "observer_owners", "observer_brokers", "observer_locations", "observer_scopes"} { + exec("CREATE TEMP TABLE " + table + " (LIKE public." + table + " INCLUDING ALL) ON COMMIT DROP") + if table != "observers" { + fk := "ALTER TABLE pg_temp." + table + " ADD FOREIGN KEY (observer_id) REFERENCES pg_temp.observers(id)" + if table != "packet_observations" { + fk += " ON DELETE CASCADE" + } + exec(fk) + } + } + store := &Store{q: sqlc.New(tx)} + cutoff := time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC) + exec(`INSERT INTO observers(public_key,display_name,last_seen,last_status_at) +SELECT decode(lpad(to_hex(i),64,'0'),'hex'),label,$1::timestamptz+seen*interval '1 second', + $1::timestamptz+status*interval '1 second' +FROM (VALUES (1,'expired',-1,NULL),(2,'boundary',0,NULL),(3,'recent',1,NULL), + (4,'recent status',-1,1),(5,'status boundary',-1,0), + (6,'packet history',-1,NULL),(7,'telemetry',-1,NULL),(8,'owned',-1,NULL)) v(i,label,seen,status)`, cutoff) + exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count) +SELECT 1,'\x01',id,'YYZ',last_seen,0,1,0 FROM observers WHERE display_name='packet history'; +INSERT INTO observer_telemetry(id,observer_id,reported_at) +SELECT 1,id,last_seen FROM observers WHERE display_name='telemetry'; +INSERT INTO observer_owners(observer_id,notes) +SELECT id,'fixture ownership without a node link' FROM observers WHERE display_name='owned'; +INSERT INTO observer_brokers(observer_id,broker_name) +SELECT id,'fixture' FROM observers WHERE display_name='expired'; +INSERT INTO observer_locations(observer_id,reported_at) +SELECT id,last_seen FROM observers WHERE display_name='expired'; +INSERT INTO observer_scopes(observer_id,scope_id) +SELECT id,1 FROM observers WHERE display_name='expired';`) + ids, err := store.DeleteOldObservers(ctx, cutoff) + if err != nil || len(ids) != 1 { + t.Fatalf("expected one unreferenced expired observer deleted, got %d: %v", len(ids), err) + } + var remaining string + if err := tx.QueryRow(ctx, "SELECT string_agg(display_name,',' ORDER BY display_name) FROM observers").Scan(&remaining); err != nil { + t.Fatal(err) + } + if remaining != "boundary,owned,packet history,recent,recent status,status boundary,telemetry" { + t.Fatalf("incorrect retained observers: %s", remaining) + } + var history, metadata int + if err := tx.QueryRow(ctx, `SELECT + (SELECT count(*) FROM packet_observations)+(SELECT count(*) FROM observer_telemetry)+(SELECT count(*) FROM observer_owners), + (SELECT count(*) FROM observer_brokers)+(SELECT count(*) FROM observer_locations)+(SELECT count(*) FROM observer_scopes)`).Scan(&history, &metadata); err != nil { + t.Fatal(err) + } + if history != 3 || metadata != 0 { + t.Fatalf("history=%d, cascading metadata=%d; want 3, 0", history, metadata) + } + if ids, err := store.DeleteOldObservers(ctx, cutoff); err != nil || len(ids) != 0 { + t.Fatalf("repeated cleanup changed protected rows: %v, %v", ids, err) + } + + // All relations below are temporary; no persistent fixtures or sequences are used. + exec("TRUNCATE pg_temp.observers CASCADE") + coalescer := presence.New(store, time.Second, time.Second) + pubkey := bytes.Repeat([]byte{42}, 32) + oldID, _, err := coalescer.UpsertObserver(ctx, pubkey) + if err != nil { + t.Fatal(err) + } + if err := coalescer.UpsertObserverBroker(ctx, oldID, "fixture"); err != nil { + t.Fatal(err) + } + // A future cutoff simulates this cached observer having aged past retention. + ids, err = coalescer.DeleteOldObservers(ctx, time.Now().Add(time.Hour)) + if err != nil || len(ids) != 1 || ids[0] != oldID { + t.Fatalf("cached expired observer was not deleted: %v, %v", ids, err) + } + newID, _, err := coalescer.UpsertObserver(ctx, pubkey) + if err != nil || newID == oldID { + t.Fatalf("returning observer reused a deleted ID: %v", err) + } + if err := coalescer.UpsertObserverBroker(ctx, newID, "fixture"); err != nil { + t.Fatal(err) + } + var brokers int + if err := tx.QueryRow(ctx, "SELECT count(*) FROM observer_brokers WHERE observer_id=$1", newID).Scan(&brokers); err != nil || brokers != 1 { + t.Fatalf("returning observer's broker was not recreated: count=%d, err=%v", brokers, err) + } + // The persisted row looks stale while fresh activity is still coalesced. + cutoff = time.Now().Add(-24 * time.Hour) + exec("UPDATE observers SET last_seen=$1 WHERE id=$2", cutoff.Add(-time.Hour), newID) + if _, _, err := coalescer.UpsertObserver(ctx, pubkey); err != nil { + t.Fatal(err) + } + if ids, err := coalescer.DeleteOldObservers(ctx, cutoff); err != nil || len(ids) != 0 { + t.Fatalf("pending presence did not protect a returning observer: %v, %v", ids, err) + } + var fresh bool + if err := tx.QueryRow(ctx, "SELECT last_seen >= $1 AND observation_count=1 FROM observers WHERE id=$2", cutoff, newID).Scan(&fresh); err != nil || !fresh { + t.Fatalf("pending presence was not persisted before cleanup: %v", err) + } + + exec("TRUNCATE pg_temp.observers CASCADE") + exec(`INSERT INTO observers(public_key,last_seen) +SELECT decode(lpad(to_hex(i),64,'0'),'hex'),$1::timestamptz-interval '1 second' +FROM generate_series(1,1005) i`, cutoff) + for _, want := range []int{1000, 5, 0} { + ids, err := store.DeleteOldObservers(ctx, cutoff) + if err != nil || len(ids) != want { + t.Fatalf("bounded cleanup removed %d observers, want %d: %v", len(ids), want, err) + } + t.Logf("bounded cleanup deleted %d observers", len(ids)) + } +} diff --git a/db/observers.go b/db/observers.go index a58569b..c8036ca 100644 --- a/db/observers.go +++ b/db/observers.go @@ -338,3 +338,7 @@ func (s *Store) IsObserverByPubkey(ctx context.Context, pubkey []byte) bool { func (s *Store) DeleteOldTelemetry(ctx context.Context, cutoff time.Time) error { return s.q.DeleteOldTelemetry(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true}) } + +func (s *Store) DeleteOldObservers(ctx context.Context, cutoff time.Time) ([]uuid.UUID, error) { + return s.q.DeleteOldObservers(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true}) +} diff --git a/db/queries/queries.sql b/db/queries/queries.sql index c039e59..9771a26 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -287,6 +287,25 @@ LIMIT $3; -- Deletes telemetry rows older than the given cutoff. Called by the cleanup goroutine. DELETE FROM observer_telemetry WHERE reported_at < $1; +-- name: DeleteOldObservers :many +-- Opt-in age-out: preserve retained history and manually recorded ownership. +-- Bound deletions per cleanup tick and skip observers being updated by ingest. +WITH expired AS ( + SELECT o.id + FROM observers o + WHERE o.last_seen < $1 + AND (o.last_status_at IS NULL OR o.last_status_at < $1) + AND NOT EXISTS (SELECT 1 FROM packet_observations po WHERE po.observer_id = o.id) + AND NOT EXISTS (SELECT 1 FROM observer_telemetry ot WHERE ot.observer_id = o.id) + AND NOT EXISTS (SELECT 1 FROM observer_owners oo WHERE oo.observer_id = o.id) + ORDER BY o.last_seen, o.id + LIMIT 1000 + FOR UPDATE OF o SKIP LOCKED +) +DELETE FROM observers o USING expired e +WHERE o.id = e.id +RETURNING o.id; + -- ============================================================ -- OBSERVER BROKERS -- ============================================================ diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index e86eb2c..c34fe77 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -71,6 +71,21 @@ func (mr *MockQuerierMockRecorder) DeleteOldNodes(ctx, lastSeen any) *gomock.Cal return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOldNodes", reflect.TypeOf((*MockQuerier)(nil).DeleteOldNodes), ctx, lastSeen) } +// DeleteOldObservers mocks base method. +func (m *MockQuerier) DeleteOldObservers(ctx context.Context, lastSeen pgtype.Timestamptz) ([]uuid.UUID, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteOldObservers", ctx, lastSeen) + ret0, _ := ret[0].([]uuid.UUID) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// DeleteOldObservers indicates an expected call of DeleteOldObservers. +func (mr *MockQuerierMockRecorder) DeleteOldObservers(ctx, lastSeen any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOldObservers", reflect.TypeOf((*MockQuerier)(nil).DeleteOldObservers), ctx, lastSeen) +} + // DeleteOldPackets mocks base method. func (m *MockQuerier) DeleteOldPackets(ctx context.Context, lastHeardAt pgtype.Timestamptz) error { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 118d1a0..1c5f8c9 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -22,6 +22,9 @@ type Querier interface { // dangling in old routes there, but ReconfirmTask already prunes stale/ambiguous routes // periodically and will clean those up on its own schedule. DeleteOldNodes(ctx context.Context, lastSeen pgtype.Timestamptz) error + // Opt-in age-out: preserve retained history and manually recorded ownership. + // Bound deletions per cleanup tick and skip observers being updated by ingest. + DeleteOldObservers(ctx context.Context, lastSeen pgtype.Timestamptz) ([]uuid.UUID, error) // Deletes packets and their observations older than the given cutoff. // packet_observations cascade-delete via FK. DeleteOldPackets(ctx context.Context, lastHeardAt pgtype.Timestamptz) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index ef7550c..8581c4c 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -40,6 +40,46 @@ func (q *Queries) DeleteOldNodes(ctx context.Context, lastSeen pgtype.Timestampt return err } +const deleteOldObservers = `-- name: DeleteOldObservers :many +WITH expired AS ( + SELECT o.id + FROM observers o + WHERE o.last_seen < $1 + AND (o.last_status_at IS NULL OR o.last_status_at < $1) + AND NOT EXISTS (SELECT 1 FROM packet_observations po WHERE po.observer_id = o.id) + AND NOT EXISTS (SELECT 1 FROM observer_telemetry ot WHERE ot.observer_id = o.id) + AND NOT EXISTS (SELECT 1 FROM observer_owners oo WHERE oo.observer_id = o.id) + ORDER BY o.last_seen, o.id + LIMIT 1000 + FOR UPDATE OF o SKIP LOCKED +) +DELETE FROM observers o USING expired e +WHERE o.id = e.id +RETURNING o.id +` + +// Opt-in age-out: preserve retained history and manually recorded ownership. +// Bound deletions per cleanup tick and skip observers being updated by ingest. +func (q *Queries) DeleteOldObservers(ctx context.Context, lastSeen pgtype.Timestamptz) ([]uuid.UUID, error) { + rows, err := q.db.Query(ctx, deleteOldObservers, lastSeen) + if err != nil { + return nil, err + } + defer rows.Close() + items := []uuid.UUID{} + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, err + } + items = append(items, id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const deleteOldPackets = `-- name: DeleteOldPackets :exec DELETE FROM packets WHERE last_heard_at < $1 ` diff --git a/internal/background/tasks.go b/internal/background/tasks.go index f2f3ec7..a19b6e4 100644 --- a/internal/background/tasks.go +++ b/internal/background/tasks.go @@ -10,6 +10,7 @@ import ( "time" "github.com/MeshCore-Beacon/beacon-server/db" + "github.com/google/uuid" ) type viewRefresher interface { @@ -83,6 +84,33 @@ func CleanupTask(store *db.Store, telemetryRetention, packetRetention, nodeDelet // table gets fully re-checked roughly daily. const reconfirmBatchSize = 750_000 +type observerCleaner interface { + DeleteOldObservers(context.Context, time.Time) ([]uuid.UUID, error) +} + +// ObserverCleanupTask ages out unreferenced observers through the presence +// coalescer, then invalidates any cached details for the deleted IDs. +func ObserverCleanupTask(store observerCleaner, deleteAfter, interval time.Duration, onDelete func(context.Context, uuid.UUID)) Task { + return Task{ + Name: "observer_cleanup", Interval: interval, + Run: func(ctx context.Context) error { + if deleteAfter <= 0 { + return nil + } + ids, err := store.DeleteOldObservers(ctx, time.Now().Add(-deleteAfter)) + if err != nil { + return err + } + if onDelete != nil { + for _, id := range ids { + onDelete(ctx, id) + } + } + return nil + }, + } +} + // ReconfirmTask returns a Task that prunes aged routes first, then reconfirms // stale and ambiguous resolved paths and neighbors, so known_routes only ever // has one writer at a time. diff --git a/internal/background/tasks_test.go b/internal/background/tasks_test.go index cee90ff..b84973a 100644 --- a/internal/background/tasks_test.go +++ b/internal/background/tasks_test.go @@ -10,6 +10,8 @@ import ( "strings" "testing" "time" + + "github.com/google/uuid" ) type refreshStub struct { @@ -17,6 +19,72 @@ type refreshStub struct { errs map[string]error } +type observerCleanupStub struct { + ids []uuid.UUID + err error + calls int + cutoff time.Time +} + +func (s *observerCleanupStub) DeleteOldObservers(ctx context.Context, cutoff time.Time) ([]uuid.UUID, error) { + s.calls++ + s.cutoff = cutoff + if err := ctx.Err(); err != nil { + return nil, err + } + return s.ids, s.err +} + +func TestObserverCleanupTask(t *testing.T) { + failure := errors.New("cleanup unavailable") + for _, tc := range []struct { + name string + retention time.Duration + err error + cancelled bool + }{ + {"disabled", 0, nil, false}, + {"negative disables", -time.Hour, nil, false}, + {"enabled", 24 * time.Hour, nil, false}, + {"failure", 24 * time.Hour, failure, false}, + {"cancelled", 24 * time.Hour, context.Canceled, true}, + } { + t.Run(tc.name, func(t *testing.T) { + store := &observerCleanupStub{ids: []uuid.UUID{uuid.New(), uuid.New()}, err: tc.err} + var invalidated []uuid.UUID + task := ObserverCleanupTask(store, tc.retention, time.Minute, func(_ context.Context, id uuid.UUID) { + invalidated = append(invalidated, id) + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if tc.cancelled { + cancel() + } + before := time.Now().Add(-tc.retention) + err := task.Run(ctx) + if !errors.Is(err, tc.err) || task.Name != "observer_cleanup" || task.Interval != time.Minute { + t.Fatalf("unexpected task outcome: %v, %+v", err, task) + } + if tc.retention <= 0 { + if store.calls != 0 || len(invalidated) != 0 { + t.Fatal("disabled retention performed work") + } + return + } + if store.calls != 1 || store.cutoff.Before(before) || store.cutoff.After(time.Now().Add(-tc.retention)) { + t.Fatal("incorrect cleanup cutoff or number of calls") + } + if tc.err != nil { + if len(invalidated) != 0 { + t.Fatal("failed cleanup invalidated cache entries") + } + } else if len(invalidated) != 2 || invalidated[0] != store.ids[0] || invalidated[1] != store.ids[1] { + t.Fatal("deleted observers were not invalidated") + } + }) + } +} + func (s *refreshStub) refresh(ctx context.Context, name string) error { s.calls = append(s.calls, name) if err := ctx.Err(); err != nil { diff --git a/internal/config/config.go b/internal/config/config.go index f742005..feffb27 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -30,6 +30,7 @@ type Config struct { Background BackgroundConfig `yaml:"background"` Presence PresenceConfig `yaml:"presence"` Nodes NodesConfig `yaml:"nodes"` + Observers ObserversConfig `yaml:"observers"` } // ResolvedConfig holds all runtime configuration with defaults applied. @@ -55,8 +56,9 @@ type ResolvedConfig struct { // NodeStaleThreshold and NodeDeleteAfter mirror ClockDriftThreshold's "0 means unset, // resolve to a default" pattern -- see NodesConfig. - NodeStaleThreshold time.Duration - NodeDeleteAfter time.Duration + NodeStaleThreshold time.Duration + NodeDeleteAfter time.Duration + ObserverDeleteAfter time.Duration } // PresenceConfig controls coalescing of presence bookkeeping writes @@ -209,6 +211,14 @@ type NodesConfig struct { DeleteAfter duration `yaml:"delete_after"` } +// ObserversConfig controls optional observer age-out. +type ObserversConfig struct { + // DeleteAfter is how long an observer must be unseen before it can be deleted. + // Omitted or nonpositive disables age-out. Retained packet observations, + // telemetry and ownership records always protect the observer from deletion. + DeleteAfter duration `yaml:"delete_after"` +} + // duration is a wrapper around time.Duration that supports YAML unmarshalling // from human-readable strings like "24h", "7d", "30d". type duration struct { @@ -333,6 +343,7 @@ func Resolve(cfg *Config) ResolvedConfig { ClockDriftThreshold: cfg.Nodes.ClockDriftThreshold.Duration, NodeStaleThreshold: cfg.Nodes.StaleThreshold.Duration, NodeDeleteAfter: cfg.Nodes.DeleteAfter.Duration, + ObserverDeleteAfter: cfg.Observers.DeleteAfter.Duration, } if r.TelemetryResolution == 0 { r.TelemetryResolution = time.Hour @@ -386,10 +397,10 @@ func Resolve(cfg *Config) ResolvedConfig { func (r ResolvedConfig) String() string { return fmt.Sprintf( - "telemetryResolution=%s telemetryRetention=%s packetRetention=%s routeRetention=%s routeGrace=%s routeMinObs=%d maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s nodeStaleThreshold=%s nodeDeleteAfter=%s", + "telemetryResolution=%s telemetryRetention=%s packetRetention=%s routeRetention=%s routeGrace=%s routeMinObs=%d maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s nodeStaleThreshold=%s nodeDeleteAfter=%s observerDeleteAfter=%s", r.TelemetryResolution, r.TelemetryRetention, r.PacketRetention, r.RouteRetention, r.RouteGrace, r.RouteMinObservations, r.MaxConnsPerIP, r.ViewRefreshInterval, r.ReconfirmInterval, r.CleanupInterval, r.PresenceFlushInterval, r.PresencePacketTTL, r.ClockDriftThreshold, - r.NodeStaleThreshold, r.NodeDeleteAfter, + r.NodeStaleThreshold, r.NodeDeleteAfter, r.ObserverDeleteAfter, ) } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 4ccd46a..75f673c 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -38,6 +38,8 @@ regions: name: British Columbia display_order: 1 iatas: [YVR] +observers: + delete_after: 720h `) f.Close() @@ -54,6 +56,9 @@ regions: if cfg.Regions[0].Slug != "bc" { t.Errorf("expected slug bc, got %s", cfg.Regions[0].Slug) } + if Resolve(cfg).ObserverDeleteAfter != 30*24*time.Hour { + t.Error("observers.delete_after was not loaded from YAML") + } } func TestLoad_InvalidYAML(t *testing.T) { @@ -104,6 +109,9 @@ func TestResolve_Defaults(t *testing.T) { if r.NodeDeleteAfter != 30*24*time.Hour { t.Errorf("expected NodeDeleteAfter 720h (same default as PacketRetention), got %v", r.NodeDeleteAfter) } + if r.ObserverDeleteAfter != 0 { + t.Errorf("observer deletion must be disabled by default, got %v", r.ObserverDeleteAfter) + } } func TestResolve_ExplicitValues(t *testing.T) { @@ -115,6 +123,7 @@ func TestResolve_ExplicitValues(t *testing.T) { cfg.Background.ViewRefresh.Duration = 2 * time.Hour cfg.Background.Reconfirm.Duration = 3 * time.Hour cfg.Background.Cleanup.Duration = 4 * time.Hour + cfg.Observers.DeleteAfter.Duration = 45 * 24 * time.Hour r := Resolve(cfg) if r.TelemetryResolution != 30*time.Minute { @@ -126,6 +135,9 @@ func TestResolve_ExplicitValues(t *testing.T) { if r.ViewRefreshInterval != 2*time.Hour { t.Errorf("expected 2h, got %v", r.ViewRefreshInterval) } + if r.ObserverDeleteAfter != 45*24*time.Hour { + t.Errorf("expected observer delete_after 1080h, got %v", r.ObserverDeleteAfter) + } } func TestResolvedConfig_String(t *testing.T) { diff --git a/internal/presence/coalescer.go b/internal/presence/coalescer.go index c0673bb..c8e3544 100644 --- a/internal/presence/coalescer.go +++ b/internal/presence/coalescer.go @@ -35,11 +35,14 @@ type Store interface { // TouchPackets applies coalesced last_heard_at bumps for the given packet // hashes in one statement. TouchPackets(ctx context.Context, hashes [][]byte, heard []time.Time) error + + DeleteOldObservers(ctx context.Context, cutoff time.Time) ([]uuid.UUID, error) } type identity struct { id uuid.UUID name string + seen time.Time } type observerBump struct { @@ -99,6 +102,8 @@ func (c *Coalescer) UpsertObserver(ctx context.Context, pubkey []byte) (uuid.UUI bump.seen = c.now() bump.count++ c.dirtyObservers[ident.id] = bump + ident.seen = bump.seen + c.identities[key] = ident c.mu.Unlock() return ident.id, ident.name, nil } @@ -109,7 +114,7 @@ func (c *Coalescer) UpsertObserver(ctx context.Context, pubkey []byte) (uuid.UUI return id, name, err } c.mu.Lock() - c.identities[key] = identity{id: id, name: name} + c.identities[key] = identity{id: id, name: name, seen: c.now()} c.mu.Unlock() return id, name, nil } @@ -191,6 +196,30 @@ func (c *Coalescer) UpdateObserverStatus(ctx context.Context, p ingest.UpdateObs return id, err } +// DeleteOldObservers persists cached activity before pruning and forgets IDs so +// returning observers write through. Include identities even if an earlier +// presence flush failed or is still in flight; GREATEST prevents older flushes +// from overwriting this freshness. A failed preparation must not delete rows. +func (c *Coalescer) DeleteOldObservers(ctx context.Context, cutoff time.Time) ([]uuid.UUID, error) { + c.mu.Lock() + observers := c.dirtyObservers + for _, ident := range c.identities { + bump := observers[ident.id] + if ident.seen.After(bump.seen) { + bump.seen = ident.seen + } + observers[ident.id] = bump + } + c.dirtyObservers = make(map[uuid.UUID]observerBump) + clear(c.identities) + clear(c.knownBrokers) + c.mu.Unlock() + if err := c.flushObservers(ctx, observers); err != nil { + return nil, err + } + return c.Store.DeleteOldObservers(ctx, cutoff) +} + // Run flushes on the configured interval until ctx is cancelled, then does a // final flush so a clean shutdown loses nothing. func (c *Coalescer) Run(ctx context.Context) { @@ -228,18 +257,8 @@ func (c *Coalescer) Flush(ctx context.Context) { } c.mu.Unlock() - if len(observers) > 0 { - ids := make([]uuid.UUID, 0, len(observers)) - seen := make([]time.Time, 0, len(observers)) - counts := make([]int32, 0, len(observers)) - for id, bump := range observers { - ids = append(ids, id) - seen = append(seen, bump.seen) - counts = append(counts, bump.count) - } - if err := c.Store.TouchObservers(ctx, ids, seen, counts); err != nil { - log.Printf("presence: flush observers failed (%d rows dropped): %v", len(ids), err) - } + if err := c.flushObservers(ctx, observers); err != nil { + log.Printf("presence: flush observers failed (%d rows dropped): %v", len(observers), err) } if len(brokers) > 0 { @@ -268,3 +287,18 @@ func (c *Coalescer) Flush(ctx context.Context) { } } } + +func (c *Coalescer) flushObservers(ctx context.Context, observers map[uuid.UUID]observerBump) error { + if len(observers) == 0 { + return nil + } + ids := make([]uuid.UUID, 0, len(observers)) + seen := make([]time.Time, 0, len(observers)) + counts := make([]int32, 0, len(observers)) + for id, bump := range observers { + ids = append(ids, id) + seen = append(seen, bump.seen) + counts = append(counts, bump.count) + } + return c.Store.TouchObservers(ctx, ids, seen, counts) +} diff --git a/internal/presence/coalescer_test.go b/internal/presence/coalescer_test.go new file mode 100644 index 0000000..0900f5f --- /dev/null +++ b/internal/presence/coalescer_test.go @@ -0,0 +1,105 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package presence + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" +) + +type retentionStore struct { + Store + id uuid.UUID + touch func(context.Context, []uuid.UUID, []time.Time, []int32) error + prune func(context.Context, time.Time) ([]uuid.UUID, error) +} + +func (s *retentionStore) UpsertObserver(context.Context, []byte) (uuid.UUID, string, error) { + return s.id, "fixture", nil +} +func (s *retentionStore) TouchObservers(ctx context.Context, ids []uuid.UUID, seen []time.Time, counts []int32) error { + return s.touch(ctx, ids, seen, counts) +} +func (s *retentionStore) DeleteOldObservers(ctx context.Context, cutoff time.Time) ([]uuid.UUID, error) { + return s.prune(ctx, cutoff) +} + +func TestObserverRetentionAbortOnFlushFailure(t *testing.T) { + for _, cause := range []error{errors.New("flush unavailable"), context.Canceled} { + t.Run(cause.Error(), func(t *testing.T) { + store := &retentionStore{id: uuid.New()} + store.touch = func(context.Context, []uuid.UUID, []time.Time, []int32) error { return cause } + store.prune = func(context.Context, time.Time) ([]uuid.UUID, error) { + t.Error("deletion attempted after failed preparation") + return nil, nil + } + c := New(store, time.Second, time.Second) + if _, _, err := c.UpsertObserver(context.Background(), []byte("fixture")); err != nil { + t.Fatal(err) + } + if _, err := c.DeleteOldObservers(context.Background(), time.Now()); !errors.Is(err, cause) { + t.Fatalf("got %v, want %v", err, cause) + } + }) + } +} + +func TestObserverRetentionPreservesActivityAfterFlush(t *testing.T) { + for _, inFlight := range []bool{false, true} { + name := "failed flush" + if inFlight { + name = "in-flight flush" + } + t.Run(name, func(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + store := &retentionStore{id: uuid.New()} + c := New(store, time.Second, time.Second) + c.now = func() time.Time { return now } + for range 2 { + if _, _, err := c.UpsertObserver(ctx, []byte("fixture")); err != nil { + t.Fatal(err) + } + } + started, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var calls atomic.Int32 + store.touch = func(_ context.Context, ids []uuid.UUID, seen []time.Time, counts []int32) error { + if calls.Add(1) == 1 { + close(started) + <-release + return errors.New("earlier flush failed") + } + if len(ids) != 1 || ids[0] != store.id || !seen[0].Equal(now) || counts[0] != 0 { + t.Errorf("cached activity was lost or counted twice: %v, %v, %v", ids, seen, counts) + } + return nil + } + store.prune = func(context.Context, time.Time) ([]uuid.UUID, error) { + if calls.Load() != 2 { + t.Error("deletion preceded the freshness write") + } + return nil, nil + } + go func() { c.Flush(ctx); close(done) }() + <-started + if !inFlight { + close(release) + <-done + } + _, err := c.DeleteOldObservers(ctx, now.Add(-time.Hour)) + if inFlight { + close(release) + <-done + } + if err != nil { + t.Fatal(err) + } + }) + } +}