From be9f38252641d3bb7869ccd8393bad0f2c4b208b Mon Sep 17 00:00:00 2001 From: efiten Date: Sat, 10 Oct 2026 10:58:33 +0200 Subject: [PATCH] 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 --- cmd/server/paths_batch_test.go | 173 +++++++++++++++++++++++++++++++++ cmd/server/resolved_index.go | 85 ++++++++++++++++ cmd/server/routes.go | 64 ++++++------ 3 files changed, 288 insertions(+), 34 deletions(-) create mode 100644 cmd/server/paths_batch_test.go diff --git a/cmd/server/paths_batch_test.go b/cmd/server/paths_batch_test.go new file mode 100644 index 00000000..5009cbf8 --- /dev/null +++ b/cmd/server/paths_batch_test.go @@ -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)) + } +} diff --git a/cmd/server/resolved_index.go b/cmd/server/resolved_index.go index 7974bbec..5b0f05a7 100644 --- a/cmd/server/resolved_index.go +++ b/cmd/server/resolved_index.go @@ -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 { diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 27d18619..066404d6 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -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()