From dfaa31d1ddb4cad06306cdf7b59d31694ea3367e Mon Sep 17 00:00:00 2001 From: n30nex Date: Sun, 6 Sep 2026 10:20:24 -0400 Subject: [PATCH] perf(db): enrich nodes after selecting the directory page (#111) --- db/nodes_integration_test.go | 185 +++++++++++++++++++++++++++++++++++ db/queries/queries.sql | 31 +++--- db/sqlc/querier.go | 1 + db/sqlc/queries.sql.go | 29 +++--- 4 files changed, 219 insertions(+), 27 deletions(-) create mode 100644 db/nodes_integration_test.go diff --git a/db/nodes_integration_test.go b/db/nodes_integration_test.go new file mode 100644 index 0000000..03ddcd0 --- /dev/null +++ b/db/nodes_integration_test.go @@ -0,0 +1,185 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + _ "embed" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "strings" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" +) + +//go:embed queries/queries.sql +var nodeTestQueries string + +// Run against a migrated PostgreSQL database with BEACON_TEST_POSTGRES_DSN. +// All fixture writes are transaction-local temporary tables, rolled back on exit. +func TestListNodesPostgres(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(), 2*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()) + for _, table := range []string{"nodes", "node_iatas", "node_neighbors", "observers", "transport_scopes"} { + if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + } + _, err = tx.Exec(ctx, ` +INSERT INTO transport_scopes (id, name, transport_key, key_fingerprint) VALUES (1, '#test', decode(repeat('00', 16), 'hex'), decode(repeat('00', 8), 'hex')); +INSERT INTO nodes (id, public_key, node_type, name, last_seen, default_scope_id, supports_multibyte_paths, supports_multibyte_traces) +SELECT md5(i::text)::uuid, decode(lpad(to_hex(i), 64, '0'), 'hex'), (i % 4 + 1)::smallint, + 'node-' || i, '2026-01-01'::timestamptz - i * interval '1 second', + CASE WHEN i % 2 = 0 THEN 1 END, i % 2 = 0, i % 3 = 0 +FROM generate_series(1, 20000) i; +INSERT INTO node_iatas (node_id, iata, last_heard) +SELECT md5(i::text)::uuid, 'YVR', '2026-01-01'::timestamptz - i * interval '1 second' +FROM generate_series(1, 19999) i; +INSERT INTO node_iatas (node_id, iata, last_heard) +SELECT md5(i::text)::uuid, 'YYJ', '2026-01-01'::timestamptz - i * interval '1 second' - interval '1 hour' +FROM generate_series(2, 19998, 2) i; +INSERT INTO node_neighbors (node_id, neighbor_id, iata) +SELECT md5(i::text)::uuid, md5((i+1)::text)::uuid, iata +FROM generate_series(1, 19999) i CROSS JOIN (VALUES ('YVR'), ('YYJ')) regions(iata); +INSERT INTO observers (id, public_key) +SELECT id, public_key FROM nodes WHERE name = 'node-10'; +ANALYZE nodes; ANALYZE node_iatas; ANALYZE node_neighbors; ANALYZE observers; ANALYZE transport_scopes; +`) + if err != nil { + t.Fatal(err) + } + query := sqlc.New(tx) + base := sqlc.ListNodesParams{Column1: int16(0), Column3: "any", Column4: "any", Column6: "", Limit: 51} + for _, tc := range []struct { + name string + change func(*sqlc.ListNodesParams) + want []int + }{ + {"page", func(p *sqlc.ListNodesParams) { p.Limit = 3 }, []int{1, 2, 3}}, + {"type", func(p *sqlc.ListNodesParams) { p.Column1 = 2; p.Limit = 3 }, []int{1, 5, 9}}, + {"multi IATA dedup", func(p *sqlc.ListNodesParams) { p.Column2 = []string{"YVR", "YYJ"}; p.Limit = 3 }, []int{1, 2, 3}}, + {"IATA", func(p *sqlc.ListNodesParams) { p.Column2 = []string{"YYJ"}; p.Limit = 3 }, []int{2, 4, 6}}, + {"unknown IATA", func(p *sqlc.ListNodesParams) { p.Column2 = []string{"ZZZ"} }, nil}, + {"scope", func(p *sqlc.ListNodesParams) { p.Column9 = "#test"; p.Limit = 3 }, []int{2, 4, 6}}, + {"capabilities", func(p *sqlc.ListNodesParams) { p.Column3 = "true"; p.Column4 = "false"; p.Limit = 3 }, []int{2, 4, 8}}, + {"name", func(p *sqlc.ListNodesParams) { p.Column6 = "NODE-19"; p.Limit = 3 }, []int{19, 190, 191}}, + {"cursor", func(p *sqlc.ListNodesParams) { + p.Column7 = pgtype.Timestamptz{Time: time.Date(2026, 1, 1, 0, 0, -3, 0, time.UTC), Valid: true} + p.Limit = 3 + }, []int{4, 5, 6}}, + {"public key", func(p *sqlc.ListNodesParams) { p.Column5, _ = hex.DecodeString(fmt.Sprintf("%064x", 10)) }, []int{10}}, + {"public key prefix", func(p *sqlc.ListNodesParams) { p.Column11 = strings.Repeat("0", 61) + "00A" }, []int{10}}, + {"no IATA", func(p *sqlc.ListNodesParams) { p.Column5, _ = hex.DecodeString(fmt.Sprintf("%064x", 20000)) }, []int{20000}}, + } { + for _, neighbors := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/neighbors=%t", tc.name, neighbors), func(t *testing.T) { + params := base + tc.change(¶ms) + params.Column10 = neighbors + rows, err := query.ListNodes(ctx, params) + if err != nil { + t.Fatal(err) + } + if len(rows) != len(tc.want) { + t.Fatalf("got %d rows, want %d", len(rows), len(tc.want)) + } + for i, number := range tc.want { + row := rows[i] + if row.Name == nil || *row.Name != fmt.Sprintf("node-%d", number) { + t.Fatalf("wrong row at %d: %v", i, row.Name) + } + var iatas []struct { + IATA string `json:"iata"` + } + if len(row.Iatas) > 0 { + if err := json.Unmarshal(row.Iatas, &iatas); err != nil { + t.Fatal(err) + } + } + wantIATAs := 1 + if number%2 == 0 { + wantIATAs = 2 + } + wantNeighbors := int64(1) + if number == 20000 { + wantIATAs, wantNeighbors = 0, 0 + } + if len(iatas) != wantIATAs || (len(iatas) > 0 && iatas[0].IATA != "YVR") { + t.Errorf("wrong IATAs: %s", row.Iatas) + } + if row.KnownNeighborCount != wantNeighbors { + t.Errorf("neighbor count %d, want %d", row.KnownNeighborCount, wantNeighbors) + } + if neighbors && len(row.NeighborIds) != int(wantNeighbors) { + t.Errorf("neighbor IDs %v", row.NeighborIds) + } + if !neighbors && len(row.NeighborIds) != 0 { + t.Errorf("unrequested neighbor IDs %v", row.NeighborIds) + } + if row.IsObserver != (number == 10) { + t.Error("wrong observer flag") + } + if number == 10 && row.ObserverID != row.ID { + t.Error("wrong observer ID") + } + } + }) + } + } + + // A small page must not aggregate every node's IATA membership. Assert work, + // not elapsed time: shared CI and Pi hosts have variable scheduling latency. + queries := strings.ReplaceAll(nodeTestQueries, "\r\n", "\n") + sql := strings.Split(strings.Split(queries, "-- name: ListNodes :many\n")[1], "\n-- name:")[0] + var planJSON []byte + err = tx.QueryRow(ctx, "EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) "+sql, + int16(0), []string(nil), "any", "any", []byte(nil), "", nil, int32(51), "", false, "").Scan(&planJSON) + if err != nil { + t.Fatal(err) + } + var plans []struct { + Plan map[string]any + ExecutionTime float64 `json:"Execution Time"` + } + if err := json.Unmarshal(planJSON, &plans); err != nil { + t.Fatal(err) + } + var visited float64 + var walk func(map[string]any) + walk = func(plan map[string]any) { + if plan["Relation Name"] == "node_iatas" { + visited += plan["Actual Rows"].(float64) * plan["Actual Loops"].(float64) + } + if children, ok := plan["Plans"].([]any); ok { + for _, child := range children { + walk(child.(map[string]any)) + } + } + } + walk(plans[0].Plan) + t.Logf("20,000 nodes / 51-row page: execution %.3f ms; IATA rows processed %.0f", plans[0].ExecutionTime, visited) + if visited > 200 { + t.Errorf("small node page processed %.0f IATA rows; budget 200", visited) + } +} diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 1e54000..b6213db 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -659,20 +659,11 @@ WHERE id = ANY($1::uuid[]); SELECT id FROM nodes WHERE public_key = $1; -- name: ListNodes :many +-- Limit the filtered node page before enriching IATA membership and neighbours. +WITH page AS ( SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.last_seen, - n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, - ts.name AS default_scope_name, - json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FILTER (WHERE ni.iata IS NOT NULL) AS iatas, - EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, - (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id, - (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count, - -- CASE short-circuits: the array_agg subquery only runs when $10 is true, - -- so requests that don't ask for neighbor IDs don't pay for it. -(CASE WHEN $10::bool THEN - (SELECT COALESCE(array_agg(DISTINCT nn.neighbor_id), '{}'::uuid[]) FROM node_neighbors nn WHERE nn.node_id = n.id) - ELSE NULL END)::uuid[] AS neighbor_ids + n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, ts.name AS default_scope_name FROM nodes n -LEFT JOIN node_iatas ni ON ni.node_id = n.id LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE ($1 = 0 OR n.node_type = $1) @@ -692,9 +683,21 @@ WHERE AND ($7::timestamptz IS NULL OR n.last_seen < $7) AND ($9::text = '' OR ts.name = $9::text) AND ($11::text = '' OR encode(n.public_key, 'hex') ILIKE $11 || '%') -GROUP BY n.id, ts.name ORDER BY n.last_seen DESC -LIMIT $8; +LIMIT $8 +) +SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.last_seen, + n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, n.default_scope_name, + (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) + FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas, + EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, + (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id, + (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count, + (CASE WHEN $10::bool THEN + (SELECT COALESCE(array_agg(DISTINCT nn.neighbor_id), '{}'::uuid[]) FROM node_neighbors nn WHERE nn.node_id = n.id) + ELSE NULL END)::uuid[] AS neighbor_ids +FROM page n +ORDER BY n.last_seen DESC; -- name: ListNodeObservations :many SELECT po.id, encode(po.packet_hash, 'hex') AS packet_hash_hex, diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 9ee115a..9e2ab2b 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -127,6 +127,7 @@ type Querier interface { // Used for WS reconnect backfill. ListMessagesAfterID(ctx context.Context, arg ListMessagesAfterIDParams) ([]ListMessagesAfterIDRow, error) ListNodeObservations(ctx context.Context, arg ListNodeObservationsParams) ([]ListNodeObservationsRow, error) + // Limit the filtered node page before enriching IATA membership and neighbours. ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNodesRow, error) ListObservationsForPacket(ctx context.Context, packetHash []byte) ([]ListObservationsForPacketRow, error) // Returns advert packets (payload_type=4) heard by a specific observer. diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 1fe4a9c..670f7ad 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2302,20 +2302,10 @@ func (q *Queries) ListNodeObservations(ctx context.Context, arg ListNodeObservat } const listNodes = `-- name: ListNodes :many +WITH page AS ( SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.last_seen, - n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, - ts.name AS default_scope_name, - json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FILTER (WHERE ni.iata IS NOT NULL) AS iatas, - EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, - (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id, - (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count, - -- CASE short-circuits: the array_agg subquery only runs when $10 is true, - -- so requests that don't ask for neighbor IDs don't pay for it. -(CASE WHEN $10::bool THEN - (SELECT COALESCE(array_agg(DISTINCT nn.neighbor_id), '{}'::uuid[]) FROM node_neighbors nn WHERE nn.node_id = n.id) - ELSE NULL END)::uuid[] AS neighbor_ids + n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, ts.name AS default_scope_name FROM nodes n -LEFT JOIN node_iatas ni ON ni.node_id = n.id LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE ($1 = 0 OR n.node_type = $1) @@ -2335,9 +2325,21 @@ WHERE AND ($7::timestamptz IS NULL OR n.last_seen < $7) AND ($9::text = '' OR ts.name = $9::text) AND ($11::text = '' OR encode(n.public_key, 'hex') ILIKE $11 || '%') -GROUP BY n.id, ts.name ORDER BY n.last_seen DESC LIMIT $8 +) +SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.last_seen, + n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, n.default_scope_name, + (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) + FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas, + EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, + (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id, + (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count, + (CASE WHEN $10::bool THEN + (SELECT COALESCE(array_agg(DISTINCT nn.neighbor_id), '{}'::uuid[]) FROM node_neighbors nn WHERE nn.node_id = n.id) + ELSE NULL END)::uuid[] AS neighbor_ids +FROM page n +ORDER BY n.last_seen DESC ` type ListNodesParams struct { @@ -2373,6 +2375,7 @@ type ListNodesRow struct { NeighborIds []uuid.UUID `json:"neighbor_ids"` } +// Limit the filtered node page before enriching IATA membership and neighbours. func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNodesRow, error) { rows, err := q.db.Query(ctx, listNodes, arg.Column1,