Files
meshcore-analyzer/cmd/server/paths_batch_test.go
T
efitenandClaude Opus 5.5 abbe19531d perf(paths): batch the resolved-path fallback for /nodes/{pk}/paths (#2223) (#2224)
Fixes #2223.

## Problem

`loadCanonicalResolvedPaths` reads the resolved path of each candidate's
longest observation in batches (#2146). A transmission whose longest
observation has no stored `resolved_path` then went through
`bestResolvedPath` on its own, with two queries:

- `fetchResolvedPathForObs` re-read that same observation, which the
batch had just read;
- `fetchResolvedPathsForTx` read all of the transmission's observations.

On a production database, 24% of the observations with a path have no
`resolved_path`, so most `/api/nodes/{pubkey}/paths` requests paid for
this once per such transmission.

## Change

- **`loadFallbackResolvedPaths` batches the fallback.** It reads those
transmissions' stored paths in one query per `resolvedPathBatchSize`
transmissions:

  ```sql
  SELECT id, transmission_id, resolved_path FROM observations
  WHERE resolved_path IS NOT NULL AND transmission_id IN (...)
  ```

On a production database the planner runs it as `SEARCH observations
USING INDEX idx_observations_tx_ts (transmission_id=?)`. It then picks
the observation `bestResolvedPath` would pick: the longest `path_json`
with a stored path, the first in snapshot order on a tie. The result
goes into the resolved-path LRU, as before.
- **`longestStoredResolvedPath`** holds that choice. `bestResolvedPath`
now uses it as well, so the rule exists in one place.
- **A failed batch still falls back.** If the first-stage batch fails,
it goes through `bestResolvedPath` per transmission, as before. A failed
fallback batch does the same for its transmissions.

The answer is the same as before, and no cache is added.

## Perf

Staging, 9 nodes, requests made from the host itself (`localhost:3001`),
so no proxy or TLS. Each node 3 times, after one warm-up round:

| | median | p90 | max | total for 27 requests |
|---|---|---|---|---|
| before | 0.174 s | 0.604 s | 0.866 s | 6.95 s |
| after | 0.079 s | 0.433 s | 0.484 s | 3.22 s |

CPU profile, 30 s each, under the same request loop:
- **Requests completed:** 180 before, 261 after.
- **CPU in `handleNodePaths` per request:** 154 ms before (27.77 s /
180), 107 ms after (27.90 s / 261).
- **Fallback cost:** `bestResolvedPath` took 13.56 s of
`loadCanonicalResolvedPaths` before. The batched
`loadFallbackResolvedPaths` takes 6.57 s.

Caveats:
- The before run was on a process that had been up longer than the after
run, so the two LRU states differ.
- One node (`060679ad…`, about 11k transmissions) did not change: 0.455
s against 0.451 s. Its time is elsewhere in the handler, and this PR
does not address it.

## Tests

- **`TestLoadCanonicalResolvedPathsBatchesFallback`:** 1,001
transmissions whose longest observation has no stored path.
  - On master: 2,002 per-transmission queries.
- Now: 0 per-transmission queries and 6 batched ones (3 for the longest
observations, 3 for the fallback), with every path taken from the
shorter observation.
- **Existing tests:**
`TestLoadCanonicalResolvedPathsMatchesBestResolvedPath` still passes. It
covers the longest observation without a path (#810), ties, nothing
stored, an empty stored path, and an observation absent from the
snapshot. The LRU and query-count tests also still pass.
- `go vet` and `go test .` in `cmd/server` pass.

No frontend change.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 11:58:28 +02:00

228 lines
8.1 KiB
Go

package main
import (
"database/sql"
"reflect"
"testing"
)
// #2146: /api/nodes/{pk}/paths looked up each candidate's canonical resolved
// path with bestResolvedPath, one or two queries per transmission. The batched
// loader must return the same path for every transmission.
func newPathsBatchStore(t *testing.T) (*PacketStore, map[int][]rpObs) {
t.Helper()
conn, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatal(err)
}
conn.SetMaxOpenConns(1)
t.Cleanup(func() { conn.Close() })
if _, err := conn.Exec(`CREATE TABLE observations (
id INTEGER PRIMARY KEY, transmission_id INTEGER, path_json TEXT, resolved_path TEXT)`); err != nil {
t.Fatal(err)
}
rows := []struct {
id, tx int
path, rp interface{}
}{
// tx 1 (#810): the longest path has no stored resolved_path.
{1, 1, `["a","b","c"]`, nil},
{2, 1, `["a","b"]`, `["aa11aa11","bb22bb22"]`},
// tx 2: equal lengths, the first in snapshot order wins.
{3, 2, `["a","b"]`, `["cc33cc33","dd44dd44"]`},
{4, 2, `["a","c"]`, `["aa11aa11","ee55ee55"]`},
// tx 3: nothing stored.
{5, 3, `["a"]`, nil},
// tx 4: an empty stored path counts as none.
{6, 4, `["a"]`, ``},
// tx 5: the match is case-insensitive.
{7, 5, `["a","b"]`, `["AA11AA11","ff66ff66"]`},
// tx 6: observation 9 is in the database but not in the store's
// snapshot; it still counts for the mention, not for the best path.
{8, 6, `["a","b"]`, `["ff66ff66","bb22bb22"]`},
{9, 6, `["a","b","c"]`, `["aa11aa11","x","y"]`},
// tx 7: the longest path wins over an earlier shorter one.
{10, 7, `["a"]`, `["bb22bb22"]`},
{11, 7, `["a","b","c"]`, `["cc33cc33","aa11aa11","dd44dd44"]`},
}
for _, r := range rows {
if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, path_json, resolved_path) VALUES (?,?,?,?)`,
r.id, r.tx, r.path, r.rp); err != nil {
t.Fatal(err)
}
}
snapshots := map[int][]rpObs{
1: {{1, `["a","b","c"]`}, {2, `["a","b"]`}},
2: {{3, `["a","b"]`}, {4, `["a","c"]`}},
3: {{5, `["a"]`}},
4: {{6, `["a"]`}},
5: {{7, `["a","b"]`}},
6: {{8, `["a","b"]`}},
7: {{10, `["a"]`}, {11, `["a","b","c"]`}},
}
return &PacketStore{db: &DB{conn: conn}}, snapshots
}
func TestLoadCanonicalResolvedPathsMatchesBestResolvedPath(t *testing.T) {
s, snapshots := newPathsBatchStore(t)
got := s.loadCanonicalResolvedPaths(snapshots)
for txID, snap := range snapshots {
want := s.bestResolvedPath(txID, snap)
if !reflect.DeepEqual(got[txID], want) {
t.Errorf("tx %d: batched %v, bestResolvedPath %v", txID, derefAll(got[txID]), derefAll(want))
}
if _, present := got[txID]; present != (want != nil) {
t.Errorf("tx %d: present in result = %v, want %v", txID, present, want != nil)
}
}
}
func derefAll(rp []*string) []string {
out := make([]string, len(rp))
for i, p := range rp {
if p != nil {
out[i] = *p
}
}
return out
}
func TestLoadCanonicalResolvedPathsBatchesQueries(t *testing.T) {
conn, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatal(err)
}
conn.SetMaxOpenConns(1)
defer conn.Close()
if _, err := conn.Exec(`CREATE TABLE observations (
id INTEGER PRIMARY KEY, transmission_id INTEGER, path_json TEXT, resolved_path TEXT)`); err != nil {
t.Fatal(err)
}
const n = 2*resolvedPathBatchSize + 1
snapshots := make(map[int][]rpObs, n)
for i := 1; i <= n; i++ {
conn.Exec(`INSERT INTO observations (id, transmission_id, path_json, resolved_path) VALUES (?,?,?,?)`,
i, i, `["a"]`, `["aa11aa11"]`)
snapshots[i] = []rpObs{{i, `["a"]`}}
}
s := &PacketStore{db: &DB{conn: conn}}
batched, single := 0, 0
s.cacheLoadHook = func(kind string) {
switch kind {
case "resolvedPathBatch":
batched++
case "resolvedPath":
single++
}
}
got := s.loadCanonicalResolvedPaths(snapshots)
if batched != 3 || single != 0 {
t.Errorf("want 3 batched and 0 single queries for %d transmissions, got %d and %d", n, batched, single)
}
if len(got) != n {
t.Errorf("want %d canonical paths, got %d", n, len(got))
}
}
// #2146: an index hit with a canonical path no longer gets the per-
// transmission SQL collision check; the aggregation decides from the
// canonical path, which is what it did after that check anyway. Here the
// longest observation (the canonical one) names A and a shorter one names B:
// both A and B are index hits, only A is on the canonical path.
func TestHandleNodePaths_CanonicalPathDecidesIndexHits(t *testing.T) {
sc := setupCollisionScenario(t, false)
mustExec(t, sc.db, `INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (43, 'BEEF', 'hash_2146', ?)`, sc.recent)
mustExec(t, sc.db, `INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, resolved_path)
VALUES (43, NULL, '["c0","aa"]', ?, ?)`, sc.recentEpoch, `["`+sc.nodeAPK+`","aaaa0000aaaa0000"]`)
mustExec(t, sc.db, `INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, resolved_path)
VALUES (43, NULL, '["c0"]', ?, ?)`, sc.recentEpoch-10, `["`+sc.nodeBPK+`"]`)
sc.reloadStore(t)
queries := 0
sc.srv.store.cacheLoadHook = func(kind string) {
if kind == "resolvedPath" {
queries++
}
}
if got := sc.query(t, sc.nodeAPK).TotalTransmissions; got != 1 {
t.Errorf("A is on the canonical path: want 1 transmission, got %d", got)
}
if got := sc.query(t, sc.nodeBPK).TotalTransmissions; got != 0 {
t.Errorf("B is only on a non-canonical observation: want 0 transmissions, got %d", got)
}
if queries != 0 {
t.Errorf("want no per-transmission resolved_path queries, got %d", queries)
}
}
// An observation already in the resolved-path LRU is served from it, as
// bestResolvedPath does, even when the stored path changed since.
func TestLoadCanonicalResolvedPathsUsesLRU(t *testing.T) {
s, snapshots := newPathsBatchStore(t)
s.apiResolvedPathLRU = make(map[int][]*string)
cached := "cached00cached00"
s.lruMu.Lock()
s.lruPut(11, []*string{&cached}) // longest observation of tx 7
s.lruMu.Unlock()
got := s.loadCanonicalResolvedPaths(map[int][]rpObs{7: snapshots[7]})
want := s.bestResolvedPath(7, snapshots[7])
if !reflect.DeepEqual(got[7], want) || len(want) != 1 || *want[0] != cached {
t.Fatalf("batched %v, bestResolvedPath %v, want the LRU entry", derefAll(got[7]), derefAll(want))
}
}
// A transmission whose longest observation has no stored resolved_path went
// through bestResolvedPath on its own: one query for that same observation,
// which the batch had just read, and one for all of the transmission's
// observations. 24% of the observations with a path had no resolved_path on a
// production database, so most /paths requests paid this per transmission.
// The fallback is batched too.
func TestLoadCanonicalResolvedPathsBatchesFallback(t *testing.T) {
conn, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatal(err)
}
conn.SetMaxOpenConns(1)
defer conn.Close()
if _, err := conn.Exec(`CREATE TABLE observations (
id INTEGER PRIMARY KEY, transmission_id INTEGER, path_json TEXT, resolved_path TEXT)`); err != nil {
t.Fatal(err)
}
const n = 2*resolvedPathBatchSize + 1
snapshots := make(map[int][]rpObs, n)
for tx := 1; tx <= n; tx++ {
longest, shorter := 2*tx-1, 2*tx
conn.Exec(`INSERT INTO observations (id, transmission_id, path_json, resolved_path) VALUES (?,?,?,NULL)`,
longest, tx, `["a","b"]`)
conn.Exec(`INSERT INTO observations (id, transmission_id, path_json, resolved_path) VALUES (?,?,?,?)`,
shorter, tx, `["a"]`, `["aa11aa11"]`)
snapshots[tx] = []rpObs{{longest, `["a","b"]`}, {shorter, `["a"]`}}
}
s := &PacketStore{db: &DB{conn: conn}}
batched, single := 0, 0
s.cacheLoadHook = func(kind string) {
switch kind {
case "resolvedPathBatch":
batched++
case "resolvedPath":
single++
}
}
got := s.loadCanonicalResolvedPaths(snapshots)
if single != 0 {
t.Errorf("want no per-transmission queries for %d fallback transmissions, got %d", n, single)
}
if batched != 6 {
t.Errorf("want 3 batched queries for the longest observations and 3 for the fallback, got %d", batched)
}
if len(got) != n {
t.Fatalf("want %d canonical paths, got %d", n, len(got))
}
for tx, rp := range got {
if len(rp) != 1 || rp[0] == nil || *rp[0] != "aa11aa11" {
t.Fatalf("tx %d: got %v, want the shorter observation's path", tx, derefAll(rp))
}
}
}