perf: known_routes retention, digest identity, and a working batched reconfirm

cmd/beacon/main.go

@@ -258,7 +258,7 @@ func main() {
 	scheduler := background.New([]background.Task{
 		background.ViewRefreshTask(store, resolved.ViewRefreshInterval),
 		background.CleanupTask(store, resolved.TelemetryRetention, resolved.PacketRetention, resolved.NodeDeleteAfter, resolved.CleanupInterval),
-		background.ReconfirmTask(store, resolved.ReconfirmInterval),
+		background.ReconfirmTask(store, resolved.RouteRetention, resolved.RouteGrace, int64(resolved.RouteMinObservations), resolved.ReconfirmInterval),
 	})
 	go scheduler.Start(ctx)
 

config.yaml.example

@@ -75,6 +75,25 @@ telemetry:
 packets:
   retention: 720h   # 30 days
 
+# Known-route retention. Routes are distilled packet history (the path a packet
+# took, hop by hop), so they may outlive packets.retention; there is no required
+# relationship between the two.
+routes:
+  # How long a route is kept after it was last observed. Every new observation
+  # of the same path resets this clock, so routes the mesh still uses never
+  # expire -- only paths nothing has traveled for this long age out.
+  retention: 336h   # 14 days
+  # Shorter window for rarely-seen routes: one observed fewer than
+  # min_observations times in total is dropped once it goes unobserved for
+  # this long. Must be shorter than retention, or it never applies.
+  grace: 168h       # 7 days
+  # Lifetime observation count a route must reach to earn the full retention
+  # window. Below it a route is treated as flood noise -- a path recorded once
+  # or twice and never confirmed -- and only kept for the grace window. The
+  # count never resets, so once a route crosses this bar it stays in the
+  # retention tier for good.
+  min_observations: 3
+
 websocket:
   max_connections_per_ip: 5 # default: 5
 

db/migrations/023_channel_messages_cascade.sql

@@ -0,0 +1,12 @@
+-- Copyright 2026 Beacon Contributors
+-- SPDX-License-Identifier: AGPL-3.0-or-later
+
+-- The packet_hash FK had no ON DELETE action, so packets carrying a channel
+-- message could never age out and DeleteOldPackets aborted. Match packet_observations.
+
+ALTER TABLE channel_messages
+  DROP CONSTRAINT IF EXISTS channel_messages_packet_hash_fkey;
+
+ALTER TABLE channel_messages
+  ADD CONSTRAINT channel_messages_packet_hash_fkey
+  FOREIGN KEY (packet_hash) REFERENCES packets(packet_hash) ON DELETE CASCADE;

db/migrations/024_known_routes_pathkey.sql

@@ -0,0 +1,43 @@
+-- Copyright 2026 Beacon Contributors
+-- SPDX-License-Identifier: AGPL-3.0-or-later
+
+-- known_routes grew unbounded (28 GB / 32.8M rows in 24 days on prod) and its
+-- UNIQUE(node_ids, iata) key indexed whole UUID arrays. Rebuild keyed on a
+-- 16-byte md5 of node_ids and add reconfirm bookkeeping. No rows are pruned
+-- here: retention is deployment config (routes.retention/grace), so the first
+-- cleanup tick after startup enforces it; a schema migration must not bake in
+-- one deployment's policy. id stays (the API serializes it) but loses its
+-- index; uniqueness lives on (iata, path_key).
+
+CREATE TABLE known_routes_new (
+  id                  BIGINT      GENERATED BY DEFAULT AS IDENTITY NOT NULL,
+  path_key            BYTEA       NOT NULL,
+  node_ids            UUID[]      NOT NULL,
+  hash_prefix         BYTEA[]     NOT NULL,
+  iata                CHAR(3)     NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
+  hop_count           INT         NOT NULL,
+  first_seen          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  last_seen           TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  observation_count   BIGINT      NOT NULL DEFAULT 1,
+  last_reconfirmed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (iata, path_key)
+);
+
+INSERT INTO known_routes_new
+  (id, path_key, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count)
+SELECT id,
+       decode(md5(array_to_string(node_ids, ',')), 'hex'),
+       node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count
+FROM known_routes;
+
+SELECT setval(pg_get_serial_sequence('known_routes_new', 'id'),
+              (SELECT COALESCE(MAX(id), 1) FROM known_routes_new));
+
+DROP TABLE known_routes;
+ALTER TABLE known_routes_new RENAME TO known_routes;
+ALTER TABLE known_routes RENAME CONSTRAINT known_routes_new_pkey TO known_routes_pkey;
+ALTER TABLE known_routes RENAME CONSTRAINT known_routes_new_iata_fkey TO known_routes_iata_fkey;
+
+CREATE INDEX idx_known_routes_hop_count ON known_routes(iata, hop_count);
+CREATE INDEX idx_known_routes_last_seen ON known_routes(last_seen DESC);
+CREATE INDEX idx_known_routes_reconfirm ON known_routes(last_reconfirmed_at);

db/queries/queries.sql

@@ -533,6 +533,13 @@ DELETE FROM nodes
 WHERE last_seen < $1
   AND id NOT IN (SELECT owner_node_id FROM observer_owners WHERE owner_node_id IS NOT NULL);
 
+-- name: DeleteOldRoutes :exec
+-- Deletes routes not observed since the retention cutoff ($1), and rarely-observed
+-- routes (observation_count < $2) not observed since the grace cutoff ($3).
+DELETE FROM known_routes
+WHERE last_seen < $1
+   OR (observation_count < $2 AND last_seen < $3);
+
 -- name: DeleteOldChannelIATAs :exec
 -- Keeps the channel IATA filter in step with packet retention.
 DELETE FROM channel_iatas WHERE last_heard < $1;
@@ -1070,12 +1077,11 @@ ORDER BY t.last_heard_at DESC;
 -- ============================================================
 
 -- name: UpsertKnownRoute :exec
--- Inserts or updates a known route (all hops resolved to high confidence).
--- node_ids and hash_prefix are ordered arrays of the resolved node UUIDs and
--- their hash bytes. last_seen is bumped on conflict.
-INSERT INTO known_routes (node_ids, hash_prefix, iata, hop_count)
-VALUES ($1, $2, $3, $4)
-ON CONFLICT (node_ids, iata) DO UPDATE SET
+-- Route identity is path_key, an md5 of node_ids computed by the caller.
+-- On conflict, observation_count and last_seen are bumped.
+INSERT INTO known_routes (path_key, node_ids, hash_prefix, iata, hop_count)
+VALUES ($1, $2, $3, $4, $5)
+ON CONFLICT (iata, path_key) DO UPDATE SET
   last_seen = NOW(),
   observation_count = known_routes.observation_count + 1;
 
@@ -1216,27 +1222,55 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_top_advertisers_by_iata;
 REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets;
 
 -- name: ReconfirmRoutes :exec
--- Delete known_routes where any hop node has departed from node_short_ids for
--- that IATA, or where any hop's prefix_4 is now ambiguous (matches >1 node).
-DELETE FROM known_routes kr
-WHERE EXISTS (
-    SELECT 1
-    FROM unnest(kr.node_ids) AS hop_node_id
-    WHERE NOT EXISTS (
-        SELECT 1 FROM node_short_ids ns
-        WHERE ns.node_id = hop_node_id
-          AND ns.iata = kr.iata
+-- Checks the $1 least-recently-reconfirmed routes: deletes those with a departed
+-- hop node or a hop prefix now matching >1 node in that IATA (length-aware:
+-- 1/2/3/4-byte hop prefixes check prefix_1/2/3/4), and stamps the survivors.
+WITH batch AS (
+    SELECT iata, path_key, node_ids, hash_prefix
+    FROM known_routes
+    ORDER BY last_reconfirmed_at
+    LIMIT $1
+),
+amb AS MATERIALIZED (
+    SELECT iata, 1 AS len, prefix_1 AS p FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1
+    UNION ALL
+    SELECT iata, 2, prefix_2 FROM node_short_ids GROUP BY iata, prefix_2 HAVING COUNT(*) > 1
+    UNION ALL
+    SELECT iata, 3, prefix_3 FROM node_short_ids GROUP BY iata, prefix_3 HAVING COUNT(*) > 1
+    UNION ALL
+    SELECT iata, 4, prefix_4 FROM node_short_ids GROUP BY iata, prefix_4 HAVING COUNT(*) > 1
+),
+dead AS (
+    SELECT b.iata, b.path_key
+    FROM batch b
+    WHERE EXISTS (
+        SELECT 1
+        FROM unnest(b.node_ids) AS hop_node_id
+        WHERE NOT EXISTS (
+            SELECT 1 FROM node_short_ids ns
+            WHERE ns.node_id = hop_node_id
+              AND ns.iata = b.iata
+        )
     )
+    UNION
+    SELECT DISTINCT b.iata, b.path_key
+    FROM batch b
+    CROSS JOIN LATERAL unnest(b.hash_prefix) AS hp
+    JOIN amb a ON a.iata = b.iata AND a.len = length(hp) AND a.p = hp
+),
+deleted AS (
+    DELETE FROM known_routes kr
+    USING dead d
+    WHERE kr.iata = d.iata AND kr.path_key = d.path_key
 )
-OR EXISTS (
-    SELECT 1
-    FROM unnest(kr.hash_prefix) AS hop_prefix
-    WHERE (
-        SELECT COUNT(*) FROM node_short_ids ns
-        WHERE ns.iata = kr.iata
-          AND ns.prefix_4 = hop_prefix
-    ) > 1
-);
+UPDATE known_routes kr
+SET last_reconfirmed_at = NOW()
+FROM batch b
+WHERE kr.iata = b.iata AND kr.path_key = b.path_key
+  AND NOT EXISTS (
+      SELECT 1 FROM dead d
+      WHERE d.iata = b.iata AND d.path_key = b.path_key
+  );
 
 -- name: ReconfirmNeighbors :exec
 -- Delete node_neighbors where the neighbor has departed from node_short_ids

db/routes.go

@@ -5,7 +5,9 @@ package db
 
 import (
 	"context"
+	"crypto/md5"
 	"encoding/hex"
+	"strings"
 	"time"
 
 	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
@@ -14,12 +16,24 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 )
 
+// routePathKey is the route's identity digest: md5 over the comma-joined
+// node UUIDs, matching Postgres's decode(md5(array_to_string(node_ids, ',')), 'hex').
+func routePathKey(nodeIDs []uuid.UUID) []byte {
+	parts := make([]string, len(nodeIDs))
+	for i, id := range nodeIDs {
+		parts[i] = id.String()
+	}
+	sum := md5.Sum([]byte(strings.Join(parts, ",")))
+	return sum[:]
+}
+
 func (s *Store) UpsertKnownRoute(ctx context.Context, nodeIDs []uuid.UUID, hashPrefix [][]byte, iata string, hopCount int32) error {
 	return s.q.UpsertKnownRoute(ctx, sqlc.UpsertKnownRouteParams{
+		PathKey:    routePathKey(nodeIDs),
 		NodeIds:    nodeIDs,
 		HashPrefix: hashPrefix,
 		Iata:       iata,
-		HopCount:   int32(hopCount),
+		HopCount:   hopCount,
 	})
 }
 
@@ -28,7 +42,7 @@ func (s *Store) ListKnownRoutes(ctx context.Context, iata string, hopCount int32
 	if !cursor.IsZero() {
 		cursorTS = pgtype.Timestamptz{Time: cursor, Valid: true}
 	}
-	rows, err := s.q.ListKnownRoutes(ctx, sqlc.ListKnownRoutesParams{
+	sqlRows, err := s.q.ListKnownRoutes(ctx, sqlc.ListKnownRoutesParams{
 		Column1: iata,
 		Column2: hopCount,
 		Column3: cursorTS,
@@ -37,6 +51,10 @@ func (s *Store) ListKnownRoutes(ctx context.Context, iata string, hopCount int32
 	if err != nil {
 		return nil, err
 	}
+	rows := make([]knownRouteRow, len(sqlRows))
+	for i, r := range sqlRows {
+		rows[i] = knownRouteRow{ID: r.ID, NodeIds: r.NodeIds, HashPrefix: r.HashPrefix, Iata: r.Iata, HopCount: r.HopCount, FirstSeen: r.FirstSeen, LastSeen: r.LastSeen, ObservationCount: r.ObservationCount}
+	}
 	ids := collectNodeIDs(rows)
 	nodes, err := s.GetNodesByIDs(ctx, ids)
 	if err != nil {
@@ -54,14 +72,18 @@ func (s *Store) SearchKnownRoutes(ctx context.Context, iata, fromHash, toHash st
 	if err != nil {
 		return nil, err
 	}
-	rows, err := s.q.SearchKnownRoutes(ctx, sqlc.SearchKnownRoutesParams{
+	sqlRows, err := s.q.SearchKnownRoutes(ctx, sqlc.SearchKnownRoutesParams{
 		Iata:    iata,
 		Column2: fromBytes,
 		Column3: toBytes,
 	})
 	if err != nil {
 		return nil, err
 	}
+	rows := make([]knownRouteRow, len(sqlRows))
+	for i, r := range sqlRows {
+		rows[i] = knownRouteRow{ID: r.ID, NodeIds: r.NodeIds, HashPrefix: r.HashPrefix, Iata: r.Iata, HopCount: r.HopCount, FirstSeen: r.FirstSeen, LastSeen: r.LastSeen, ObservationCount: r.ObservationCount}
+	}
 	ids := collectNodeIDs(rows)
 	nodes, err := s.GetNodesByIDs(ctx, ids)
 	if err != nil {
@@ -109,13 +131,17 @@ func (s *Store) SearchKnownRoutes(ctx context.Context, iata, fromHash, toHash st
 }
 
 func (s *Store) GetKnownRoutesByNode(ctx context.Context, iata string, nodeID uuid.UUID) ([]api.KnownRoute, error) {
-	rows, err := s.q.GetKnownRoutesByNode(ctx, sqlc.GetKnownRoutesByNodeParams{
+	sqlRows, err := s.q.GetKnownRoutesByNode(ctx, sqlc.GetKnownRoutesByNodeParams{
 		Iata:    iata,
 		Column2: nodeID,
 	})
 	if err != nil {
 		return nil, err
 	}
+	rows := make([]knownRouteRow, len(sqlRows))
+	for i, r := range sqlRows {
+		rows[i] = knownRouteRow{ID: r.ID, NodeIds: r.NodeIds, HashPrefix: r.HashPrefix, Iata: r.Iata, HopCount: r.HopCount, FirstSeen: r.FirstSeen, LastSeen: r.LastSeen, ObservationCount: r.ObservationCount}
+	}
 	ids := collectNodeIDs(rows)
 	nodes, err := s.GetNodesByIDs(ctx, ids)
 	if err != nil {
@@ -260,8 +286,20 @@ func (s *Store) SearchCrossIATARoutes(ctx context.Context, fromHash, fromIATA, t
 	return results, nil
 }
 
-func (s *Store) ReconfirmRoutes(ctx context.Context) error {
-	return s.q.ReconfirmRoutes(ctx)
+// ReconfirmRoutes checks the batchSize least-recently-reconfirmed routes,
+// deleting stale or ambiguous ones and stamping the survivors.
+func (s *Store) ReconfirmRoutes(ctx context.Context, batchSize int32) error {
+	return s.q.ReconfirmRoutes(ctx, batchSize)
+}
+
+// DeleteOldRoutes prunes routes per the retention rule: unconditionally past
+// retentionCutoff, and past graceCutoff when observed fewer than minObservations times.
+func (s *Store) DeleteOldRoutes(ctx context.Context, retentionCutoff time.Time, minObservations int64, graceCutoff time.Time) error {
+	return s.q.DeleteOldRoutes(ctx, sqlc.DeleteOldRoutesParams{
+		LastSeen:         pgtype.Timestamptz{Time: retentionCutoff, Valid: true},
+		ObservationCount: minObservations,
+		LastSeen_2:       pgtype.Timestamptz{Time: graceCutoff, Valid: true},
+	})
 }
 
 // extractFromNode returns the portion of a route starting at the given node.
@@ -274,7 +312,20 @@ func extractFromNode(hops []api.RouteHop, nodeID uuid.UUID) []api.RouteHop {
 	return hops
 }
 
-func toKnownRoutes(rows []sqlc.KnownRoute, nodes map[uuid.UUID]*api.ResolvedNode) []api.KnownRoute {
+// knownRouteRow normalizes the per-query sqlc row structs (identical
+// columns, distinct generated types) so the helpers below share one body.
+type knownRouteRow struct {
+	ID               int64
+	NodeIds          []uuid.UUID
+	HashPrefix       [][]byte
+	Iata             string
+	HopCount         int32
+	FirstSeen        pgtype.Timestamptz
+	LastSeen         pgtype.Timestamptz
+	ObservationCount int64
+}
+
+func toKnownRoutes(rows []knownRouteRow, nodes map[uuid.UUID]*api.ResolvedNode) []api.KnownRoute {
 	items := make([]api.KnownRoute, 0, len(rows))
 	for _, r := range rows {
 		hops := make([]api.RouteHop, 0, len(r.NodeIds))
@@ -301,7 +352,7 @@ func toKnownRoutes(rows []sqlc.KnownRoute, nodes map[uuid.UUID]*api.ResolvedNode
 	return items
 }
 
-func collectNodeIDs(rows []sqlc.KnownRoute) []uuid.UUID {
+func collectNodeIDs(rows []knownRouteRow) []uuid.UUID {
 	seen := make(map[uuid.UUID]struct{})
 	var ids []uuid.UUID
 	for _, r := range rows {