diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 9c46794..ccc8492 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -1481,15 +1481,17 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets; -- name: RefreshObserverActivity :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_observer_activity_hourly; --- name: ReconfirmRoutes :exec --- Checks the $1 least-recently-reconfirmed routes: deletes those with a departed +-- 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. -WITH batch AS ( - SELECT iata, path_key, node_ids, hash_prefix - FROM known_routes - ORDER BY last_reconfirmed_at - LIMIT $1 +WITH batch AS MATERIALIZED ( + SELECT r.iata, r.path_key, r.node_ids, r.hash_prefix + FROM known_routes r + WHERE r.last_reconfirmed_at < @before + ORDER BY r.last_reconfirmed_at + 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 @@ -1522,15 +1524,18 @@ deleted AS ( DELETE FROM known_routes kr USING dead d WHERE kr.iata = d.iata AND kr.path_key = d.path_key +), +updated AS ( + UPDATE known_routes kr + SET last_reconfirmed_at = GREATEST(NOW(), @before::timestamptz) + 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 + ) ) -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 - ); +SELECT count(*) FROM batch; -- name: ReconfirmNeighbors :exec -- Delete node_neighbors where the neighbor has departed from node_short_ids diff --git a/db/reconfirm_integration_test.go b/db/reconfirm_integration_test.go new file mode 100644 index 0000000..7aeed34 --- /dev/null +++ b/db/reconfirm_integration_test.go @@ -0,0 +1,324 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "bytes" + "context" + "fmt" + "os" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +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 { + t.Fatalf("first batch: %d, %v", n, err) + } + gate, err := pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer gate.Rollback(context.Background()) + key := time.Now().UnixNano() + if _, err := gate.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, key); err != nil { + t.Fatal(err) + } + _, err = pool.Exec(ctx, fmt.Sprintf(` +CREATE FUNCTION pause_second_route() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN IF NEW.path_key=int4send(2) THEN PERFORM pg_advisory_xact_lock(%d); END IF; RETURN NEW; END $$; +CREATE TRIGGER pause_second_route BEFORE UPDATE ON known_routes FOR EACH ROW EXECUTE FUNCTION pause_second_route();`, key)) + if err != nil { + t.Fatal(err) + } + finished := make(chan error, 1) + go func() { + _, err := store.ReconfirmRoutes(ctx, 1, before) + finished <- err + }() + deadline := time.Now().Add(3 * time.Second) + for { + var waiting bool + if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_locks WHERE locktype='advisory' AND NOT granted AND database=(SELECT oid FROM pg_database WHERE datname=current_database()))`).Scan(&waiting); err != nil { + t.Fatal(err) + } + if waiting { + break + } + if time.Now().After(deadline) { + t.Fatal("second batch did not reach the test gate") + } + time.Sleep(time.Millisecond) + } + upsertCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) + defer cancel() + err = sqlc.New(pool).UpsertKnownRoute(upsertCtx, sqlc.UpsertKnownRouteParams{ + PathKey: []byte{0, 0, 0, 1}, NodeIds: []uuid.UUID{uuid.MustParse("00000000-0000-0000-0000-000000000001")}, + HashPrefix: [][]byte{{0x11}}, Iata: "SEA", HopCount: 1, + }) + if err != nil { + t.Fatalf("completed batch still blocks ingest: %v", err) + } + if err := gate.Rollback(ctx); err != nil { + t.Fatal(err) + } + if err := <-finished; err != nil { + t.Fatal(err) + } + var count int + if err := pool.QueryRow(ctx, `SELECT observation_count FROM known_routes WHERE path_key=int4send(1)`).Scan(&count); err != nil || count != 2 { + t.Fatalf("concurrent observation increment lost: count %d, error %v", count, err) + } +} + +func TestReconfirmPreservesRouteDataPostgres(t *testing.T) { + ctx, pool, store := reconfirmPool(t) + before := time.Now().UTC().Truncate(time.Microsecond) + _, err := pool.Exec(ctx, ` +TRUNCATE known_routes, node_short_ids; +INSERT INTO node_short_ids (node_id,iata,prefix_4) VALUES + (md5('n1')::uuid,'SEA','\x11223344'), (md5('n2')::uuid,'SEA','\x11223355'), + (md5('n3')::uuid,'SEA','\xaabbccdd'), (md5('n4')::uuid,'SEA','\xaabbccdd'), + (md5('n5')::uuid,'SEA','\x55667788'), (md5('n6')::uuid,'SFO','\x55667799'); +INSERT INTO known_routes (id,path_key,node_ids,hash_prefix,iata,hop_count,first_seen,last_seen,observation_count,last_reconfirmed_at) +SELECT id,int4send(id),ARRAY[md5(node)::uuid],ARRAY[decode(prefix,'hex')],iata,1, + '2025-12-01','2026-01-01',42,'2026-01-01' +FROM (VALUES + (1,'n1','11223344','SEA'), (2,'n2','11223355','SEA'), + (3,'n5','55','SEA'), (4,'n5','5566','SEA'), (5,'n5','556677','SEA'), (6,'n5','55667788','SEA'), + (7,'n1','11','SEA'), (8,'n1','1122','SEA'), (9,'n1','112233','SEA'), (10,'n3','aabbccdd','SEA'), + (11,'missing','55','SEA'), (12,'n6','55','SEA'), + (13,'n1','11223344','SEA'), (14,'n5','55667788','SEA'), (15,'n6','55','SFO'), + (16,'n5','55','SEA'), (17,'n5','55','SEA') +) AS routes(id,node,prefix,iata); +UPDATE known_routes SET node_ids=node_ids || md5('n5')::uuid, hash_prefix=hash_prefix || '\x55667788'::bytea, hop_count=2 WHERE id=13; +UPDATE known_routes SET node_ids=node_ids || md5('missing')::uuid, hash_prefix=hash_prefix || '\x99'::bytea, hop_count=2 WHERE id=14;`) + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `UPDATE known_routes SET last_reconfirmed_at=$1::timestamptz + CASE WHEN id=16 THEN interval '1 hour' ELSE interval '0' END WHERE id IN (16,17)`, before); err != nil { + t.Fatal(err) + } + snapshot := func() map[int][]byte { + t.Helper() + rows, err := pool.Query(ctx, `SELECT id, to_jsonb(r)-'last_reconfirmed_at' FROM known_routes r`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + result := map[int][]byte{} + for rows.Next() { + var id int + var data []byte + if err := rows.Scan(&id, &data); err != nil { + t.Fatal(err) + } + result[id] = data + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + return result + } + original := snapshot() + var checked int64 + for range 10 { + n, err := store.ReconfirmRoutes(ctx, 3, before) + if err != nil { + t.Fatal(err) + } + checked += n + if n == 0 { + break + } + } + if checked != 15 { + t.Fatalf("checked %d routes; want 15 once each, including deleted routes", checked) + } + after := snapshot() + want := []int{1, 2, 3, 4, 5, 6, 13, 15, 16, 17} + if len(after) != len(want) { + t.Fatalf("surviving routes: %v", after) + } + for _, id := range want { + if !bytes.Equal(original[id], after[id]) { + t.Errorf("route %d lost or changed fields: before %s, after %s", id, original[id], after[id]) + } + } + var stamped int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_routes WHERE id NOT IN (16,17) AND last_reconfirmed_at >= $1`, before).Scan(&stamped); err != nil { + t.Fatal(err) + } + if stamped != 8 { + t.Errorf("validated survivors = %d, want 8", stamped) + } +} + +func TestReconfirmCancelledBatchRollsBackPostgres(t *testing.T) { + ctx, pool, store := reconfirmPool(t) + before := time.Now() + if _, err := store.ReconfirmRoutes(ctx, 1, before); err != nil { + t.Fatal(err) + } + _, err := pool.Exec(ctx, ` +CREATE FUNCTION pause_reconfirm() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN PERFORM pg_sleep(1); RETURN NEW; END $$; +CREATE TRIGGER pause_reconfirm BEFORE UPDATE ON known_routes FOR EACH ROW EXECUTE FUNCTION pause_reconfirm();`) + if err != nil { + t.Fatal(err) + } + batchCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) + defer cancel() + if _, err := store.ReconfirmRoutes(batchCtx, 1, before); err == nil { + t.Fatal("expected batch cancellation") + } + // Waiting for this DDL also waits for cancellation to release the row lock. + if _, err := pool.Exec(ctx, `DROP TRIGGER pause_reconfirm ON known_routes`); err != nil { + t.Fatal(err) + } + var total, stamped int + if err := pool.QueryRow(ctx, `SELECT count(*),count(*) FILTER (WHERE last_reconfirmed_at >= $1) FROM known_routes`, before).Scan(&total, &stamped); err != nil { + t.Fatal(err) + } + 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 { + t.Fatalf("cancelled route was not available to retry: %d, %v", n, err) + } +} + +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 { + t.Fatalf("first batch: %d, %v", n, err) + } + if n, err := store.ReconfirmRoutes(ctx, 2, before); err != nil || n != 0 { + t.Fatalf("completed routes consumed the run budget twice: %d, %v", n, err) + } +} + +func TestReconfirmCancellationDoesNotPartiallyDeletePostgres(t *testing.T) { + ctx, pool, store := reconfirmPool(t) + _, err := pool.Exec(ctx, ` +UPDATE known_routes SET node_ids=ARRAY[md5('departed')::uuid] WHERE path_key=int4send(1); +CREATE FUNCTION pause_valid_route() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN PERFORM pg_sleep(1); RETURN NEW; END $$; +CREATE TRIGGER pause_valid_route BEFORE UPDATE ON known_routes FOR EACH ROW EXECUTE FUNCTION pause_valid_route();`) + if err != nil { + t.Fatal(err) + } + before := time.Now() + batchCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) + defer cancel() + if _, err := store.ReconfirmRoutes(batchCtx, 2, before); err == nil { + t.Fatal("expected batch cancellation") + } + if _, err := pool.Exec(ctx, `DROP TRIGGER pause_valid_route ON known_routes`); err != nil { + t.Fatal(err) + } + var count int + 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 { + 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 { + t.Fatalf("retry lost the valid route: %d, %v", count, err) + } +} + +// A private schema lets independent connections exercise row locks. +func reconfirmPool(t *testing.T) (context.Context, *pgxpool.Pool, *Store) { + t.Helper() + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN for PostgreSQL tests") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + t.Cleanup(cancel) + admin, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { admin.Close(context.Background()) }) + schema := pgx.Identifier{"reconfirm_" + uuid.NewString()}.Sanitize() + if _, err := admin.Exec(ctx, "CREATE SCHEMA "+schema); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if _, err := admin.Exec(context.Background(), "DROP SCHEMA "+schema+" CASCADE"); err != nil { + t.Error(err) + } + }) + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + cfg.MaxConns = 4 + cfg.ConnConfig.RuntimeParams["search_path"] = schema + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + _, err = pool.Exec(ctx, ` +CREATE TABLE known_routes (LIKE public.known_routes INCLUDING ALL); +CREATE TABLE node_short_ids (LIKE public.node_short_ids INCLUDING ALL); +INSERT INTO node_short_ids (node_id, iata, prefix_4) VALUES + ('00000000-0000-0000-0000-000000000001', 'SEA', '\x11223344'); +INSERT INTO known_routes (path_key,node_ids,hash_prefix,iata,hop_count,last_reconfirmed_at) +SELECT int4send(i), ARRAY['00000000-0000-0000-0000-000000000001'::uuid], ARRAY['\x11'::bytea], 'SEA', 1, + '2026-01-01'::timestamptz + i * interval '1 second' +FROM generate_series(1,2) i;`) + if err != nil { + t.Fatal(err) + } + return ctx, pool, New(pool, 0, 0) +} + +func TestReconfirmSkipsBusyRoutePostgres(t *testing.T) { + ctx, pool, store := reconfirmPool(t) + busy, err := pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer busy.Rollback(context.Background()) + if _, err := busy.Exec(ctx, `SELECT 1 FROM known_routes WHERE path_key=int4send(1) FOR UPDATE`); err != nil { + t.Fatal(err) + } + batchCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) + defer cancel() + if _, err := store.ReconfirmRoutes(batchCtx, 1, time.Now()); err != nil { + t.Fatalf("maintenance should skip the busy route and validate the next one: %v", err) + } + var checked, preserved int + if err := pool.QueryRow(ctx, `SELECT count(*) FILTER (WHERE last_reconfirmed_at > '2026-02-01'), count(*) FROM known_routes`).Scan(&checked, &preserved); err != nil { + t.Fatal(err) + } + if checked != 1 || preserved != 2 { + t.Fatalf("checked %d, preserved %d; want 1 and 2", checked, preserved) + } + if err := busy.Rollback(ctx); err != nil { + t.Fatal(err) + } + if _, err := store.ReconfirmRoutes(ctx, 1, time.Now()); 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 { + t.Fatal(err) + } + if checked != 2 { + t.Fatalf("deferred route was not validated after release: checked %d", checked) + } +} diff --git a/db/routes.go b/db/routes.go index a629bcb..187e07e 100644 --- a/db/routes.go +++ b/db/routes.go @@ -288,8 +288,11 @@ func (s *Store) SearchCrossIATARoutes(ctx context.Context, fromHash, fromIATA, t // 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) +func (s *Store) ReconfirmRoutes(ctx context.Context, batchSize int32, before time.Time) (int64, error) { + return s.q.ReconfirmRoutes(ctx, sqlc.ReconfirmRoutesParams{ + BatchSize: batchSize, + Before: pgtype.Timestamptz{Time: before, Valid: true}, + }) } const routeDeleteBatch = 10000 diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 6a665e7..aeae62b 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -1253,17 +1253,18 @@ func (mr *MockQuerierMockRecorder) ReconfirmNeighbors(ctx any) *gomock.Call { } // ReconfirmRoutes mocks base method. -func (m *MockQuerier) ReconfirmRoutes(ctx context.Context, limit int32) error { +func (m *MockQuerier) ReconfirmRoutes(ctx context.Context, arg db.ReconfirmRoutesParams) (int64, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "ReconfirmRoutes", ctx, limit) - ret0, _ := ret[0].(error) - return ret0 + ret := m.ctrl.Call(m, "ReconfirmRoutes", ctx, arg) + ret0, _ := ret[0].(int64) + ret1, _ := ret[1].(error) + return ret0, ret1 } // ReconfirmRoutes indicates an expected call of ReconfirmRoutes. -func (mr *MockQuerierMockRecorder) ReconfirmRoutes(ctx, limit any) *gomock.Call { +func (mr *MockQuerierMockRecorder) ReconfirmRoutes(ctx, arg any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReconfirmRoutes", reflect.TypeOf((*MockQuerier)(nil).ReconfirmRoutes), ctx, limit) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReconfirmRoutes", reflect.TypeOf((*MockQuerier)(nil).ReconfirmRoutes), ctx, arg) } // RefreshHourlyStats mocks base method. diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 232eae3..1547c1c 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -211,10 +211,10 @@ type Querier interface { // Delete node_neighbors where the neighbor has departed from node_short_ids // for that IATA, or where its prefix_4 is now ambiguous. ReconfirmNeighbors(ctx context.Context) error - // Checks the $1 least-recently-reconfirmed routes: deletes those with a departed + // 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. - ReconfirmRoutes(ctx context.Context, limit int32) error + ReconfirmRoutes(ctx context.Context, arg ReconfirmRoutesParams) (int64, error) RefreshHourlyStats(ctx context.Context) error RefreshObserverActivity(ctx context.Context) error RefreshPathStats(ctx context.Context) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 38d299c..c047b44 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -3865,12 +3865,14 @@ func (q *Queries) ReconfirmNeighbors(ctx context.Context) error { return err } -const reconfirmRoutes = `-- name: ReconfirmRoutes :exec -WITH batch AS ( - SELECT iata, path_key, node_ids, hash_prefix - FROM known_routes - ORDER BY last_reconfirmed_at - LIMIT $1 +const reconfirmRoutes = `-- name: ReconfirmRoutes :one +WITH batch AS MATERIALIZED ( + SELECT r.iata, r.path_key, r.node_ids, r.hash_prefix + FROM known_routes r + WHERE r.last_reconfirmed_at < $1 + ORDER BY r.last_reconfirmed_at + 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 @@ -3903,23 +3905,33 @@ deleted AS ( DELETE FROM known_routes kr USING dead d WHERE kr.iata = d.iata AND kr.path_key = d.path_key +), +updated AS ( + UPDATE known_routes kr + SET last_reconfirmed_at = GREATEST(NOW(), $1::timestamptz) + 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 + ) ) -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 - ) +SELECT count(*) FROM batch ` -// Checks the $1 least-recently-reconfirmed routes: deletes those with a departed +type ReconfirmRoutesParams struct { + Before pgtype.Timestamptz `json:"before"` + BatchSize int32 `json:"batch_size"` +} + +// 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. -func (q *Queries) ReconfirmRoutes(ctx context.Context, limit int32) error { - _, err := q.db.Exec(ctx, reconfirmRoutes, limit) - return err +func (q *Queries) ReconfirmRoutes(ctx context.Context, arg ReconfirmRoutesParams) (int64, error) { + row := q.db.QueryRow(ctx, reconfirmRoutes, arg.Before, arg.BatchSize) + var count int64 + err := row.Scan(&count) + return count, err } const refreshHourlyStats = `-- name: RefreshHourlyStats :exec diff --git a/internal/background/reconfirm_test.go b/internal/background/reconfirm_test.go new file mode 100644 index 0000000..2012688 --- /dev/null +++ b/internal/background/reconfirm_test.go @@ -0,0 +1,95 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package background + +import ( + "context" + "errors" + "testing" + "testing/synctest" + "time" +) + +type reconfirmStore struct { + run func(context.Context, int32, time.Time) (int64, error) + neighbors bool +} + +func (*reconfirmStore) DeleteOldRoutes(context.Context, time.Time, int64, time.Time) error { + return nil +} + +func (s *reconfirmStore) ReconfirmRoutes(ctx context.Context, n int32, before time.Time) (int64, error) { + return s.run(ctx, n, before) +} + +func (s *reconfirmStore) ReconfirmNeighbors(context.Context) error { + s.neighbors = true + return nil +} + +func TestReconfirmCoverage(t *testing.T) { + for _, tc := range []struct { + name string + available, wantLeft int64 + }{ + {"empty", 0, 0}, + {"partial last batch", 1501, 0}, + {"full run budget", 750010, 10}, + } { + t.Run(tc.name, func(t *testing.T) { + left := tc.available + var cutoff time.Time + s := &reconfirmStore{run: func(ctx context.Context, n int32, before time.Time) (int64, error) { + if n <= 0 || n > 1000 { + t.Fatalf("transaction requested %d routes", n) + } + if cutoff.IsZero() { + cutoff = before + } else if !cutoff.Equal(before) { + t.Fatal("cutoff advanced, allowing completed routes to be revisited") + } + checked := min(int64(n), left) + left -= checked + return checked, ctx.Err() + }} + if err := ReconfirmTask(s, time.Hour, time.Minute, 3, time.Hour).Run(context.Background()); err != nil { + t.Fatal(err) + } + if left != tc.wantLeft || !s.neighbors { + t.Fatalf("remaining %d, neighbor cleanup %t", left, s.neighbors) + } + }) + } +} + +func TestReconfirmStopsOnFailure(t *testing.T) { + want := errors.New("database unavailable") + calls := 0 + s := &reconfirmStore{run: func(context.Context, int32, time.Time) (int64, error) { + calls++ + if calls == 1 { + return 1000, nil + } + return 0, want + }} + err := ReconfirmTask(s, time.Hour, time.Minute, 3, time.Hour).Run(context.Background()) + if !errors.Is(err, want) || calls != 2 || s.neighbors { + t.Fatalf("failure was not propagated: error %v, calls %d, neighbors %t", err, calls, s.neighbors) + } +} + +func TestReconfirmBatchDeadline(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s := &reconfirmStore{run: func(ctx context.Context, _ int32, _ time.Time) (int64, error) { + <-ctx.Done() + return 0, ctx.Err() + }} + start := time.Now() + err := ReconfirmTask(s, time.Hour, time.Minute, 3, time.Hour).Run(context.Background()) + if !errors.Is(err, context.DeadlineExceeded) || time.Since(start) > 5*time.Second || s.neighbors { + t.Fatalf("batch exceeded its deadline: error %v, duration %s, neighbors %t", err, time.Since(start), s.neighbors) + } + }) +} diff --git a/internal/background/tasks.go b/internal/background/tasks.go index 64c4297..7d82fdf 100644 --- a/internal/background/tasks.go +++ b/internal/background/tasks.go @@ -86,9 +86,18 @@ func CleanupTask(store *db.Store, telemetryRetention, packetRetention, nodeDelet } } -// reconfirmBatchSize bounds per-tick reconfirm work; at hourly ticks a 16M-row -// table gets fully re-checked roughly daily. -const reconfirmBatchSize = 750_000 +// Limit lock lifetime while retaining the per-run work budget. +const ( + reconfirmRunLimit = 750_000 + reconfirmBatchSize = 1_000 + reconfirmBatchTimeout = 5 * time.Second +) + +type routeMaintainer interface { + DeleteOldRoutes(context.Context, time.Time, int64, time.Time) error + ReconfirmRoutes(context.Context, int32, time.Time) (int64, error) + ReconfirmNeighbors(context.Context) error +} type observerCleaner interface { DeleteOldObservers(context.Context, time.Time) ([]uuid.UUID, error) @@ -118,9 +127,8 @@ func ObserverCleanupTask(store observerCleaner, deleteAfter, interval time.Durat } // ReconfirmTask returns a Task that prunes aged routes first, then reconfirms -// stale and ambiguous resolved paths and neighbors, so known_routes only ever -// has one writer at a time. -func ReconfirmTask(store *db.Store, routeRetention, routeGrace time.Duration, routeMinObservations int64, interval time.Duration) Task { +// stale and ambiguous resolved paths and neighbors in order. +func ReconfirmTask(store routeMaintainer, routeRetention, routeGrace time.Duration, routeMinObservations int64, interval time.Duration) Task { return Task{ Name: "reconfirm", Interval: interval, @@ -129,8 +137,18 @@ func ReconfirmTask(store *db.Store, routeRetention, routeGrace time.Duration, ro if err := store.DeleteOldRoutes(ctx, now.Add(-routeRetention), routeMinObservations, now.Add(-routeGrace)); err != nil { return fmt.Errorf("route retention: %w", err) } - if err := store.ReconfirmRoutes(ctx, reconfirmBatchSize); err != nil { - return fmt.Errorf("routes: %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) + cancel() + if err != nil { + return fmt.Errorf("routes: %w", err) + } + remaining -= n + if n < limit { + break + } } if err := store.ReconfirmNeighbors(ctx); err != nil { return fmt.Errorf("neighbors: %w", err)