mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-23 09:52:32 +00:00
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.
This commit is contained in:
1 parent
a1aff810e9
commit
4ad869d711
13 files changed
+175
-26
No files matched your search
@@ -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);
|
||||
|
||||
+14
-7
@@ -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
|
||||
|
||||
+2
-1
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+16
-7
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
},
|
||||
}))
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Vendored
+1
-1
@@ -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
|
||||
}
|
||||
|
||||
|
||||
Vendored
+2
-2
@@ -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].
|
||||
|
||||
Reference in new issue
Block a user