dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-09-09 21:45:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(db): enrich nodes after selecting the directory page (#111)

Browse Source
This commit is contained in:
n30nex
2026-09-06 10:20:24 -04:00
committed by GitHub
parent b5ab592534
commit dfaa31d1dd
4 changed files with 219 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/nodes_integration_test.go
+185
View File
@@ -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(&params)
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)
}
}
db/queries/queries.sql
+17 -14
View File
@@ -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,
db/sqlc/querier.go
+1
View File
@@ -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.
db/sqlc/queries.sql.go
+16 -13
View File
@@ -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,