perf(db): compute route ambiguity once per reconfirm run

The four-way GROUP BY over node_short_ids ran once per 1000-route batch, up to 750 times a run, and could eat the whole 5s batch budget. The task now fetches the set once and passes it to each batch.

db/queries/queries.sql

@@ -1482,10 +1482,21 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets;
 -- name: RefreshObserverActivity :exec
 REFRESH MATERIALIZED VIEW CONCURRENTLY mv_observer_activity_hourly;
 
+-- name: AmbiguousPrefixes :many
+-- Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run.
+SELECT iata::text AS iata, 1::int AS len, prefix_1 AS prefix FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1
+UNION ALL
+SELECT iata::text, 2::int, prefix_2 FROM node_short_ids GROUP BY iata, prefix_2 HAVING COUNT(*) > 1
+UNION ALL
+SELECT iata::text, 3::int, prefix_3 FROM node_short_ids GROUP BY iata, prefix_3 HAVING COUNT(*) > 1
+UNION ALL
+SELECT iata::text, 4::int, prefix_4 FROM node_short_ids GROUP BY iata, prefix_4 HAVING COUNT(*) > 1;
+
 -- name: ReconfirmRoutes :one
 -- Checks one batch of 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.
+-- 1/2/3/4-byte hop prefixes check prefix_1/2/3/4; ambiguity set supplied by AmbiguousPrefixes),
+-- and stamps the survivors.
 WITH batch AS MATERIALIZED (
     SELECT r.iata, r.path_key, r.node_ids, r.hash_prefix
     FROM known_routes r
@@ -1494,14 +1505,9 @@ WITH batch AS MATERIALIZED (
     LIMIT @batch_size
     FOR UPDATE OF r SKIP LOCKED
 ),
-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
+amb AS (
+    SELECT a.iata::char(3) AS iata, a.len, a.p
+    FROM ROWS FROM (unnest(@amb_iata::text[]), unnest(@amb_len::int[]), unnest(@amb_prefix::bytea[])) AS a(iata, len, p)
 ),
 dead AS (
     SELECT b.iata, b.path_key

db/reconfirm_integration_test.go

@@ -17,10 +17,19 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
+func ambiguity(t *testing.T, ctx context.Context, store *Store) AmbiguousPrefixes {
+	t.Helper()
+	amb, err := store.AmbiguousPrefixes(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return amb
+}
+
 func TestReconfirmReleasesEarlierBatchPostgres(t *testing.T) {
 	ctx, pool, store := reconfirmPool(t)
 	before := time.Now()
-	if n, err := store.ReconfirmRoutes(ctx, 1, before); err != nil || n != 1 {
+	if n, err := store.ReconfirmRoutes(ctx, 1, before, ambiguity(t, ctx, store)); err != nil || n != 1 {
 		t.Fatalf("first batch: %d, %v", n, err)
 	}
 	gate, err := pool.Begin(ctx)
@@ -40,8 +49,9 @@ CREATE TRIGGER pause_second_route BEFORE UPDATE ON known_routes FOR EACH ROW EXE
 		t.Fatal(err)
 	}
 	finished := make(chan error, 1)
+	amb := ambiguity(t, ctx, store)
 	go func() {
-		_, err := store.ReconfirmRoutes(ctx, 1, before)
+		_, err := store.ReconfirmRoutes(ctx, 1, before, amb)
 		finished <- err
 	}()
 	deadline := time.Now().Add(3 * time.Second)
@@ -131,7 +141,7 @@ UPDATE known_routes SET node_ids=node_ids || md5('missing')::uuid, hash_prefix=h
 	original := snapshot()
 	var checked int64
 	for range 10 {
-		n, err := store.ReconfirmRoutes(ctx, 3, before)
+		n, err := store.ReconfirmRoutes(ctx, 3, before, ambiguity(t, ctx, store))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -165,7 +175,7 @@ UPDATE known_routes SET node_ids=node_ids || md5('missing')::uuid, hash_prefix=h
 func TestReconfirmCancelledBatchRollsBackPostgres(t *testing.T) {
 	ctx, pool, store := reconfirmPool(t)
 	before := time.Now()
-	if _, err := store.ReconfirmRoutes(ctx, 1, before); err != nil {
+	if _, err := store.ReconfirmRoutes(ctx, 1, before, ambiguity(t, ctx, store)); err != nil {
 		t.Fatal(err)
 	}
 	_, err := pool.Exec(ctx, `
@@ -177,7 +187,7 @@ CREATE TRIGGER pause_reconfirm BEFORE UPDATE ON known_routes FOR EACH ROW EXECUT
 	}
 	batchCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
 	defer cancel()
-	if _, err := store.ReconfirmRoutes(batchCtx, 1, before); err == nil {
+	if _, err := store.ReconfirmRoutes(batchCtx, 1, before, ambiguity(t, ctx, store)); err == nil {
 		t.Fatal("expected batch cancellation")
 	}
 	// Waiting for this DDL also waits for cancellation to release the row lock.
@@ -191,7 +201,7 @@ CREATE TRIGGER pause_reconfirm BEFORE UPDATE ON known_routes FOR EACH ROW EXECUT
 	if total != 2 || stamped != 1 {
 		t.Fatalf("cancellation changed committed work: total %d, checked %d", total, stamped)
 	}
-	if n, err := store.ReconfirmRoutes(ctx, 1, before); err != nil || n != 1 {
+	if n, err := store.ReconfirmRoutes(ctx, 1, before, ambiguity(t, ctx, store)); err != nil || n != 1 {
 		t.Fatalf("cancelled route was not available to retry: %d, %v", n, err)
 	}
 }
@@ -199,10 +209,10 @@ CREATE TRIGGER pause_reconfirm BEFORE UPDATE ON known_routes FOR EACH ROW EXECUT
 func TestReconfirmFutureCutoffDoesNotRepeatPostgres(t *testing.T) {
 	ctx, _, store := reconfirmPool(t)
 	before := time.Now().Add(time.Hour)
-	if n, err := store.ReconfirmRoutes(ctx, 2, before); err != nil || n != 2 {
+	if n, err := store.ReconfirmRoutes(ctx, 2, before, ambiguity(t, ctx, store)); err != nil || n != 2 {
 		t.Fatalf("first batch: %d, %v", n, err)
 	}
-	if n, err := store.ReconfirmRoutes(ctx, 2, before); err != nil || n != 0 {
+	if n, err := store.ReconfirmRoutes(ctx, 2, before, ambiguity(t, ctx, store)); err != nil || n != 0 {
 		t.Fatalf("completed routes consumed the run budget twice: %d, %v", n, err)
 	}
 }
@@ -220,7 +230,7 @@ CREATE TRIGGER pause_valid_route BEFORE UPDATE ON known_routes FOR EACH ROW EXEC
 	before := time.Now()
 	batchCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
 	defer cancel()
-	if _, err := store.ReconfirmRoutes(batchCtx, 2, before); err == nil {
+	if _, err := store.ReconfirmRoutes(batchCtx, 2, before, ambiguity(t, ctx, store)); err == nil {
 		t.Fatal("expected batch cancellation")
 	}
 	if _, err := pool.Exec(ctx, `DROP TRIGGER pause_valid_route ON known_routes`); err != nil {
@@ -230,7 +240,7 @@ CREATE TRIGGER pause_valid_route BEFORE UPDATE ON known_routes FOR EACH ROW EXEC
 	if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_routes`).Scan(&count); err != nil || count != 2 {
 		t.Fatalf("cancelled batch partially deleted routes: %d, %v", count, err)
 	}
-	if n, err := store.ReconfirmRoutes(ctx, 2, before); err != nil || n != 2 {
+	if n, err := store.ReconfirmRoutes(ctx, 2, before, ambiguity(t, ctx, store)); err != nil || n != 2 {
 		t.Fatalf("retry did not process both routes: %d, %v", n, err)
 	}
 	if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_routes WHERE path_key=int4send(2)`).Scan(&count); err != nil || count != 1 {
@@ -299,7 +309,7 @@ func TestReconfirmSkipsBusyRoutePostgres(t *testing.T) {
 	}
 	batchCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
 	defer cancel()
-	if _, err := store.ReconfirmRoutes(batchCtx, 1, time.Now()); err != nil {
+	if _, err := store.ReconfirmRoutes(batchCtx, 1, time.Now(), ambiguity(t, ctx, store)); err != nil {
 		t.Fatalf("maintenance should skip the busy route and validate the next one: %v", err)
 	}
 	var checked, preserved int
@@ -312,7 +322,7 @@ func TestReconfirmSkipsBusyRoutePostgres(t *testing.T) {
 	if err := busy.Rollback(ctx); err != nil {
 		t.Fatal(err)
 	}
-	if _, err := store.ReconfirmRoutes(ctx, 1, time.Now()); err != nil {
+	if _, err := store.ReconfirmRoutes(ctx, 1, time.Now(), ambiguity(t, ctx, store)); err != nil {
 		t.Fatal(err)
 	}
 	if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_routes WHERE last_reconfirmed_at > '2026-02-01'`).Scan(&checked); err != nil {

db/routes.go

@@ -287,12 +287,36 @@ func (s *Store) SearchCrossIATARoutes(ctx context.Context, fromHash, fromIATA, t
 	return results, nil
 }
 
+// AmbiguousPrefixes is the per-IATA set of hop prefixes that resolve to more than one node.
+type AmbiguousPrefixes struct {
+	IATAs    []string
+	Lens     []int32
+	Prefixes [][]byte
+}
+
+func (s *Store) AmbiguousPrefixes(ctx context.Context) (AmbiguousPrefixes, error) {
+	rows, err := s.q.AmbiguousPrefixes(ctx)
+	if err != nil {
+		return AmbiguousPrefixes{}, err
+	}
+	amb := AmbiguousPrefixes{IATAs: make([]string, 0, len(rows)), Lens: make([]int32, 0, len(rows)), Prefixes: make([][]byte, 0, len(rows))}
+	for _, r := range rows {
+		amb.IATAs = append(amb.IATAs, r.Iata)
+		amb.Lens = append(amb.Lens, r.Len)
+		amb.Prefixes = append(amb.Prefixes, r.Prefix)
+	}
+	return amb, nil
+}
+
 // 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, before time.Time) (int64, error) {
+func (s *Store) ReconfirmRoutes(ctx context.Context, batchSize int32, before time.Time, amb AmbiguousPrefixes) (int64, error) {
 	return s.q.ReconfirmRoutes(ctx, sqlc.ReconfirmRoutesParams{
 		BatchSize: batchSize,
 		Before:    pgtype.Timestamptz{Time: before, Valid: true},
+		AmbIata:   amb.IATAs,
+		AmbLen:    amb.Lens,
+		AmbPrefix: amb.Prefixes,
 	})
 }
 

db/sqlc/mock/querier.go

@@ -43,6 +43,21 @@ func (m *MockQuerier) EXPECT() *MockQuerierMockRecorder {
 	return m.recorder
 }
 
+// AmbiguousPrefixes mocks base method.
+func (m *MockQuerier) AmbiguousPrefixes(ctx context.Context) ([]db.AmbiguousPrefixesRow, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AmbiguousPrefixes", ctx)
+	ret0, _ := ret[0].([]db.AmbiguousPrefixesRow)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// AmbiguousPrefixes indicates an expected call of AmbiguousPrefixes.
+func (mr *MockQuerierMockRecorder) AmbiguousPrefixes(ctx any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AmbiguousPrefixes", reflect.TypeOf((*MockQuerier)(nil).AmbiguousPrefixes), ctx)
+}
+
 // CreateAccount mocks base method.
 func (m *MockQuerier) CreateAccount(ctx context.Context, name string) (db.Account, error) {
 	m.ctrl.T.Helper()

db/sqlc/querier.go

@@ -12,6 +12,8 @@ import (
 )
 
 type Querier interface {
+	// Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run.
+	AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow, error)
 	CreateAccount(ctx context.Context, name string) (Account, error)
 	// Lock the current row before deciding the outcome, including when another
 	// deactivation commits while this statement is waiting for its row lock.
@@ -223,7 +225,8 @@ type Querier interface {
 	ReconfirmNeighbors(ctx context.Context) error
 	// Checks one batch of 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.
+	// 1/2/3/4-byte hop prefixes check prefix_1/2/3/4; ambiguity set supplied by AmbiguousPrefixes),
+	// and stamps the survivors.
 	ReconfirmRoutes(ctx context.Context, arg ReconfirmRoutesParams) (int64, error)
 	RefreshHourlyStats(ctx context.Context) error
 	RefreshObserverActivity(ctx context.Context) error

db/sqlc/queries.sql.go

@@ -12,6 +12,43 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 )
 
+const ambiguousPrefixes = `-- name: AmbiguousPrefixes :many
+SELECT iata::text AS iata, 1::int AS len, prefix_1 AS prefix FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1
+UNION ALL
+SELECT iata::text, 2::int, prefix_2 FROM node_short_ids GROUP BY iata, prefix_2 HAVING COUNT(*) > 1
+UNION ALL
+SELECT iata::text, 3::int, prefix_3 FROM node_short_ids GROUP BY iata, prefix_3 HAVING COUNT(*) > 1
+UNION ALL
+SELECT iata::text, 4::int, prefix_4 FROM node_short_ids GROUP BY iata, prefix_4 HAVING COUNT(*) > 1
+`
+
+type AmbiguousPrefixesRow struct {
+	Iata   string `json:"iata"`
+	Len    int32  `json:"len"`
+	Prefix []byte `json:"prefix"`
+}
+
+// Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run.
+func (q *Queries) AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow, error) {
+	rows, err := q.db.Query(ctx, ambiguousPrefixes)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []AmbiguousPrefixesRow{}
+	for rows.Next() {
+		var i AmbiguousPrefixesRow
+		if err := rows.Scan(&i.Iata, &i.Len, &i.Prefix); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const createAccount = `-- name: CreateAccount :one
 INSERT INTO accounts (name) VALUES ($1)
 ON CONFLICT (name) WHERE deactivated_at IS NULL DO NOTHING
@@ -3972,14 +4009,9 @@ WITH batch AS MATERIALIZED (
     LIMIT $2
     FOR UPDATE OF r SKIP LOCKED
 ),
-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
+amb AS (
+    SELECT a.iata::char(3) AS iata, a.len, a.p
+    FROM ROWS FROM (unnest($3::text[]), unnest($4::int[]), unnest($5::bytea[])) AS a(iata, len, p)
 ),
 dead AS (
     SELECT b.iata, b.path_key
@@ -4020,13 +4052,23 @@ SELECT count(*) FROM batch
 type ReconfirmRoutesParams struct {
 	Before    pgtype.Timestamptz `json:"before"`
 	BatchSize int32              `json:"batch_size"`
+	AmbIata   []string           `json:"amb_iata"`
+	AmbLen    []int32            `json:"amb_len"`
+	AmbPrefix [][]byte           `json:"amb_prefix"`
 }
 
 // Checks one batch of 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.
+// 1/2/3/4-byte hop prefixes check prefix_1/2/3/4; ambiguity set supplied by AmbiguousPrefixes),
+// and stamps the survivors.
 func (q *Queries) ReconfirmRoutes(ctx context.Context, arg ReconfirmRoutesParams) (int64, error) {
-	row := q.db.QueryRow(ctx, reconfirmRoutes, arg.Before, arg.BatchSize)
+	row := q.db.QueryRow(ctx, reconfirmRoutes,
+		arg.Before,
+		arg.BatchSize,
+		arg.AmbIata,
+		arg.AmbLen,
+		arg.AmbPrefix,
+	)
 	var count int64
 	err := row.Scan(&count)
 	return count, err

internal/background/reconfirm_test.go

@@ -9,10 +9,13 @@ import (
 	"testing"
 	"testing/synctest"
 	"time"
+
+	"github.com/MeshCore-Beacon/beacon-server/db"
 )
 
 type reconfirmStore struct {
 	run       func(context.Context, int32, time.Time) (int64, error)
+	ambCalls  int
 	neighbors bool
 }
 
@@ -20,7 +23,12 @@ func (*reconfirmStore) DeleteOldRoutes(context.Context, time.Time, int64, time.T
 	return nil
 }
 
-func (s *reconfirmStore) ReconfirmRoutes(ctx context.Context, n int32, before time.Time) (int64, error) {
+func (s *reconfirmStore) AmbiguousPrefixes(context.Context) (db.AmbiguousPrefixes, error) {
+	s.ambCalls++
+	return db.AmbiguousPrefixes{}, nil
+}
+
+func (s *reconfirmStore) ReconfirmRoutes(ctx context.Context, n int32, before time.Time, _ db.AmbiguousPrefixes) (int64, error) {
 	return s.run(ctx, n, before)
 }
 
@@ -60,6 +68,9 @@ func TestReconfirmCoverage(t *testing.T) {
 			if left != tc.wantLeft || !s.neighbors {
 				t.Fatalf("remaining %d, neighbor cleanup %t", left, s.neighbors)
 			}
+			if s.ambCalls != 1 {
+				t.Fatalf("ambiguity computed %d times, want once per run", s.ambCalls)
+			}
 		})
 	}
 }

internal/background/tasks.go

@@ -95,7 +95,8 @@ const (
 
 type routeMaintainer interface {
 	DeleteOldRoutes(context.Context, time.Time, int64, time.Time) error
-	ReconfirmRoutes(context.Context, int32, time.Time) (int64, error)
+	AmbiguousPrefixes(context.Context) (db.AmbiguousPrefixes, error)
+	ReconfirmRoutes(context.Context, int32, time.Time, db.AmbiguousPrefixes) (int64, error)
 	ReconfirmNeighbors(context.Context) error
 }
 
@@ -137,10 +138,14 @@ func ReconfirmTask(store routeMaintainer, routeRetention, routeGrace time.Durati
 			if err := store.DeleteOldRoutes(ctx, now.Add(-routeRetention), routeMinObservations, now.Add(-routeGrace)); err != nil {
 				return fmt.Errorf("route retention: %w", err)
 			}
+			amb, err := store.AmbiguousPrefixes(ctx)
+			if err != nil {
+				return fmt.Errorf("ambiguous prefixes: %w", err)
+			}
 			for remaining := int64(reconfirmRunLimit); remaining > 0; {
 				limit := min(int64(reconfirmBatchSize), remaining)
 				batchCtx, cancel := context.WithTimeout(ctx, reconfirmBatchTimeout)
-				n, err := store.ReconfirmRoutes(batchCtx, int32(limit), now)
+				n, err := store.ReconfirmRoutes(batchCtx, int32(limit), now, amb)
 				cancel()
 				if err != nil {
 					return fmt.Errorf("routes: %w", err)