diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index cdfdfa2..3a3a80a 100644 --- a/cmd/beacon/main.go +++ b/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) diff --git a/config.yaml.example b/config.yaml.example index 9897bf9..d99f4de 100644 --- a/config.yaml.example +++ b/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 diff --git a/db/nodes.go b/db/nodes.go index fb4353c..38db9a6 100644 --- a/db/nodes.go +++ b/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) +} diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 965ff51..935d22d 100644 --- a/db/queries/queries.sql +++ b/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; diff --git a/db/routes.go b/db/routes.go index d1aef44..3fd1783 100644 --- a/db/routes.go +++ b/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 { diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 4855a47..c385605 100644 --- a/db/sqlc/queries.sql.go +++ b/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 ` diff --git a/internal/background/tasks.go b/internal/background/tasks.go index 67df0f0..6006cce 100644 --- a/internal/background/tasks.go +++ b/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 + }, + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 658069c..50002f1 100644 --- a/internal/config/config.go +++ b/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"`