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()