diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 328960e..77f1c51 100644 --- a/db/queries/queries.sql +++ b/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 diff --git a/db/reconfirm_integration_test.go b/db/reconfirm_integration_test.go index 7aeed34..0b52206 100644 --- a/db/reconfirm_integration_test.go +++ b/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 { diff --git a/db/routes.go b/db/routes.go index 090d2a0..69183fb 100644 --- a/db/routes.go +++ b/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, }) } diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 8a9a7ef..82d1660 100644 --- a/db/sqlc/mock/querier.go +++ b/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() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index ef65fd7..6d699a9 100644 --- a/db/sqlc/querier.go +++ b/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 diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 3b3def7..dcd52c2 100644 --- a/db/sqlc/queries.sql.go +++ b/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 diff --git a/internal/background/reconfirm_test.go b/internal/background/reconfirm_test.go index 2012688..991a7c4 100644 --- a/internal/background/reconfirm_test.go +++ b/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) + } }) } } diff --git a/internal/background/tasks.go b/internal/background/tasks.go index 7d82fdf..951dca7 100644 --- a/internal/background/tasks.go +++ b/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)