perf(store): move runtime SQL outside packet-store locks

cmd/server/chunked_load.go

@@ -412,9 +412,8 @@ func (s *PacketStore) LoadChunked(chunkSize int) error {
 	// PR #1643 R1 munger #1: cold load uses unique_prefix-only gate, so
 	// the neighbor graph is no longer consulted here (affinity-tier
 	// resolution against ≤168h-old observations would silently mis-attribute).
-	s.mu.RLock()
-	_, relayPM := s.getCachedNodesAndPM()
-	s.mu.RUnlock()
+	nodeSnapshot := s.snapshotPathNodes()
+	relayPM := nodeSnapshot.pm
 	var coldLoadAmbiguousHopsSkipped int
 
 	for {
@@ -516,7 +515,7 @@ func (s *PacketStore) LoadChunked(chunkSize int) error {
 	})
 	s.buildSubpathIndex()
 	s.buildPathHopIndex()
-	s.buildDistanceIndex()
+	s.buildDistanceIndex(nodeSnapshot)
 	if s.hotStartupHours > 0 {
 		s.oldestLoaded = hotCutoffStr
 	} else if len(s.packets) > 0 {

cmd/server/clock_skew.go

@@ -827,11 +827,10 @@ func (s *PacketStore) computeFleetClockSkewForArea(area string) []*NodeClockSkew
 		areaNodes = s.resolveAreaNodes(area)
 	}
 
+	// Refresh node metadata before taking the store lock.
+	allNodes, _ := s.getCachedNodesAndPM()
 	s.mu.RLock()
 	defer s.mu.RUnlock()
-
-	// Build name/role lookup from DB cache (requires s.mu held).
-	allNodes, _ := s.getCachedNodesAndPM()
 	nameMap := make(map[string]nodeInfo, len(allNodes))
 	for _, ni := range allNodes {
 		nameMap[ni.PublicKey] = ni

cmd/server/coverage_test.go

@@ -1843,7 +1843,7 @@ func TestEnrichObs(t *testing.T) {
 		t.Skip("no observations loaded")
 	}
 
-	enriched := store.enrichObs(obs)
+	enriched := store.GetPacketByID(obs.ID)
 	if enriched["observer_id"] == nil {
 		t.Error("expected observer_id")
 	}

cmd/server/db_test.go

@@ -3252,7 +3252,7 @@ func TestPerObservationRawHexEnrich(t *testing.T) {
 
 	// Check enriched observations
 	for _, obs := range tx.Observations {
-		m := store.enrichObs(obs)
+		m := store.GetPacketByID(obs.ID)
 		rh, _ := m["raw_hex"].(string)
 		if obs.RawHex != "" {
 			// Observer A: should get per-observation raw_hex

cmd/server/hash_migrate.go

@@ -132,9 +132,10 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration
 			store.WaitIndexesReady(30 * time.Second)
 			indexesReady = true
 		}
+		metadata := store.snapshotPathNodes()
 		store.mu.Lock()
 		held := time.Now()
-		r, m := store.applyContentHashUpdates(updates)
+		r, m := store.applyContentHashUpdates(updates, metadata)
 		hold.record(time.Since(held))
 		store.mu.Unlock()
 		rehashed += r
@@ -201,7 +202,7 @@ func staleContentHashes(s *PacketStore, batch []*StoreTx) []hashUpdate {
 // applyContentHashUpdates rehashes the transmissions of updates and merges the
 // ones that collide. Returns how many it rehashed and how many duplicates it
 // merged away. Must hold s.mu for writing.
-func (s *PacketStore) applyContentHashUpdates(updates []hashUpdate) (rehashed, merged int) {
+func (s *PacketStore) applyContentHashUpdates(updates []hashUpdate, snapshots ...pathNodeSnapshot) (rehashed, merged int) {
 	// The batch was chosen under a read lock: drop what changed since
 	// (eviction, a concurrent rehash).
 	live := updates[:0]
@@ -243,7 +244,7 @@ func (s *PacketStore) applyContentHashUpdates(updates []hashUpdate) (rehashed, m
 		s.moveObservations(winner, loser, m)
 		m.add(winner, loser)
 	}
-	merged = s.finishHashMerge(m)
+	merged = s.finishHashMerge(m, s.pathNodeSnapshotLocked(snapshots))
 	return len(live), merged
 }
 
@@ -433,7 +434,7 @@ func (m *hashMerge) add(winner, loser *StoreTx) {
 // finishHashMerge removes every merged-away transmission from every index and
 // from trackedBytes, then refreshes the survivors (best observation, charge,
 // path-derived indexes). Returns the number of duplicates removed.
-func (s *PacketStore) finishHashMerge(m *hashMerge) int {
+func (s *PacketStore) finishHashMerge(m *hashMerge, metadata pathNodeSnapshot) int {
 	if len(m.losers) == 0 {
 		return 0
 	}
@@ -518,7 +519,7 @@ func (s *PacketStore) finishHashMerge(m *hashMerge) int {
 		}
 	}
 	if len(distChanged) > 0 {
-		s.updateDistanceIndexForTxs(distChanged)
+		s.updateDistanceIndexForTxs(distChanged, metadata)
 	}
 
 	s.invalidateRelayStatsCache()

cmd/server/node_activity.go

@@ -22,17 +22,6 @@ type NodeHealthStats struct {
 	LastAdvert         *string  `json:"lastAdvert"`
 }
 
-// relayPrefixMapLocked reuses the existing bounded node cache. Missing node
-// metadata fails closed for short hashes (e.g. during startup/schema failure).
-// Caller holds s.mu. There is no per-node or per-packet SQL query here.
-func (s *PacketStore) relayPrefixMapLocked() *prefixMap {
-	if s.db == nil || s.db.conn == nil {
-		return s.nodePM
-	}
-	_, pm := s.getCachedNodesAndPM()
-	return pm
-}
-
 // confirmedRelayKey deliberately does not use graph/geo/fallback guesses.
 // A full raw identity or a unique wire prefix is evidence; a colliding prefix
 // remains unknown even if a heuristic previously indexed it under a full key.

cmd/server/repeater_enrich_bulk.go

@@ -65,12 +65,13 @@ func (s *PacketStore) GetRepeaterRelayInfoMap(windowHours float64) map[string]Re
 // avoiding collided-prefix fanout. Memory: O(unique tx + keys + distinct hop
 // tokens), bounded by store eviction.
 func (s *PacketStore) computeRepeaterRelayInfoMap(windowHours float64) map[string]RepeaterRelayInfo {
+	metadata := s.snapshotPathNodes()
 	// Everything that reads store data happens under the read lock: ingest
 	// appends to and eviction compacts byPathHop/byNode slices in place, so a
 	// copied slice header is not a stable snapshot, and StoreTx observations
 	// are mutable. The unlocked phase reads only values owned by this call.
 	s.mu.RLock()
-	pm := s.relayPrefixMapLocked()
+	pm := metadata.pm
 
 	// Per transmission: its relay entry values and the identity-safe relay
 	// keys confirmed by its raw observed flood paths, computed once.

cmd/server/repeater_liveness.go

@@ -151,8 +151,8 @@ func newRelayEntry(tx *StoreTx, fromPrefix bool) relayEntry {
 // Candidate membership is never proof: persisted/live resolution can include
 // heuristic guesses. Every candidate is verified against its raw observation
 // paths, without target-biased resolve. Each transmission is checked once.
-func (s *PacketStore) collectRelayEntriesLocked(key string) []relayEntry {
-	m := newRelayKeyMatcher(key, s.relayPrefixMapLocked())
+func (s *PacketStore) collectRelayEntriesLocked(key string, snapshots ...pathNodeSnapshot) []relayEntry {
+	m := newRelayKeyMatcher(key, s.pathNodeSnapshotLocked(snapshots).pm)
 	var entries []relayEntry
 	if !m.possible() {
 		return entries
@@ -309,9 +309,10 @@ func (s *PacketStore) GetRepeaterRelayInfo(pubkey string, windowHours float64) R
 		return RepeaterRelayInfo{WindowHours: windowHours}
 	}
 	key := strings.ToLower(pubkey)
+	metadata := s.snapshotPathNodes()
 
 	s.mu.RLock()
-	entries := s.collectRelayEntriesLocked(key)
+	entries := s.collectRelayEntriesLocked(key, metadata)
 	s.mu.RUnlock()
 
 	return computeRelayInfoFromEntries(entries, windowHours)

cmd/server/repeater_usefulness.go

@@ -125,6 +125,7 @@ func (s *PacketStore) GetRepeaterNodeStatsBatch(pubkeys []string, windowHours fl
 	if len(pubkeys) == 0 {
 		return result
 	}
+	metadata := s.snapshotPathNodes()
 
 	type nodeSnap struct {
 		entries []relayEntry
@@ -144,7 +145,7 @@ func (s *PacketStore) GetRepeaterNodeStatsBatch(pubkeys []string, windowHours fl
 	var ids []int
 	for _, pk := range pubkeys {
 		key := strings.ToLower(pk)
-		entries := s.collectRelayEntriesLocked(key)
+		entries := s.collectRelayEntriesLocked(key, metadata)
 		var relayed int
 		relayed, ids = countDistinctNonAdvert(s.byPathHop[key], ids)
 		snaps[pk] = nodeSnap{entries: entries, relayed: relayed}

cmd/server/resolved_index.go

@@ -19,6 +19,72 @@ import (
 	"time"
 )
 
+type resolvedPathObservation struct {
+	id       int
+	pathJSON string
+}
+
+// A job owns immutable lookup inputs and a response setter, not a StoreTx or
+// a shared observation slice. Setters only mutate this request's response.
+type resolvedPathJob struct {
+	forTx        bool
+	txID         int
+	observations []resolvedPathObservation
+	obsID        int
+	set          func([]*string)
+}
+
+func (s *PacketStore) applyResolvedPathJobs(jobs []resolvedPathJob) {
+	for _, job := range jobs {
+		var rp []*string
+		if job.forTx {
+			rp = s.resolvedPathForSnapshot(job.txID, job.observations)
+		} else {
+			rp = s.fetchResolvedPathForObs(job.obsID)
+		}
+		if rp != nil {
+			job.set(rp)
+		}
+	}
+}
+
+// Same longest-present-path choice and LRU behavior as the live-store helper,
+// but all inputs were copied under s.mu before SQL is attempted.
+func (s *PacketStore) resolvedPathForSnapshot(txID int, observations []resolvedPathObservation) []*string {
+	return s.bestResolvedPathAt(txID, len(observations), func(i int) resolvedPathObservation { return observations[i] }, s.lruNow())
+}
+
+func (s *PacketStore) bestResolvedPathAt(txID, count int, at func(int) resolvedPathObservation, now int64) []*string {
+	if count == 0 {
+		return nil
+	}
+	longest := at(0)
+	for i := 1; i < count; i++ {
+		obs := at(i)
+		if pathLen(obs.pathJSON) > pathLen(longest.pathJSON) {
+			longest = obs
+		}
+	}
+	if rp := s.fetchResolvedPathForObsAt(longest.id, now); rp != nil {
+		return rp
+	}
+	paths := s.fetchResolvedPathsForTx(txID)
+	var best []*string
+	bestLen, bestID := -1, 0
+	for i := 0; i < count; i++ {
+		obs := at(i)
+		if rp := paths[obs.id]; rp != nil && pathLen(obs.pathJSON) > bestLen {
+			best, bestLen, bestID = rp, pathLen(obs.pathJSON), obs.id
+		}
+	}
+	if bestID != 0 {
+		s.lruMu.Lock()
+		s.lruPut(bestID, best)
+		s.lruMu.Unlock()
+	}
+	return best
+}
+
 // resolvedPubkeyHash computes a fast 64-bit hash for membership index keying.
 // Uses FNV-1a from stdlib — good distribution, no external dependency.
 func resolvedPubkeyHash(pk string) uint64 {
@@ -321,50 +387,9 @@ func (s *PacketStore) fetchResolvedPathForTxBest(tx *StoreTx) []*string {
 // tx.Observations under that lock (#470). Ingest only appends, so the
 // elements of such a snapshot never change.
 func (s *PacketStore) fetchResolvedPathForTxBestAt(txID int, observations []*StoreObs, now int64) []*string {
-	if len(observations) == 0 {
-		return nil
-	}
-	// Fast path: try the longest-path obs first via the LRU/SQL helper.
-	longest := observations[0]
-	longestLen := pathLen(longest.PathJSON)
-	for _, obs := range observations[1:] {
-		if l := pathLen(obs.PathJSON); l > longestLen {
-			longest = obs
-			longestLen = l
-		}
-	}
-	if rp := s.fetchResolvedPathForObsAt(longest.ID, now); rp != nil {
-		return rp
-	}
-	// Fallback: longest-path obs has no stored resolved_path. Query all
-	// observations for this tx and pick the one with the longest path_json
-	// that actually has a stored resolved_path.
-	rpMap := s.fetchResolvedPathsForTx(txID)
-	if len(rpMap) == 0 {
-		return nil
-	}
-	var bestRP []*string
-	bestObsID := 0
-	bestLen := -1
-	for _, obs := range observations {
-		rp, ok := rpMap[obs.ID]
-		if !ok || rp == nil {
-			continue
-		}
-		if l := pathLen(obs.PathJSON); l > bestLen {
-			bestLen = l
-			bestRP = rp
-			bestObsID = obs.ID
-		}
-	}
-	// Populate LRU so repeat lookups for this tx don't re-issue the multi-row
-	// SQL fallback (e.g. dashboard polling /api/nodes/{pk}/health).
-	if bestRP != nil && bestObsID != 0 {
-		s.lruMu.Lock()
-		s.lruPut(bestObsID, bestRP)
-		s.lruMu.Unlock()
-	}
-	return bestRP
+	return s.bestResolvedPathAt(txID, len(observations), func(i int) resolvedPathObservation {
+		return resolvedPathObservation{id: observations[i].ID, pathJSON: observations[i].PathJSON}
+	}, now)
 }
 
 // --- Simple LRU cache for resolved paths ---

cmd/server/routes.go

@@ -2344,8 +2344,8 @@ func (s *Server) handleNodePaths(w http.ResponseWriter, r *http.Request) {
 		prefix1 = prefix1[:2]
 	}
 
-	s.store.mu.RLock()
 	_, pm := s.store.getCachedNodesAndPM()
+	s.store.mu.RLock()
 
 	// Collect candidate transmissions from the index, deduplicating by tx ID.
 	// confirmedByFullKey tracks TXs found via the full-pubkey index key — these are
@@ -3284,9 +3284,7 @@ func (s *Server) handleResolveHops(w http.ResponseWriter, r *http.Request) {
 	// Get the server's prefix map for resolveWithContext.
 	var pm *prefixMap
 	if s.store != nil {
-		s.store.mu.RLock()
 		_, pm = s.store.getCachedNodesAndPM()
-		s.store.mu.RUnlock()
 	}
 
 	for _, hop := range hops {