From 8ba8a3ac5c908a84c59b7a4f435d5650326e35d9 Mon Sep 17 00:00:00 2001 From: dborup Date: Fri, 16 Oct 2026 11:31:44 +0200 Subject: [PATCH] perf(store): move runtime SQL outside packet-store locks --- cmd/server/chunked_load.go | 7 +- cmd/server/clock_skew.go | 5 +- cmd/server/coverage_test.go | 2 +- cmd/server/db_test.go | 2 +- cmd/server/hash_migrate.go | 11 +- cmd/server/node_activity.go | 11 - cmd/server/repeater_enrich_bulk.go | 3 +- cmd/server/repeater_liveness.go | 7 +- cmd/server/repeater_usefulness.go | 3 +- cmd/server/resolved_index.go | 113 +++-- cmd/server/routes.go | 4 +- cmd/server/store.go | 430 +++++++++++------- .../store_lock_sql_invariant_412_test.go | 377 +++++++++++++++ cmd/server/store_memory_diagnostics_test.go | 4 +- cmd/server/store_sql_unlocked_412_test.go | 335 ++++++++++++++ 15 files changed, 1070 insertions(+), 244 deletions(-) create mode 100644 cmd/server/store_lock_sql_invariant_412_test.go create mode 100644 cmd/server/store_sql_unlocked_412_test.go diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index dd626716a..aa732d97f 100644 --- a/cmd/server/chunked_load.go +++ b/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 { diff --git a/cmd/server/clock_skew.go b/cmd/server/clock_skew.go index 1eb5fbb4c..bdea65f91 100644 --- a/cmd/server/clock_skew.go +++ b/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 diff --git a/cmd/server/coverage_test.go b/cmd/server/coverage_test.go index eae9e07fa..18a48563d 100644 --- a/cmd/server/coverage_test.go +++ b/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") } diff --git a/cmd/server/db_test.go b/cmd/server/db_test.go index b592a1d0d..79310972b 100644 --- a/cmd/server/db_test.go +++ b/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 diff --git a/cmd/server/hash_migrate.go b/cmd/server/hash_migrate.go index 6b573ee8d..6309bb444 100644 --- a/cmd/server/hash_migrate.go +++ b/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() diff --git a/cmd/server/node_activity.go b/cmd/server/node_activity.go index fa6fc14e6..66e3c8250 100644 --- a/cmd/server/node_activity.go +++ b/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. diff --git a/cmd/server/repeater_enrich_bulk.go b/cmd/server/repeater_enrich_bulk.go index bfca1afd6..d8b54f307 100644 --- a/cmd/server/repeater_enrich_bulk.go +++ b/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. diff --git a/cmd/server/repeater_liveness.go b/cmd/server/repeater_liveness.go index ee8f187fb..c90676503 100644 --- a/cmd/server/repeater_liveness.go +++ b/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) diff --git a/cmd/server/repeater_usefulness.go b/cmd/server/repeater_usefulness.go index 76324c45d..2060d1dcc 100644 --- a/cmd/server/repeater_usefulness.go +++ b/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} diff --git a/cmd/server/resolved_index.go b/cmd/server/resolved_index.go index 2f3dca209..b603016bb 100644 --- a/cmd/server/resolved_index.go +++ b/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 --- diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 66833f6c3..c4d115fbb 100644 --- a/cmd/server/routes.go +++ b/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 { diff --git a/cmd/server/store.go b/cmd/server/store.go index 37ba45e29..14bc5cd77 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -303,6 +303,8 @@ type PacketStore struct { nodeCache []nodeInfo nodePM *prefixMap nodeCacheTime time.Time + nodeCacheGen uint64 // cacheMu; invalidation must win over an in-flight SQL read + nodeCacheLoad singleflight.Group // Per-store dedupe set for one-shot schema-degradation warnings. Field // (not package-level) so each test gets a fresh state — see #1199 item 5. schemaDegradationLogged sync.Map @@ -1310,9 +1312,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { // cursor below: getCachedNodesAndPM issues its own DB query, which // would deadlock against the open cursor on a single-connection SQLite // pool. - s.mu.RLock() _, relayPM := s.getCachedNodesAndPM() - s.mu.RUnlock() var coldLoadAmbiguousHopsSkipped int rows, err := s.db.conn.Query(chunkSQL, qArgs...) @@ -2311,8 +2311,10 @@ func (s *PacketStore) QueryPackets(q PacketQuery) *PacketResult { } } atomic.AddInt64(&s.queryCount, 1) + filters := s.preparePacketFilters(q) s.mu.RLock() - defer s.mu.RUnlock() + var jobs []resolvedPathJob + defer func() { s.mu.RUnlock(); s.applyResolvedPathJobs(jobs) }() if q.Limit <= 0 { q.Limit = 50 @@ -2321,7 +2323,7 @@ func (s *PacketStore) QueryPackets(q PacketQuery) *PacketResult { q.Order = "DESC" } - results := s.filterPackets(q) + results := s.filterPacketsPrepared(q, filters) total := len(results) // #1345: order by ingest id, not insertion-into-s.packets order. After @@ -2352,7 +2354,7 @@ func (s *PacketStore) QueryPackets(q PacketQuery) *PacketResult { packets := make([]map[string]interface{}, 0, pageSize) if q.Order == "ASC" { for _, tx := range results[start : start+pageSize] { - packets = append(packets, s.txToMapWithRP(tx, q.ExpandObservations)) + packets = append(packets, s.txToMapDeferredRP(tx, &jobs, q.ExpandObservations)) } } else { // DESC: newest items are at the tail; page 0 = last pageSize items reversed @@ -2362,7 +2364,7 @@ func (s *PacketStore) QueryPackets(q PacketQuery) *PacketResult { startIdx = 0 } for i := endIdx - 1; i >= startIdx; i-- { - packets = append(packets, s.txToMapWithRP(results[i], q.ExpandObservations)) + packets = append(packets, s.txToMapDeferredRP(results[i], &jobs, q.ExpandObservations)) } } return &PacketResult{Packets: packets, Total: total} @@ -2419,8 +2421,9 @@ func (s *PacketStore) QueryGroupedPackets(q PacketQuery) *PacketResult { tx *StoreTx latest string } + filters := s.preparePacketFilters(q) s.mu.RLock() - results := s.filterPackets(q) + results := s.filterPacketsPrepared(q, filters) sorted := make([]latestTx, len(results)) for i, tx := range results { sorted[i] = latestTx{tx: tx, latest: tx.LatestSeen} @@ -3026,43 +3029,49 @@ func (s *PacketStore) GetStoreMemoryBreakdown() *StoreMemoryBreakdown { // GetTransmissionByID returns a transmission by its DB ID, formatted as a map. func (s *PacketStore) GetTransmissionByID(id int) map[string]interface{} { s.mu.RLock() - defer s.mu.RUnlock() + var jobs []resolvedPathJob + defer func() { s.mu.RUnlock(); s.applyResolvedPathJobs(jobs) }() tx := s.byTxID[id] if tx == nil { return nil } - return s.txToMapWithRP(tx, true) + return s.txToMapDeferredRP(tx, &jobs, true) } // GetPacketByHash returns a transmission by content hash. func (s *PacketStore) GetPacketByHash(hash string) map[string]interface{} { s.mu.RLock() - defer s.mu.RUnlock() + var jobs []resolvedPathJob + defer func() { s.mu.RUnlock(); s.applyResolvedPathJobs(jobs) }() tx := s.byHash[strings.ToLower(hash)] if tx == nil { return nil } - return s.txToMapWithRP(tx, true) + return s.txToMapDeferredRP(tx, &jobs, true) } // GetPacketByID returns an observation (enriched with transmission fields) by observation ID. func (s *PacketStore) GetPacketByID(id int) map[string]interface{} { s.mu.RLock() - defer s.mu.RUnlock() + var jobs []resolvedPathJob + defer func() { s.mu.RUnlock(); s.applyResolvedPathJobs(jobs) }() obs := s.byObsID[id] if obs == nil { return nil } - return s.enrichObs(obs) + m := enrichObsFields(obs, snapObsTx(s.byTxID[obs.TransmissionID])) + jobs = append(jobs, resolvedPathJob{obsID: obs.ID, set: func(rp []*string) { m["resolved_path"] = rp }}) + return m } // GetObservationsForHash returns all observations for a hash, enriched with transmission fields. func (s *PacketStore) GetObservationsForHash(hash string) []map[string]interface{} { s.mu.RLock() - defer s.mu.RUnlock() + var jobs []resolvedPathJob + defer func() { s.mu.RUnlock(); s.applyResolvedPathJobs(jobs) }() tx := s.byHash[strings.ToLower(hash)] if tx == nil { @@ -3071,7 +3080,9 @@ func (s *PacketStore) GetObservationsForHash(hash string) []map[string]interface result := make([]map[string]interface{}, 0, len(tx.Observations)) for _, obs := range tx.Observations { - result = append(result, s.enrichObs(obs)) + m := enrichObsFields(obs, snapObsTx(s.byTxID[obs.TransmissionID])) + jobs = append(jobs, resolvedPathJob{obsID: obs.ID, set: func(rp []*string) { m["resolved_path"] = rp }}) + result = append(result, m) } return result } @@ -3099,9 +3110,6 @@ func (s *PacketStore) GetTimestamps(since string) []string { // QueryMultiNodePackets filters packets matching any of the given pubkeys. func (s *PacketStore) QueryMultiNodePackets(pubkeys []string, limit, offset int, order, since, until string) *PacketResult { - s.mu.RLock() - defer s.mu.RUnlock() - if len(pubkeys) == 0 { return &PacketResult{Packets: []map[string]interface{}{}, Total: 0} } @@ -3113,6 +3121,9 @@ func (s *PacketStore) QueryMultiNodePackets(pubkeys []string, limit, offset int, for i, pk := range pubkeys { resolved[i] = s.db.resolveNodePubkey(pk) } + s.mu.RLock() + var jobs []resolvedPathJob + defer func() { s.mu.RUnlock(); s.applyResolvedPathJobs(jobs) }() // Use byNode index instead of scanning all packets (O(indexed) vs O(all×pubkeys×json)). hashSet := make(map[string]bool) @@ -3152,7 +3163,7 @@ func (s *PacketStore) QueryMultiNodePackets(pubkeys []string, limit, offset int, packets := make([]map[string]interface{}, 0, pageSize) if order == "ASC" { for _, tx := range filtered[offset : offset+pageSize] { - packets = append(packets, s.txToMapWithRP(tx)) + packets = append(packets, s.txToMapDeferredRP(tx, &jobs)) } } else { endIdx := total - offset @@ -3161,7 +3172,7 @@ func (s *PacketStore) QueryMultiNodePackets(pubkeys []string, limit, offset int, startIdx = 0 } for i := endIdx - 1; i >= startIdx; i-- { - packets = append(packets, s.txToMapWithRP(filtered[i])) + packets = append(packets, s.txToMapDeferredRP(filtered[i], &jobs)) } } return &PacketResult{Packets: packets, Total: total} @@ -3326,6 +3337,16 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac if len(tempRows) == 0 { return nil, sinceID } + rows.Close() // a limit break can leave the sole DB connection checked out + allNodes, cachedPM := s.getCachedNodesAndPM() + var observerGPS map[string][2]float64 + if s.config != nil && len(s.config.Areas) > 0 { + ids := make([]string, 0, len(tempRows)) + for _, row := range tempRows { + ids = append(ids, strings.ToLower(strings.TrimSpace(row.observerID))) + } + observerGPS = s.db.gpsByPubkeysExact(ids) + } // Now lock and merge into store s.mu.Lock() @@ -3338,7 +3359,6 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac // Hoist getCachedNodesAndPM() once before the observation loop to avoid // per-observation function calls (review item #1). - _, cachedPM := s.getCachedNodesAndPM() // Hoist atomic graph.Load() out of the per-row loop too (PR #1208 // carmack #1) — one Load per ingest call, not one per row. cachedGraph := s.graph.Load() @@ -3482,7 +3502,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac // Incrementally update precomputed distance index with new transmissions if len(broadcastTxs) > 0 { - allNodes, pm := s.getCachedNodesAndPM() + pm := cachedPM nodeByPk := make(map[string]*nodeInfo, len(allNodes)) repeaterSet := make(map[string]bool) for i := range allNodes { @@ -3569,7 +3589,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac // only after the next full reload. Only a unique_prefix match // with a known position sets it — never a guess. if entryPrefix := pathFirstHop(obs.PathJSON); entryPrefix != "" { - if label, ok := s.resolveEntryPointArea([]string{entryPrefix}); ok { + if label, ok := s.resolveEntryPointAreaWithPM([]string{entryPrefix}, cachedPM); ok { pkt["area"] = label } } else if obs.ObserverID != "" && s.db != nil && s.config != nil && len(s.config.Areas) > 0 { @@ -3579,7 +3599,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac // annotateMessageAreas (routes.go), not extended to a // multi-hop path that just failed to resolve. pk := strings.ToLower(strings.TrimSpace(obs.ObserverID)) - if gps, ok := s.db.gpsByPubkeysExact([]string{pk})[pk]; ok { + if gps, ok := observerGPS[pk]; ok { if label, ok := AreaForPoint(gps[0], gps[1], s.config.Areas); ok { pkt["area"] = label } @@ -3757,6 +3777,9 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] if len(obsRows) == 0 { return nil } + rows.Close() + metadata := s.snapshotPathNodes() + pm := metadata.pm s.mu.Lock() defer s.mu.Unlock() @@ -3769,7 +3792,6 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] var obsRPMap map[int][]*string // obsID → resolved path (decode-window) // Hoist getCachedNodesAndPM() before the loop — same pattern as IngestNewFromDB (review fix #1). - _, pm := s.getCachedNodesAndPM() graphRef := s.graph.Load() hopsSeen := make(map[string]bool) // reused across observations; cleared per use @@ -3896,7 +3918,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] // Same live entry-point area resolution as IngestNewFromDB above -- // see the comment there. if entryPrefix := pathFirstHop(obs.PathJSON); entryPrefix != "" { - if label, ok := s.resolveEntryPointArea([]string{entryPrefix}); ok { + if label, ok := s.resolveEntryPointAreaWithPM([]string{entryPrefix}, pm); ok { pkt["area"] = label } } @@ -3947,7 +3969,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] } } if len(changedTxs) > 0 { - s.updateDistanceIndexForTxs(changedTxs) + s.updateDistanceIndexForTxs(changedTxs, metadata) } if len(updatedTxs) > 0 { @@ -4020,7 +4042,37 @@ func (s *PacketStore) MaxObservationID() int { // --- Internal filter/query helpers --- // filterPackets applies PacketQuery filters to the in-memory packet list. +// packetFilters holds only immutable DB-derived filter state. Prepare it before +// taking s.mu: even a cold region/area cache or a node-name lookup can wait +// for SQLite's connection pool. +type packetFilters struct { + regionObservers map[string]bool + areaNodes map[string]bool + nodePK string +} + +func (s *PacketStore) preparePacketFilters(q PacketQuery) packetFilters { + var f packetFilters + if q.Region != "" { + f.regionObservers = s.resolveRegionObservers(q.Region) + } + if q.Area != "" { + f.areaNodes = s.resolveAreaNodes(q.Area) + } + if q.Node != "" { + f.nodePK = s.db.resolveNodePubkey(q.Node) + } + return f +} + +// filterPackets is retained for package-local callers. No caller may hold +// s.mu while preparing DB-derived filters; production entry points use the +// prepared variant below and take their lock around that variant only. func (s *PacketStore) filterPackets(q PacketQuery) []*StoreTx { + return s.filterPacketsPrepared(q, s.preparePacketFilters(q)) +} + +func (s *PacketStore) filterPacketsPrepared(q PacketQuery, f packetFilters) []*StoreTx { // Fast path: single-key index lookups if q.Hash != "" && q.Type == nil && q.ExcludeTypes == 0 && q.Route == nil && q.Observer == "" && q.Region == "" && q.Area == "" && q.Node == "" && q.Channel == "" && q.Since == "" && q.Until == "" { @@ -4070,26 +4122,21 @@ func (s *PacketStore) filterPackets(q PacketQuery) []*StoreTx { } // Pre-compute region observer set. - var regionObservers map[string]bool + regionObservers := f.regionObservers if q.Region != "" { - regionObservers = s.resolveRegionObservers(q.Region) if len(regionObservers) == 0 { return nil } } // Pre-compute area node set. - var areaNodes map[string]bool - if q.Area != "" { - areaNodes = s.resolveAreaNodes(q.Area) - } + areaNodes := f.areaNodes // Pre-compute node filter parameters. - var nodePK string + nodePK := f.nodePK var nodeHashSet map[string]bool hasNode := q.Node != "" if hasNode { - nodePK = s.db.resolveNodePubkey(q.Node) indexed := s.byNode[nodePK] nodeHashSet = make(map[string]bool, len(indexed)) for _, tx := range indexed { @@ -4306,12 +4353,15 @@ func (s *PacketStore) resolveEntryPointArea(prefixes []string) (label string, ok if s == nil || s.config == nil || len(s.config.Areas) == 0 || len(prefixes) == 0 { return "", false } - // getCachedNodesAndPM guards itself with its own cacheMu -- it never - // touches s.mu. Do not wrap this call in s.mu.RLock(): callers like - // IngestNewFromDB/IngestNewObservations invoke this while already - // holding s.mu.Lock(), and Go's RWMutex is not reentrant, so an - // RLock here would deadlock against the caller's own write lock. _, pm := s.getCachedNodesAndPM() + return s.resolveEntryPointAreaWithPM(prefixes, pm) +} + +// Store-locked ingest callers pass the node snapshot prepared before locking. +func (s *PacketStore) resolveEntryPointAreaWithPM(prefixes []string, pm *prefixMap) (label string, ok bool) { + if s == nil || s.config == nil || len(s.config.Areas) == 0 { + return "", false + } if pm == nil { return "", false } @@ -4416,7 +4466,7 @@ func iataMatchesRegion(iata, regionParam string) bool { // // #804: feeds analytics region-attribution so a multi-byte repeater whose // flood adverts get relayed across regions is still attributed to its home. -func (s *PacketStore) computeNodeHomeRegions() map[string]string { +func (s *PacketStore) observerIATAs() map[string]string { // Build observer → IATA map. observers table is small (≪ packets), so a // single DB read here is acceptable; resolveRegionObservers does similar. obsIATA := make(map[string]string, 64) @@ -4429,6 +4479,12 @@ func (s *PacketStore) computeNodeHomeRegions() map[string]string { } } } + return obsIATA +} + +// computeNodeHomeRegions reads only the locked store and an immutable observer +// snapshot prepared before taking s.mu. +func (s *PacketStore) computeNodeHomeRegions(obsIATA map[string]string) map[string]string { if len(obsIATA) == 0 { return map[string]string{} } @@ -4514,16 +4570,6 @@ func (s *PacketStore) computeNodeHomeRegions() map[string]string { return out } -// enrichObs returns a map with observation fields + transmission fields. -// Looks up the transmission in s.byTxID itself — safe only when the caller -// already holds s.mu (directly, or via a defer'd RLock spanning the call). -// Callers that snapshot observations under RLock and then release it before -// iterating (e.g. handleObserverAnalytics, #1830) must use enrichObsWithTx -// with the transmission resolved during that same snapshot instead. -func (s *PacketStore) enrichObs(obs *StoreObs) map[string]interface{} { - return s.enrichObsWithTx(obs, snapObsTx(s.byTxID[obs.TransmissionID])) -} - // obsTx is an observation's transmission with the fields of it that the // content-hash migration (#215) rewrites under s.mu.Lock while HTTP is // served, copied when it was resolved (#470). The fields enrichObsWithTx @@ -4553,6 +4599,16 @@ func snapObsTx(tx *StoreTx) obsTx { // each observation's transmission (snapObsTx) during their RLock-held // snapshot and pass it in here. func (s *PacketStore) enrichObsWithTx(obs *StoreObs, t obsTx) map[string]interface{} { + m := enrichObsFields(obs, t) + if rp := s.fetchResolvedPathForObs(obs.ID); rp != nil { + m["resolved_path"] = rp + } + return m +} + +// enrichObsFields is SQL-free. The caller must snapshot the transmission's +// migrated hash/route and obs.TransmissionID under s.mu (#475). +func enrichObsFields(obs *StoreObs, t obsTx) map[string]interface{} { m := map[string]interface{}{ "id": obs.ID, "timestamp": strOrNil(obs.Timestamp), @@ -4565,12 +4621,6 @@ func (s *PacketStore) enrichObsWithTx(obs *StoreObs, t obsTx) map[string]interfa "score": intPtrOrNil(obs.Score), "path_json": strOrNil(obs.PathJSON), } - // On-demand SQL fetch for resolved_path - rp := s.fetchResolvedPathForObs(obs.ID) - if rp != nil { - m["resolved_path"] = rp - } - if tx := t.tx; tx != nil { m["hash"] = strOrNil(t.hash) // raw_hex comes from the transmission (#881). obs.RawHex is no longer @@ -4648,23 +4698,22 @@ func txToMap(tx *StoreTx, includeObservations ...bool) map[string]interface{} { return m } -// txToMapWithRP is like txToMap but also fetches resolved_path on demand from the store. -func (s *PacketStore) txToMapWithRP(tx *StoreTx, includeObservations ...bool) map[string]interface{} { +// txToMapDeferredRP builds response fields and lookup inputs while s.mu is +// held. The jobs run only after unlocking. No mutable StoreTx is read by them. +func (s *PacketStore) txToMapDeferredRP(tx *StoreTx, jobs *[]resolvedPathJob, includeObservations ...bool) map[string]interface{} { m := txToMap(tx, includeObservations...) - // On-demand SQL fetch for resolved_path - rp := s.fetchResolvedPathForTxBest(tx) - if rp != nil { - m["resolved_path"] = rp + observations := make([]resolvedPathObservation, len(tx.Observations)) + for i, obs := range tx.Observations { + observations[i] = resolvedPathObservation{id: obs.ID, pathJSON: obs.PathJSON} } + *jobs = append(*jobs, resolvedPathJob{forTx: true, txID: tx.ID, observations: observations, set: func(rp []*string) { m["resolved_path"] = rp }}) // Also add resolved_path to observation sub-maps if present if len(includeObservations) > 0 && includeObservations[0] { if obsList, ok := m["observations"].([]map[string]interface{}); ok { for i, o := range tx.Observations { if i < len(obsList) { - obsRP := s.fetchResolvedPathForObs(o.ID) - if obsRP != nil { - obsList[i]["resolved_path"] = obsRP - } + om := obsList[i] + *jobs = append(*jobs, resolvedPathJob{obsID: o.ID, set: func(rp []*string) { om["resolved_path"] = rp }}) } } } @@ -5209,7 +5258,7 @@ func hasEvictedHopPrefix(key string, prefixes map[string]bool, lens []int) bool // transmissions and recomputes them. Builds lookup maps once, amortising the // cost across all changed txs in a single ingest cycle. Must be called with // s.mu held. -func (s *PacketStore) updateDistanceIndexForTxs(txs []*StoreTx) { +func (s *PacketStore) updateDistanceIndexForTxs(txs []*StoreTx, snapshots ...pathNodeSnapshot) { // Remove old records for all changed txs first. removeSet := make(map[*StoreTx]bool, len(txs)) for _, tx := range txs { @@ -5218,7 +5267,8 @@ func (s *PacketStore) updateDistanceIndexForTxs(txs []*StoreTx) { s.compactDistIndex(removeSet) // Build lookup maps once. - allNodes, pm := s.getCachedNodesAndPM() + meta := s.pathNodeSnapshotLocked(snapshots) + allNodes, pm := meta.nodes, meta.pm nodeByPk := make(map[string]*nodeInfo, len(allNodes)) repeaterSet := make(map[string]bool) for i := range allNodes { @@ -5436,8 +5486,9 @@ func (s *PacketStore) runDistanceIndexBuild() { s.distanceBuildHook() // test seam: hold the build window open } + metadata := s.snapshotPathNodes() s.mu.Lock() - s.buildDistanceIndex() + s.buildDistanceIndex(metadata) // Keep this read inside the s.mu section of the build: the load // completion bumps the generation under s.mu, so here it matches the // data just read. Read after the Unlock, a load completing in between @@ -5478,8 +5529,9 @@ func (s *PacketStore) runDistanceIndexBuild() { // buildDistanceIndex precomputes haversine distances for all packets. // Must be called with s.mu held (Lock). -func (s *PacketStore) buildDistanceIndex() { - allNodes, pm := s.getCachedNodesAndPM() +func (s *PacketStore) buildDistanceIndex(snapshots ...pathNodeSnapshot) { + meta := s.pathNodeSnapshotLocked(snapshots) + allNodes, pm := meta.nodes, meta.pm nodeByPk := make(map[string]*nodeInfo, len(allNodes)) repeaterSet := make(map[string]bool) for i := range allNodes { @@ -5724,8 +5776,9 @@ func (s *PacketStore) EvictStaleWithRP(rpBatch map[int][]string) int { return s.evictStaleInternal(rpBatch) } -// EvictStale runs eviction, fetching resolved pubkeys inline (SQL under lock). -// Prefer RunEviction() which batches the SQL outside the lock. +// EvictStale is a SQL-free legacy/test entry point without a resolved-pubkey +// batch. Prefer RunEviction(), which prefetches that cleanup evidence outside +// the lock; a nil batch skips that portion of byNode/nodeHashes cleanup. // Must be called under s.mu.Lock. func (s *PacketStore) EvictStale() int { return s.evictStaleInternal(nil) @@ -6314,12 +6367,12 @@ func (s *PacketStore) GetChannels(region string) []map[string]interface{} { hasRegion bool } - // Copy only the fields needed — release the lock before JSON unmarshal. - s.mu.RLock() + // Prepare the SQL-backed filter before taking the store snapshot. var regionObs map[string]bool if region != "" { regionObs = s.resolveRegionObservers(region) } + s.mu.RLock() grpTxts := s.byPayloadType[5] snapshots := make([]txSnapshot, 0, len(grpTxts)) for _, tx := range grpTxts { @@ -6468,11 +6521,11 @@ func (s *PacketStore) GetChannels(region string) []map[string]interface{} { // GetEncryptedChannels returns undecryptable GRP_TXT channels from in-memory packets. func (s *PacketStore) GetEncryptedChannels(region string) []map[string]interface{} { - s.mu.RLock() var regionObs map[string]bool if region != "" { regionObs = s.resolveRegionObservers(region) } + s.mu.RLock() grpTxts := s.byPayloadType[5] type encInfo struct { @@ -6546,9 +6599,6 @@ func (s *PacketStore) GetEncryptedChannels(region string) []map[string]interface // GetChannelMessages returns deduplicated messages for a channel from in-memory packets. func (s *PacketStore) GetChannelMessages(channelHash string, limit, offset int, region ...string) ([]map[string]interface{}, int) { - s.mu.RLock() - defer s.mu.RUnlock() - if limit <= 0 { limit = 100 } @@ -6565,6 +6615,8 @@ func (s *PacketStore) GetChannelMessages(channelHash string, limit, offset int, regionParam = region[0] } regionObs := s.resolveRegionObservers(regionParam) + s.mu.RLock() + defer s.mu.RUnlock() // Iterate type-5 packets oldest-first (byPayloadType is ASC = oldest first) type decodedMsg struct { @@ -6833,13 +6885,12 @@ func (s *PacketStore) computeAnalyticsChannels(region, area string, window TimeW areaNodes = s.resolveAreaNodes(area) } - s.mu.RLock() - defer s.mu.RUnlock() - var regionObs map[string]bool if region != "" { regionObs = s.resolveRegionObservers(region) } + s.mu.RLock() + defer s.mu.RUnlock() type decodedGrp struct { Type string `json:"type"` @@ -7108,15 +7159,14 @@ func (s *PacketStore) computeAnalyticsRF(region, area string, window TimeWindow) areaNodes = s.resolveAreaNodes(area) } - s.mu.RLock() - defer s.mu.RUnlock() - ptNames := payloadTypeNames var regionObs map[string]bool if region != "" { regionObs = s.resolveRegionObservers(region) } + s.mu.RLock() + defer s.mu.RUnlock() // Collect all observations matching the region estCap := s.totalObs @@ -7844,8 +7894,9 @@ func buildPrefixMap(nodes []nodeInfo) *prefixMap { return pm } -// getCachedNodesAndPM returns cached node list and prefix map, rebuilding if stale. -// Must be called with s.mu held (RLock or Lock). +// getCachedNodesAndPM returns an immutable node snapshot, rebuilding if stale. +// Call WITHOUT s.mu: cold/expired reads wait for SQL. A request keeps its +// snapshot even if invalidated concurrently, but never republishes it then. func (s *PacketStore) getCachedNodesAndPM() ([]nodeInfo, *prefixMap) { s.cacheMu.Lock() if s.nodeCache != nil && time.Since(s.nodeCacheTime) < 30*time.Second { @@ -7855,27 +7906,44 @@ func (s *PacketStore) getCachedNodesAndPM() ([]nodeInfo, *prefixMap) { } s.cacheMu.Unlock() - nodes := s.getAllNodes() - pm := buildPrefixMap(nodes) - // Issue #1290: exclude observers that advertised `repeat:off` from - // the path-hop candidate set. Failure is non-fatal — we log via the - // schema-degradation channel and proceed with an empty filter (i.e. - // pre-#1290 behavior). - if s.db != nil && s.db.conn != nil { - if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil { - pm.markNonRelay(pks) - } else { - s.logSchemaDegradationOnce("observers.can_relay read failed; path-hop disambiguator will not filter listener-only observers: " + err.Error()) - } + type nodeSnapshot struct { + nodes []nodeInfo + pm *prefixMap } + v, _, _ := s.nodeCacheLoad.Do("nodes", func() (interface{}, error) { + s.cacheMu.Lock() + if s.nodeCache != nil && time.Since(s.nodeCacheTime) < 30*time.Second { + out := nodeSnapshot{s.nodeCache, s.nodePM} + s.cacheMu.Unlock() + return out, nil + } + gen := s.nodeCacheGen + s.cacheMu.Unlock() + nodes := s.getAllNodes() + pm := buildPrefixMap(nodes) + // Issue #1290: exclude observers that advertised `repeat:off` from + // the path-hop candidate set. Failure is non-fatal — we log via the + // schema-degradation channel and proceed with an empty filter (i.e. + // pre-#1290 behavior). + if s.db != nil && s.db.conn != nil { + if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil { + pm.markNonRelay(pks) + } else { + s.logSchemaDegradationOnce("observers.can_relay read failed; path-hop disambiguator will not filter listener-only observers: " + err.Error()) + } + } - s.cacheMu.Lock() - s.nodeCache = nodes - s.nodePM = pm - s.nodeCacheTime = time.Now() - s.cacheMu.Unlock() - - return nodes, pm + s.cacheMu.Lock() + if s.nodeCacheGen == gen { + s.nodeCache = nodes + s.nodePM = pm + s.nodeCacheTime = time.Now() + } + s.cacheMu.Unlock() + return nodeSnapshot{nodes, pm}, nil + }) + out := v.(nodeSnapshot) + return out.nodes, out.pm } // InvalidateNodeCache forces the next getCachedNodesAndPM call to rebuild. @@ -7884,9 +7952,47 @@ func (s *PacketStore) InvalidateNodeCache() { s.nodeCache = nil s.nodePM = nil s.nodeCacheTime = time.Time{} + s.nodeCacheGen++ s.cacheMu.Unlock() } +// cachedNodesAndPM only reads published metadata and never performs SQL. +// Internal store-locked index helpers use it after their entry point warmed +// getCachedNodesAndPM outside s.mu. An invalidated cache fails closed. +func (s *PacketStore) cachedNodesAndPM() ([]nodeInfo, *prefixMap) { + s.cacheMu.RLock() + nodes, pm := s.nodeCache, s.nodePM + s.cacheMu.RUnlock() + return nodes, pm +} + +type pathNodeSnapshot struct { + nodes []nodeInfo + pm *prefixMap +} + +// snapshotPathNodes runs before the caller takes s.mu. Retaining this value +// across invalidation prevents a store-locked index build from seeing a nil +// cache after it was warmed and erasing otherwise valid distance/relay facts. +func (s *PacketStore) snapshotPathNodes() pathNodeSnapshot { + if s.db == nil || s.db.conn == nil { + nodes, pm := s.cachedNodesAndPM() + return pathNodeSnapshot{nodes, pm} + } + nodes, pm := s.getCachedNodesAndPM() + return pathNodeSnapshot{nodes, pm} +} + +// Production callers supply their captured snapshot. The default is only for +// package-local, store-locked test helpers which seed metadata directly. +func (s *PacketStore) pathNodeSnapshotLocked(snapshots []pathNodeSnapshot) pathNodeSnapshot { + if len(snapshots) != 0 { + return snapshots[0] + } + nodes, pm := s.cachedNodesAndPM() + return pathNodeSnapshot{nodes, pm} +} + func (pm *prefixMap) resolve(hop string) *nodeInfo { h := strings.ToLower(hop) candidates := pm.m[h] @@ -8234,9 +8340,6 @@ func (s *PacketStore) computeAnalyticsTopology(region, area string, window TimeW areaNodes = s.resolveAreaNodes(area) } - s.mu.RLock() - defer s.mu.RUnlock() - var regionObs map[string]bool if region != "" { regionObs = s.resolveRegionObservers(region) @@ -8244,6 +8347,14 @@ func (s *PacketStore) computeAnalyticsTopology(region, area string, window TimeW allNodes, pm := s.getCachedNodesAndPM() _ = allNodes // only pm is needed for topology + uniqueNodes := 0 + if s.db != nil { + if stats, err := s.db.GetStats(); err == nil { + uniqueNodes = stats.TotalNodes + } + } + s.mu.RLock() + defer s.mu.RUnlock() // Materialize the filtered tx slice ONCE — both the context-build pass // and the main aggregation pass need the same window+region predicate. @@ -8754,14 +8865,6 @@ func (s *PacketStore) computeAnalyticsTopology(region, area string, window TimeW bestPathList = bestPathList[:50] } - // Use DB 7-day active node count (matches /api/stats totalNodes) - uniqueNodes := 0 - if s.db != nil { - if stats, err := s.db.GetStats(); err == nil { - uniqueNodes = stats.TotalNodes - } - } - return map[string]interface{}{ "uniqueNodes": uniqueNodes, "avgHops": avgHops, @@ -9233,9 +9336,6 @@ func (s *PacketStore) computeAnalyticsHashSizesWithCapability(region, area strin } func (s *PacketStore) computeAnalyticsHashSizes(region, area string) map[string]interface{} { - s.mu.RLock() - defer s.mu.RUnlock() - var regionObs map[string]bool if region != "" { regionObs = s.resolveRegionObservers(region) @@ -9245,18 +9345,20 @@ func (s *PacketStore) computeAnalyticsHashSizes(region, area string) map[string] areaNodes = s.resolveAreaNodes(area) } + obsIATA := s.observerIATAs() + allNodes, pm := s.getCachedNodesAndPM() + s.mu.RLock() + defer s.mu.RUnlock() // #804: derive each node's HOME region from zero-hop direct adverts (the // most authoritative location signal — those packets cannot have been // relayed). When non-empty, multi-byte node attribution prefers this // over observer-region. Falls back to observer-region when unknown. - nodeHomeRegion := s.computeNodeHomeRegions() + nodeHomeRegion := s.computeNodeHomeRegions(obsIATA) attributionMethod := "observer" if region != "" && len(nodeHomeRegion) > 0 { attributionMethod = "repeater" } - allNodes, pm := s.getCachedNodesAndPM() - // Build pubkey→role map for filtering by node type. nodeRoleByPK := make(map[string]string, len(allNodes)) for _, n := range allNodes { @@ -10445,13 +10547,15 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string areaNodes = s.resolveAreaNodes(area) } + var regionObs map[string]bool + if region != "" { + regionObs = s.resolveRegionObservers(region) + } + _, pm := s.getCachedNodesAndPM() s.mu.RLock() - defer s.mu.RUnlock() - // Region filtering var regionNodeKeys map[string]bool if region != "" { - regionObs := s.resolveRegionObservers(region) if regionObs != nil { regionalHashes := make(map[string]bool) for obsID := range regionObs { @@ -10474,6 +10578,7 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string } } } + s.mu.RUnlock() // Get nodes from DB — fetch more when filtering so we don't under-fill after exclusions queryLimit := limit @@ -10510,10 +10615,11 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string break } } - // Release the connection before the store work: the early break above - // leaves rows open, and relayPrefixMapLocked may query the node table - // again (a single-connection pool would block on it forever). + // Release the connection before the store-only aggregation, including + // when the early break above left rows open. rows.Close() + s.mu.RLock() + defer s.mu.RUnlock() // Only cap to limit in the global (no-filter) case; area/region returns full filtered set if regionNodeKeys != nil && areaNodes == nil && len(nodes) > limit { nodes = nodes[:limit] @@ -10521,7 +10627,6 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string todayStart := time.Now().UTC().Truncate(24 * time.Hour).Format(time.RFC3339) results := make([]map[string]interface{}, 0, len(nodes)) - pm := s.relayPrefixMapLocked() relaySeen := make(map[int]struct{}) // request-local relay-evidence dedupe scratch heard := newHeardByAccumulator("", pm) // reset per node, storage reused @@ -10601,9 +10706,25 @@ func (s *PacketStore) GetNodeHealth(pubkey string, now time.Time) (map[string]in s.bgErrMu.RLock() loadCoverage := s.loadCoverageRatio s.bgErrMu.RUnlock() + _, pm := s.getCachedNodesAndPM() + nonRelaySet := map[string]struct{}{} + seenSet := map[string]struct{}{} + if s.db != nil && s.db.conn != nil { + if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil { + for _, pk := range pks { + nonRelaySet[strings.ToLower(pk)] = struct{}{} + } + } + if pks, err := s.db.GetCanRelaySeenObserverPubkeys(); err == nil { + for _, pk := range pks { + seenSet[strings.ToLower(pk)] = struct{}{} + } + } + } s.mu.RLock() - defer s.mu.RUnlock() + var jobs []resolvedPathJob + defer func() { s.mu.RUnlock(); s.applyResolvedPathJobs(jobs) }() packets := s.byNode[pubkey] activityKey := strings.ToLower(pubkey) @@ -10613,7 +10734,6 @@ func (s *PacketStore) GetNodeHealth(pubkey string, now time.Time) (map[string]in var packetsToday int var totalHops, hopCount int var lastHeard, lastAdvert string - pm := s.relayPrefixMapLocked() heard := newHeardByAccumulator(pubkey, pm) totalObservations := 0 @@ -10644,24 +10764,10 @@ func (s *PacketStore) GetNodeHealth(pubkey string, now time.Time) (map[string]in // Two case conventions on the same upstream string would be a // latent regression waiting for any refactor that touches the // observer-id normalization layer. - nonRelaySet := map[string]struct{}{} // PR #1624 MAJOR-2: tri-state badge needs to distinguish "confirmed // repeater" (seen=1, can_relay=1) from "unknown" (seen=0). Build // the set of observers we have NO repeat-field record for so the // badge is nil/omitted for them — matches nodes.js:679 tri-state. - seenSet := map[string]struct{}{} - if s.db != nil && s.db.conn != nil { - if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil { - for _, pk := range pks { - nonRelaySet[strings.ToLower(pk)] = struct{}{} - } - } - if pks, err := s.db.GetCanRelaySeenObserverPubkeys(); err == nil { - for _, pk := range pks { - seenSet[strings.ToLower(pk)] = struct{}{} - } - } - } direct, relay := heard.lists(func(id string) *bool { idLower := strings.ToLower(id) if _, seen := seenSet[idLower]; !seen { @@ -10684,7 +10790,7 @@ func (s *PacketStore) GetNodeHealth(pubkey string, now time.Time) (map[string]in } recentPackets := make([]map[string]interface{}, 0, recentLimit) for i := len(packets) - 1; i >= len(packets)-recentLimit; i-- { - p := s.txToMapWithRP(packets[i]) + p := s.txToMapDeferredRP(packets[i], &jobs) delete(p, "observations") recentPackets = append(recentPackets, p) } @@ -10978,10 +11084,14 @@ func (s *PacketStore) GetAnalyticsSubpathsWithWindow(region string, minLen, maxL // region-filters. Mirrors computeSubpathsSlow but with the window predicate // inlined and without requiring a non-empty region. func (s *PacketStore) computeAnalyticsSubpathsWindowed(region string, minLen, maxLen, limit int, window TimeWindow) map[string]interface{} { + _, pm := s.getCachedNodesAndPM() + var regionObs map[string]bool + if region != "" { + regionObs = s.resolveRegionObservers(region) + } s.mu.RLock() defer s.mu.RUnlock() - _, pm := s.getCachedNodesAndPM() contextPubkeys := buildAggregateHopContextPubkeys(s.packets, pm) hopCache := make(map[string]*nodeInfo) graph := s.graph.Load() @@ -11000,11 +11110,6 @@ func (s *PacketStore) computeAnalyticsSubpathsWindowed(region string, minLen, ma return hop } - var regionObs map[string]bool - if region != "" { - regionObs = s.resolveRegionObservers(region) - } - subpathCounts := make(map[string]*subpathAccum) totalPaths := 0 @@ -11121,9 +11226,9 @@ func (s *PacketStore) GetAnalyticsSubpathsBulk(region string, groups []subpathGr } s.cacheMu.Unlock() - // Single scan: bucket by hop length into per-group accumulators. - s.mu.RLock() + // Refresh DB-backed metadata before the store-only scan. _, pm := s.getCachedNodesAndPM() + s.mu.RLock() // Aggregate hop-disambiguation context across all packets so the // resolver's tiers 1 and 2 light up even on this bulk-aggregate path // (the index iterates raw subpath strings, not per-tx). See #1197. @@ -11216,10 +11321,11 @@ func (a *subpathAccum) add(raw string, n int) { } func (s *PacketStore) computeAnalyticsSubpaths(region string, minLen, maxLen, limit int) map[string]interface{} { + _, pm := s.getCachedNodesAndPM() + regionObs := s.resolveRegionObservers(region) s.mu.RLock() defer s.mu.RUnlock() - _, pm := s.getCachedNodesAndPM() // Aggregate hop-disambiguation context across all packets — bulk // aggregator over s.spIndex / per-tx fallback both need it. See #1197. contextPubkeys := buildAggregateHopContextPubkeys(s.packets, pm) @@ -11243,7 +11349,7 @@ func (s *PacketStore) computeAnalyticsSubpaths(region string, minLen, maxLen, li // For region queries fall back to packet iteration (region filtering // requires per-transmission observer checks). if region != "" { - return s.computeSubpathsSlow(region, minLen, maxLen, limit, resolveHop) + return s.computeSubpathsSlow(regionObs, minLen, maxLen, limit, resolveHop) } // Fast path: read from precomputed raw-hop subpath index. @@ -11273,8 +11379,7 @@ func (s *PacketStore) computeAnalyticsSubpaths(region string, minLen, maxLen, li // computeSubpathsSlow is the original O(N) packet-iteration path, used only // for region-filtered queries where we must check per-transmission observers. -func (s *PacketStore) computeSubpathsSlow(region string, minLen, maxLen, limit int, resolveHop func(string) string) map[string]interface{} { - regionObs := s.resolveRegionObservers(region) +func (s *PacketStore) computeSubpathsSlow(regionObs map[string]bool, minLen, maxLen, limit int, resolveHop func(string) string) map[string]interface{} { subpathCounts := make(map[string]*subpathAccum) totalPaths := 0 @@ -11418,11 +11523,10 @@ type subpathDetailObserver struct { // SubpathDetail gathers the parts of a subpath detail. func (s *PacketStore) SubpathDetail(rawHops []string) *subpathDetailParts { + _, pm := s.getCachedNodesAndPM() s.mu.RLock() defer s.mu.RUnlock() - _, pm := s.getCachedNodesAndPM() - // Build the subpath key the same way the index does (lowercase, comma-joined) spKey := strings.ToLower(strings.Join(rawHops, ",")) @@ -11687,8 +11791,6 @@ func (s *PacketStore) LoadCoverageRatio() float64 { // HopCandidates returns, per lower-cased hop prefix, the nodes that prefix can // stand for (the resolver's candidate set). Values are copies. func (s *PacketStore) HopCandidates(prefixes []string) map[string][]nodeInfo { - s.mu.RLock() - defer s.mu.RUnlock() _, pm := s.getCachedNodesAndPM() out := make(map[string][]nodeInfo, len(prefixes)) for _, p := range prefixes { diff --git a/cmd/server/store_lock_sql_invariant_412_test.go b/cmd/server/store_lock_sql_invariant_412_test.go new file mode 100644 index 000000000..8c8bc403e --- /dev/null +++ b/cmd/server/store_lock_sql_invariant_412_test.go @@ -0,0 +1,377 @@ +package main + +import ( + "go/ast" + "go/parser" + "go/token" + "os" + "path/filepath" + "sort" + "strings" + "testing" +) + +// #412: no request/refresh SQL while PacketStore.mu is held. The read pool has 4 +// connections; a store-lock holder that waits for one stalls the ingest +// writer, and behind a pending writer every other reader of the store. +// +// This test parses the package and fails when code inside a store.mu +// section calls, directly or through other package functions, a function +// that queries SQLite. Calls are resolved by name only (no type checking), +// which is enough for this package's receiver conventions. + +// The lexical scan sees calls in a deferred closure before its outer +// function ends. applyResolvedPathJobs is SQL-capable, but every such closure +// releases s.mu first; TestStoreSQLDoesNotHoldPacketLock412 exercises those +// calls against the real blocked SQL pool. hasRouteMask is a pure atomic +// schema flag read, not a query (schemaFlag.get's documented contract). +var sqlUnderLockAllowedCallees = map[string]bool{ + "PacketStore.applyResolvedPathJobs": true, + "DB.hasRouteMask": true, +} + +// Legacy startup Load intentionally fills the store under its exclusive +// publication lock before HTTP serving. Chunked startup/load methods may +// unlock in nested blocks which this lexical scanner cannot model. Keep +// these explicit exceptions; do not claim that startup loading is SQL-free. +var sqlUnderLockAllowedHolders = map[string]bool{ + "PacketStore.Load": true, + "PacketStore.LoadChunked": true, + "PacketStore.loadChunk": true, + "PacketStore.scanAndMergeChunk": true, +} + +func TestNoSQLUnderStoreLock(t *testing.T) { + files, err := filepath.Glob("*.go") + if err != nil { + t.Fatal(err) + } + srcs := map[string]string{} + for _, f := range files { + if strings.HasSuffix(f, "_test.go") { + continue + } + b, err := os.ReadFile(f) + if err != nil { + t.Fatal(err) + } + srcs[f] = string(b) + } + hits, regions := findSQLUnderStoreLock(t, srcs) + // Guard against a scan that silently stops matching anything. + if regions < 40 { + t.Fatalf("found only %d store.mu sections; the scan no longer recognises the locking code", regions) + } + for _, h := range hits { + t.Errorf("SQL under store.mu: %s", h) + } + assertResolvedPathJobDefers412(t, srcs) +} + +// Do not let the lexical scan's one SQL-capable callee exception become a +// loophole: every call must be the second statement of an unlock-then-apply +// defer. Runtime convoy tests independently verify the actual behavior. +func assertResolvedPathJobDefers412(t *testing.T, srcs map[string]string) { + t.Helper() + seen := 0 + for filename, src := range srcs { + fset := token.NewFileSet() + f, err := parser.ParseFile(fset, filename, src, 0) + if err != nil { + t.Fatal(err) + } + allowed := map[*ast.CallExpr]bool{} + ast.Inspect(f, func(n ast.Node) bool { + d, ok := n.(*ast.DeferStmt) + if !ok { + return true + } + body, ok := d.Call.Fun.(*ast.FuncLit) + if !ok || len(body.Body.List) != 2 { + return true + } + first, ok := body.Body.List[0].(*ast.ExprStmt) + if !ok { + return true + } + unlock, ok := first.X.(*ast.CallExpr) + if !ok || lockScanExpr(unlock.Fun) != "s.mu.RUnlock" { + return true + } + second, ok := body.Body.List[1].(*ast.ExprStmt) + if !ok { + return true + } + apply, ok := second.X.(*ast.CallExpr) + if ok && lockScanExpr(apply.Fun) == "s.applyResolvedPathJobs" { + allowed[apply] = true + } + return true + }) + ast.Inspect(f, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + se, ok := call.Fun.(*ast.SelectorExpr) + if !ok || se.Sel.Name != "applyResolvedPathJobs" { + return true + } + seen++ + if !allowed[call] { + t.Errorf("%s: resolved-path jobs must run in an unlock-then-apply defer", fset.Position(call.Pos())) + } + return true + }) + } + if seen < 5 { + t.Fatalf("only %d resolved-path application sites found; scanner lost the callers", seen) + } +} + +// The scan itself must catch a direct and an indirect query under the lock, +// and ignore one after the unlock. +func TestNoSQLUnderStoreLockDetects(t *testing.T) { + src := `package main +type PacketStore struct{} +type DB struct{} +func (db *DB) GetThing() {} +func (s *PacketStore) helper() { s.db.GetThing() } +func (s *PacketStore) Direct() { + s.mu.RLock() + defer s.mu.RUnlock() + s.db.conn.Query("SELECT 1") +} +func (s *PacketStore) Indirect() { + s.mu.RLock() + s.helper() + s.mu.RUnlock() +} +func (s *PacketStore) After() { + s.mu.RLock() + x := 1 + s.mu.RUnlock() + s.helper() + _ = x +} +` + hits, _ := findSQLUnderStoreLock(t, map[string]string{"x.go": src}) + got := strings.Join(hits, "\n") + for _, want := range []string{"PacketStore.Direct", "PacketStore.Indirect calls PacketStore.helper"} { + if !strings.Contains(got, want) { + t.Errorf("scan missed %q; hits:\n%s", want, got) + } + } + if strings.Contains(got, "PacketStore.After") { + t.Errorf("scan flagged a query after the unlock; hits:\n%s", got) + } +} + +var sqlMethodNames = map[string]bool{ + "Query": true, "QueryRow": true, "QueryContext": true, "QueryRowContext": true, + "Exec": true, "ExecContext": true, "Prepare": true, "Begin": true, "BeginTx": true, +} + +func lockScanExpr(e ast.Expr) string { + switch x := e.(type) { + case *ast.Ident: + return x.Name + case *ast.SelectorExpr: + return lockScanExpr(x.X) + "." + x.Sel.Name + case *ast.CallExpr: + return lockScanExpr(x.Fun) + "()" + } + return "?" +} + +func lockScanRecv(d *ast.FuncDecl) string { + if d.Recv == nil || len(d.Recv.List) == 0 { + return "" + } + t := d.Recv.List[0].Type + if s, ok := t.(*ast.StarExpr); ok { + t = s.X + } + if id, ok := t.(*ast.Ident); ok { + return id.Name + } + return "" +} + +// findSQLUnderStoreLock returns one line per call made inside a store.mu +// section that reaches SQL, and the number of sections it found. +func findSQLUnderStoreLock(t *testing.T, srcs map[string]string) (hits []string, regions int) { + t.Helper() + fset := token.NewFileSet() + type fn struct { + decl *ast.FuncDecl + recv string + } + funcs := map[string]fn{} + for name, src := range srcs { + f, err := parser.ParseFile(fset, name, src, 0) + if err != nil { + t.Fatal(err) + } + for _, d := range f.Decls { + fd, ok := d.(*ast.FuncDecl) + if !ok || fd.Body == nil { + continue + } + recv := lockScanRecv(fd) + funcs[recv+"."+fd.Name.Name] = fn{fd, recv} + } + } + resolve := func(c *ast.CallExpr, recv string) string { + switch f := c.Fun.(type) { + case *ast.Ident: + if _, ok := funcs["."+f.Name]; ok { + return "." + f.Name + } + case *ast.SelectorExpr: + x := lockScanExpr(f.X) + var typ string + switch { + case x == "s" && recv != "": + typ = recv + case strings.HasSuffix(x, "store"): + typ = "PacketStore" + case strings.HasSuffix(x, "db"): + typ = "DB" + } + if _, ok := funcs[typ+"."+f.Sel.Name]; ok { + return typ + "." + f.Sel.Name + } + } + return "" + } + + // via[k] is the next step from k towards a function that runs SQL. + via := map[string]string{} + edges := map[string][]string{} + for k, f := range funcs { + if f.recv == "DB" { + via[k] = k + } + ast.Inspect(f.decl.Body, func(n ast.Node) bool { + c, ok := n.(*ast.CallExpr) + if !ok { + return true + } + if se, ok := c.Fun.(*ast.SelectorExpr); ok && sqlMethodNames[se.Sel.Name] && strings.HasSuffix(lockScanExpr(se.X), "conn") { + via[k] = k + } + if callee := resolve(c, f.recv); callee != "" { + edges[k] = append(edges[k], callee) + } + return true + }) + } + for k := range sqlUnderLockAllowedCallees { + delete(via, k) + } + for changed := true; changed; { + changed = false + for k, es := range edges { + if _, ok := via[k]; ok || sqlUnderLockAllowedCallees[k] { + continue + } + for _, e := range es { + if _, ok := via[e]; ok { + via[k] = e + changed = true + break + } + } + } + } + chain := func(k string) string { + out := []string{k} + for i := 0; i < 10 && via[k] != k; i++ { + k = via[k] + out = append(out, k) + } + return strings.Join(out, " -> ") + } + + isStoreMu := func(x ast.Expr, recv string) bool { + s := lockScanExpr(x) + return s == "s.store.mu" || s == "store.mu" || (s == "s.mu" && recv == "PacketStore") + } + isUnlock := func(st ast.Stmt) bool { + es, ok := st.(*ast.ExprStmt) + if !ok { + return false + } + c, ok := es.X.(*ast.CallExpr) + if !ok { + return false + } + s := lockScanExpr(c.Fun) + return strings.HasSuffix(s, "mu.RUnlock") || strings.HasSuffix(s, "mu.Unlock") + } + + for k, f := range funcs { + if sqlUnderLockAllowedHolders[k] { + continue + } + var spans [][2]token.Pos + ast.Inspect(f.decl.Body, func(n ast.Node) bool { + b, ok := n.(*ast.BlockStmt) + if !ok { + return true + } + for i, st := range b.List { + es, ok := st.(*ast.ExprStmt) + if !ok { + continue + } + c, ok := es.X.(*ast.CallExpr) + if !ok { + continue + } + se, ok := c.Fun.(*ast.SelectorExpr) + if !ok || (se.Sel.Name != "RLock" && se.Sel.Name != "Lock") || !isStoreMu(se.X, f.recv) { + continue + } + end := b.End() // a deferred unlock holds it to the end of the block + for _, st2 := range b.List[i+1:] { + if isUnlock(st2) { + end = st2.Pos() + break + } + } + spans = append(spans, [2]token.Pos{c.Pos(), end}) + } + return true + }) + regions += len(spans) + ast.Inspect(f.decl.Body, func(n ast.Node) bool { + c, ok := n.(*ast.CallExpr) + if !ok { + return true + } + inside := false + for _, sp := range spans { + if c.Pos() > sp[0] && c.Pos() < sp[1] { + inside = true + } + } + if !inside { + return true + } + if se, ok := c.Fun.(*ast.SelectorExpr); ok && sqlMethodNames[se.Sel.Name] && strings.HasSuffix(lockScanExpr(se.X), "conn") { + hits = append(hits, fset.Position(c.Pos()).String()+" "+k+" queries directly") + return true + } + if callee := resolve(c, f.recv); callee != "" { + if _, ok := via[callee]; ok { + hits = append(hits, fset.Position(c.Pos()).String()+" "+k+" calls "+chain(callee)) + } + } + return true + }) + } + sort.Strings(hits) + return hits, regions +} diff --git a/cmd/server/store_memory_diagnostics_test.go b/cmd/server/store_memory_diagnostics_test.go index f7829cea1..dd9ac5325 100644 --- a/cmd/server/store_memory_diagnostics_test.go +++ b/cmd/server/store_memory_diagnostics_test.go @@ -164,9 +164,7 @@ func TestObsRawHexNotRetainedOnLoad(t *testing.T) { } // The read path must still serve raw_hex via the parent-tx fallback. - store.mu.RLock() - m := store.enrichObs(obs) - store.mu.RUnlock() + m := store.GetPacketByID(obs.ID) rh, _ := m["raw_hex"].(string) if rh != txHex { t.Errorf("enrichObs raw_hex: want parent tx fallback %q, got %q", txHex, rh) diff --git a/cmd/server/store_sql_unlocked_412_test.go b/cmd/server/store_sql_unlocked_412_test.go new file mode 100644 index 000000000..b15984a6c --- /dev/null +++ b/cmd/server/store_sql_unlocked_412_test.go @@ -0,0 +1,335 @@ +package main + +import ( + "context" + "fmt" + "runtime" + "sort" + "testing" + "time" +) + +// Exercise the real SQL boundary: setupTestDB has exactly one connection. +// Keeping it checked out makes the next query wait in database/sql rather +// than relying on a sleep or on a test double which never executes the query. +func TestStoreSQLDoesNotHoldPacketLock412(t *testing.T) { + for _, name := range []string{"hash", "transmission", "observation", "observations", "packets", "expanded", "multi-node", "grouped", "node-filter", "region-cold", "region-expired", "area-cold", "area-expired", "nodes-cold", "nodes-expired", "health", "bulk-health", "subpaths", "channels"} { + t.Run(name, func(t *testing.T) { + srv, _ := setupTestServer(t) + txID := seedConfirmTx(t, srv, "sql_unlocked_412", `["aa","bb"]`, `["aabbccdd11223344","eeff00112233aabb"]`) + store := reloadConfirmStore(t, srv) + resetLRU(store) + store.mu.RLock() + obsID := store.byTxID[txID].Observations[0].ID + store.mu.RUnlock() + var request func() + switch name { + case "hash": + request = func() { store.GetPacketByHash("sql_unlocked_412") } + case "transmission": + request = func() { store.GetTransmissionByID(txID) } + case "observation": + request = func() { store.GetPacketByID(obsID) } + case "observations": + request = func() { store.GetObservationsForHash("sql_unlocked_412") } + case "packets": + request = func() { store.QueryPackets(PacketQuery{Limit: 1}) } + case "expanded": + request = func() { store.QueryPackets(PacketQuery{Limit: 1, ExpandObservations: true}) } + case "multi-node": + request = func() { store.QueryMultiNodePackets([]string{confirmTestTarget}, 1, 0, "DESC", "", "") } + case "grouped": + request = func() { store.QueryGroupedPackets(PacketQuery{Limit: 1}) } + case "node-filter": + request = func() { store.QueryPackets(PacketQuery{Node: "TestRepeater", Limit: 1}) } + case "region-cold", "region-expired": + if name == "region-expired" { + store.resolveRegionObservers("SJC") + store.regionObsCacheTime = time.Now().Add(-time.Hour) + } + request = func() { store.QueryPackets(PacketQuery{Region: "SJC", Limit: 1}) } + case "area-cold", "area-expired": + store.config = &Config{Areas: map[string]AreaEntry{"TEST": {Label: "Test", Polygon: [][2]float64{{0, 0}, {1, 0}, {1, 1}, {0, 1}}}}} + if name == "area-expired" { + store.resolveAreaNodes("TEST") + store.areaNodeCacheTimes["TEST"] = time.Now().Add(-time.Hour) + } + request = func() { store.QueryPackets(PacketQuery{Area: "TEST", Limit: 1}) } + case "nodes-cold", "nodes-expired": + store.cacheMu.Lock() + if name == "nodes-cold" { + store.nodeCache = nil + } + store.nodeCacheTime = time.Now().Add(-time.Hour) + store.cacheMu.Unlock() + request = func() { store.computeAnalyticsTopology("", "", TimeWindow{}) } + case "health": + request = func() { store.GetNodeHealth(confirmTestTarget, time.Now()) } + case "bulk-health": + request = func() { store.GetBulkHealth(10, "", "") } + case "subpaths": + request = func() { store.computeAnalyticsSubpaths("SJC", 2, 8, 10) } + case "channels": + request = func() { store.GetChannelMessages("public", 10, 0, "SJC") } + } + connection, err := srv.db.conn.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + // Release SQL before joining the request even when an assertion fails. + done := make(chan struct{}) + defer func() { + connection.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("request did not finish after releasing SQL") + } + }() + waitCount := srv.db.conn.Stats().WaitCount + go func() { defer close(done); request() }() + deadline := time.Now().Add(3 * time.Second) + for srv.db.conn.Stats().WaitCount == waitCount { + select { + case <-done: + t.Fatal("request never attempted SQL") + default: + } + if time.Now().After(deadline) { + t.Fatal("request never reached the occupied SQL pool") + } + runtime.Gosched() + } + writer := make(chan struct{}) + go func() { store.mu.Lock(); store.mu.Unlock(); close(writer) }() + select { + case <-writer: + case <-time.After(time.Second): + t.Error("SQL wait held PacketStore.mu and blocked an unrelated writer") + } + }) + } +} + +func TestNodeCacheInvalidationWinsOverSQL412(t *testing.T) { + db := setupTestDB(t) + seedTestData(t, db) + store := NewPacketStore(db, nil) + connection, err := db.conn.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + defer func() { + connection.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("cache load did not finish") + } + }() + waits := db.conn.Stats().WaitCount + go func() { defer close(done); store.getCachedNodesAndPM() }() + deadline := time.Now().Add(3 * time.Second) + for db.conn.Stats().WaitCount == waits { + if time.Now().After(deadline) { + t.Fatal("node cache did not reach SQL") + } + runtime.Gosched() + } + store.InvalidateNodeCache() + connection.Close() + <-done + store.cacheMu.RLock() + if store.nodeCache != nil || store.nodePM != nil || !store.nodeCacheTime.IsZero() { + t.Error("invalidated SQL snapshot was republished") + } + store.cacheMu.RUnlock() + if nodes, pm := store.getCachedNodesAndPM(); len(nodes) == 0 || pm == nil { + t.Fatal("next request did not rebuild invalidated cache") + } +} + +func TestPacketDetailResolvedPathZeroAndNegativeIDs412(t *testing.T) { + srv, _ := setupTestServer(t) + for _, id := range []int{0, -5} { + hash := fmt.Sprintf("edge_id_%d", id) + if _, err := srv.db.conn.Exec(`INSERT INTO transmissions(id,raw_hex,hash,first_seen,route_type,payload_type,decoded_json) VALUES (?,'FF01',?,'2026-01-01T00:00:00Z',1,4,'{}')`, id, hash); err != nil { + t.Fatal(err) + } + if _, err := srv.db.conn.Exec(`INSERT INTO observations(transmission_id,observer_idx,path_json,timestamp,resolved_path) VALUES (?,1,'["aa","bb"]',1767225600,'["aabbccdd11223344","eeff00112233aabb"]')`, id); err != nil { + t.Fatal(err) + } + } + store := reloadConfirmStore(t, srv) + for _, id := range []int{0, -5} { + m := store.GetTransmissionByID(id) + if m == nil { + t.Fatalf("missing tx %d", id) + } + rp, _ := m["resolved_path"].([]*string) + if len(rp) != 2 || rp[0] == nil || *rp[0] != confirmTestTarget { + t.Errorf("tx %d resolved_path = %#v", id, m["resolved_path"]) + } + } +} + +func TestDistanceIndexUsesCapturedNodeSnapshot412(t *testing.T) { + late := time.Now().UTC().Add(-72 * time.Hour) + early := late.Add(-time.Hour) + s := hm288Setup(t, true, late.Format(time.RFC3339), early.Format(time.RFC3339), `["05","06"]`, `["05","06"]`) + metadata := s.snapshotPathNodes() + s.mu.RLock() + wantHops, wantPaths := len(s.distHops), len(s.distPaths) + s.mu.RUnlock() + if wantHops == 0 || wantPaths == 0 { + t.Fatal("fixture has no distance records") + } + // This is the precise warm -> invalidation -> lock interleaving. The + // helper must use the captured value, not re-read an empty global cache. + s.InvalidateNodeCache() + s.mu.Lock() + s.buildDistanceIndex(metadata) + gotHops, gotPaths := len(s.distHops), len(s.distPaths) + s.mu.Unlock() + if gotHops != wantHops || gotPaths != wantPaths { + t.Fatalf("invalidating cache lost distance facts: hops %d->%d paths %d->%d", wantHops, gotHops, wantPaths, gotPaths) + } +} + +func TestPacketDetailSnapshotSurvivesConcurrentMutation412(t *testing.T) { + srv, _ := setupTestServer(t) + txID := seedConfirmTx(t, srv, "stable_412", `["aa","bb"]`, `["aabbccdd11223344","eeff00112233aabb"]`) + store := reloadConfirmStore(t, srv) + resetLRU(store) + connection, err := srv.db.conn.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + defer func() { + connection.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("detail did not finish") + } + }() + waits := srv.db.conn.Stats().WaitCount + var result struct { + hash string + route int + observations int + } + go func() { + defer close(done) + m := store.GetTransmissionByID(txID) + result.hash, _ = m["hash"].(string) + result.route, _ = m["route_type"].(int) + result.observations = len(m["observations"].([]map[string]interface{})) + }() + deadline := time.Now().Add(3 * time.Second) + for srv.db.conn.Stats().WaitCount == waits { + if time.Now().After(deadline) { + t.Fatal("detail did not reach SQL") + } + runtime.Gosched() + } + // Mirror the hash migration and an append while the response's SQL is + // blocked. A response owns its original snapshot, not mutable tx fields. + mutated := make(chan struct{}) + go func() { + store.mu.Lock() + tx := store.byTxID[txID] + tx.Hash = "migrated_412" + route := 2 + tx.RouteType = &route + tx.Observations = append(tx.Observations, &StoreObs{ID: 999, TransmissionID: txID, PathJSON: `["cc"]`}) + store.mu.Unlock() + close(mutated) + }() + select { + case <-mutated: + case <-time.After(time.Second): + t.Fatal("detail blocks concurrent mutation") + } + connection.Close() + <-done + if result.hash != "stable_412" || result.route != 1 || result.observations != 1 { + t.Fatalf("mixed snapshot: %+v", result) + } +} + +// The benchmark measures locked snapshot overhead and allocations without a +// DB delay. Convoy safety itself is enforced by the deterministic test above. +func BenchmarkPacketDetailSnapshot412(b *testing.B) { + for _, n := range []int{30000} { + b.Run(fmt.Sprint(n), func(b *testing.B) { + s := NewPacketStore(nil, nil) + for i := 1; i <= n; i++ { + tx := &StoreTx{ID: i, Hash: fmt.Sprintf("%016x", i), FirstSeen: "2026-01-01T00:00:00Z", Observations: []*StoreObs{{ID: i, TransmissionID: i, PathJSON: `["aa","bb"]`}}} + txGetParsedPath(tx) + s.byHash[tx.Hash] = tx + s.byTxID[i] = tx + s.packets = append(s.packets, tx) + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.GetTransmissionByID(1) + } + }) + } +} + +// Real database/sql pool contention with a controlled 20 ms connection hold. +// The writer applies the same observation/header mutation as live ingest +// (applyObservation470), not merely a mutex probe. Both versions wait for the +// same SQL hold overall; writer-us metrics isolate the user-visible convoy. +func BenchmarkPacketSQLWriterDelay412(b *testing.B) { + db := setupTestDB(b) + s := NewPacketStore(db, nil) + for i := 1; i <= 30000; i++ { + tx := &StoreTx{ID: i, Hash: fmt.Sprintf("%016x", i), FirstSeen: "2026-01-01T00:00:00Z", Observations: []*StoreObs{{ID: i, TransmissionID: i, PathJSON: `["aa","bb"]`}}} + txGetParsedPath(tx) + s.byHash[tx.Hash], s.byTxID[i] = tx, tx + s.packets = append(s.packets, tx) + } + latencies := make([]float64, 0, b.N) + b.ResetTimer() + for i := 0; i < b.N; i++ { + resetLRU(s) + connection, err := db.conn.Conn(context.Background()) + if err != nil { + b.Fatal(err) + } + waits := db.conn.Stats().WaitCount + done := make(chan struct{}) + go func() { s.GetTransmissionByID(1); close(done) }() + deadline := time.Now().Add(3 * time.Second) + for db.conn.Stats().WaitCount == waits { + if time.Now().After(deadline) { + connection.Close() + b.Fatal("reader did not reach SQL") + } + runtime.Gosched() + } + timer := time.AfterFunc(20*time.Millisecond, func() { connection.Close() }) + written := make(chan struct{}) + started := time.Now() + go func() { + applyObservation470(s, s.byTxID[1], &StoreObs{ID: 30001 + i, TransmissionID: 1, PathJSON: `["aa","bb"]`, Timestamp: "2026-01-01T00:00:01Z"}) + close(written) + }() + <-written + latencies = append(latencies, float64(time.Since(started).Nanoseconds())/1000) + <-done + timer.Stop() + } + b.StopTimer() + sort.Float64s(latencies) + if len(latencies) > 0 { + b.ReportMetric(latencies[len(latencies)/2], "writer-p50-us") + b.ReportMetric(latencies[(len(latencies)-1)*95/100], "writer-p95-us") + } +}