diff --git a/cmd/server/neighbor_graph.go b/cmd/server/neighbor_graph.go index 19ecc63b..9e407dfc 100644 --- a/cmd/server/neighbor_graph.go +++ b/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) diff --git a/cmd/server/observer_analytics.go b/cmd/server/observer_analytics.go index 005cf70d..b4994b2d 100644 --- a/cmd/server/observer_analytics.go +++ b/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, + } +} diff --git a/cmd/server/packet_scope_name_test.go b/cmd/server/packet_scope_name_test.go index cbf920b7..27f6ae2c 100644 --- a/cmd/server/packet_scope_name_test.go +++ b/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 { diff --git a/cmd/server/routes.go b/cmd/server/routes.go index b215c960..f84adf2a 100644 --- a/cmd/server/routes.go +++ b/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}) diff --git a/cmd/server/store.go b/cmd/server/store.go index 9915e35d..366cbf8c 100644 --- a/cmd/server/store.go +++ b/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 } diff --git a/cmd/server/store_read_race_2237_test.go b/cmd/server/store_read_race_2237_test.go new file mode 100644 index 00000000..6334ddf2 --- /dev/null +++ b/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) + } + }) + } +}