From 4ad869d711b41ff6bb3ef726098b5c5b329d8531 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Fri, 2 Oct 2026 18:59:09 -0400 Subject: [PATCH] fix(db): page known routes by (last_seen, id) The cursor was the last row's last_seen in ms, so routes sharing that millisecond (common within one upsert batch) never came back. GET /routes takes an optional cursorId, backed by keyset indexes; cursor alone keeps working as before. --- db/migrations/001_baseline.sql | 4 ++ db/queries/queries.sql | 21 ++++-- db/routes.go | 3 +- db/routes_integration_test.go | 2 +- db/routes_pagination_integration_test.go | 78 +++++++++++++++++++++++ db/sqlc/queries.sql.go | 23 +++++-- internal/api/handlers/limits_test.go | 2 +- internal/api/handlers/routes.go | 17 ++++- internal/api/handlers/routes_test.go | 36 ++++++++++- internal/api/handlers/stub_reader_test.go | 6 +- internal/api/reader.go | 3 +- internal/cache/cache_test.go | 2 +- internal/cache/reader.go | 4 +- 13 files changed, 175 insertions(+), 26 deletions(-) create mode 100644 db/routes_pagination_integration_test.go diff --git a/db/migrations/001_baseline.sql b/db/migrations/001_baseline.sql index 71e42f2..4126ad7 100644 --- a/db/migrations/001_baseline.sql +++ b/db/migrations/001_baseline.sql @@ -761,6 +761,10 @@ CREATE INDEX idx_known_routes_hop_count ON known_routes USING btree (iata, hop_c CREATE INDEX idx_known_routes_iata_last_seen ON known_routes USING btree (iata, last_seen DESC); +CREATE INDEX idx_known_routes_iata_keyset ON known_routes USING btree (iata, date_trunc('milliseconds', last_seen, 'UTC') DESC, id DESC); + +CREATE INDEX idx_known_routes_keyset ON known_routes USING btree (date_trunc('milliseconds', last_seen, 'UTC') DESC, id DESC); + CREATE INDEX idx_known_routes_last_seen ON known_routes USING btree (last_seen DESC); CREATE INDEX idx_known_routes_reconfirm ON known_routes USING btree (last_reconfirmed_at); diff --git a/db/queries/queries.sql b/db/queries/queries.sql index ecdcffe..079593c 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -1462,13 +1462,18 @@ ON CONFLICT (iata, path_key) DO UPDATE SET -- Only one branch runs. Keep the IATA range ordered by the composite index: -- generic plans can otherwise prefer scanning the global timestamp index. -- The text equality preserves exact input matching, including trailing spaces. -( +-- Pages by (last_seen to the ms, id) so routes sharing the cursor's millisecond +-- aren't skipped; cursor id 0 keeps the plain timestamp cursor. +SELECT r.id, r.node_ids, r.hash_prefix, r.iata, r.hop_count, r.first_seen, r.last_seen, r.observation_count +FROM (( SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count FROM known_routes WHERE $1 = '' AND ($2 = 0 OR hop_count = $2) - AND ($3::timestamptz IS NULL OR last_seen < $3) -ORDER BY last_seen DESC + AND ($3::timestamptz IS NULL + OR date_trunc('milliseconds', last_seen, 'UTC') < $3 + OR ($5::bigint > 0 AND date_trunc('milliseconds', last_seen, 'UTC') = $3 AND id < $5)) +ORDER BY date_trunc('milliseconds', last_seen, 'UTC') DESC, id DESC LIMIT $4 ) UNION ALL @@ -1478,11 +1483,13 @@ FROM known_routes WHERE $1 <> '' AND iata::text = $1 AND iata >= $1::bpchar AND iata <= $1::bpchar AND ($2 = 0 OR hop_count = $2) - AND ($3::timestamptz IS NULL OR last_seen < $3) -ORDER BY iata, last_seen DESC + AND ($3::timestamptz IS NULL + OR date_trunc('milliseconds', last_seen, 'UTC') < $3 + OR ($5::bigint > 0 AND date_trunc('milliseconds', last_seen, 'UTC') = $3 AND id < $5)) +ORDER BY iata, date_trunc('milliseconds', last_seen, 'UTC') DESC, id DESC LIMIT $4 -) -ORDER BY last_seen DESC +)) r +ORDER BY date_trunc('milliseconds', r.last_seen, 'UTC') DESC, r.id DESC LIMIT $4; -- name: SearchKnownRoutes :many diff --git a/db/routes.go b/db/routes.go index 69183fb..73b17c5 100644 --- a/db/routes.go +++ b/db/routes.go @@ -37,7 +37,7 @@ func (s *Store) UpsertKnownRoute(ctx context.Context, nodeIDs []uuid.UUID, hashP }) } -func (s *Store) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, limit int32) ([]api.KnownRoute, error) { +func (s *Store) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, cursorID int64, limit int32) ([]api.KnownRoute, error) { var cursorTS pgtype.Timestamptz if !cursor.IsZero() { cursorTS = pgtype.Timestamptz{Time: cursor, Valid: true} @@ -47,6 +47,7 @@ func (s *Store) ListKnownRoutes(ctx context.Context, iata string, hopCount int32 Column2: hopCount, Column3: cursorTS, Limit: limit, + Column5: cursorID, }) if err != nil { return nil, err diff --git a/db/routes_integration_test.go b/db/routes_integration_test.go index 6d6addf..7e6594c 100644 --- a/db/routes_integration_test.go +++ b/db/routes_integration_test.go @@ -111,7 +111,7 @@ ANALYZE known_routes;`) } for _, iata := range []string{"GPT", "YYZ", ""} { var planJSON []byte - if err := tx.QueryRow(ctx, "EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE route_page('"+iata+"',0,NULL,50)").Scan(&planJSON); err != nil { + if err := tx.QueryRow(ctx, "EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE route_page('"+iata+"',0,NULL,50,0)").Scan(&planJSON); err != nil { t.Fatal(err) } var plans []struct { diff --git a/db/routes_pagination_integration_test.go b/db/routes_pagination_integration_test.go new file mode 100644 index 0000000..8151329 --- /dev/null +++ b/db/routes_pagination_integration_test.go @@ -0,0 +1,78 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "os" + "slices" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/jackc/pgx/v5" +) + +// Routes upserted in one batch share last_seen; paging must not skip any of them. +func TestKnownRoutePaginationTiesPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN for the PostgreSQL regression test") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + applyBaseline(t, ctx, tx) + // Seven routes in one millisecond (two with sub-ms offsets), one older, one in another IATA. + if _, err := tx.Exec(ctx, ` +INSERT INTO iata_codes (iata) VALUES ('YYZ'), ('YOW') ON CONFLICT DO NOTHING; +INSERT INTO known_routes (id,path_key,node_ids,hash_prefix,iata,hop_count,first_seen,last_seen) +SELECT i, decode(md5(i::text),'hex'), ARRAY[]::uuid[], ARRAY['\x01'::bytea], + CASE WHEN i = 9 THEN 'YOW' ELSE 'YYZ' END, 2, + '2026-01-01 00:00:00+00'::timestamptz, + CASE WHEN i = 8 THEN '2026-01-01 00:00:04+00'::timestamptz + WHEN i IN (3, 6) THEN '2026-01-01 00:00:05.000400+00'::timestamptz + ELSE '2026-01-01 00:00:05+00'::timestamptz END +FROM generate_series(1,9) i;`); err != nil { + t.Fatal(err) + } + store := &Store{q: sqlc.New(tx)} + for _, iata := range []string{"", "YYZ"} { + t.Run("iata="+iata, func(t *testing.T) { + var got []int64 + var cursor time.Time + var cursorID int64 + for page := 0; page < 10; page++ { + routes, err := store.ListKnownRoutes(ctx, iata, 0, cursor, cursorID, 3) + if err != nil { + t.Fatal(err) + } + if len(routes) == 0 { + break + } + for _, r := range routes { + got = append(got, r.ID) + } + last := routes[len(routes)-1] + cursor, cursorID = time.UnixMilli(last.LastSeen), last.ID + } + want := []int64{7, 6, 5, 4, 3, 2, 1, 8} + if iata == "" { + want = []int64{9, 7, 6, 5, 4, 3, 2, 1, 8} + } + if !slices.Equal(got, want) { + t.Fatalf("paged ids = %v, want %v", got, want) + } + }) + } +} diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 435bb94..df93bc2 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2968,13 +2968,16 @@ func (q *Queries) ListIATAs(ctx context.Context) ([]IataCode, error) { } const listKnownRoutes = `-- name: ListKnownRoutes :many -( +SELECT r.id, r.node_ids, r.hash_prefix, r.iata, r.hop_count, r.first_seen, r.last_seen, r.observation_count +FROM (( SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count FROM known_routes WHERE $1 = '' AND ($2 = 0 OR hop_count = $2) - AND ($3::timestamptz IS NULL OR last_seen < $3) -ORDER BY last_seen DESC + AND ($3::timestamptz IS NULL + OR date_trunc('milliseconds', last_seen, 'UTC') < $3 + OR ($5::bigint > 0 AND date_trunc('milliseconds', last_seen, 'UTC') = $3 AND id < $5)) +ORDER BY date_trunc('milliseconds', last_seen, 'UTC') DESC, id DESC LIMIT $4 ) UNION ALL @@ -2984,11 +2987,13 @@ FROM known_routes WHERE $1 <> '' AND iata::text = $1 AND iata >= $1::bpchar AND iata <= $1::bpchar AND ($2 = 0 OR hop_count = $2) - AND ($3::timestamptz IS NULL OR last_seen < $3) -ORDER BY iata, last_seen DESC + AND ($3::timestamptz IS NULL + OR date_trunc('milliseconds', last_seen, 'UTC') < $3 + OR ($5::bigint > 0 AND date_trunc('milliseconds', last_seen, 'UTC') = $3 AND id < $5)) +ORDER BY iata, date_trunc('milliseconds', last_seen, 'UTC') DESC, id DESC LIMIT $4 -) -ORDER BY last_seen DESC +)) r +ORDER BY date_trunc('milliseconds', r.last_seen, 'UTC') DESC, r.id DESC LIMIT $4 ` @@ -2997,6 +3002,7 @@ type ListKnownRoutesParams struct { Column2 interface{} `json:"column_2"` Column3 pgtype.Timestamptz `json:"column_3"` Limit int32 `json:"limit"` + Column5 int64 `json:"column_5"` } type ListKnownRoutesRow struct { @@ -3013,12 +3019,15 @@ type ListKnownRoutesRow struct { // Only one branch runs. Keep the IATA range ordered by the composite index: // generic plans can otherwise prefer scanning the global timestamp index. // The text equality preserves exact input matching, including trailing spaces. +// Pages by (last_seen to the ms, id) so routes sharing the cursor's millisecond +// aren't skipped; cursor id 0 keeps the plain timestamp cursor. func (q *Queries) ListKnownRoutes(ctx context.Context, arg ListKnownRoutesParams) ([]ListKnownRoutesRow, error) { rows, err := q.db.Query(ctx, listKnownRoutes, arg.Column1, arg.Column2, arg.Column3, arg.Limit, + arg.Column5, ) if err != nil { return nil, err diff --git a/internal/api/handlers/limits_test.go b/internal/api/handlers/limits_test.go index b526769..e904e5c 100644 --- a/internal/api/handlers/limits_test.go +++ b/internal/api/handlers/limits_test.go @@ -99,7 +99,7 @@ func TestListLimits(t *testing.T) { got = limit return nil, nil }, - listKnownRoutes: func(ctx context.Context, iata string, hopCount int32, cursor time.Time, limit int32) ([]api.KnownRoute, error) { + listKnownRoutes: func(ctx context.Context, iata string, hopCount int32, cursor time.Time, _ int64, limit int32) ([]api.KnownRoute, error) { calls++ got = limit return nil, nil diff --git a/internal/api/handlers/routes.go b/internal/api/handlers/routes.go index 77f1117..86a7c86 100644 --- a/internal/api/handlers/routes.go +++ b/internal/api/handlers/routes.go @@ -34,6 +34,7 @@ func RoutesRouter(reader api.Reader) http.Handler { // @Param iata query string false "Filter by IATA code" // @Param hopCount query int false "Filter by exact hop count" // @Param cursor query int false "Epoch ms timestamp of last item for pagination" +// @Param cursorId query int false "id of the last item; with cursor, also returns later routes sharing that millisecond" // @Param limit query int false "Max results (default 50); must be positive, values above 200 are clamped" minimum(1) maximum(200) // @Failure 400 {object} handlers.APIError // @Success 200 {object} []api.KnownRoute @@ -54,12 +55,26 @@ func listKnownRoutes(reader api.Reader) http.HandlerFunc { cursor = time.UnixMilli(ms) } } + // cursorId breaks ties within the cursor's millisecond. + var cursorID int64 + if v := r.URL.Query().Get("cursorId"); v != "" { + id, err := strconv.ParseInt(v, 10, 64) + if err != nil || id <= 0 { + respondError(w, http.StatusBadRequest, "cursorId must be a positive integer") + return + } + if cursor.IsZero() { + respondError(w, http.StatusBadRequest, "cursorId requires cursor") + return + } + cursorID = id + } limit, err := parseLimit(r, 50) if err != nil { respondError(w, http.StatusBadRequest, err.Error()) return } - routes, err := reader.ListKnownRoutes(r.Context(), iata, hopCount, cursor, limit) + routes, err := reader.ListKnownRoutes(r.Context(), iata, hopCount, cursor, cursorID, limit) if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return diff --git a/internal/api/handlers/routes_test.go b/internal/api/handlers/routes_test.go index 699d545..9860d6d 100644 --- a/internal/api/handlers/routes_test.go +++ b/internal/api/handlers/routes_test.go @@ -60,10 +60,44 @@ func TestSearchCrossIATARoutes_MissingParams(t *testing.T) { } } +func TestListKnownRoutes_CursorID(t *testing.T) { + var gotCursor time.Time + var gotID int64 + r := chi.NewRouter() + r.Get("/routes", listKnownRoutes(stubReader{ + listKnownRoutes: func(_ context.Context, _ string, _ int32, cursor time.Time, cursorID int64, _ int32) ([]api.KnownRoute, error) { + gotCursor, gotID = cursor, cursorID + return nil, nil + }, + })) + for _, tc := range []struct { + query string + code int + id int64 + }{ + {"cursor=1700000000000&cursorId=42", http.StatusOK, 42}, + {"cursor=1700000000000", http.StatusOK, 0}, + {"cursor=1700000000000&cursorId=0", http.StatusBadRequest, 0}, + {"cursor=1700000000000&cursorId=x", http.StatusBadRequest, 0}, + {"cursorId=42", http.StatusBadRequest, 0}, + } { + gotID = 0 + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/routes?"+tc.query, nil)) + if w.Code != tc.code { + t.Errorf("%s: code %d, want %d", tc.query, w.Code, tc.code) + continue + } + if tc.code == http.StatusOK && (gotID != tc.id || gotCursor.UnixMilli() != 1700000000000) { + t.Errorf("%s: cursor %v id %d", tc.query, gotCursor, gotID) + } + } +} + func TestListKnownRoutes_OK(t *testing.T) { r := chi.NewRouter() r.Get("/routes", listKnownRoutes(stubReader{ - listKnownRoutes: func(_ context.Context, _ string, _ int32, _ time.Time, _ int32) ([]api.KnownRoute, error) { + listKnownRoutes: func(_ context.Context, _ string, _ int32, _ time.Time, _ int64, _ int32) ([]api.KnownRoute, error) { return []api.KnownRoute{{IATA: "YVR"}}, nil }, })) diff --git a/internal/api/handlers/stub_reader_test.go b/internal/api/handlers/stub_reader_test.go index 6ae0ae4..efb5cb8 100644 --- a/internal/api/handlers/stub_reader_test.go +++ b/internal/api/handlers/stub_reader_test.go @@ -59,7 +59,7 @@ type stubReader struct { getScopeByName func(ctx context.Context, name string) (*api.ScopeDetail, error) listTraceTags func(ctx context.Context, iatas []string, scope, traceType string, since, until time.Time, cursor time.Time, cursorTag string, limit int32) ([]api.TraceTagSummary, error) getTraceByTag func(ctx context.Context, tag string) (*api.TraceDetail, error) - listKnownRoutes func(ctx context.Context, iata string, hopCount int32, cursor time.Time, limit int32) ([]api.KnownRoute, error) + listKnownRoutes func(ctx context.Context, iata string, hopCount int32, cursor time.Time, cursorID int64, limit int32) ([]api.KnownRoute, error) searchKnownRoutes func(ctx context.Context, iata, fromHash, toHash string) ([]api.KnownRoute, error) getKnownRoutesByNode func(ctx context.Context, iata string, nodeID uuid.UUID) ([]api.KnownRoute, error) getCrossIATANeighbors func(ctx context.Context, nodeID uuid.UUID, iata string) ([]api.NodeNeighbor, error) @@ -369,9 +369,9 @@ func (s stubReader) GetTraceByTag(ctx context.Context, tag string) (*api.TraceDe return nil, nil } -func (s stubReader) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, limit int32) ([]api.KnownRoute, error) { +func (s stubReader) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, cursorID int64, limit int32) ([]api.KnownRoute, error) { if s.listKnownRoutes != nil { - return s.listKnownRoutes(ctx, iata, hopCount, cursor, limit) + return s.listKnownRoutes(ctx, iata, hopCount, cursor, cursorID, limit) } return nil, nil } diff --git a/internal/api/reader.go b/internal/api/reader.go index 36c67e0..dc71e63 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -225,7 +225,8 @@ type Reader interface { GetTraceByTag(ctx context.Context, tag string) (*TraceDetail, error) // ListKnownRoutes returns known routes filtered by IATA and optional hop count. - ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, limit int32) ([]KnownRoute, error) + // A nonzero cursorID breaks ties within the cursor's millisecond. + ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, cursorID int64, limit int32) ([]KnownRoute, error) // SearchKnownRoutes returns known routes containing a path from source to destination hash. SearchKnownRoutes(ctx context.Context, iata, fromHash, toHash string) ([]KnownRoute, error) diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 3af439d..b5dcc7f 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -208,7 +208,7 @@ func (s *stubReader) ListPacketsAfterID(_ context.Context, _ int64, _, _ int16, return nil, nil } -func (s *stubReader) ListKnownRoutes(_ context.Context, _ string, _ int32, _ time.Time, _ int32) ([]api.KnownRoute, error) { +func (s *stubReader) ListKnownRoutes(_ context.Context, _ string, _ int32, _ time.Time, _ int64, _ int32) ([]api.KnownRoute, error) { return nil, nil } diff --git a/internal/cache/reader.go b/internal/cache/reader.go index 372cae6..e491152 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -355,8 +355,8 @@ func (cr *CachedReader) ListPacketsAfterID(ctx context.Context, afterObservation } // ListKnownRoutes implements [api.Reader]. -func (cr *CachedReader) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, limit int32) ([]api.KnownRoute, error) { - return cr.inner.ListKnownRoutes(ctx, iata, hopCount, cursor, limit) +func (cr *CachedReader) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, cursorID int64, limit int32) ([]api.KnownRoute, error) { + return cr.inner.ListKnownRoutes(ctx, iata, hopCount, cursor, cursorID, limit) } // SearchKnownRoutes implements [api.Reader].