feat: add route and neighbor reconfirmation task

prunes stale and ambiguous routes and neighbors

keep that data clean

cmd/beacon/main.go

@@ -106,12 +106,17 @@ func main() {
 	if viewRefreshInterval == 0 {
 		viewRefreshInterval = time.Hour
 	}
+	reconfirmInterval := cfg.Background.Reconfirm.Duration
+	if reconfirmInterval == 0 {
+		reconfirmInterval = time.Hour
+	}
 	cleanupInterval := cfg.Background.Cleanup.Duration
 	if cleanupInterval == 0 {
 		cleanupInterval = time.Hour
 	}
-	log.Printf("config: loaded — telemetryResolution=%s telemetryRetention=%s packetRetention=%s maxConnsPerIP=%d viewRefresh=%s cleanup=%s",
-		telemetryResolution, telemetryRetention, packetRetention, maxConnsPerIP, viewRefreshInterval, cleanupInterval)
+
+	log.Printf("config: loaded — telemetryResolution=%s telemetryRetention=%s packetRetention=%s maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s",
+		telemetryResolution, telemetryRetention, packetRetention, maxConnsPerIP, viewRefreshInterval, reconfirmInterval, cleanupInterval)
 
 	// ── Hub ──────────────────────────────────────────────────────────────────
 	h := hub.New()
@@ -263,6 +268,7 @@ func main() {
 	scheduler := background.New([]background.Task{
 		background.ViewRefreshTask(store, viewRefreshInterval),
 		background.CleanupTask(store, telemetryRetention, packetRetention, cleanupInterval),
+		background.ReconfirmTask(store, reconfirmInterval),
 	})
 	go scheduler.Start(ctx)
 

config.yaml.example

@@ -122,4 +122,5 @@ cache:
 # flowing. Back off to 1h or more once stable.
 #background:
 #  view_refresh: 1h   # default: 1h
+#  reconfirm: 1h      # default: 1h
 #  cleanup: 1h        # default: 1h

db/nodes.go

@@ -245,3 +245,7 @@ func (s *Store) GetNodeNeighbors(ctx context.Context, nodeID uuid.UUID) ([]api.N
 	}
 	return items, nil
 }
+
+func (s *Store) ReconfirmNeighbors(ctx context.Context) error {
+	return s.q.ReconfirmNeighbors(ctx)
+}

db/queries/queries.sql

@@ -937,3 +937,45 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_top_nodes_by_iata;
 
 -- name: RefreshRadioPresets :exec
 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
+    )
+)
+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
+);
+
+-- name: ReconfirmNeighbors :exec
+-- Delete node_neighbors where the neighbor has departed from node_short_ids
+-- for that IATA, or where its prefix_4 is now ambiguous.
+DELETE FROM node_neighbors nn
+WHERE NOT EXISTS (
+    SELECT 1 FROM node_short_ids ns
+    WHERE ns.node_id = nn.neighbor_id
+      AND ns.iata = nn.iata
+)
+OR (
+    SELECT COUNT(*) FROM node_short_ids ns
+    WHERE ns.iata = nn.iata
+      AND ns.prefix_4 = (
+          SELECT prefix_4 FROM node_short_ids
+          WHERE node_id = nn.neighbor_id
+            AND iata = nn.iata
+      )
+) > 1;

db/routes.go

@@ -259,6 +259,10 @@ 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)
+}
+
 // extractFromNode returns the portion of a route starting at the given node.
 func extractFromNode(hops []api.RouteHop, nodeID uuid.UUID) []api.RouteHop {
 	for i, hop := range hops {

db/sqlc/queries.sql.go

@@ -2704,6 +2704,60 @@ func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([
 	return items, nil
 }
 
+const reconfirmNeighbors = `-- name: ReconfirmNeighbors :exec
+DELETE FROM node_neighbors nn
+WHERE NOT EXISTS (
+    SELECT 1 FROM node_short_ids ns
+    WHERE ns.node_id = nn.neighbor_id
+      AND ns.iata = nn.iata
+)
+OR (
+    SELECT COUNT(*) FROM node_short_ids ns
+    WHERE ns.iata = nn.iata
+      AND ns.prefix_4 = (
+          SELECT prefix_4 FROM node_short_ids
+          WHERE node_id = nn.neighbor_id
+            AND iata = nn.iata
+      )
+) > 1
+`
+
+// Delete node_neighbors where the neighbor has departed from node_short_ids
+// for that IATA, or where its prefix_4 is now ambiguous.
+func (q *Queries) ReconfirmNeighbors(ctx context.Context) error {
+	_, err := q.db.Exec(ctx, reconfirmNeighbors)
+	return err
+}
+
+const reconfirmRoutes = `-- name: ReconfirmRoutes :exec
+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
+    )
+)
+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
+)
+`
+
+// 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).
+func (q *Queries) ReconfirmRoutes(ctx context.Context) error {
+	_, err := q.db.Exec(ctx, reconfirmRoutes)
+	return err
+}
+
 const refreshHourlyStats = `-- name: RefreshHourlyStats :exec
 REFRESH MATERIALIZED VIEW CONCURRENTLY mv_hourly_iata_stats
 `

internal/background/tasks.go

@@ -5,6 +5,7 @@ package background
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"time"
 
@@ -47,3 +48,22 @@ func CleanupTask(store *db.Store, telemetryRetention, packetRetention, interval
 		},
 	}
 }
+
+// ReconfirmTask returns a Task that prunes stale and ambiguous resolved paths
+// and neighbors. Runs after routes to ensure neighbors are cleaned against
+// already-reconfirmed path data.
+func ReconfirmTask(store *db.Store, interval time.Duration) Task {
+	return Task{
+		Name:     "reconfirm",
+		Interval: interval,
+		Run: func(ctx context.Context) error {
+			if err := store.ReconfirmRoutes(ctx); err != nil {
+				return fmt.Errorf("routes: %w", err)
+			}
+			if err := store.ReconfirmNeighbors(ctx); err != nil {
+				return fmt.Errorf("neighbors: %w", err)
+			}
+			return nil
+		},
+	}
+}

internal/config/config.go

@@ -33,6 +33,9 @@ type BackgroundConfig struct {
 	// Defaults to 1h if not set.
 	ViewRefresh duration `yaml:"view_refresh"`
 
+	// Reconfirm prunes stale and ambiguous resolved paths and neigbors.
+	Reconfirm duration `yaml:"reconfirm"`
+
 	// Cleanup is how often old telemetry and packet rows are pruned.
 	// Defaults to 1h if not set.
 	Cleanup duration `yaml:"cleanup"`