mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-23 11:52:25 +00:00
perf(server): /paths reads canonical paths in batches and skips redundant collision checks (#2157)
Refs #2146 (item: `/api/nodes/:pubkey/paths`). ## Situation Per-step timing in a staging build (3 most-connected nodes, 2 calls each) showed where a `/paths` request spends its time: | Step | Time | |---|---| | build candidates under the read lock | 73-115 ms | | collision check: one `INSTR(LOWER(resolved_path), ?)` query per index hit, 13,367-16,790 per call | 1.3-2.7 s (first call after start: 13.5 s) | | canonical resolved_path per survivor (`bestResolvedPath`, one or two queries each) | 1.7-2.9 s | | aggregation under the read lock | 82-120 ms | So 3.2-4.1 s of a 3.4-5.1 s request was SQL. On a production database (busiest node, 27k transmissions, read-only): the collision check costs 1.85 s one by one and still 1.69 s batched, because SQLite reads and lowercases every stored path either way. Reading the canonical path (the longest observation's stored path) costs 0.18 s one by one and 0.09 s batched by id. ## My changes in this PR - **The collision check runs only where it decides something.** The aggregation keeps a transmission with a canonical path only if that path names the node (`if !containsTarget { continue }`). If no stored path names the node, the canonical one does not either. So for index hits with a canonical path the check changes nothing, and it now runs only for index hits without one, as before. - **Canonical paths are read in batches.** `loadCanonicalResolvedPaths` reads the stored path of each candidate's longest observation, one query per 500 transmissions. When that path is missing, empty or unparseable it calls `bestResolvedPath` for that transmission, so the result is the same as before. - The snapshot of observation IDs and `path_json` strings is taken in the existing candidate read lock (string copies, no parsing). The PR has two commits: a first version that batched both queries and did not get faster on staging, and the change described above, whose message has the measurements. Squash-merging is fine. ## Measurements Staging, after the ingestor's startup `ANALYZE`, same probe for both builds (ms): | | without this PR | with this PR | |---|---|---| | node 1 (degree 55), call 1 / call 2 | 22,237 / 3,364 | 1,793 / 1,686 | | node 2 (degree 47) | 4,427 / 3,615 | 1,828 / 1,809 | | node 3 (degree 46) | 5,171 / 4,035 | 3,050 / 2,408 | | 4 concurrent clients over 40 nodes, 120 s: calls / p50 / p95 | 73 / 6,717 / 15,160 | 125 / 2,861 / 10,630 | Both staging builds also contain #2147 and #2155. ## Tests - `TestLoadCanonicalResolvedPathsMatchesBestResolvedPath` compares the batched result with `bestResolvedPath` on seven cases: the #810 case (longest observation without a stored path), equal lengths, an empty stored path, case, an observation missing from the snapshot, longest wins, nothing stored. - `TestLoadCanonicalResolvedPathsBatchesQueries`: 1,001 transmissions take 3 batched queries and no per-transmission lookups. - `TestHandleNodePaths_CanonicalPathDecidesIndexHits`: the longest observation names A, a shorter one names B; both are index hits. A gets the transmission, B does not, with no per-transmission query. Run against master, the A/B results are the same and only the query-count check fails (1 there, 0 here). - The existing `/paths` tests pass unchanged (#810, #1278, #1352 collision and fallback cases, anchor bias). - `cmd/server`: `go test ./...` and `go test -race` green. ## Not in this PR - `nodeInResolvedPathViaIndex` still calls the per-transmission check; it has no production caller, only tests. - The concurrent p95 is still 10.6 s on staging. What is left there is the 4-connection read pool shared with other endpoints; a staging A/B with 8 connections gave no gain (`/paths` throughput dropped from 133-146 to 81-103 calls per 120 s), so the pool stays at 4. --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
efitenandClaude Opus 5.5 authored and GitHub committed
1 parent
3e21b179ec
commit
be9f382526
3 files changed
+288
-34
No files matched your search
@@ -0,0 +1,173 @@
|
||||
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))
|
||||
}
|
||||
}
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"database/sql"
|
||||
"hash/fnv"
|
||||
"log"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -167,6 +168,90 @@ func (s *PacketStore) confirmResolvedPathContains(txID int, pubkey string) bool
|
||||
return count > 0
|
||||
}
|
||||
|
||||
// resolvedPathBatchSize bounds the IN list of loadCanonicalResolvedPaths.
|
||||
const resolvedPathBatchSize = 500
|
||||
|
||||
// loadCanonicalResolvedPaths returns, per transmission in snapshots, the
|
||||
// resolved path bestResolvedPath returns, without bestResolvedPath's one or
|
||||
// two queries per transmission (#2146). It takes the path of each
|
||||
// transmission's longest observation from the resolved-path LRU, as
|
||||
// bestResolvedPath does, and reads the rest in one query per
|
||||
// resolvedPathBatchSize transmissions. A transmission whose longest observation has no usable
|
||||
// stored path, or whose batch failed, goes through bestResolvedPath itself,
|
||||
// so the answer is the same. Transmissions with no stored path are absent.
|
||||
// Must be called without s.mu held.
|
||||
func (s *PacketStore) loadCanonicalResolvedPaths(snapshots map[int][]rpObs) map[int][]*string {
|
||||
out := make(map[int][]*string, len(snapshots))
|
||||
longestTx := make(map[int]int, len(snapshots)) // longest observation ID → transmission ID
|
||||
var fallback []int
|
||||
for txID, observations := range snapshots {
|
||||
if len(observations) == 0 {
|
||||
continue
|
||||
}
|
||||
longest, longestLen := observations[0].id, pathLen(observations[0].pathJSON)
|
||||
for _, o := range observations[1:] {
|
||||
if l := pathLen(o.pathJSON); l > longestLen {
|
||||
longest, longestLen = o.id, l
|
||||
}
|
||||
}
|
||||
longestTx[longest] = txID
|
||||
}
|
||||
if s.db == nil || s.db.conn == nil {
|
||||
return out
|
||||
}
|
||||
// The LRU first, as bestResolvedPath does, so /paths shows the same path
|
||||
// as the packets page for an observation that page already cached.
|
||||
ids := make([]int, 0, len(longestTx))
|
||||
s.lruMu.RLock()
|
||||
for id, txID := range longestTx {
|
||||
if rp, ok := s.apiResolvedPathLRU[id]; ok && rp != nil {
|
||||
out[txID] = rp
|
||||
continue
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
s.lruMu.RUnlock()
|
||||
sort.Ints(ids)
|
||||
for start := 0; start < len(ids); start += resolvedPathBatchSize {
|
||||
chunk := ids[start:min(start+resolvedPathBatchSize, len(ids))]
|
||||
args := make([]interface{}, len(chunk))
|
||||
for i, id := range chunk {
|
||||
args[i] = id
|
||||
}
|
||||
s.beforeCacheLoad("resolvedPathBatch")
|
||||
rows, err := s.db.conn.Query(`SELECT id, resolved_path FROM observations WHERE id IN (?`+
|
||||
strings.Repeat(",?", len(chunk)-1)+`)`, args...)
|
||||
if err != nil {
|
||||
for _, id := range chunk {
|
||||
fallback = append(fallback, longestTx[id])
|
||||
}
|
||||
continue
|
||||
}
|
||||
for rows.Next() {
|
||||
var obsID int
|
||||
var raw sql.NullString
|
||||
if rows.Scan(&obsID, &raw) != nil || !raw.Valid {
|
||||
continue
|
||||
}
|
||||
if rp := unmarshalResolvedPath(raw.String); rp != nil {
|
||||
out[longestTx[obsID]] = rp
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
for _, id := range chunk {
|
||||
if _, ok := out[longestTx[id]]; !ok {
|
||||
fallback = append(fallback, longestTx[id])
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, txID := range fallback {
|
||||
if rp := s.bestResolvedPath(txID, snapshots[txID]); rp != nil {
|
||||
out[txID] = rp
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// fetchResolvedPathsForTx fetches resolved_path from SQLite for all observations
|
||||
// of a transmission. Used for on-demand API responses and eviction cleanup.
|
||||
func (s *PacketStore) fetchResolvedPathsForTx(txID int) map[int][]*string {
|
||||
|
||||
+30
-34
@@ -1905,13 +1905,17 @@ func (s *Server) handleNodePaths(w http.ResponseWriter, r *http.Request) {
|
||||
// We lean on resolved_path (from neighbor affinity graph) to disambiguate.
|
||||
//
|
||||
// Collect candidate IDs and index membership under the read lock, then release
|
||||
// the lock before running SQL queries (confirmResolvedPathContains does disk I/O).
|
||||
// the lock before reading the stored resolved paths (disk I/O).
|
||||
type candidateCheck struct {
|
||||
tx *StoreTx
|
||||
hasReverse bool
|
||||
inIndex bool
|
||||
}
|
||||
checks := make([]candidateCheck, len(candidates))
|
||||
// The stored resolved paths are read after s.mu is released, so they get
|
||||
// a snapshot of each candidate's observations instead of tx.Observations,
|
||||
// which ingest appends to under the write lock.
|
||||
rpSnapshots := make(map[int][]rpObs, len(candidates))
|
||||
for i, tx := range candidates {
|
||||
cc := candidateCheck{tx: tx}
|
||||
if !s.store.useResolvedPathIndex {
|
||||
@@ -1929,39 +1933,14 @@ func (s *Server) handleNodePaths(w http.ResponseWriter, r *http.Request) {
|
||||
// If not in index at all, it's a definite no
|
||||
}
|
||||
checks[i] = cc
|
||||
}
|
||||
s.store.mu.RUnlock()
|
||||
|
||||
// Now run SQL checks outside the lock for candidates that need confirmation.
|
||||
confirmedBySQL := make(map[int]bool)
|
||||
filtered := candidates[:0]
|
||||
for _, cc := range checks {
|
||||
if cc.inIndex {
|
||||
filtered = append(filtered, cc.tx)
|
||||
} else if cc.hasReverse {
|
||||
if s.store.confirmResolvedPathContains(cc.tx.ID, lowerPK) {
|
||||
filtered = append(filtered, cc.tx)
|
||||
confirmedBySQL[cc.tx.ID] = true
|
||||
}
|
||||
if cc.inIndex || cc.hasReverse {
|
||||
rpSnapshots[tx.ID] = snapshotRPObs(tx)
|
||||
}
|
||||
// else: not in index → exclude
|
||||
}
|
||||
candidates = filtered
|
||||
|
||||
// The canonical resolved_path lookup below runs without s.mu, so it reads
|
||||
// a snapshot of each surviving candidate's observations instead of
|
||||
// tx.Observations, which ingest appends to under the write lock. Taken in
|
||||
// a short read lock of its own so only the survivors are copied.
|
||||
s.store.mu.RLock()
|
||||
rpSnapshots := make(map[int][]rpObs, len(candidates))
|
||||
for _, tx := range candidates {
|
||||
rpSnapshots[tx.ID] = snapshotRPObs(tx)
|
||||
}
|
||||
s.store.mu.RUnlock()
|
||||
|
||||
// #1278: Read the CANONICAL persisted resolved_path for each surviving
|
||||
// candidate OUTSIDE s.mu (bestResolvedPath takes lruMu; the
|
||||
// lock-ordering contract forbids acquiring lruMu under s.mu).
|
||||
// #1278: the CANONICAL persisted resolved_path of each candidate, the one
|
||||
// bestResolvedPath returns, read in batches (#2146).
|
||||
//
|
||||
// Option A from the issue: the packets page renders each tx via
|
||||
// bestResolvedPath. For /api/nodes/{pk}/paths to stay
|
||||
@@ -1973,12 +1952,29 @@ func (s *Server) handleNodePaths(w http.ResponseWriter, r *http.Request) {
|
||||
// Falls back to biased re-resolve only when a tx has no persisted
|
||||
// resolved_path (older data / async backfill incomplete); in that case
|
||||
// there's no canonical answer to be consistent with.
|
||||
canonicalRP := make(map[int][]*string, len(candidates))
|
||||
for _, tx := range candidates {
|
||||
if rp := s.store.bestResolvedPath(tx.ID, rpSnapshots[tx.ID]); rp != nil {
|
||||
canonicalRP[tx.ID] = rp
|
||||
canonicalRP := s.store.loadCanonicalResolvedPaths(rpSnapshots)
|
||||
|
||||
// Collision check for candidates found through the resolved-pubkey hash
|
||||
// index (hasReverse). A candidate with a canonical path needs none: the
|
||||
// aggregation below keeps it only if that path names the node, and if no
|
||||
// stored path names it the canonical one does not either. Only one without
|
||||
// a canonical path is checked against SQL, as before.
|
||||
confirmedBySQL := make(map[int]bool)
|
||||
filtered := candidates[:0]
|
||||
for _, cc := range checks {
|
||||
if cc.inIndex {
|
||||
filtered = append(filtered, cc.tx)
|
||||
} else if cc.hasReverse {
|
||||
if _, ok := canonicalRP[cc.tx.ID]; ok {
|
||||
filtered = append(filtered, cc.tx)
|
||||
} else if s.store.confirmResolvedPathContains(cc.tx.ID, lowerPK) {
|
||||
filtered = append(filtered, cc.tx)
|
||||
confirmedBySQL[cc.tx.ID] = true
|
||||
}
|
||||
}
|
||||
// else: not in index → exclude
|
||||
}
|
||||
candidates = filtered
|
||||
|
||||
// Re-acquire read lock for the aggregation phase that reads store data.
|
||||
s.store.mu.RLock()
|
||||
|
||||
Reference in new issue
Block a user