feat(retention): age out unreferenced observers when enabled (#126)

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

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 ──────────────────────────────────────────────────────────

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

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

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

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
 -- ============================================================

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()

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

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
 `

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.

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 {

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

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) {

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

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