fix(server): read StoreTx state under s.mu in grouped packets and other readers (#2237) (#2246)

Fixes #2237.

Ingest appends to `tx.Observations` and rewrites a transmission's header
fields (`LatestSeen`, observer, path, counts) under `s.mu.Lock`. Several
readers touched that state without the lock. This PR follows the
approach of dborup/CoreScope#475, the reporter's fix: copy what each
reader needs under the RLock it already takes, and sort or encode
outside it. That fix does not apply to upstream as is (the fork has
diverged), so this is a rewrite against current master.

## What changes, per site

| Site | Before | After |
|---|---|---|
| `QueryGroupedPackets` (`store.go`) | Sort read `tx.LatestSeen` after
`RUnlock`; the page (`groupedTxsToPage`, `storeTxDistinctIatas`) was
built from the shared cache with no lock | The sort key is copied under
the existing RLock (`groupedEntry`). The cache holds those entries.
`groupedPage` copies only the requested page's rows under a short RLock;
the maps are built after it |
| `txGetParsedPath` (`store.go`) | Filled `tx.parsedPath` lazily under
RLock, so concurrent readers wrote the same fields |
`pickBestObservation` fills it through `setTxPath` where the path is
written, under the write lock or before the tx is shared |
| `BuildFromStoreWithOptions` (`neighbor_graph.go`) | Ranged
`tx.Observations` after releasing the lock | Each tx's `Observations`
slice header is copied with the packet list |
| `/api/audio-lab/buckets` (`routes.go`) | Read path, observer and count
after `RUnlock` | The sort stays outside the lock (`RawHex` never
changes); the picked rows are read under a second short RLock |
| `trackedBytes` (`/api/stats`, perf stats) | Read with no lock | Read
under `s.mu` |
| Observer analytics (`routes.go`, `observer_analytics.go`) | Read
`tx.Hash` after `RUnlock`; the content-hash migration rewrites it |
`txByID` holds copies made under the existing RLock
(`txAnalyticsFields`) |

`/api/nodes/{pk}/paths` and `GetNodeHopAnalytics`, items 2 in the issue,
are not changed: on current master they already copy under the lock
(`snapshotRPObs`) or run inside it.

Appending to `tx.Observations` never changes the elements an already
copied slice header covers, so copying the header is enough there.

## Tests

`store_read_race_2237_test.go` adds one `-race` regression test per
site. Each runs an ingest-style writer under `s.mu.Lock` against the
reader:

- `TestQueryGroupedPacketsIngestRace2237` (cold and cached paths)
- `TestTxGetParsedPathReadersRace2237`
- `TestBuildFromStoreIngestRace2237`
- `TestAudioLabBucketsIngestRace2237`
- `TestPerfStoreStatsTrackedBytesRace2237`
- `TestStatsTrackedBytesRace2237`
- `TestObserverAnalyticsHashMigrationRace2237`

Plus `TestGroupedPageReflectsHeaderAfterIngest2237` (no race detector):
header fields, `distinct_iatas`, count and order after ingest.

How I checked them:
- With the fix, `go test -race -count=3 -run 'Race2237'` passes.
- Each site reverted on its own, one at a time, makes its test report
`DATA RACE` (between 1 and 24 reports per site).
- `go test ./...` in `cmd/server` passes; `go vet` and `gofmt -l` are
clean.

## Performance

`QueryGroupedPackets` is on the packets page's hot path.
`BenchmarkQueryGroupedPackets2237`, 30K transmissions, median of 8
alternating runs per binary (Windows, noisy: single runs varied up to
2x):

| Case | Before (ns/op) | After (ns/op) | B/op before → after |
|---|---|---|---|
| cold, limit 50 | 1,015,512 | 1,022,879 (+1%) | 1.62 MB → 2.10 MB |
| cold, limit 1000 | 1,711,237 | 1,793,032 (+5%) | 2.93 MB → 3.64 MB |
| cached, limit 50 | 29,439 | 32,608 (+11%, about 3 µs) | 69 KB → 82 KB
|
| cached, limit 1000 | 606,970 | 662,204 (+9%) | 1.38 MB → 1.61 MB |

The extra cost is copying the page's rows. The lock is held for that
copy only (at most `limit` rows), never for the sort or for building the
maps.

## Not done

- `BuildFromStoreWithOptions` still calls `tx.ParsedDecoded()` outside
the lock, and `tx_rewrite_feed.go` resets `decodedOnce` /
`parsedDecoded` under the write lock. That pair is still a race. Fixing
it means parsing under the RLock on the first build, so I left it for a
follow-up rather than widen this PR.
- I ran the new tests under `-race`, not the whole package.
- No load test against a live server; backend only, no browser check.

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

cmd/server/neighbor_graph.go

@@ -276,10 +276,19 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG
 		}
 	}
 
+	// Snapshot what we need under lock: the packet list and each tx's
+	// Observations slice header. Ingest appends to tx.Observations under
+	// store.mu (#2237); appending never changes the elements a copied
+	// header covers.
+	type graphTx struct {
+		tx           *StoreTx
+		observations []*StoreObs
+	}
 	store.mu.RLock()
-	// Snapshot what we need under lock.
-	packets := make([]*StoreTx, len(store.packets))
-	copy(packets, store.packets)
+	packets := make([]graphTx, len(store.packets))
+	for i, tx := range store.packets {
+		packets[i] = graphTx{tx: tx, observations: tx.Observations}
+	}
 	store.mu.RUnlock()
 
 	// Build prefix map for candidate resolution.
@@ -300,7 +309,8 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG
 	lowerCache := make(map[string]string, 256)
 
 	// Phase 1: Extract edges from every transmission + observation.
-	for _, tx := range packets {
+	for _, p := range packets {
+		tx := p.tx
 		isAdvert := tx.PayloadType != nil && *tx.PayloadType == PayloadADVERT
 		isAnonReq := tx.PayloadType != nil && *tx.PayloadType == PayloadANON_REQ
 		// #1777: ANON_REQ's ephemeralPubKey is a full Ed25519 pubkey — the
@@ -317,7 +327,7 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG
 			fromLower = cachedToLower(lowerCache, fromNode)
 		}
 
-		for _, obs := range tx.Observations {
+		for _, obs := range p.observations {
 			path := parsePathJSON(obs.PathJSON)
 			observerPK := cachedToLower(lowerCache, obs.ObserverID)
 

cmd/server/observer_analytics.go

@@ -210,3 +210,22 @@ func buildRecentPackets(store *PacketStore, filtered []*StoreObs, limit int, txB
 	}
 	return out
 }
+
+// txAnalyticsFields returns a detached copy of the tx fields the helpers in
+// this file read (hash, raw hex, payload and route type, decoded JSON), or
+// nil for a nil tx. Caller holds store.mu: the content-hash migration and the
+// rewrite feed change tx.Hash and tx.DecodedJSON under it, and the helpers
+// run after the handler released it (#2237).
+func txAnalyticsFields(tx *StoreTx) *StoreTx {
+	if tx == nil {
+		return nil
+	}
+	return &StoreTx{
+		ID:          tx.ID,
+		Hash:        tx.Hash,
+		RawHex:      tx.RawHex,
+		PayloadType: tx.PayloadType,
+		RouteType:   tx.RouteType,
+		DecodedJSON: tx.DecodedJSON,
+	}
+}

cmd/server/packet_scope_name_test.go

@@ -133,7 +133,7 @@ func TestPacketDetailExposesScopeName(t *testing.T) {
 // --- grouped view ---
 //
 // The Packets tab defaults to "Group by Hash", which is served by a separate
-// mapper (groupedTxsToPage in the store, a dedicated query in the DB fallback).
+// mapper (groupedRowsToPage in the store, a dedicated query in the DB fallback).
 // The Scope column reads scope_name off those rows, so both paths must carry it.
 
 func TestGroupedTxsToPageCarriesScopeName(t *testing.T) {
@@ -144,7 +144,11 @@ func TestGroupedTxsToPageCarriesScopeName(t *testing.T) {
 		{ID: 2, Hash: "bb", ScopeName: &unmatched},
 		{ID: 3, Hash: "cc", ScopeName: nil},
 	}
-	res := groupedTxsToPage(txs, len(txs), 0, len(txs))
+	rows := make([]groupedRow, len(txs))
+	for i, tx := range txs {
+		rows[i] = snapshotGroupedRow(tx)
+	}
+	res := groupedRowsToPage(rows, len(txs))
 	want := []interface{}{"#belgium", "", nil}
 	for i, w := range want {
 		if got := res.Packets[i]["scope_name"]; got != w {

cmd/server/routes.go

@@ -927,7 +927,9 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) {
 		// share a single 1s-cached snapshot to amortize ReadMemStats cost.
 		var storeDataMB float64
 		if s.store != nil {
+			s.store.mu.RLock()
 			storeDataMB = s.store.trackedMemoryMB()
+			s.store.mu.RUnlock()
 		}
 		mem := s.getMemorySnapshot(storeDataMB)
 
@@ -3156,10 +3158,13 @@ func (s *Server) handleObserverAnalytics(w http.ResponseWriter, r *http.Request)
 	// used to via s.store.byTxID[...] and enrichObs() — races with those
 	// writers. Keyed by TransmissionID (not by observation index) since
 	// multiple observations can share one transmission.
+	// #2237: the values are copies (txAnalyticsFields), since the
+	// content-hash migration and the rewrite feed change tx.Hash and
+	// tx.DecodedJSON under the lock while the helpers below run without it.
 	txByID := make(map[int]*StoreTx, len(obsSnapshot))
 	for _, obs := range obsSnapshot {
 		if _, ok := txByID[obs.TransmissionID]; !ok {
-			txByID[obs.TransmissionID] = s.store.byTxID[obs.TransmissionID]
+			txByID[obs.TransmissionID] = txAnalyticsFields(s.store.byTxID[obs.TransmissionID])
 		}
 	}
 	s.store.mu.RUnlock()
@@ -3396,15 +3401,26 @@ func (s *Server) handleAudioLabBuckets(w http.ResponseWriter, r *http.Request) {
 		}
 		s.store.mu.RUnlock()
 
+		// RawHex never changes once a tx is in the store, so the sort runs
+		// outside the lock.
 		for typeName, pkts := range byType {
 			sort.Slice(pkts, func(i, j int) bool {
 				return len(pkts[i].RawHex) < len(pkts[j].RawHex)
 			})
 			count := min(8, len(pkts))
-			picked := make([]AudioLabPacket, 0, count)
-			for i := 0; i < count; i++ {
-				idx := (i * len(pkts)) / count
-				tx := pkts[idx]
+			picks := make([]*StoreTx, count)
+			for i := range picks {
+				picks[i] = pkts[(i*len(pkts))/count]
+			}
+			byType[typeName] = picks
+		}
+
+		// Ingest and the rewrite feed change the picked packets' path,
+		// observer, count and decoded JSON under s.store.mu (#2237).
+		s.store.mu.RLock()
+		for typeName, picks := range byType {
+			picked := make([]AudioLabPacket, 0, len(picks))
+			for _, tx := range picks {
 				pt := 0
 				if tx.PayloadType != nil {
 					pt = *tx.PayloadType
@@ -3422,6 +3438,7 @@ func (s *Server) handleAudioLabBuckets(w http.ResponseWriter, r *http.Request) {
 			}
 			buckets[typeName] = picked
 		}
+		s.store.mu.RUnlock()
 	}
 
 	writeJSON(w, AudioLabBucketsResponse{Buckets: buckets})

cmd/server/store.go

@@ -307,7 +307,7 @@ type PacketStore struct {
 	groupedCacheMu    sync.Mutex
 	groupedCacheKey   string
 	groupedCacheExp   time.Time
-	groupedCacheTxs   []*StoreTx // sorted by LatestSeen DESC
+	groupedCacheTxs   []groupedEntry // sorted by LatestSeen DESC
 	groupedCacheTotal int
 	// Short-lived cache for GetChannels (avoids repeated full scan + JSON unmarshal)
 	channelsCacheMu  sync.Mutex
@@ -1823,6 +1823,7 @@ func (s *PacketStore) loadBackgroundChunks() {
 // and sets it as the transmission's display observation.
 func pickBestObservation(tx *StoreTx) {
 	if len(tx.Observations) == 0 {
+		setTxPath(tx, tx.PathJSON)
 		return
 	}
 	best := tx.Observations[0]
@@ -1839,9 +1840,21 @@ func pickBestObservation(tx *StoreTx) {
 	tx.ObserverIATA = best.ObserverIATA
 	tx.SNR = best.SNR
 	tx.RSSI = best.RSSI
-	tx.PathJSON = best.PathJSON
 	tx.Direction = best.Direction
-	tx.pathParsed = false // invalidate cached parsed path
+	setTxPath(tx, best.PathJSON)
+}
+
+// setTxPath sets tx's display path and fills its parsed-path cache. It runs
+// where tx is written (s.mu held for writing, or tx not yet in the store),
+// so readers calling txGetParsedPath under s.mu.RLock find the cache full
+// instead of filling it concurrently (#2237).
+func setTxPath(tx *StoreTx, pathJSON string) {
+	if tx.pathParsed && tx.PathJSON == pathJSON {
+		return
+	}
+	tx.PathJSON = pathJSON
+	tx.parsedPath = parsePathJSON(pathJSON)
+	tx.pathParsed = true
 }
 
 // rechargeTx re-estimates tx and returns the change since it was last charged,
@@ -2167,26 +2180,30 @@ func (s *PacketStore) QueryGroupedPackets(q PacketQuery) *PacketResult {
 		cachedTxs := s.groupedCacheTxs
 		cachedTotal := s.groupedCacheTotal
 		s.groupedCacheMu.Unlock()
-		return groupedTxsToPage(cachedTxs, cachedTotal, q.Offset, q.Limit)
+		return s.groupedPage(cachedTxs, cachedTotal, q.Offset, q.Limit)
 	}
 	s.groupedCacheMu.Unlock()
 
-	// Collect StoreTx pointers under read lock; sort outside it.
+	// Collect StoreTx pointers and their sort keys under read lock; sort
+	// outside it. Ingest rewrites LatestSeen under s.mu, so the key is
+	// copied rather than read during the sort (#2237).
 	nodePK := s.resolveQueryNode(q.Node)
 	s.mu.RLock()
 	results := s.filterPackets(q, nodePK)
-	txs := make([]*StoreTx, len(results))
-	copy(txs, results)
+	txs := make([]groupedEntry, len(results))
+	for i, tx := range results {
+		txs[i] = groupedEntry{tx: tx, latest: tx.LatestSeen}
+	}
 	s.mu.RUnlock()
 
 	total := len(txs)
 
 	// Full sort by LatestSeen DESC so the cached slice supports all page offsets.
-	sort.Slice(txs, func(i, j int) bool {
-		return txs[i].LatestSeen > txs[j].LatestSeen
+	slices.SortFunc(txs, func(a, b groupedEntry) int {
+		return strings.Compare(b.latest, a.latest)
 	})
 
-	// Cache the sorted StoreTx slice (not maps) — lightweight and reusable for any page.
+	// Cache the sorted entries (not maps) — lightweight and reusable for any page.
 	s.groupedCacheMu.Lock()
 	s.groupedCacheTxs = txs
 	s.groupedCacheTotal = total
@@ -2194,7 +2211,7 @@ func (s *PacketStore) QueryGroupedPackets(q PacketQuery) *PacketResult {
 	s.groupedCacheExp = time.Now().Add(3 * time.Second)
 	s.groupedCacheMu.Unlock()
 
-	return groupedTxsToPage(txs, total, q.Offset, q.Limit)
+	return s.groupedPage(txs, total, q.Offset, q.Limit)
 }
 
 // pagePacketResult returns a window of a PacketResult without re-allocating the slice.
@@ -2210,9 +2227,30 @@ func pagePacketResult(r *PacketResult, offset, limit int) *PacketResult {
 	return &PacketResult{Packets: r.Packets[offset:end], Total: total}
 }
 
-// groupedTxsToPage builds map representations only for the requested page of sorted StoreTx pointers.
-// This avoids allocating maps for all 30K+ transmissions when only 50 are needed.
-func groupedTxsToPage(txs []*StoreTx, total, offset, limit int) *PacketResult {
+// groupedEntry is one transmission of the sorted grouped list, with the
+// LatestSeen it was sorted by.
+type groupedEntry struct {
+	tx     *StoreTx
+	latest string
+}
+
+// groupedRow is a copy of the StoreTx fields one grouped-packets row shows.
+type groupedRow struct {
+	hash, firstSeen, latest                string
+	observerID, observerName, observerIATA string
+	pathJSON, rawHex, decodedJSON          string
+	observationCount, observerCount        int
+	payloadType, routeType                 *int
+	snr, rssi                              *float64
+	scopeName                              *string
+	observations                           []*StoreObs // slice header only, see snapshotGroupedRow
+}
+
+// groupedPage builds the requested page of txs (sorted by LatestSeen DESC).
+// txs may be the shared grouped cache, and ingest rewrites the header fields
+// and appends to tx.Observations under s.mu, so the page's rows are copied
+// under s.mu.RLock; the maps are built after it is released (#2237).
+func (s *PacketStore) groupedPage(txs []groupedEntry, total, offset, limit int) *PacketResult {
 	if offset >= len(txs) {
 		return &PacketResult{Packets: []map[string]interface{}{}, Total: total}
 	}
@@ -2221,33 +2259,69 @@ func groupedTxsToPage(txs []*StoreTx, total, offset, limit int) *PacketResult {
 		end = len(txs)
 	}
 	page := txs[offset:end]
+	rows := make([]groupedRow, len(page))
+	s.mu.RLock()
+	for i, e := range page {
+		rows[i] = snapshotGroupedRow(e.tx)
+	}
+	s.mu.RUnlock()
+	return groupedRowsToPage(rows, total)
+}
 
-	packets := make([]map[string]interface{}, len(page))
-	for i, tx := range page {
+// snapshotGroupedRow copies the row fields of tx. Caller holds s.mu, or
+// owns tx. Of tx.Observations only the slice header is copied: ingest
+// appends to it, which never changes the elements the header covers.
+func snapshotGroupedRow(tx *StoreTx) groupedRow {
+	return groupedRow{
+		hash:             tx.Hash,
+		firstSeen:        tx.FirstSeen,
+		latest:           tx.LatestSeen,
+		observerID:       tx.ObserverID,
+		observerName:     tx.ObserverName,
+		observerIATA:     tx.ObserverIATA,
+		pathJSON:         tx.PathJSON,
+		rawHex:           tx.RawHex,
+		decodedJSON:      tx.DecodedJSON,
+		observationCount: tx.ObservationCount,
+		observerCount:    tx.UniqueObserverCount,
+		payloadType:      tx.PayloadType,
+		routeType:        tx.RouteType,
+		snr:              tx.SNR,
+		rssi:             tx.RSSI,
+		scopeName:        tx.ScopeName,
+		observations:     tx.Observations,
+	}
+}
+
+// groupedRowsToPage builds map representations only for the requested page of rows.
+// This avoids allocating maps for all 30K+ transmissions when only 50 are needed.
+func groupedRowsToPage(rows []groupedRow, total int) *PacketResult {
+	packets := make([]map[string]interface{}, len(rows))
+	for i, r := range rows {
 		// #1189 R2: compute distinct IATA set across all observations.
 		// Frontend uses this in the default collapsed view to show CROSS-region
 		// reception at a glance — see groupedObserverIataBadgesHtml in
 		// public/packets.js.
-		distinctIatas := storeTxDistinctIatas(tx)
+		distinctIatas := distinctObserverIatas(r.observerIATA, r.observations)
 		m := map[string]interface{}{
-			"hash":              strOrNil(tx.Hash),
-			"first_seen":        strOrNil(tx.FirstSeen),
-			"count":             tx.ObservationCount,
-			"observer_count":    tx.UniqueObserverCount,
-			"observation_count": tx.ObservationCount,
-			"latest":            strOrNil(tx.LatestSeen),
-			"observer_id":       strOrNil(tx.ObserverID),
-			"observer_name":     strOrNil(tx.ObserverName),
-			"observer_iata":     strOrNil(tx.ObserverIATA),
+			"hash":              strOrNil(r.hash),
+			"first_seen":        strOrNil(r.firstSeen),
+			"count":             r.observationCount,
+			"observer_count":    r.observerCount,
+			"observation_count": r.observationCount,
+			"latest":            strOrNil(r.latest),
+			"observer_id":       strOrNil(r.observerID),
+			"observer_name":     strOrNil(r.observerName),
+			"observer_iata":     strOrNil(r.observerIATA),
 			"distinct_iatas":    distinctIatas,
-			"path_json":         strOrNil(tx.PathJSON),
-			"payload_type":      intPtrOrNil(tx.PayloadType),
-			"route_type":        intPtrOrNil(tx.RouteType),
-			"raw_hex":           strOrNil(tx.RawHex),
-			"decoded_json":      strOrNil(tx.DecodedJSON),
-			"snr":               floatPtrOrNil(tx.SNR),
-			"rssi":              floatPtrOrNil(tx.RSSI),
-			"scope_name":        strPtrOrNil(tx.ScopeName),
+			"path_json":         strOrNil(r.pathJSON),
+			"payload_type":      intPtrOrNil(r.payloadType),
+			"route_type":        intPtrOrNil(r.routeType),
+			"raw_hex":           strOrNil(r.rawHex),
+			"decoded_json":      strOrNil(r.decodedJSON),
+			"snr":               floatPtrOrNil(r.snr),
+			"rssi":              floatPtrOrNil(r.rssi),
+			"scope_name":        strPtrOrNil(r.scopeName),
 		}
 		// resolved_path omitted for grouped view (cold path, not worth SQL round-trip)
 		packets[i] = m
@@ -2256,22 +2330,20 @@ func groupedTxsToPage(txs []*StoreTx, total, offset, limit int) *PacketResult {
 	return &PacketResult{Packets: packets, Total: total}
 }
 
-// storeTxDistinctIatas (#1189 R2) returns a sorted, deduped list of observer
-// IATA codes for a StoreTx, excluding empty values. Returns an empty
-// (non-nil) []string when the tx has no IATA'd observations so JSON
-// serialization stays consistent across the in-memory store and SQL
-// fallback paths (db.go's parseDistinctIatasCSV does the same).
-func storeTxDistinctIatas(tx *StoreTx) []string {
-	if tx == nil {
-		return []string{}
-	}
+// distinctObserverIatas (#1189 R2) returns a sorted, deduped list of observer
+// IATA codes for a transmission, from its header observer and its
+// observations, excluding empty values. Returns an empty (non-nil) []string
+// when there are none so JSON serialization stays consistent across the
+// in-memory store and SQL fallback paths (db.go's parseDistinctIatasCSV does
+// the same).
+func distinctObserverIatas(headerIATA string, observations []*StoreObs) []string {
 	seen := make(map[string]bool)
 	// Include the header observer's IATA (some hot-path StoreTx records the
 	// chosen observer fields directly without re-populating Observations).
-	if tx.ObserverIATA != "" {
-		seen[tx.ObserverIATA] = true
+	if headerIATA != "" {
+		seen[headerIATA] = true
 	}
-	for _, o := range tx.Observations {
+	for _, o := range observations {
 		if o != nil && o.ObserverIATA != "" {
 			seen[o.ObserverIATA] = true
 		}
@@ -2422,10 +2494,10 @@ func (s *PacketStore) GetPerfStoreStats() map[string]interface{} {
 
 	// Distinct advert pubkey count — precomputed incrementally (see trackAdvertPubkey).
 	advertByObsCount := len(s.advertPubkeys)
+	trackedMB := math.Round(s.trackedMemoryMB()*10) / 10
 	s.mu.RUnlock()
 
 	estimatedMB := math.Round(s.estimatedMemoryMB()*10) / 10
-	trackedMB := math.Round(s.trackedMemoryMB()*10) / 10
 
 	evicted := atomic.LoadInt64(&s.evicted)
 
@@ -2609,6 +2681,11 @@ func (s *PacketStore) GetPerfStoreStatsTyped() PerfPacketStoreStats {
 	retentionHours := s.retentionHours
 
 	advertByObsCount := len(s.advertPubkeys)
+	trackedMB := math.Round(s.trackedMemoryMB()*10) / 10
+	var avgBytesPerPacket int64
+	if totalLoaded > 0 {
+		avgBytesPerPacket = s.trackedBytes / int64(totalLoaded)
+	}
 	s.mu.RUnlock()
 
 	s.bgErrMu.RLock()
@@ -2617,12 +2694,6 @@ func (s *PacketStore) GetPerfStoreStatsTyped() PerfPacketStoreStats {
 	s.bgErrMu.RUnlock()
 
 	estimatedMB := math.Round(s.estimatedMemoryMB()*10) / 10
-	trackedMB := math.Round(s.trackedMemoryMB()*10) / 10
-
-	var avgBytesPerPacket int64
-	if totalLoaded > 0 {
-		avgBytesPerPacket = s.trackedBytes / int64(totalLoaded)
-	}
 
 	return PerfPacketStoreStats{
 		TotalLoaded:       totalLoaded,
@@ -3549,8 +3620,8 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 				removeTxFromPathHopIndex(s.byPathHop, tx)
 				tx.parsedPath, tx.pathParsed = saved, savedFlag
 			}
-			// pickBestObservation already set pathParsed=false so
-			// addTxToPathHopIndex re-parses the new path.
+			// pickBestObservation already parsed the new path into
+			// tx.parsedPath for addTxToPathHopIndex.
 			addTxToPathHopIndex(s.byPathHop, tx)
 			// #1164: coalesce — one invalidate after the loop, not per-tx.
 			pathHopMutated = true
@@ -4318,6 +4389,9 @@ func resolvePayloadTypeName(pt *int) string {
 }
 
 // txGetParsedPath returns cached parsed path hops, parsing on first call.
+// Readers must not be the first caller (#2237): a tx's cache is filled under
+// the write lock where it is created (rechargeTx, via estimateStoreTxBytes)
+// and where its path changes (pickBestObservation, via setTxPath).
 func txGetParsedPath(tx *StoreTx) []string {
 	if tx.pathParsed {
 		return tx.parsedPath
@@ -5029,6 +5103,7 @@ func (s *PacketStore) estimatedMemoryMB() float64 {
 }
 
 // trackedMemoryMB returns the self-accounted packet store memory in MB.
+// Caller holds s.mu: ingest and eviction update trackedBytes under it.
 func (s *PacketStore) trackedMemoryMB() float64 {
 	return float64(s.trackedBytes) / 1048576.0
 }

cmd/server/store_read_race_2237_test.go

@@ -0,0 +1,267 @@
+package main
+
+import (
+	"fmt"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+)
+
+// Issue #2237: HTTP readers read StoreTx state that ingest rewrites under
+// s.mu.Lock (Observations, LatestSeen, the header observer/path fields,
+// the parsed-path cache, trackedBytes, and the hash migration's tx.Hash)
+// without holding s.mu. Each test runs an ingest-style writer concurrently
+// with one reader; they only fail under `go test -race`.
+
+// ingestObs2237 applies one ingest step to tx the way IngestNewObservations
+// does: append an observation, advance LatestSeen, re-pick the best
+// observation (the path grows each step, so the header path changes) and
+// charge trackedBytes, all under s.mu.Lock.
+func ingestObs2237(s *PacketStore, tx *StoreTx, n int) {
+	hops := make([]string, n%8+4)
+	for i := range hops {
+		hops[i] = fmt.Sprintf(`"%02x"`, (n+i)%256)
+	}
+	obs := &StoreObs{
+		ID:             10_000_000 + n,
+		TransmissionID: tx.ID,
+		ObserverID:     fmt.Sprintf("race-obs-%d", n),
+		ObserverName:   "race",
+		ObserverIATA:   fmt.Sprintf("I%02d", n%50),
+		PathJSON:       "[" + strings.Join(hops, ",") + "]",
+		Timestamp:      time.Now().UTC().Add(time.Duration(n) * time.Millisecond).Format(time.RFC3339Nano),
+	}
+	s.mu.Lock()
+	tx.addObservation(obs)
+	if obs.Timestamp > tx.LatestSeen {
+		tx.LatestSeen = obs.Timestamp
+	}
+	pickBestObservation(tx)
+	s.trackedBytes += estimateStoreObsBytes(obs)
+	s.mu.Unlock()
+}
+
+// runIngestAgainst2237 runs steps ingest steps over txs in a goroutine and
+// calls read until the writer is done.
+func runIngestAgainst2237(t *testing.T, s *PacketStore, txs []*StoreTx, steps int, read func(i int)) {
+	t.Helper()
+	runAgainst2237(steps, func(i int) { ingestObs2237(s, txs[i%len(txs)], i) }, read)
+}
+
+// runAgainst2237 calls write(0..steps-1) in a goroutine and read in the
+// caller until the writer is done. The writer starts after the first read,
+// and read runs at least minReads2237 times, so the two always overlap.
+func runAgainst2237(steps int, write, read func(i int)) {
+	read(0)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := 0; i < steps; i++ {
+			write(i)
+		}
+	}()
+	for i := 1; ; i++ {
+		if i > minReads2237 {
+			select {
+			case <-done:
+				return
+			default:
+			}
+		}
+		read(i)
+	}
+}
+
+const minReads2237 = 20
+
+func makeRaceStore2237(n int) *PacketStore {
+	s := makeTestStore(n, time.Now().Add(-time.Hour), 1)
+	for _, tx := range s.packets {
+		tx.LatestSeen = tx.FirstSeen
+		tx.RawHex = "0102030405"
+		tx.ObserverID = tx.Observations[0].ObserverID
+	}
+	return s
+}
+
+func TestQueryGroupedPacketsIngestRace2237(t *testing.T) {
+	s := makeRaceStore2237(20)
+	runIngestAgainst2237(t, s, s.packets, 300, func(i int) {
+		if i%2 == 0 {
+			// Alternate the cold path (filter + sort) with the warm cache hit.
+			s.groupedCacheMu.Lock()
+			s.groupedCacheTxs = nil
+			s.groupedCacheMu.Unlock()
+		}
+		res := s.QueryGroupedPackets(PacketQuery{Limit: 50})
+		if res.Total != 20 {
+			t.Errorf("total = %d, want 20", res.Total)
+		}
+	})
+}
+
+// TestGroupedPageReflectsHeaderAfterIngest2237 pins that a row still shows
+// the header fields pickBestObservation chose, and the distinct IATAs of
+// every observation, after the snapshot refactor.
+func TestGroupedPageReflectsHeaderAfterIngest2237(t *testing.T) {
+	s := makeRaceStore2237(3)
+	tx := s.packets[2]
+	ingestObs2237(s, tx, 5) // 9 hops, I05
+	res := s.QueryGroupedPackets(PacketQuery{Limit: 50})
+	var row map[string]interface{}
+	for _, p := range res.Packets {
+		if p["hash"] == tx.Hash {
+			row = p
+		}
+	}
+	if row == nil {
+		t.Fatalf("tx %s missing from page", tx.Hash)
+	}
+	if row["observer_id"] != "race-obs-5" {
+		t.Errorf("observer_id = %v, want race-obs-5", row["observer_id"])
+	}
+	if row["path_json"] != tx.PathJSON || row["observation_count"] != 3 || row["latest"] != tx.LatestSeen {
+		t.Errorf("row = %v, want path %s, 3 observations, latest %s", row, tx.PathJSON, tx.LatestSeen)
+	}
+	if got := fmt.Sprint(row["distinct_iatas"]); got != "[I05]" {
+		t.Errorf("distinct_iatas = %s, want [I05]", got)
+	}
+	if res.Packets[0]["hash"] != tx.Hash {
+		t.Errorf("first row = %v, want the most recently heard tx %s", res.Packets[0]["hash"], tx.Hash)
+	}
+}
+
+// Two readers under s.mu.RLock call txGetParsedPath while ingest changes
+// the best path: a lazy parse there is two readers writing tx.parsedPath.
+func TestTxGetParsedPathReadersRace2237(t *testing.T) {
+	s := makeRaceStore2237(1)
+	tx := s.packets[0]
+	readParsed := func() []string {
+		s.mu.RLock()
+		defer s.mu.RUnlock()
+		return txGetParsedPath(tx)
+	}
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			readParsed()
+		}
+	}()
+	runIngestAgainst2237(t, s, s.packets, 300, func(int) { readParsed() })
+	close(stop)
+	wg.Wait()
+	if got, want := fmt.Sprint(readParsed()), fmt.Sprint(parsePathJSON(tx.PathJSON)); got != want {
+		t.Fatalf("parsed path %s, want %s", got, want)
+	}
+}
+
+func TestBuildFromStoreIngestRace2237(t *testing.T) {
+	var txs []*StoreTx
+	for i := 0; i < 10; i++ {
+		txs = append(txs, ngMakeTx(i+1, 4, ngFromNodeJSON("aaaa1111"), []*StoreObs{
+			ngMakeObs("obs00001", `["bbbb"]`, time.Now().UTC().Format(time.RFC3339), ngFloatPtr(-10)),
+		}))
+	}
+	s := ngTestStore(nil, txs)
+	runIngestAgainst2237(t, s, txs, 300, func(int) {
+		BuildFromStoreWithOptions(s, BuildOptions{MaxEdgeKm: DefaultMaxEdgeKm})
+	})
+}
+
+func TestAudioLabBucketsIngestRace2237(t *testing.T) {
+	s := makeRaceStore2237(20)
+	srv := &Server{store: s}
+	runIngestAgainst2237(t, s, s.packets, 300, func(int) {
+		w := httptest.NewRecorder()
+		srv.handleAudioLabBuckets(w, httptest.NewRequest("GET", "/api/audio-lab/buckets", nil))
+		if w.Code != 200 {
+			t.Errorf("status %d", w.Code)
+		}
+	})
+}
+
+func TestPerfStoreStatsTrackedBytesRace2237(t *testing.T) {
+	s := makeRaceStore2237(5)
+	runIngestAgainst2237(t, s, s.packets, 300, func(int) {
+		_ = s.GetPerfStoreStats()
+		_ = s.GetPerfStoreStatsTyped()
+	})
+}
+
+func TestStatsTrackedBytesRace2237(t *testing.T) {
+	srv, router := setupTestServer(t)
+	s := srv.store
+	s.mu.RLock()
+	txs := append([]*StoreTx(nil), s.packets...)
+	s.mu.RUnlock()
+	runIngestAgainst2237(t, s, txs, 300, func(int) {
+		srv.statsMu.Lock()
+		srv.statsCache = nil
+		srv.statsMu.Unlock()
+		w := httptest.NewRecorder()
+		router.ServeHTTP(w, httptest.NewRequest("GET", "/api/stats", nil))
+		if w.Code != 200 {
+			t.Errorf("status %d", w.Code)
+		}
+	})
+}
+
+// The content-hash migration (hash_migrate.go) rewrites tx.Hash under
+// s.mu.Lock while the observer analytics handler builds its response.
+func TestObserverAnalyticsHashMigrationRace2237(t *testing.T) {
+	s := makeRaceStore2237(20)
+	srv := &Server{store: s}
+	runAgainst2237(300, func(i int) {
+		s.mu.Lock()
+		s.packets[i%len(s.packets)].Hash = fmt.Sprintf("migrated%04d", i)
+		s.mu.Unlock()
+	}, func(int) {
+		req := mux.SetURLVars(httptest.NewRequest("GET", "/api/observers/obs0/analytics", nil), map[string]string{"id": "obs0"})
+		w := httptest.NewRecorder()
+		srv.handleObserverAnalytics(w, req)
+		if w.Code != 200 {
+			t.Errorf("status %d", w.Code)
+		}
+	})
+}
+
+// BenchmarkQueryGroupedPackets2237 measures the grouped packets page on a
+// 30K-transmission store: cold (filter + sort + page) and warm (3s cache hit,
+// page only).
+func BenchmarkQueryGroupedPackets2237(b *testing.B) {
+	s := makeRaceStore2237(30000)
+	for _, limit := range []int{50, 1000} {
+		q := PacketQuery{Limit: limit}
+		b.Run(fmt.Sprintf("cold/limit=%d", limit), func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				s.groupedCacheMu.Lock()
+				s.groupedCacheTxs = nil
+				s.groupedCacheMu.Unlock()
+				s.QueryGroupedPackets(q)
+			}
+		})
+		b.Run(fmt.Sprintf("warm/limit=%d", limit), func(b *testing.B) {
+			b.ReportAllocs()
+			s.QueryGroupedPackets(q)
+			for i := 0; i < b.N; i++ {
+				s.groupedCacheMu.Lock()
+				s.groupedCacheExp = time.Now().Add(time.Hour)
+				s.groupedCacheMu.Unlock()
+				s.QueryGroupedPackets(q)
+			}
+		})
+	}
+}