From 20aef204eca86dc7f1ef193cd348d19bc6e6be2b Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 11:41:47 +0200 Subject: [PATCH 1/4] test(store): observer metadata string sharing, red Relates to #489. The tests and the plumbing for a bounded sharing pool of the four observer metadata strings (observer ID, name, IATA code, direction), with the pool's lookup stubbed out: intern() returns every value as the row scan handed it over, which is exactly what master does. So the measurement harness and the benchmarks in this commit produce the "before" numbers, and every assertion about sharing, GC stability, saturation, the pool's own accounting and the per-observation charge is red here. Covered: all five StoreObs constructors (hot Load, LoadChunked, loadChunk/background fill, IngestNewFromDB, IngestNewObservations), the #215 hash merge, eviction/accounting, metadata changes, concurrency and the lock discipline (#495). Co-Authored-By: Claude Opus 5 --- cmd/server/chunked_load.go | 17 +- cmd/server/obs_strings_489_bench_test.go | 81 +++ cmd/server/obs_strings_489_data_test.go | 453 +++++++++++++ cmd/server/obs_strings_paths_489_test.go | 769 +++++++++++++++++++++++ cmd/server/obs_strings_pool.go | 184 ++++++ cmd/server/obs_strings_pool_489_test.go | 314 +++++++++ cmd/server/store.go | 178 +++--- 7 files changed, 1921 insertions(+), 75 deletions(-) create mode 100644 cmd/server/obs_strings_489_bench_test.go create mode 100644 cmd/server/obs_strings_489_data_test.go create mode 100644 cmd/server/obs_strings_paths_489_test.go create mode 100644 cmd/server/obs_strings_pool.go create mode 100644 cmd/server/obs_strings_pool_489_test.go diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index 4068b9a54..85d3d3e64 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -655,7 +655,13 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold if obsID.Valid { oid := int(obsID.Int64) - obsIDStr := nullStrVal(observerID) + // #489: one canonical copy of each observer ID, name, IATA code + // and direction, as in Load. The chunk merge holds s.mu, but the + // pool is uncontended here: a chunk brings at most a handful of + // values the pool does not already hold. + om := s.obsStrPool.internObs(nullStrVal(observerID), nullStrVal(observerName), + nullStrVal(observerIATA), nullStrVal(direction)) + obsIDStr := om.observerID obsPJ := nullStrVal(pathJSON) dup, knownObserver := tx.lookupObservation(obsIDStr, obsPJ) @@ -666,10 +672,11 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold obs := &StoreObs{ ID: oid, TransmissionID: txID, - ObserverID: obsIDStr, - ObserverName: nullStrVal(observerName), - ObserverIATA: nullStrVal(observerIATA), - Direction: nullStrVal(direction), + ObserverID: om.observerID, + ObserverName: om.observerName, + ObserverIATA: om.observerIATA, + Direction: om.direction, + sharedStrs: om.shared, SNR: nullFloatPtr(snr), RSSI: nullFloatPtr(rssi), Score: nullIntPtr(score), diff --git a/cmd/server/obs_strings_489_bench_test.go b/cmd/server/obs_strings_489_bench_test.go new file mode 100644 index 000000000..ff24503b7 --- /dev/null +++ b/cmd/server/obs_strings_489_bench_test.go @@ -0,0 +1,81 @@ +package main + +import ( + "fmt" + "testing" +) + +// Pool micro-benchmarks for #489: the steady state is a hit, which must take +// only the read lock and allocate nothing. The parallel variant is the +// contention case (background fill, live ingest and a chunk merge can all +// intern at once). +// +// go test -run '^$' -bench ObsStringPool -benchtime 2s . + +func benchPoolKeys(n int) []string { + keys := make([]string, n) + for i := range keys { + keys[i] = fmt.Sprintf("%064x", i) + } + return keys +} + +func BenchmarkObsStringPoolHit(b *testing.B) { + p := newObsStringPool() + keys := benchPoolKeys(140) + for _, k := range keys { + p.intern(k) + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + p.intern(keys[i%len(keys)]) + } +} + +func BenchmarkObsStringPoolHitParallel(b *testing.B) { + p := newObsStringPool() + keys := benchPoolKeys(140) + for _, k := range keys { + p.intern(k) + } + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + i := 0 + for pb.Next() { + p.intern(keys[i%len(keys)]) + i++ + } + }) +} + +// BenchmarkObsStringInternObs is one observation's worth of interning: the +// four fields a row scan hands over. +func BenchmarkObsStringInternObs(b *testing.B) { + p := newObsStringPool() + id, name, iata, dir := obs489ObserverID(7), obs489ObserverName(7), obs489IATA(7), "RX" + p.internObs(id, name, iata, dir) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + p.internObs(id, name, iata, dir) + } +} + +// BenchmarkObsStringPoolMiss is the cold-load case: a value the pool does not +// hold yet, which takes the write lock and clones. +func BenchmarkObsStringPoolMiss(b *testing.B) { + keys := benchPoolKeys(obsStringPoolMax) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + b.StopTimer() + p := newObsStringPool() + b.StartTimer() + for _, k := range keys { + p.intern(k) + } + } + b.ReportMetric(float64(len(keys)), "values/op") +} diff --git a/cmd/server/obs_strings_489_data_test.go b/cmd/server/obs_strings_489_data_test.go new file mode 100644 index 000000000..84465a268 --- /dev/null +++ b/cmd/server/obs_strings_489_data_test.go @@ -0,0 +1,453 @@ +package main + +import ( + "database/sql" + "encoding/json" + "fmt" + "math" + "math/rand" + "os" + "path/filepath" + "runtime" + "sort" + "strconv" + "strings" + "testing" + "time" + "unsafe" +) + +// stringDataPtr is the address of s's backing bytes. Two strings share one +// copy exactly when their data pointers are equal, which is what the sharing +// pool must achieve and what a clone must break. +func stringDataPtr(s string) uintptr { + if s == "" { + return 0 + } + return uintptr(unsafe.Pointer(unsafe.StringData(s))) +} + +// Synthetic data and measurement harness for #489 (bounded sharing of the +// observer metadata strings). It compiles against master too, so the same +// generator produces the "before" and the "after" numbers. +// +// The shape follows the large fork instance: about 65 observers and 2.5M +// observations, i.e. a handful of distinct observer IDs, names, IATA codes +// and directions repeated across hundreds of thousands of rows. Observations +// per transmission follow a log-normal distribution with the production +// median and spread (median 15, p90 ~35, p99 ~70). + +const ( + obs489Observers = 65 + obs489Median = 15.0 + obs489Sigma = 0.8 +) + +// obs489ObserverID is a 64-hex pubkey, the dominant per-observation string. +func obs489ObserverID(k int) string { + return fmt.Sprintf("%064x", uint64(k+1)*0x9e3779b97f4a7c15) +} + +// obs489ObserverName returns a station name of a realistic length (8 to 26 +// bytes), so the name term of the measurement is not a single fixed size. +func obs489ObserverName(k int) string { + towns := []string{"Nord", "Soenderbro", "Havnen", "Vestervig", "Oesterskov", "Midtby", "Lufthavn"} + kinds := []string{"Repeater", "Room", "Companion", "Sensor"} + return fmt.Sprintf("%s %s %02d", towns[k%len(towns)], kinds[(k/len(towns))%len(kinds)], k) +} + +// obs489IATA returns a 3-letter code, empty for every seventh observer (the +// fixture has observers without one). +func obs489IATA(k int) string { + if k%7 == 6 { + return "" + } + codes := []string{"CPH", "AAR", "BLL", "AAL", "ODE", "RNN", "EBJ", "SGD"} + return codes[k%len(codes)] +} + +// obs489Direction is "RX" or "TX"; a tenth of the rows carry neither. +func obs489Direction(j int) string { + switch { + case j%10 == 9: + return "" + case j%3 == 0: + return "TX" + default: + return "RX" + } +} + +type obs489Row struct { + id int + observer int // index into the observer pool + dir string + path []string + resolved []string + ts time.Time +} + +type obs489Tx struct { + id int + hash string + firstSeen time.Time + rows []obs489Row +} + +// obs489Data builds transmissions until it holds at least numObs observation +// rows, each row one observer on one path. Every row of a transmission gets a +// distinct observer/path pair, so none of them is dropped by the #411 dedup +// and the observation count is the row count. +// +// The first half of the transmissions are 3 days old and the second half one +// hour old, so a store with HotStartupHours=2 loads the recent half hot and +// the old half through the background fill, and a retention window below an +// hour evicts every one of them. +// +// Observation ids are assigned in two phases, as for #411: every +// transmission's first row gets an id before any later row, so the upsert +// path can ingest the first rows as new transmissions and the rest as +// observations of transmissions it already holds. +func obs489Data(numObs int, seed int64) []obs489Tx { + rng := rand.New(rand.NewSource(seed)) + now := time.Now().UTC().Truncate(time.Second) + var txs []obs489Tx + total, i := 0, 0 + for total < numObs { + i++ + rows := int(math.Round(math.Exp(math.Log(obs489Median) + obs489Sigma*rng.NormFloat64()))) + if rows < 1 { + rows = 1 + } + if total+rows > numObs { + rows = numObs - total + } + fs := now.Add(-time.Hour + time.Duration(total)*time.Millisecond) + if total*2 < numObs { + fs = now.Add(-72*time.Hour + time.Duration(total)*time.Millisecond) + } + tx := obs489Tx{id: i, hash: fmt.Sprintf("%016x", uint64(i)*0x2545f4914f6cdd1d), firstSeen: fs} + for j := 0; j < rows; j++ { + obs := (j + rng.Intn(obs489Observers)) % obs489Observers + hops := 1 + (i+j)%3 + path := make([]string, hops) + resolved := make([]string, hops) + for h := range path { + path[h] = acct113Hop(i, h, j) + resolved[h] = acct113PK(i + h*7 + j) + } + tx.rows = append(tx.rows, obs489Row{observer: obs, dir: obs489Direction(j), + path: path, resolved: resolved, ts: fs.Add(time.Duration(j) * time.Millisecond)}) + } + total += rows + txs = append(txs, tx) + } + id := 0 + for i := range txs { + id++ + txs[i].rows[0].id = id + } + for i := range txs { + for j := 1; j < len(txs[i].rows); j++ { + id++ + txs[i].rows[j].id = id + } + } + return txs +} + +// obs489MaxObsID is the highest observation id obs489Data assigned. +func obs489MaxObsID(txs []obs489Tx) int { + max := 0 + for _, tx := range txs { + for _, r := range tx.rows { + if r.id > max { + max = r.id + } + } + } + return max +} + +// obs489Create writes the v3 schema (observer_idx + observers join, what the +// fork instances run) and the observer pool. +func obs489Create(tb testing.TB, dbPath string) *sql.DB { + tb.Helper() + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + tb.Fatal(err) + } + for _, q := range []string{ + `CREATE TABLE transmissions (id INTEGER PRIMARY KEY, raw_hex TEXT, hash TEXT, first_seen TEXT, + route_type INTEGER, payload_type INTEGER, payload_version INTEGER, decoded_json TEXT)`, + `CREATE TABLE observations (id INTEGER PRIMARY KEY, transmission_id INTEGER, observer_idx INTEGER, + direction TEXT, snr REAL, rssi REAL, score INTEGER, path_json TEXT, + timestamp INTEGER, resolved_path TEXT)`, + `CREATE TABLE observers (rowid INTEGER PRIMARY KEY, id TEXT, name TEXT, iata TEXT)`, + `CREATE TABLE nodes (pubkey TEXT PRIMARY KEY, name TEXT, role TEXT, lat REAL, lon REAL, + last_seen TEXT, first_seen TEXT, frequency REAL)`, + `CREATE TABLE schema_version (version INTEGER)`, + `INSERT INTO schema_version (version) VALUES (3)`, + `CREATE INDEX idx_tx_first_seen ON transmissions(first_seen)`, + `CREATE INDEX idx_obs_tx ON observations(transmission_id)`, + } { + if _, err := conn.Exec(q); err != nil { + tb.Fatalf("setup: %v\n%s", err, q) + } + } + for k := 0; k < obs489Observers; k++ { + if _, err := conn.Exec(`INSERT INTO observers (rowid, id, name, iata) VALUES (?, ?, ?, ?)`, + k+1, obs489ObserverID(k), obs489ObserverName(k), obs489IATA(k)); err != nil { + tb.Fatal(err) + } + } + return conn +} + +// obs489Write inserts txs; rows selects which observation rows of each are +// written (nil: all of them). +func obs489Write(tb testing.TB, conn *sql.DB, txs []obs489Tx, withTx bool, rows func(j int) bool) { + tb.Helper() + dbtx, err := conn.Begin() + if err != nil { + tb.Fatal(err) + } + txStmt, err := dbtx.Prepare(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, ?, ?, ?, 1, 5, 1, ?)`) + if err != nil { + tb.Fatal(err) + } + obsStmt, err := dbtx.Prepare(`INSERT INTO observations (id, transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp, resolved_path) + VALUES (?, ?, ?, ?, -5.0, -90.0, 5, ?, ?, ?)`) + if err != nil { + tb.Fatal(err) + } + for _, tx := range txs { + if withTx { + decoded := fmt.Sprintf(`{"type":"GRP_TXT","channel":"#test","text":"message number %d","sender":"node%d"}`, tx.id, tx.id%50) + if _, err := txStmt.Exec(tx.id, fmt.Sprintf("1500%036d", tx.id), tx.hash, tx.firstSeen.Format(time.RFC3339), decoded); err != nil { + tb.Fatal(err) + } + } + for j, r := range tx.rows { + if rows != nil && !rows(j) { + continue + } + pj, _ := json.Marshal(r.path) + rj, _ := json.Marshal(r.resolved) + if _, err := obsStmt.Exec(r.id, tx.id, r.observer+1, r.dir, string(pj), r.ts.Unix(), string(rj)); err != nil { + tb.Fatal(err) + } + } + } + if err := dbtx.Commit(); err != nil { + tb.Fatal(err) + } +} + +// obs489DB writes txs in full to a new database file and returns its path. +func obs489DB(tb testing.TB, txs []obs489Tx) string { + tb.Helper() + dbPath := filepath.Join(tb.TempDir(), "obs489.db") + conn := obs489Create(tb, dbPath) + obs489Write(tb, conn, txs, true, nil) + conn.Close() + return dbPath +} + +// obs489StringStats is what a store's observations hold in the four observer +// metadata fields: the logical bytes (one copy per observation) against the +// bytes of the distinct values, which is what a perfect sharing pool retains. +type obs489StringStats struct { + obs int + logicalBytes int64 + distinctBytes int64 + distinctValues int + pointerDistinct int // distinct string data pointers (actual copies held) + perField map[string]int +} + +// obs489StatsOfStore walks every observation of s and returns its string +// stats. It counts distinct VALUES and distinct string data POINTERS: with no +// sharing the pointer count is one per observation, with sharing it collapses +// to the value count. +func obs489StatsOfStore(s *PacketStore) obs489StringStats { + out := obs489StringStats{perField: map[string]int{}} + values := map[string]bool{} + ptrs := map[uintptr]bool{} + fieldValues := map[string]map[string]bool{ + "observerID": {}, "observerName": {}, "observerIATA": {}, "direction": {}, + } + s.mu.RLock() + defer s.mu.RUnlock() + for _, tx := range s.packets { + for _, o := range tx.Observations { + out.obs++ + for name, v := range map[string]string{ + "observerID": o.ObserverID, "observerName": o.ObserverName, + "observerIATA": o.ObserverIATA, "direction": o.Direction, + } { + if v == "" { + continue + } + out.logicalBytes += int64(len(v)) + fieldValues[name][v] = true + if !values[v] { + values[v] = true + out.distinctBytes += int64(len(v)) + } + ptrs[stringDataPtr(v)] = true + } + } + } + out.distinctValues = len(values) + out.pointerDistinct = len(ptrs) + for name, vs := range fieldValues { + out.perField[name] = len(vs) + } + return out +} + +// TestObsStrings489_MeasureFixture reports the observer metadata strings held +// after loading test-fixtures/e2e-fixture.db. It compiles against master too. +// +// CORESCOPE_OBS489_MEASURE=fixture go test -run ObsStrings489_MeasureFixture -v +func TestObsStrings489_MeasureFixture(t *testing.T) { + if os.Getenv("CORESCOPE_OBS489_MEASURE") == "" { + t.Skip("set CORESCOPE_OBS489_MEASURE=fixture to run") + } + src := filepath.Join("..", "..", "test-fixtures", "e2e-fixture.db") + raw, err := os.ReadFile(src) + if err != nil { + t.Skipf("fixture unavailable: %v", err) + } + dbPath := filepath.Join(t.TempDir(), "fixture.db") + if err := os.WriteFile(dbPath, raw, 0o600); err != nil { + t.Fatal(err) + } + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + defer db.conn.Close() + store, heap := heapOf(func() *PacketStore { + s := NewPacketStore(db, &PacketStoreConfig{}) + // main.go's startup order: the neighbour graph is loaded before the + // packet store (Load panics otherwise, see the #1643 invariant). + s.graph.Store(loadNeighborEdgesFromDB(db.conn)) + if err := s.Load(); err != nil { + t.Fatal(err) + } + s.WaitIndexesReady(60 * time.Second) + return s + }) + st := obs489StatsOfStore(store) + t.Logf("fixture: tx=%d obs=%d heap=%.2f MiB trackedBytes=%.2f MiB", + len(store.packets), st.obs, float64(heap)/1048576, float64(store.trackedBytes)/1048576) + t.Logf("fixture: observer strings logical=%d B distinct=%d B (%d values, %.1f%% duplicate) copies=%d perField=%v", + st.logicalBytes, st.distinctBytes, st.distinctValues, + 100*float64(st.logicalBytes-st.distinctBytes)/float64(st.logicalBytes), st.pointerDistinct, st.perField) + runtime.KeepAlive(store) +} + +// TestObsStrings489_Measure loads a synthetic store for each observation count +// in CORESCOPE_OBS489_MEASURE and reports heap after load, trackedBytes, the +// retained observer-string bytes and the Load wall time. +// +// CORESCOPE_OBS489_MEASURE=250000,1000000,2500000 go test -run ObsStrings489_Measure$ -v -timeout 60m +func TestObsStrings489_Measure(t *testing.T) { + sizes := os.Getenv("CORESCOPE_OBS489_MEASURE") + if sizes == "" { + t.Skip("set CORESCOPE_OBS489_MEASURE=250000 to run") + } + for _, field := range strings.Split(sizes, ",") { + n, err := strconv.Atoi(strings.TrimSpace(field)) + if err != nil { + continue + } + txs := obs489Data(n, 489) + dist := make([]int, 0, len(txs)) + for _, tx := range txs { + dist = append(dist, len(tx.rows)) + } + sort.Ints(dist) + t.Logf("n=%d tx=%d obs/tx mean=%.1f p50=%d p90=%d p99=%d max=%d", + n, len(txs), float64(n)/float64(len(txs)), + obs411Pct(dist, .5), obs411Pct(dist, .9), obs411Pct(dist, .99), dist[len(dist)-1]) + + dbPath := obs489DB(t, txs) + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + var t0 time.Time + var loadMS int64 + store, heap := heapOf(func() *PacketStore { + s := NewPacketStore(db, &PacketStoreConfig{}) + t0 = time.Now() + if err := s.Load(); err != nil { + t.Fatal(err) + } + loadMS = time.Since(t0).Milliseconds() + s.WaitIndexesReady(10 * time.Minute) + return s + }) + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + st := obs489StatsOfStore(store) + t.Logf("n=%d heap=%.1f MiB trackedBytes=%.1f MiB sysMB=%.0f load=%d ms obs=%d", + n, float64(heap)/1048576, float64(store.trackedBytes)/1048576, + float64(ms.Sys)/1048576, loadMS, st.obs) + t.Logf("n=%d observer strings logical=%.2f MiB distinct=%d B (%d values, %.2f%% duplicate) copies=%d perField=%v", + n, float64(st.logicalBytes)/1048576, st.distinctBytes, st.distinctValues, + 100*float64(st.logicalBytes-st.distinctBytes)/float64(st.logicalBytes), st.pointerDistinct, st.perField) + runtime.KeepAlive(store) + db.conn.Close() + store = nil + runtime.GC() + } +} + +// BenchmarkLoad_Obs489 is a hot Load of obs489Bench observations with the +// production observation-per-transmission distribution, plus heap after load +// as a custom metric. It compiles against master and against the red commit, +// so the same benchmark gives the before and after numbers. +// +// go test -run '^$' -bench Load_Obs489 -benchtime 3x -count 6 . +const obs489Bench = 250000 + +func BenchmarkLoad_Obs489(b *testing.B) { + dbPath := obs489DB(b, obs489Data(obs489Bench, 489)) + db, err := OpenDB(dbPath) + if err != nil { + b.Fatal(err) + } + defer db.conn.Close() + b.ReportAllocs() + b.ResetTimer() + var heap, tracked int64 + for i := 0; i < b.N; i++ { + b.StopTimer() + runtime.GC() + runtime.GC() + var m0 runtime.MemStats + runtime.ReadMemStats(&m0) + b.StartTimer() + s := NewPacketStore(db, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + b.Fatal(err) + } + b.StopTimer() + s.WaitIndexesReady(10 * time.Minute) + runtime.GC() + runtime.GC() + var m1 runtime.MemStats + runtime.ReadMemStats(&m1) + heap += int64(m1.HeapAlloc) - int64(m0.HeapAlloc) + tracked += s.trackedBytes + runtime.KeepAlive(s) + b.StartTimer() + } + b.ReportMetric(float64(heap)/float64(b.N)/1048576, "heapMiB") + b.ReportMetric(float64(tracked)/float64(b.N)/1048576, "trackedMiB") +} diff --git a/cmd/server/obs_strings_paths_489_test.go b/cmd/server/obs_strings_paths_489_test.go new file mode 100644 index 000000000..6a6638f36 --- /dev/null +++ b/cmd/server/obs_strings_paths_489_test.go @@ -0,0 +1,769 @@ +package main + +import ( + "fmt" + "path/filepath" + "reflect" + "runtime" + "sort" + "sync" + "testing" + "time" +) + +// Issue #489: the observer ID, name, IATA code and direction of every +// observation come from a SQL join, so every row scan allocated its own copy. +// The store now shares one canonical copy of each value through a bounded +// pool. These tests cover every construction path, the merge, eviction and +// accounting, metadata changes, GC stability and the lock discipline. + +func obs489Open(t *testing.T, dbPath string, cfg *PacketStoreConfig) *PacketStore { + t.Helper() + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.conn.Close() }) + return NewPacketStore(db, cfg) +} + +func obs489Ready(t *testing.T, s *PacketStore) *PacketStore { + t.Helper() + if !s.WaitIndexesReady(60 * time.Second) { + t.Fatal("background index builds did not finish") + } + return s +} + +func obs489IngestAll(s *PacketStore) { + for since := s.MaxTransmissionID(); ; { + _, next := s.IngestNewFromDB(since, 97) + if next <= since { + return + } + since = next + } +} + +func obs489IngestObsAll(s *PacketStore) { + for since := s.MaxObservationID(); ; { + s.IngestNewObservations(since, 101) + next := s.MaxObservationID() + if next <= since { + return + } + since = next + } +} + +type obs489Path struct { + name string + build func(t *testing.T, txs []obs489Tx) *PacketStore +} + +// obs489PathList is every path that constructs a StoreObs: the hot cold load, +// the chunked load, the background fill, live transmission ingest and live +// observation upsert. +func obs489PathList() []obs489Path { + return []obs489Path{ + {"hot Load", func(t *testing.T, txs []obs489Tx) *PacketStore { + s := obs489Open(t, obs489DB(t, txs), &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + return obs489Ready(t, s) + }}, + {"LoadChunked", func(t *testing.T, txs []obs489Tx) *PacketStore { + s := obs489Open(t, obs489DB(t, txs), &PacketStoreConfig{RetentionHours: 168}) + if err := s.RunStartupLoad(53); err != nil { + t.Fatal(err) + } + return obs489Ready(t, s) + }}, + {"background fill", func(t *testing.T, txs []obs489Tx) *PacketStore { + s := obs489Open(t, obs489DB(t, txs), &PacketStoreConfig{RetentionHours: 168, HotStartupHours: 2}) + if err := s.RunStartupLoad(53); err != nil { + t.Fatal(err) + } + return obs489Ready(t, s) + }}, + {"ingest", func(t *testing.T, txs []obs489Tx) *PacketStore { + s := obs489Open(t, obs489DB(t, txs), &PacketStoreConfig{}) + obs489IngestAll(s) + return s + }}, + {"upsert", func(t *testing.T, txs []obs489Tx) *PacketStore { + // Each transmission arrives with its first observation; the rest + // reach the store as observations of transmissions it holds. + dbPath := filepath.Join(t.TempDir(), "obs489.db") + conn := obs489Create(t, dbPath) + defer conn.Close() + obs489Write(t, conn, txs, true, func(j int) bool { return j == 0 }) + s := obs489Open(t, dbPath, &PacketStoreConfig{}) + obs489IngestAll(s) + obs489Write(t, conn, txs, false, func(j int) bool { return j > 0 }) + obs489IngestObsAll(s) + return s + }}, + } +} + +// obs489Set is a set small enough for a unit test but wide enough to repeat +// every observer many times and to push a few transmissions past the #411 +// dedup threshold. +func obs489Set() []obs489Tx { return obs489Data(4000, 489) } + +// obs489Sharing walks the store and reports, per metadata field, how many +// distinct values it holds and how many distinct copies of them, plus the +// values themselves. +type obs489Sharing struct { + obs int + values map[string]map[string]int // field → value → number of distinct copies +} + +func obs489SharingOf(s *PacketStore) obs489Sharing { + out := obs489Sharing{values: map[string]map[string]int{}} + ptrs := map[string]map[string]map[uintptr]bool{} + s.mu.RLock() + defer s.mu.RUnlock() + for _, tx := range s.packets { + for _, o := range tx.Observations { + out.obs++ + for field, v := range map[string]string{ + "observerID": o.ObserverID, "observerName": o.ObserverName, + "observerIATA": o.ObserverIATA, "direction": o.Direction, + } { + if v == "" { + continue + } + if ptrs[field] == nil { + ptrs[field] = map[string]map[uintptr]bool{} + } + if ptrs[field][v] == nil { + ptrs[field][v] = map[uintptr]bool{} + } + ptrs[field][v][stringDataPtr(v)] = true + } + } + } + for field, byValue := range ptrs { + out.values[field] = map[string]int{} + for v, set := range byValue { + out.values[field][v] = len(set) + } + } + return out +} + +// obs489WantValues is the set of non-empty metadata values the generator +// writes, per field. +func obs489WantValues(txs []obs489Tx) map[string]map[string]bool { + out := map[string]map[string]bool{ + "observerID": {}, "observerName": {}, "observerIATA": {}, "direction": {}, + } + for _, tx := range txs { + for _, r := range tx.rows { + add := func(field, v string) { + if v != "" { + out[field][v] = true + } + } + add("observerID", obs489ObserverID(r.observer)) + add("observerName", obs489ObserverName(r.observer)) + add("observerIATA", obs489IATA(r.observer)) + add("direction", r.dir) + } + } + return out +} + +// obs489CheckShared asserts that every observation carries the value its row +// wrote and that the store holds exactly one copy of each distinct value. +func obs489CheckShared(t *testing.T, s *PacketStore, want map[string]map[string]bool, where string) { + t.Helper() + got := obs489SharingOf(s) + for field, wantValues := range want { + gotValues := got.values[field] + if len(gotValues) != len(wantValues) { + t.Errorf("%s: %s holds %d distinct values, want %d", where, field, len(gotValues), len(wantValues)) + } + for v, copies := range gotValues { + if !wantValues[v] { + t.Errorf("%s: %s holds unexpected value %q", where, field, v) + } + if copies != 1 { + t.Errorf("%s: %s value %q is held in %d separate copies, want 1 shared copy", where, field, v, copies) + } + } + } + // The pool holds exactly the distinct non-empty values, no more. + distinct := map[string]bool{} + for _, values := range want { + for v := range values { + distinct[v] = true + } + } + if n := s.obsStrPool.entries(); n != len(distinct) { + t.Errorf("%s: the pool holds %d entries for %d distinct values", where, n, len(distinct)) + } +} + +func TestObsStringsSharedOnEveryConstructionPath_489(t *testing.T) { + txs := obs489Set() + want := obs489WantValues(txs) + if len(want["observerID"]) != obs489Observers { + t.Fatalf("setup: %d distinct observer IDs, want %d", len(want["observerID"]), obs489Observers) + } + // The background-fill path only exercises loadChunk when some rows sit + // outside its 2h hot window and some inside it. + old := 0 + for _, tx := range txs { + if time.Since(tx.firstSeen) > 2*time.Hour { + old++ + } + } + if old == 0 || old == len(txs) { + t.Fatalf("setup: %d of %d transmissions lie outside the 2h hot window", old, len(txs)) + } + type ref struct { + tracked int64 + pool int64 + totalObs int + keys map[string][]string + } + var first ref + for i, p := range obs489PathList() { + s := p.build(t, txs) + if got := obs489SharingOf(s).obs; got == 0 { + t.Fatalf("%s: store holds no observations", p.name) + } + obs489CheckShared(t, s, want, p.name) + acct113Check(t, s, p.name) + + // Every observation's charge must exclude the shared bytes. + s.mu.RLock() + keys := map[string][]string{} + for _, tx := range s.packets { + for _, o := range tx.Observations { + if got := ownedObsStringBytes(o); got != 0 { + t.Errorf("%s: observation %d is charged %d B for metadata the pool holds", p.name, o.ID, got) + } + keys[tx.Hash] = append(keys[tx.Hash], fmt.Sprintf("%s|%s|%s|%s|%s", + o.ObserverID, o.ObserverName, o.ObserverIATA, o.Direction, o.PathJSON)) + } + sort.Strings(keys[tx.Hash]) + } + cur := ref{tracked: s.trackedBytes, pool: s.obsStrPool.retainedBytes(), totalObs: s.totalObs, keys: keys} + s.mu.RUnlock() + + if cur.pool <= 0 { + t.Errorf("%s: the pool reports no retention", p.name) + } + if i == 0 { + first = cur + continue + } + if !reflect.DeepEqual(cur.keys, first.keys) { + t.Errorf("%s: the observations differ from hot Load", p.name) + } + if cur.tracked != first.tracked || cur.totalObs != first.totalObs || cur.pool != first.pool { + t.Errorf("%s: trackedBytes/totalObs/pool = %d/%d/%d, hot Load %d/%d/%d", + p.name, cur.tracked, cur.totalObs, cur.pool, first.tracked, first.totalObs, first.pool) + } + } +} + +// trackedBytes must hold the actual retention, not one logical copy of the +// metadata per observation, and the pool's own retention must be accounted +// for on top of it. +func TestObsStringsTrackedBytesHoldsSharedBytesOnce_489(t *testing.T) { + txs := obs489Set() + s := obs489Open(t, obs489DB(t, txs), &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + obs489Ready(t, s) + + s.mu.RLock() + defer s.mu.RUnlock() + // What the same store would be charged if every observation owned its + // own copy of the four fields, i.e. the charge before #489. + var logical int64 + obsCount := 0 + for _, tx := range s.packets { + for _, o := range tx.Observations { + obsCount++ + logical += int64(len(o.ObserverID) + len(o.ObserverName) + len(o.ObserverIATA) + len(o.Direction)) + } + } + if obsCount < 1000 { + t.Fatalf("setup: only %d observations", obsCount) + } + if logical < 1<<17 { + t.Fatalf("setup: only %d B of logical metadata; the test would not show the difference", logical) + } + // What trackedBytes really holds for the metadata: only the bytes the + // observations own, i.e. nothing the pool holds for them. + var owned int64 + for _, tx := range s.packets { + for _, o := range tx.Observations { + owned += ownedObsStringBytes(o) + } + } + if owned*10 > logical { + t.Errorf("observations are charged %d B of the %d B of metadata; the shared bytes are still charged per observation", owned, logical) + } + pool := s.obsStrPool.retainedBytes() + if pool <= 0 { + t.Fatal("the pool reports no retention of its own, so nothing is accounted for the shared copies") + } + if pool*10 > logical { + t.Errorf("the pool retains %d B against %d B logical: the sharing is barely saving anything", pool, logical) + } + if got := s.accountedBytes(); got != s.trackedBytes+pool { + t.Fatalf("accountedBytes = %d, want trackedBytes %d + pool %d", got, s.trackedBytes, pool) + } + if want := float64(s.accountedBytes()) / 1048576.0; s.trackedMemoryMB() != want { + t.Fatalf("trackedMemoryMB = %f, want %f (the pool must be in the reported figure)", s.trackedMemoryMB(), want) + } + t.Logf("trackedBytes=%d pool=%d owned metadata=%d logical metadata=%d (saved %d B of charge)", + s.trackedBytes, pool, owned, logical, logical-owned-pool) +} + +// Eviction credits exactly what ingest charged: trackedBytes returns to zero +// and only the pool's bounded retention is left standing. +func TestObsStringsEvictionCreditsExactly_489(t *testing.T) { + txs := obs489Data(2000, 4890) + for _, p := range obs489PathList() { + t.Run(p.name, func(t *testing.T) { + s := p.build(t, txs) + acct113Check(t, s, "after the fill") + pool := s.obsStrPool.retainedBytes() + if pool <= 0 { + t.Fatal("setup: the pool reports no retention") + } + s.mu.Lock() + s.retentionHours = 0.5 // every row is at least an hour old + s.mu.Unlock() + for i := 0; i < 40 && len(s.packets) > 0; i++ { + if s.RunEviction() == 0 { + break + } + } + s.mu.RLock() + defer s.mu.RUnlock() + if len(s.packets) != 0 { + t.Fatalf("%d transmissions survived eviction", len(s.packets)) + } + if s.trackedBytes != 0 { + t.Errorf("trackedBytes = %d after evicting everything, want 0", s.trackedBytes) + } + if got := s.obsStrPool.retainedBytes(); got != pool { + t.Errorf("pool retention changed to %d on eviction, want the %d it held (the canonical copies outlive the observations)", got, pool) + } + if got := s.accountedBytes(); got != pool { + t.Errorf("accountedBytes = %d with an empty store, want the pool's %d", got, pool) + } + }) + } +} + +// A renamed observer is a new value: it gets its own entry, and the +// observations heard before the rename keep the name they were heard with. +func TestObsStringsMetadataChangeLeavesOlderObservations_489(t *testing.T) { + txs := obs489Data(300, 4891) + dbPath := filepath.Join(t.TempDir(), "obs489.db") + conn := obs489Create(t, dbPath) + defer conn.Close() + obs489Write(t, conn, txs, true, nil) + + s := obs489Open(t, dbPath, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + obs489Ready(t, s) + + oldName := obs489ObserverName(0) + before := map[int]string{} + var oldCopy string + s.mu.RLock() + for _, o := range s.byObsID { + if o.ObserverName == oldName { + before[o.ID] = o.ObserverName + oldCopy = o.ObserverName + } + } + s.mu.RUnlock() + if len(before) == 0 { + t.Fatal("setup: no observation carries observer 0's name") + } + entriesBefore := s.obsStrPool.entries() + + // The ingestor renames observer 0 and a new observation arrives. + const newName = "Renamed Repeater 00" + if _, err := conn.Exec(`UPDATE observers SET name = ? WHERE rowid = 1`, newName); err != nil { + t.Fatal(err) + } + nextObs := 0 + for _, tx := range txs { + for _, r := range tx.rows { + if r.id > nextObs { + nextObs = r.id + } + } + } + nextObs++ + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp, resolved_path) + VALUES (?, ?, 1, 'RX', -5.0, -90.0, 5, '["zz"]', ?, NULL)`, + nextObs, txs[0].id, time.Now().UTC().Unix()); err != nil { + t.Fatal(err) + } + obs489IngestObsAll(s) + + s.mu.RLock() + defer s.mu.RUnlock() + fresh := s.byObsID[nextObs] + if fresh == nil { + t.Fatal("the new observation was not ingested") + } + if fresh.ObserverName != newName { + t.Errorf("new observation name = %q, want %q", fresh.ObserverName, newName) + } + if fresh.sharedStrs&obsSharedObserverName == 0 { + t.Error("the new name was not shared") + } + for id, want := range before { + o := s.byObsID[id] + if o == nil { + t.Fatalf("observation %d disappeared", id) + } + if o.ObserverName != want { + t.Errorf("observation %d name changed to %q, want the %q it was heard with", id, o.ObserverName, want) + } + if !sameCopy(o.ObserverName, oldCopy) { + t.Errorf("observation %d no longer shares the old name's copy", id) + } + } + if sameCopy(fresh.ObserverName, oldCopy) { + t.Error("the renamed observer shares the old name's copy") + } + if got := s.obsStrPool.entries(); got != entriesBefore+1 { + t.Errorf("pool holds %d entries after one rename, want %d (the old name stays for the older observations)", got, entriesBefore+1) + } + acct113Check(t, s, "after the rename") +} + +// The sharing must survive garbage collections: an observation ingested after +// a GC shares the copies the cold load put in the pool. +func TestObsStringsSharingSurvivesGC_489(t *testing.T) { + txs := obs489Data(300, 4892) + dbPath := filepath.Join(t.TempDir(), "obs489.db") + conn := obs489Create(t, dbPath) + defer conn.Close() + obs489Write(t, conn, txs, true, nil) + s := obs489Open(t, dbPath, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + obs489Ready(t, s) + + var loaded *StoreObs + s.mu.RLock() + for _, o := range s.byObsID { + if o.ObserverName == obs489ObserverName(0) && o.Direction == "RX" { + loaded = o + break + } + } + s.mu.RUnlock() + if loaded == nil { + t.Fatal("setup: no RX observation by observer 0") + } + for i := 0; i < 3; i++ { + runtime.GC() + } + + nextObs := 0 + for _, tx := range txs { + for _, r := range tx.rows { + if r.id > nextObs { + nextObs = r.id + } + } + } + nextObs++ + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp, resolved_path) + VALUES (?, ?, 1, 'RX', -5.0, -90.0, 5, '["yy"]', ?, NULL)`, + nextObs, txs[0].id, time.Now().UTC().Unix()); err != nil { + t.Fatal(err) + } + obs489IngestObsAll(s) + + s.mu.RLock() + defer s.mu.RUnlock() + fresh := s.byObsID[nextObs] + if fresh == nil { + t.Fatal("the new observation was not ingested") + } + for _, f := range []struct { + name string + got, loaded string + }{ + {"observer ID", fresh.ObserverID, loaded.ObserverID}, + {"observer name", fresh.ObserverName, loaded.ObserverName}, + {"IATA", fresh.ObserverIATA, loaded.ObserverIATA}, + {"direction", fresh.Direction, loaded.Direction}, + } { + if f.got != f.loaded { + t.Fatalf("setup: %s differs (%q vs %q)", f.name, f.got, f.loaded) + } + if !sameCopy(f.got, f.loaded) { + t.Errorf("%s: the observation ingested after the GC holds a second copy", f.name) + } + } +} + +// The hash merge (#215) moves observations between transmissions. They keep +// their canonical copies and their charges, and a dropped duplicate is +// credited exactly. +func TestObsStringsHashMergeKeepsSharingAndCharges_489(t *testing.T) { + d := hm215Create(t) + d.ballast(t) + d.stale(t, 3, 3, 3) + s := hm215Open(t, d) + + before := map[string]uintptr{} + s.mu.RLock() + for _, tx := range s.packets { + for _, o := range tx.Observations { + before[o.ObserverID] = stringDataPtr(o.ObserverID) + if o.sharedStrs&(obsSharedObserverID|obsSharedObserverName|obsSharedDirection) != + obsSharedObserverID|obsSharedObserverName|obsSharedDirection { + t.Fatalf("setup: observation %d carries unshared metadata (mask %#b)", o.ID, o.sharedStrs) + } + } + } + s.mu.RUnlock() + if len(before) == 0 { + t.Fatal("setup: no observations") + } + poolBefore := s.obsStrPool.retainedBytes() + + hm215Migrate(s, 2) + acct113Check(t, s, "after the merge") + + s.mu.RLock() + defer s.mu.RUnlock() + moved := 0 + for _, tx := range s.packets { + for _, o := range tx.Observations { + moved++ + if got := stringDataPtr(o.ObserverID); got != before[o.ObserverID] { + t.Errorf("observation %d holds a new copy of %q after the merge", o.ID, o.ObserverID) + } + if got := ownedObsStringBytes(o); got != 0 { + t.Errorf("observation %d is charged %d B for shared metadata after the merge", o.ID, got) + } + } + } + if moved == 0 { + t.Fatal("no observations survived the merge") + } + if got := s.obsStrPool.retainedBytes(); got != poolBefore { + t.Errorf("the merge changed the pool's retention from %d to %d", poolBefore, got) + } +} + +// A full pool must not break ingest: the observations are correct, unshared, +// and charged for the bytes they then own. +func TestObsStringsSaturatedPoolStillIngests_489(t *testing.T) { + txs := obs489Data(600, 4893) + s := obs489Open(t, obs489DB(t, txs), &PacketStoreConfig{}) + // Saturate before the load, so every observation takes the fallback. + for i := 0; i < obsStringPoolMax; i++ { + s.obsStrPool.intern(fmt.Sprintf("filler-%07d", i)) + } + if err := s.Load(); err != nil { + t.Fatal(err) + } + obs489Ready(t, s) + + want := obs489WantValues(txs) + s.mu.RLock() + defer s.mu.RUnlock() + seen := map[string]bool{} + charged := int64(0) + for _, tx := range s.packets { + for _, o := range tx.Observations { + if !want["observerID"][o.ObserverID] { + t.Fatalf("observation %d carries observer ID %q, which no row wrote", o.ID, o.ObserverID) + } + if o.sharedStrs != 0 { + t.Fatalf("observation %d reports shared metadata (%#b) though the pool was full", o.ID, o.sharedStrs) + } + seen[o.ObserverID] = true + charged += ownedObsStringBytes(o) + } + } + if len(seen) != obs489Observers { + t.Errorf("%d distinct observers survived a full pool, want %d", len(seen), obs489Observers) + } + if charged == 0 { + t.Error("a full pool charged nothing for the bytes the observations own") + } + if s.obsStrPool.entries() != obsStringPoolMax { + t.Errorf("pool holds %d entries, want the cap %d", s.obsStrPool.entries(), obsStringPoolMax) + } + acct113Check(t, s, "after a load through a full pool") +} + +// Lock discipline (#495): the live paths intern while scanning rows, so a +// goroutine parked inside the pool does not hold s.mu. A store writer must +// get the lock while an ingest is stuck on a pool miss. +func TestObsStringPoolNeverBlocksStoreWriters_489(t *testing.T) { + txs := obs489Data(200, 4894) + dbPath := filepath.Join(t.TempDir(), "obs489.db") + conn := obs489Create(t, dbPath) + defer conn.Close() + obs489Write(t, conn, txs, true, nil) + s := obs489Open(t, dbPath, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + obs489Ready(t, s) + + nextTx := len(txs) + 1 + nextObs := obs489MaxObsID(txs) + 1 + now := time.Now().UTC() + for _, tc := range []struct { + name string + add func(t *testing.T) + run func() + }{ + {"IngestNewFromDB", func(t *testing.T) { + raw := fmt.Sprintf("1500%036d", nextTx) + if _, err := conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, ?, ?, ?, 1, 5, 1, '{"type":"GRP_TXT"}')`, + nextTx, raw, fmt.Sprintf("fresh-%d", nextTx), now.Format(time.RFC3339)); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp, resolved_path) + VALUES (?, ?, 1, 'RX', -5.0, -90.0, 5, '["aa"]', ?, NULL)`, + nextObs, nextTx, now.Unix()); err != nil { + t.Fatal(err) + } + nextTx++ + nextObs++ + }, func() { obs489IngestAll(s) }}, + {"IngestNewObservations", func(t *testing.T) { + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp, resolved_path) + VALUES (?, ?, 1, 'RX', -5.0, -90.0, 5, '["bb"]', ?, NULL)`, + nextObs, txs[0].id, now.Unix()); err != nil { + t.Fatal(err) + } + nextObs++ + }, func() { obs489IngestObsAll(s) }}, + } { + t.Run(tc.name, func(t *testing.T) { + // A value the pool does not hold yet, so the ingest must take the + // pool's write lock to intern it. + if _, err := conn.Exec(`UPDATE observers SET name = ? WHERE rowid = 1`, + "Fresh Name "+tc.name); err != nil { + t.Fatal(err) + } + tc.add(t) + + // Park every would-be interner inside the pool. + s.obsStrPool.mu.Lock() + done := make(chan struct{}) + go func() { defer close(done); tc.run() }() + + // The ingest is now either finished or stuck in the pool. Either + // way it must not be holding s.mu: a writer gets the lock. + locked := make(chan struct{}) + go func() { + s.mu.Lock() + s.mu.Unlock() + close(locked) + }() + select { + case <-locked: + case <-time.After(10 * time.Second): + s.obsStrPool.mu.Unlock() + t.Fatal("a store writer waited on s.mu while an ingest was parked in the sharing pool: the pool is being used under the store lock") + } + s.obsStrPool.mu.Unlock() + select { + case <-done: + case <-time.After(30 * time.Second): + t.Fatal("the ingest did not finish after the pool was released") + } + }) + } + acct113Check(t, s, "after the parked ingests") +} + +// Concurrency: live ingest, eviction and readers at once, under -race. +func TestObsStringsConcurrentIngestAndReaders_489(t *testing.T) { + txs := obs489Data(1500, 4895) + dbPath := filepath.Join(t.TempDir(), "obs489.db") + conn := obs489Create(t, dbPath) + defer conn.Close() + obs489Write(t, conn, txs, true, func(j int) bool { return j == 0 }) + s := obs489Open(t, dbPath, &PacketStoreConfig{}) + obs489IngestAll(s) + obs489Write(t, conn, txs, false, func(j int) bool { return j > 0 }) + + var wg sync.WaitGroup + stop := make(chan struct{}) + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + s.mu.RLock() + for _, tx := range s.packets { + for _, o := range tx.Observations { + _ = o.ObserverID + o.ObserverName + o.ObserverIATA + o.Direction + _ = ownedObsStringBytes(o) + } + } + s.mu.RUnlock() + } + }() + } + wg.Add(1) + go func() { + defer wg.Done() + s.obsStrPool.internObs(obs489ObserverID(3), "Concurrent Name", "ZZZ", "RX") + }() + obs489IngestObsAll(s) + close(stop) + wg.Wait() + + want := obs489WantValues(txs) + obs489CheckSharedAllowingExtra(t, s, want) + acct113Check(t, s, "after the concurrent run") +} + +// obs489CheckSharedAllowingExtra is obs489CheckShared without the pool-entry +// count, for a store whose pool a test also interned into directly. +func obs489CheckSharedAllowingExtra(t *testing.T, s *PacketStore, want map[string]map[string]bool) { + t.Helper() + got := obs489SharingOf(s) + for field, wantValues := range want { + for v, copies := range got.values[field] { + if !wantValues[v] { + t.Errorf("%s holds unexpected value %q", field, v) + } + if copies != 1 { + t.Errorf("%s value %q is held in %d separate copies, want 1", field, v, copies) + } + } + } +} diff --git a/cmd/server/obs_strings_pool.go b/cmd/server/obs_strings_pool.go new file mode 100644 index 000000000..71aea84c0 --- /dev/null +++ b/cmd/server/obs_strings_pool.go @@ -0,0 +1,184 @@ +package main + +import ( + "sync" + "sync/atomic" +) + +// Bounded sharing of the observer metadata strings (#489). +// +// Every observation the store holds carries its observer's ID (a 64-hex +// pubkey), name, IATA code and direction. All four come out of a SQL join, so +// a row scan allocates a fresh copy of each for every observation. The large +// fork instance holds about 2.5M observations from about 65 observers, so the +// same ~140 distinct values were stored ten million times: 207 MiB of logical +// string content against 5.3 KiB of distinct values (measured -- see +// TestObsStrings489_Measure and the PR description). +// +// obsStringPool hands out one canonical copy of each value, so equal values +// share immutable storage. It is owned by the PacketStore, so each store +// (and each test) has its own, and it has its own lock: no load or ingest +// path waits for the pool while holding s.mu. The two live paths intern +// while scanning rows, before they take s.mu at all +// (TestObsStringPoolNeverBlocksStoreWriters_489). +// +// Policy, and why: +// +// - The canonical copies are held for the life of the store. A strongly +// held copy is what makes the sharing survive a garbage collection: a +// weak table (unique.Make) hands back a second copy for a value interned +// again after a GC. +// - A new value is cloned, so a short string scanned out of a large buffer +// never pins that buffer. +// - The table is bounded (obsStringPoolMax). Past the cap a value is +// returned exactly as it came in: the observation is correct and simply +// unshared. A full pool never rejects an observation. +// - Nothing is ever replaced or mutated in place. A renamed observer is a +// different value, which gets its own entry, so observations recorded +// before the rename keep the name they were heard with. +const ( + // obsStringPoolMax bounds the table. The steady state is one entry per + // distinct value: 65 observer IDs + 65 names + 8 IATA codes + 2 + // directions = 140 on the large instance, 48 on the e2e fixture. The cap + // leaves 58x headroom for observer growth and for renames (a rename adds + // one entry, and the old name stays so observations heard before it keep + // sharing their own), and bounds the pool's own retention to 2.2 MB of + // charge (about 0.7 MB in fact, see obsStringEntryBytes). + obsStringPoolMax = 8192 + + // obsStringEntryBytes is one entry's cost on top of the canonical copy's + // own bytes: two string headers in the slot, the slot's share of the + // group control word and the table's slack. Measured on Go swiss maps + // (TestObsStringPoolRetainedBytes_489): a table filled to its capacity + // costs 80 B per entry, and the steady state -- 140 entries in a table + // sized for 256 -- costs 208 B per entry because the table's footprint + // is fixed below the hint. The higher figure is charged, so the charge + // is an upper bound at every fill: at the cap it holds 2.2 MB for about + // 0.7 MB of real retention, and over-charging can only make eviction + // trigger earlier, never later. + obsStringEntryBytes = 208 +) + +// Bits of StoreObs.sharedStrs: set when that field's value is a canonical +// copy the pool holds, i.e. when the observation does not own those bytes. +// estimateStoreObsBytes reads them, so the charge of an observation stays a +// pure function of the observation and eviction credits exactly what ingest +// charged. +const ( + obsSharedObserverID uint8 = 1 << iota + obsSharedObserverName + obsSharedObserverIATA + obsSharedDirection +) + +// obsStringPool is the store's bounded canonical table of observer metadata +// strings. A nil pool is a valid disabled pool (every value stays unshared), +// which is what a test store built as &PacketStore{} gets. +type obsStringPool struct { + mu sync.RWMutex + table map[string]string + // bytes is the pool's own retention: per entry the canonical copy's + // content plus obsStringEntryBytes. Written under mu, read without it + // by the accounting path, which runs under s.mu. + bytes atomic.Int64 + // interned counts successful shares; full counts the calls the cap sent + // back unshared. Diagnostics only (and the saturation test). + interned atomic.Int64 + full atomic.Int64 +} + +func newObsStringPool() *obsStringPool { + return &obsStringPool{table: make(map[string]string, 256)} +} + +// retainedBytes is what the pool itself holds. It only grows, and no eviction +// credits it: the canonical copies outlive the observations that share them. +func (p *obsStringPool) retainedBytes() int64 { + if p == nil { + return 0 + } + return p.bytes.Load() +} + +// entries is the number of canonical copies held. +func (p *obsStringPool) entries() int { + if p == nil { + return 0 + } + p.mu.RLock() + defer p.mu.RUnlock() + return len(p.table) +} + +// intern returns the canonical copy of s and whether it is shared. A hit +// takes only the read lock and allocates nothing. The empty string is never +// interned: it holds no bytes and costs nothing to repeat. +func (p *obsStringPool) intern(s string) (string, bool) { + if p == nil || s == "" { + return s, false + } + // RED COMMIT: the table is not consulted yet, so every value stays the + // row scan's own copy -- exactly what master does. The green commit + // replaces this body with the real lookup. + return s, false +} + +// obsStrings is one observation's observer metadata after interning, with the +// mask of the fields the pool now holds. Carried from a row scan to the +// StoreObs the scan builds. +type obsStrings struct { + observerID, observerName, observerIATA, direction string + shared uint8 +} + +// internObs interns the four observer metadata strings of one observation. +// Call it where the row is scanned -- for the live ingest paths that is +// before s.mu is taken, so the pool is never waited on under the store lock. +func (p *obsStringPool) internObs(observerID, observerName, observerIATA, direction string) obsStrings { + var out obsStrings + var shared bool + if out.observerID, shared = p.intern(observerID); shared { + out.shared |= obsSharedObserverID + } + if out.observerName, shared = p.intern(observerName); shared { + out.shared |= obsSharedObserverName + } + if out.observerIATA, shared = p.intern(observerIATA); shared { + out.shared |= obsSharedObserverIATA + } + if out.direction, shared = p.intern(direction); shared { + out.shared |= obsSharedDirection + } + return out +} + +// apply copies the interned strings and their shared mask onto obs. Used by +// the two live ingest paths, which intern before the lock and build the +// StoreObs under it. +func (m obsStrings) apply(obs *StoreObs) { + obs.ObserverID, obs.ObserverName = m.observerID, m.observerName + obs.ObserverIATA, obs.Direction = m.observerIATA, m.direction + obs.sharedStrs = m.shared +} + +// ownedObsStringBytes is the observer metadata content estimateStoreObsBytes +// charges obs for. A field the pool holds costs the observation only its +// string header, which storeObsBaseBytes already covers, so only the fields +// obs owns outright are charged. The pool's own retention is charged once, +// separately (accountedBytes). +func ownedObsStringBytes(obs *StoreObs) int64 { + var n int64 + if obs.sharedStrs&obsSharedObserverID == 0 { + n += int64(len(obs.ObserverID)) + } + if obs.sharedStrs&obsSharedObserverName == 0 { + n += int64(len(obs.ObserverName)) + } + if obs.sharedStrs&obsSharedObserverIATA == 0 { + n += int64(len(obs.ObserverIATA)) + } + if obs.sharedStrs&obsSharedDirection == 0 { + n += int64(len(obs.Direction)) + } + return n +} diff --git a/cmd/server/obs_strings_pool_489_test.go b/cmd/server/obs_strings_pool_489_test.go new file mode 100644 index 000000000..ed157b5a7 --- /dev/null +++ b/cmd/server/obs_strings_pool_489_test.go @@ -0,0 +1,314 @@ +package main + +import ( + "fmt" + "runtime" + "strings" + "sync" + "testing" + "unsafe" +) + +// Issue #489: obsStringPool is the store's bounded, concurrency-safe table of +// canonical observer metadata strings. These are its unit properties; the +// load/ingest/merge paths are covered in obs_strings_paths_489_test.go. + +// sameCopy reports whether a and b are literally the same bytes in memory, +// not merely equal strings. This is what "shared" means. +func sameCopy(a, b string) bool { return stringDataPtr(a) == stringDataPtr(b) } + +func TestObsStringPoolSharesOneCopy_489(t *testing.T) { + p := newObsStringPool() + // Two equal values with different backing arrays, as two row scans give. + a := strings.Repeat("ab", 32) + b := strings.Repeat("ab", 32) + if sameCopy(a, b) { + t.Fatal("setup: the two inputs already share one copy") + } + ca, sharedA := p.intern(a) + cb, sharedB := p.intern(b) + if !sharedA || !sharedB { + t.Fatalf("intern reported shared=%v/%v, want both shared", sharedA, sharedB) + } + if ca != a || cb != a { + t.Fatalf("intern returned %q/%q, want %q", ca, cb, a) + } + if !sameCopy(ca, cb) { + t.Fatal("the two observations hold two copies of the same value; the pool shared nothing") + } + if got := p.entries(); got != 1 { + t.Fatalf("pool holds %d entries for one distinct value, want 1", got) + } +} + +func TestObsStringPoolClonesNewValues_489(t *testing.T) { + p := newObsStringPool() + // A row scan can hand over a short string that is a slice of a much + // larger buffer. Retaining it as-is would pin the whole buffer. + const big = 1 << 20 + buf := strings.Repeat("z", big-2) + "OK" + sub := buf[big-2:] + if len(sub) != 2 { + t.Fatalf("setup: substring is %d bytes", len(sub)) + } + c, shared := p.intern(sub) + if !shared || c != "OK" { + t.Fatalf("intern(sub) = %q, %v; want \"OK\", true", c, shared) + } + if sameCopy(c, sub) { + t.Fatal("the pool retained the substring itself, pinning its 1 MiB backing buffer") + } + // The canonical copy must sit outside the big buffer entirely. + lo := stringDataPtr(buf) + if got := stringDataPtr(c); got >= lo && got < lo+uintptr(big) { + t.Fatalf("canonical copy at %#x lies inside the 1 MiB buffer at %#x", got, lo) + } + // With the copy outside the buffer, dropping the buffer frees all 1 MiB + // of it and the canonical copy still reads correctly. + buf, sub = "", "" + runtime.GC() + runtime.GC() + if c != "OK" { + t.Fatalf("canonical copy changed to %q after the buffer was dropped", c) + } + if again, _ := p.intern("OK"); !sameCopy(again, c) { + t.Fatal("the pool lost its copy when the buffer was dropped") + } +} + +func TestObsStringPoolSurvivesGC_489(t *testing.T) { + p := newObsStringPool() + first, _ := p.intern(strings.Repeat("c4", 32)) + ptr := stringDataPtr(first) + // A weakly held table (unique.Make) hands back a second copy here. + for i := 0; i < 3; i++ { + runtime.GC() + } + again, shared := p.intern(strings.Repeat("c4", 32)) + if !shared { + t.Fatal("a value interned before a GC came back unshared") + } + if stringDataPtr(again) != ptr { + t.Fatal("a value interned again after a GC came back as a second copy; the sharing does not survive collection") + } + if got := p.entries(); got != 1 { + t.Fatalf("pool holds %d entries after the GC, want 1", got) + } + runtime.KeepAlive(first) +} + +func TestObsStringPoolSaturationFallsBackToPlainStrings_489(t *testing.T) { + p := newObsStringPool() + for i := 0; i < obsStringPoolMax; i++ { + if _, shared := p.intern(fmt.Sprintf("value-%06d", i)); !shared { + t.Fatalf("value %d was not shared while the pool had room", i) + } + } + if got := p.entries(); got != obsStringPoolMax { + t.Fatalf("pool holds %d entries, want the cap %d", got, obsStringPoolMax) + } + bytesAtCap := p.retainedBytes() + + // Past the cap: the value comes back exactly as it came in, unshared. + fresh := strings.Repeat("ff", 32) + c, shared := p.intern(fresh) + if shared { + t.Fatal("a full pool reported a value as shared") + } + if c != fresh || !sameCopy(c, fresh) { + t.Fatalf("a full pool returned %q (same copy: %v), want the input unchanged", c, sameCopy(c, fresh)) + } + if got := p.entries(); got != obsStringPoolMax { + t.Fatalf("pool grew past the cap to %d entries", got) + } + if got := p.retainedBytes(); got != bytesAtCap { + t.Fatalf("a full pool retained %d more bytes", got-bytesAtCap) + } + // Values already held keep being shared after saturation. + held, shared := p.intern("value-000007") + if !shared || held != "value-000007" { + t.Fatalf("intern of a held value after saturation = %q, %v", held, shared) + } + if p.full.Load() != 1 { + t.Fatalf("full counter = %d, want 1", p.full.Load()) + } +} + +func TestObsStringPoolConcurrentCallsShareOneCopy_489(t *testing.T) { + p := newObsStringPool() + const goroutines, values = 16, 64 + got := make([][]string, goroutines) + var wg sync.WaitGroup + start := make(chan struct{}) + for g := 0; g < goroutines; g++ { + wg.Add(1) + go func(g int) { + defer wg.Done() + out := make([]string, values) + <-start + for v := 0; v < values; v++ { + // A fresh backing array per goroutine per value. + s := fmt.Sprintf("%064x", v) + c, shared := p.intern(s) + if !shared { + t.Errorf("goroutine %d: value %d unshared", g, v) + } + out[v] = c + } + got[g] = out + }(g) + } + close(start) + wg.Wait() + for v := 0; v < values; v++ { + want := got[0][v] + for g := 1; g < goroutines; g++ { + if got[g][v] != want { + t.Fatalf("goroutine %d value %d = %q, want %q", g, v, got[g][v], want) + } + if !sameCopy(got[g][v], want) { + t.Fatalf("value %d: goroutine %d holds a second copy", v, g) + } + } + } + if n := p.entries(); n != values { + t.Fatalf("pool holds %d entries for %d distinct values", n, values) + } +} + +func TestObsStringPoolHitDoesNotAllocate_489(t *testing.T) { + p := newObsStringPool() + key := strings.Repeat("7e", 32) + p.intern(key) + if n := testing.AllocsPerRun(200, func() { p.intern(key) }); n != 0 { + t.Fatalf("a pool hit allocates %.1f times per call, want 0", n) + } + m := obsStrings{} + if n := testing.AllocsPerRun(200, func() { + m = p.internObs(key, "Name", "CPH", "RX") + }); n != 0 { + t.Fatalf("internObs allocates %.1f times per call once warm, want 0", n) + } + if m.shared != obsSharedObserverID|obsSharedObserverName|obsSharedObserverIATA|obsSharedDirection { + t.Fatalf("shared mask = %#b, want all four fields shared", m.shared) + } +} + +func TestObsStringPoolEmptyAndNil_489(t *testing.T) { + var nilPool *obsStringPool + if c, shared := nilPool.intern("abc"); c != "abc" || shared { + t.Fatalf("nil pool intern = %q, %v; want the input unshared", c, shared) + } + if nilPool.retainedBytes() != 0 || nilPool.entries() != 0 { + t.Fatal("nil pool reports retention") + } + m := nilPool.internObs("a", "b", "c", "d") + if m.shared != 0 || m.observerID != "a" || m.direction != "d" { + t.Fatalf("nil pool internObs = %+v", m) + } + + p := newObsStringPool() + if c, shared := p.intern(""); c != "" || shared { + t.Fatalf("intern(\"\") = %q, %v; want the empty string unshared", c, shared) + } + if p.entries() != 0 { + t.Fatal("the empty string took an entry") + } + // An observation with no IATA and no direction is charged nothing for + // them and marks neither as shared. + m = p.internObs("id", "name", "", "") + if m.shared != obsSharedObserverID|obsSharedObserverName { + t.Fatalf("shared mask = %#b, want only ID and name", m.shared) + } + obs := &StoreObs{} + m.apply(obs) + if got := ownedObsStringBytes(obs); got != 0 { + t.Fatalf("ownedObsStringBytes = %d for a fully shared observation, want 0", got) + } +} + +// Accounting: the pool charges every entry its content plus one entry +// overhead, and that charge is an upper bound on what the table really +// retains at every fill (obsStringEntryBytes's doc comment). +func TestObsStringPoolRetainedBytes_489(t *testing.T) { + p := newObsStringPool() + var content int64 + for i := 0; i < 2000; i++ { + v := fmt.Sprintf("%064x", i) + p.intern(v) + content += int64(len(v)) + } + want := content + 2000*obsStringEntryBytes + if got := p.retainedBytes(); got != want { + t.Fatalf("retainedBytes = %d, want content %d + %d entries x %d = %d", got, content, 2000, obsStringEntryBytes, want) + } + // Re-interning held values must not charge again. + for i := 0; i < 2000; i++ { + p.intern(fmt.Sprintf("%064x", i)) + } + if got := p.retainedBytes(); got != want { + t.Fatalf("retainedBytes grew to %d on repeat interning, want %d", got, want) + } + + // Upper bound against the real thing: a map[string]string of the same + // shape, measured the way obsStringEntryBytes was. + keys := make([]string, 2000) + for i := range keys { + keys[i] = fmt.Sprintf("%064x", i) + } + var m0, m1 runtime.MemStats + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m0) + table := make(map[string]string, 256) + for _, k := range keys { + c := strings.Clone(k) + table[c] = c + } + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m1) + actual := int64(m1.HeapAlloc) - int64(m0.HeapAlloc) + runtime.KeepAlive(table) + runtime.KeepAlive(keys) + if actual <= 0 { + t.Skipf("heap measurement unusable (%d B)", actual) + } + if want < actual { + t.Fatalf("the pool charges %d B for what really costs %d B: the charge must be an upper bound", want, actual) + } + if want > 3*actual { + t.Fatalf("the pool charges %d B for %d B of real retention; obsStringEntryBytes = %d is more than 3x off", want, actual, obsStringEntryBytes) + } +} + +func TestObsStringPoolChurnStaysBounded_489(t *testing.T) { + p := newObsStringPool() + // Twice the cap in distinct values: every one of them must come back + // correct, and the table must stop at the cap. + for i := 0; i < 2*obsStringPoolMax; i++ { + v := fmt.Sprintf("rename-%07d", i) + c, _ := p.intern(v) + if c != v { + t.Fatalf("value %d came back as %q", i, c) + } + } + if got := p.entries(); got != obsStringPoolMax { + t.Fatalf("pool holds %d entries after 2x the cap of distinct values, want %d", got, obsStringPoolMax) + } + maxBytes := int64(obsStringPoolMax) * (obsStringEntryBytes + 64) + if got := p.retainedBytes(); got > maxBytes { + t.Fatalf("pool retains %d B, above the %d B the cap allows for 64-byte values", got, maxBytes) + } +} + +// StoreObs must not grow: sharedStrs uses the padding wireRoute already sits +// in (see TestStoreObsLayoutFitsWireRouteInPadding_76). +func TestStoreObsLayoutFitsSharedStrsInPadding_489(t *testing.T) { + if unsafe.Sizeof(uintptr(0)) != 8 { + t.Skip("layout pinned for 64-bit builds") + } + if got := unsafe.Sizeof(StoreObs{}); got != 200 { + t.Fatalf("unsafe.Sizeof(StoreObs{}) = %d, want 200: sharedStrs must use existing padding", got) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index ac38e8cc8..787f0f0fb 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -115,6 +115,12 @@ type StoreObs struct { // tx.RouteType. The frame itself is not kept (#881). It fills padding // after tsParsedOK, so StoreObs does not grow. wireRoute uint8 + // sharedStrs marks which of ObserverID, ObserverName, ObserverIATA and + // Direction are canonical copies the store's bounded sharing pool holds + // (#489, obs_strings_pool.go), so estimateStoreObsBytes charges only + // the bytes this observation owns. It uses the same padding block as + // wireRoute, so StoreObs stays 200 bytes. + sharedStrs uint8 } // ParsedTime returns the parsed Timestamp, caching the result. @@ -208,6 +214,7 @@ type PacketStore struct { maxTxID int // highest transmission_id in store maxObsID int // highest observation_id in store byObserver map[string][]*StoreObs // observer_id → observations + obsStrPool *obsStringPool // shared observer metadata strings (#489); nil = disabled byNode map[string][]*StoreTx // pubkey → transmissions nodeHashes map[string]map[string]bool // pubkey → Set byPathHop map[string][]*StoreTx // lowercase hop/pubkey → transmissions with that hop in path @@ -827,6 +834,7 @@ func NewPacketStore(db *DB, cfg *PacketStoreConfig, cacheTTLs ...map[string]inte byTxID: make(map[int]*StoreTx, 65536), byObsID: make(map[int]*StoreObs, 65536), byObserver: make(map[string][]*StoreObs), + obsStrPool: newObsStringPool(), byNode: make(map[string][]*StoreTx), byPathHop: make(map[string][]*StoreTx), nodeHashes: make(map[string]map[string]bool), @@ -1120,7 +1128,14 @@ func (s *PacketStore) Load() error { if obsID.Valid { oid := int(obsID.Int64) - obsIDStr := nullStrVal(observerID) + // #489: one canonical copy of each observer ID, name, IATA code + // and direction, so the observations share them instead of + // holding the row scan's own copy each. Interned before the + // dedup check so byObserver and the index keys below are the + // canonical copies too. + om := s.obsStrPool.internObs(nullStrVal(observerID), nullStrVal(observerName), + nullStrVal(observerIATA), nullStrVal(direction)) + obsIDStr := om.observerID obsPJ := nullStrVal(pathJSON) // Dedup: skip if same observer + same path already loaded @@ -1132,10 +1147,11 @@ func (s *PacketStore) Load() error { obs := &StoreObs{ ID: oid, TransmissionID: txID, - ObserverID: obsIDStr, - ObserverName: nullStrVal(observerName), - ObserverIATA: nullStrVal(observerIATA), - Direction: nullStrVal(direction), + ObserverID: om.observerID, + ObserverName: om.observerName, + ObserverIATA: om.observerIATA, + Direction: om.direction, + sharedStrs: om.shared, SNR: nullFloatPtr(snr), RSSI: nullFloatPtr(rssi), Score: nullIntPtr(score), @@ -1431,7 +1447,12 @@ func (s *PacketStore) loadChunkWhere(where string, qArgs []interface{}, label st if obsID.Valid { oid := int(obsID.Int64) - obsIDStr := nullStrVal(observerID) + // #489: shared observer metadata, as in Load. The background + // fill scans and builds its batch outside s.mu, so this never + // waits for the pool under the store lock. + om := s.obsStrPool.internObs(nullStrVal(observerID), nullStrVal(observerName), + nullStrVal(observerIATA), nullStrVal(direction)) + obsIDStr := om.observerID obsPJ := nullStrVal(pathJSON) dup, knownObserver := tx.lookupObservation(obsIDStr, obsPJ) @@ -1442,10 +1463,11 @@ func (s *PacketStore) loadChunkWhere(where string, qArgs []interface{}, label st obs := &StoreObs{ ID: oid, TransmissionID: txID, - ObserverID: obsIDStr, - ObserverName: nullStrVal(observerName), - ObserverIATA: nullStrVal(observerIATA), - Direction: nullStrVal(direction), + ObserverID: om.observerID, + ObserverName: om.observerName, + ObserverIATA: om.observerIATA, + Direction: om.direction, + sharedStrs: om.shared, SNR: nullFloatPtr(snr), RSSI: nullFloatPtr(rssi), Score: nullIntPtr(score), @@ -3011,7 +3033,12 @@ func (s *PacketStore) GetPerfStoreStatsTyped() PerfPacketStoreStats { // per field. For fields that are inline struct members (RawHex, DecodedJSON, // PathJSON, observer strings) that header is ALSO part of storeTxBaseBytes / // storeObsBaseBytes, so the per-component totals are a deliberate UPPER BOUND, -// not additive with TotalTxEstimatedMB. struct/index/map overhead, the neighbor +// not additive with TotalTxEstimatedMB. ObsStringsMB is the loosest of them +// since #489: the observer ID, name, IATA code and direction of every +// observation are canonical copies the store's bounded pool holds, so the +// bytes it counts per observation are the same few hundred values over and +// over. TrackedMB (PerfPacketStoreStats) is the figure that reflects the +// actual retention, pool included (accountedBytes). struct/index/map overhead, the neighbor // graph and analytics caches are excluded entirely — which is why the total // here sits below goHeapInuseMB. func (s *PacketStore) GetStoreMemoryBreakdown() *StoreMemoryBreakdown { @@ -3287,17 +3314,18 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac // Scan into temp structures type tempRow struct { - txID int - rawHex, hash, firstSeen, decodedJSON string - routeType, payloadType *int - obsID *int - observerID, observerName, observerIATA, direction, pathJSON, obsTS string - obsRawHex string - resolvedPath persistedRelayPath - scopeName string - routeMask sql.NullInt64 - snr, rssi *float64 - score *int + txID int + rawHex, hash, firstSeen, decodedJSON string + routeType, payloadType *int + obsID *int + strs obsStrings // interned observer metadata (#489) + pathJSON, obsTS string + obsRawHex string + resolvedPath persistedRelayPath + scopeName string + routeMask sql.NullInt64 + snr, rssi *float64 + score *int } var tempRows []tempRow @@ -3346,17 +3374,16 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac } tr := tempRow{ - txID: txID, - rawHex: nullStrVal(rawHex), - hash: nullStrVal(hash), - firstSeen: nullStrVal(firstSeen), - decodedJSON: nullStrVal(decodedJSON), - routeType: nullIntPtr(routeType), - payloadType: nullIntPtr(payloadType), - observerID: nullStrVal(observerID), - observerName: nullStrVal(observerName), - observerIATA: nullStrVal(observerIATA), - direction: nullStrVal(direction), + txID: txID, + rawHex: nullStrVal(rawHex), + hash: nullStrVal(hash), + firstSeen: nullStrVal(firstSeen), + decodedJSON: nullStrVal(decodedJSON), + routeType: nullIntPtr(routeType), + payloadType: nullIntPtr(payloadType), + // #489: interned here, while the rows are scanned, so the merge + // section below never waits for the sharing pool under s.mu. + strs: s.obsStrPool.internObs(nullStrVal(observerID), nullStrVal(observerName), nullStrVal(observerIATA), nullStrVal(direction)), pathJSON: nullStrVal(pathJSON), obsTS: nullStrVal(obsTimestamp), obsRawHex: nullStrVal(obsRawHex), @@ -3383,7 +3410,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac 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))) + ids = append(ids, strings.ToLower(strings.TrimSpace(row.strs.observerID))) } observerGPS = s.db.gpsByPubkeysExact(ids) } @@ -3456,7 +3483,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac if r.obsID != nil { oid := *r.obsID - dup, knownObserver := tx.lookupObservation(r.observerID, r.pathJSON) + dup, knownObserver := tx.lookupObservation(r.strs.observerID, r.pathJSON) if dup { continue } @@ -3464,10 +3491,6 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac obs := &StoreObs{ ID: oid, TransmissionID: r.txID, - ObserverID: r.observerID, - ObserverName: r.observerName, - ObserverIATA: r.observerIATA, - Direction: r.direction, SNR: r.snr, RSSI: r.rssi, Score: r.score, @@ -3475,15 +3498,16 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac Timestamp: normalizeTimestamp(r.obsTS), wireRoute: wireRouteFromRawHex(r.obsRawHex), } + r.strs.apply(obs) // Relay-hop indexes: same input as Load (#158 follow-up). // Decode-window discipline: decode once, feed consumers, never // store on struct. - s.indexObservationRelayHops(tx, r.resolvedPath, r.pathJSON, r.observerID, cachedPM, hopsSeen, nil) + s.indexObservationRelayHops(tx, r.resolvedPath, r.pathJSON, r.strs.observerID, cachedPM, hopsSeen, nil) // The live broadcast still shows the server's own resolution. var rpForBroadcast []*string if r.pathJSON != "" && r.pathJSON != "[]" && cachedPM != nil { - rpForBroadcast = resolvePathForObs(r.pathJSON, r.observerID, tx, cachedPM, cachedGraph) + rpForBroadcast = resolvePathForObs(r.pathJSON, r.strs.observerID, tx, cachedPM, cachedGraph) } // Stash rpForBroadcast for later broadcast/persist (keyed by obs ID) if rpForBroadcast != nil { @@ -3503,8 +3527,8 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac if oid > s.maxObsID { s.maxObsID = oid } - if r.observerID != "" { - s.byObserver[r.observerID] = append(s.byObserver[r.observerID], obs) + if r.strs.observerID != "" { + s.byObserver[r.strs.observerID] = append(s.byObserver[r.strs.observerID], obs) } s.totalObs++ s.trackedBytes += estimateStoreObsBytes(obs) @@ -3743,10 +3767,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] type obsRow struct { obsID int txID int - observerID string - observerName string - observerIATA string - direction string + strs obsStrings // interned observer metadata (#489) snr, rssi *float64 score *int pathJSON string @@ -3782,12 +3803,13 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] } obsRows = append(obsRows, obsRow{ - obsID: oid, - txID: txID, - observerID: nullStrVal(observerID), - observerName: nullStrVal(observerName), - observerIATA: nullStrVal(observerIATA), - direction: nullStrVal(direction), + obsID: oid, + txID: txID, + // #489: interned while the rows are scanned, before s.mu is + // taken below, so the merge never waits for the sharing pool + // under the store lock. + strs: s.obsStrPool.internObs(nullStrVal(observerID), nullStrVal(observerName), + nullStrVal(observerIATA), nullStrVal(direction)), snr: nullFloatPtr(snr), rssi: nullFloatPtr(rssi), score: nullIntPtr(score), @@ -3834,7 +3856,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] tx.mergeRouteMask(r.routeMask) // Dedup by observer + path - dup, knownObserver := tx.lookupObservation(r.observerID, r.pathJSON) + dup, knownObserver := tx.lookupObservation(r.strs.observerID, r.pathJSON) if dup { continue } @@ -3842,10 +3864,6 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] obs := &StoreObs{ ID: r.obsID, TransmissionID: r.txID, - ObserverID: r.observerID, - ObserverName: r.observerName, - ObserverIATA: r.observerIATA, - Direction: r.direction, SNR: r.snr, RSSI: r.rssi, Score: r.score, @@ -3856,14 +3874,15 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] Timestamp: normalizeTimestamp(r.timestamp), wireRoute: wireRouteFromRawHex(r.rawHex), } + r.strs.apply(obs) // Relay-hop indexes: same input as Load (#158 follow-up). // Decode-window discipline: decode, feed consumers, don't store on struct. - s.indexObservationRelayHops(tx, r.resolvedPath, r.pathJSON, r.observerID, pm, hopsSeen, nil) + s.indexObservationRelayHops(tx, r.resolvedPath, r.pathJSON, r.strs.observerID, pm, hopsSeen, nil) // The live broadcast still shows the server's own resolution (review item #2). var obsResolvedPath []*string if r.pathJSON != "" && r.pathJSON != "[]" && pm != nil { - obsResolvedPath = resolvePathForObs(r.pathJSON, r.observerID, tx, pm, graphRef) + obsResolvedPath = resolvePathForObs(r.pathJSON, r.strs.observerID, tx, pm, graphRef) } // Stash for broadcast/persist if obsResolvedPath != nil { @@ -3884,8 +3903,8 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] if r.obsID > s.maxObsID { s.maxObsID = r.obsID } - if r.observerID != "" { - s.byObserver[r.observerID] = append(s.byObserver[r.observerID], obs) + if r.strs.observerID != "" { + s.byObserver[r.strs.observerID] = append(s.byObserver[r.strs.observerID], obs) } s.totalObs++ s.trackedBytes += estimateStoreObsBytes(obs) @@ -5568,7 +5587,10 @@ func estimateStoreTxBytesTypical(numObs int) int64 { base += hops * perPathHopBytes base += resolvedRelayBytes(typicalResolvedRelays) // Add observation costs - obsBase := int64(storeObsBaseBytes) + 30 + 30 + 60 + 25 // observer ID + name + path + timestamp + // path + timestamp only: a typical observation's observer ID, name, IATA + // and direction are shared through the pool (#489), so they cost it + // nothing beyond the string headers storeObsBaseBytes already holds. + obsBase := int64(storeObsBaseBytes) + 60 + 25 obsBase += int64(numIndexesPerObs * indexEntryBytes) // No per-obs ResolvedPath overhead (#800) base += int64(numObs) * obsBase @@ -5579,8 +5601,11 @@ func estimateStoreTxBytesTypical(numObs int) int64 { // ResolvedPath membership index overhead is tracked separately. func estimateStoreObsBytes(obs *StoreObs) int64 { base := int64(storeObsBaseBytes) - base += int64(len(obs.PathJSON) + len(obs.ObserverID) + len(obs.ObserverName) + - len(obs.ObserverIATA) + len(obs.Direction) + len(obs.RawHex) + len(obs.Timestamp)) + base += int64(len(obs.PathJSON) + len(obs.RawHex) + len(obs.Timestamp)) + // #489: the four observer metadata fields are charged only when this + // observation owns their bytes. A value the sharing pool holds is charged + // once, by the pool itself (accountedBytes). + base += ownedObsStringBytes(obs) base += int64(numIndexesPerObs * indexEntryBytes) // ResolvedPath field removed (#800) — no per-obs RP overhead return base @@ -5609,10 +5634,21 @@ func (s *PacketStore) estimatedMemoryMB() float64 { return s.estMemVal } +// accountedBytes is everything the store's self-accounting holds: the +// per-transmission and per-observation charges (trackedBytes) plus the +// observer-string sharing pool's own retention (#489). The pool term is never +// credited by eviction -- the canonical copies outlive the observations that +// share them -- so it is a floor the memory budget must respect rather than +// part of what an eviction pass can free. +// Caller holds s.mu: ingest and eviction update trackedBytes under s.mu.Lock. +func (s *PacketStore) accountedBytes() int64 { + return s.trackedBytes + s.obsStrPool.retainedBytes() +} + // trackedMemoryMB returns the self-accounted packet store memory in MB. // Caller holds s.mu: ingest and eviction update trackedBytes under s.mu.Lock. func (s *PacketStore) trackedMemoryMB() float64 { - return float64(s.trackedBytes) / 1048576.0 + return float64(s.accountedBytes()) / 1048576.0 } // EvictStale removes packets older than the retention window and/or exceeding @@ -5634,10 +5670,11 @@ func (s *PacketStore) evictionCandidateTxIDs() []int { if s.maxMemoryMB > 0 { highWatermark := int64(s.maxMemoryMB) * 1048576 lowWatermark := int64(float64(highWatermark) * 0.85) - if s.trackedBytes > highWatermark && len(s.packets) > 0 { + accounted := s.accountedBytes() + if accounted > highWatermark && len(s.packets) > 0 { var bytesToEvict int64 memCutoff := cutoffIdx - for memCutoff < len(s.packets) && (s.trackedBytes-bytesToEvict) > lowWatermark { + for memCutoff < len(s.packets) && (accounted-bytesToEvict) > lowWatermark { tx := s.packets[memCutoff] bytesToEvict += s.txChargedBytes(tx) for _, obs := range tx.Observations { @@ -5703,11 +5740,12 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { if s.maxMemoryMB > 0 { highWatermark := int64(s.maxMemoryMB) * 1048576 lowWatermark := int64(float64(highWatermark) * 0.85) - if s.trackedBytes > highWatermark && len(s.packets) > 0 { + accounted := s.accountedBytes() + if accounted > highWatermark && len(s.packets) > 0 { // Evict from head until trackedBytes would drop below low watermark var bytesToEvict int64 memCutoff := cutoffIdx - for memCutoff < len(s.packets) && (s.trackedBytes-bytesToEvict) > lowWatermark { + for memCutoff < len(s.packets) && (accounted-bytesToEvict) > lowWatermark { tx := s.packets[memCutoff] bytesToEvict += s.txChargedBytes(tx) for _, obs := range tx.Observations { From 25472d3d594982c2455586fbcad99693b0f0f1ea Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 11:44:28 +0200 Subject: [PATCH 2/4] perf(store): bounded sharing of observer metadata strings Relates to #489. intern() now consults the bounded canonical table: a hit takes only the read lock and allocates nothing, a miss clones the value (so a short string scanned out of a large buffer cannot pin it) and keeps the copy for the life of the store, and past obsStringPoolMax entries the value is returned exactly as it came in -- unshared, never rejected. Co-Authored-By: Claude Opus 5 --- cmd/server/obs_strings_paths_489_test.go | 6 +++-- cmd/server/obs_strings_pool.go | 28 ++++++++++++++++++++---- 2 files changed, 28 insertions(+), 6 deletions(-) diff --git a/cmd/server/obs_strings_paths_489_test.go b/cmd/server/obs_strings_paths_489_test.go index 6a6638f36..7d3fdbcd3 100644 --- a/cmd/server/obs_strings_paths_489_test.go +++ b/cmd/server/obs_strings_paths_489_test.go @@ -399,8 +399,10 @@ func TestObsStringsMetadataChangeLeavesOlderObservations_489(t *testing.T) { } entriesBefore := s.obsStrPool.entries() - // The ingestor renames observer 0 and a new observation arrives. - const newName = "Renamed Repeater 00" + // The ingestor renames observer 0 and a new observation arrives. The new + // name shares a prefix with the old one, so a pool that matched values + // loosely instead of exactly would hand the old copy back here. + const newName = "Nord Repeater 00 (renamed)" if _, err := conn.Exec(`UPDATE observers SET name = ? WHERE rowid = 1`, newName); err != nil { t.Fatal(err) } diff --git a/cmd/server/obs_strings_pool.go b/cmd/server/obs_strings_pool.go index 71aea84c0..922bf4acd 100644 --- a/cmd/server/obs_strings_pool.go +++ b/cmd/server/obs_strings_pool.go @@ -1,6 +1,7 @@ package main import ( + "strings" "sync" "sync/atomic" ) @@ -117,10 +118,29 @@ func (p *obsStringPool) intern(s string) (string, bool) { if p == nil || s == "" { return s, false } - // RED COMMIT: the table is not consulted yet, so every value stays the - // row scan's own copy -- exactly what master does. The green commit - // replaces this body with the real lookup. - return s, false + p.mu.RLock() + c, ok := p.table[s] + p.mu.RUnlock() + if ok { + p.interned.Add(1) + return c, true + } + p.mu.Lock() + defer p.mu.Unlock() + if c, ok := p.table[s]; ok { + p.interned.Add(1) + return c, true + } + if len(p.table) >= obsStringPoolMax { + p.full.Add(1) + return s, false + } + // Clone so a value scanned out of a larger buffer does not pin it. + c = strings.Clone(s) + p.table[c] = c + p.bytes.Add(int64(len(c)) + obsStringEntryBytes) + p.interned.Add(1) + return c, true } // obsStrings is one observation's observer metadata after interning, with the From 5fdcf59519ce06c3257073c6d9f5a3ec6eb35846 Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 11:51:40 +0200 Subject: [PATCH 3/4] perf(store): shard the observer string pool Relates to #489. BenchmarkObsStringPoolHitParallel showed the read path, not the map, as the bottleneck under concurrency: on one shared RWMutex a hit cost 9 ns alone but 34 ns with two goroutines and 84 ns with ten, and a per-hit atomic counter made it worse still. The table is now cut into 16 independently locked, cache-line-padded shards and the per-hit counter is gone, so a hit costs 11 ns alone, 11 ns with two goroutines and 22 ns with ten. The cap stays exact: an insert reserves its slot in one shared counter before taking it. Co-Authored-By: Claude Opus 5 --- cmd/server/obs_strings_paths_489_test.go | 13 +++- cmd/server/obs_strings_pool.go | 87 +++++++++++++++++------- cmd/server/obs_strings_pool_489_test.go | 11 ++- 3 files changed, 77 insertions(+), 34 deletions(-) diff --git a/cmd/server/obs_strings_paths_489_test.go b/cmd/server/obs_strings_paths_489_test.go index 7d3fdbcd3..a75b3fd08 100644 --- a/cmd/server/obs_strings_paths_489_test.go +++ b/cmd/server/obs_strings_paths_489_test.go @@ -676,7 +676,14 @@ func TestObsStringPoolNeverBlocksStoreWriters_489(t *testing.T) { tc.add(t) // Park every would-be interner inside the pool. - s.obsStrPool.mu.Lock() + for i := range s.obsStrPool.shards { + s.obsStrPool.shards[i].mu.Lock() + } + unpark := func() { + for i := range s.obsStrPool.shards { + s.obsStrPool.shards[i].mu.Unlock() + } + } done := make(chan struct{}) go func() { defer close(done); tc.run() }() @@ -691,10 +698,10 @@ func TestObsStringPoolNeverBlocksStoreWriters_489(t *testing.T) { select { case <-locked: case <-time.After(10 * time.Second): - s.obsStrPool.mu.Unlock() + unpark() t.Fatal("a store writer waited on s.mu while an ingest was parked in the sharing pool: the pool is being used under the store lock") } - s.obsStrPool.mu.Unlock() + unpark() select { case <-done: case <-time.After(30 * time.Second): diff --git a/cmd/server/obs_strings_pool.go b/cmd/server/obs_strings_pool.go index 922bf4acd..fd281b75b 100644 --- a/cmd/server/obs_strings_pool.go +++ b/cmd/server/obs_strings_pool.go @@ -47,6 +47,10 @@ const ( // charge (about 0.7 MB in fact, see obsStringEntryBytes). obsStringPoolMax = 8192 + // obsStringPoolShards is how many independently locked slices the table + // is cut into. See obsStringShard for the contention numbers that set it. + obsStringPoolShards = 16 + // obsStringEntryBytes is one entry's cost on top of the canonical copy's // own bytes: two string headers in the slot, the slot's share of the // group control word and the table's slack. Measured on Go swiss maps @@ -72,24 +76,59 @@ const ( obsSharedDirection ) +// obsStringShard is one slice of the table with its own lock. Measured with +// BenchmarkObsStringPoolHitParallel (140 held values, hits only): on one +// shared RWMutex a hit costs 9 ns alone, 34 ns with two goroutines on it and +// 84 ns with ten -- the read lock's own cache line, not the map, is the +// bottleneck. Over 16 shards it costs 11 ns alone, 11 ns with two and 22 ns +// with ten, so the cold load, the background fill and live ingest can intern +// at once without waiting on each other. The 2 ns a shard lookup adds to a +// serial hit is 24 ms over the 10M calls a 2.5M-observation load makes. +// The padding keeps one shard's lock off the next one's cache line. +type obsStringShard struct { + mu sync.RWMutex + table map[string]string + _ [64 - 32]byte +} + // obsStringPool is the store's bounded canonical table of observer metadata // strings. A nil pool is a valid disabled pool (every value stays unshared), // which is what a test store built as &PacketStore{} gets. type obsStringPool struct { - mu sync.RWMutex - table map[string]string + shards [obsStringPoolShards]obsStringShard + // count is the number of canonical copies held, across all shards. An + // insert reserves its slot here before taking it, so the cap holds + // exactly however many shards insert at once. + count atomic.Int64 // bytes is the pool's own retention: per entry the canonical copy's - // content plus obsStringEntryBytes. Written under mu, read without it - // by the accounting path, which runs under s.mu. + // content plus obsStringEntryBytes. bytes atomic.Int64 - // interned counts successful shares; full counts the calls the cap sent - // back unshared. Diagnostics only (and the saturation test). - interned atomic.Int64 - full atomic.Int64 + // full counts the calls the cap sent back unshared. It is only touched + // past saturation, so it costs a hit nothing; a per-hit counter would be + // a contended cache line on the one path that must stay cheap. + full atomic.Int64 } func newObsStringPool() *obsStringPool { - return &obsStringPool{table: make(map[string]string, 256)} + p := &obsStringPool{} + for i := range p.shards { + p.shards[i].table = make(map[string]string, 16) + } + return p +} + +// obsStringShardOf picks s's shard. It reads the length and up to four bytes +// from each end, so a 2-byte direction costs the same as a 64-hex pubkey, +// and the same value always lands on the same shard -- which is all the +// sharding needs for correctness. The spread is a performance property only +// (BenchmarkObsStringPoolHitParallel measures it). +func obsStringShardOf(s string) uint { + h := uint(len(s)) * 2654435761 + for i := 0; i < 4 && i < len(s); i++ { + h = h*31 + uint(s[i]) + h = h*31 + uint(s[len(s)-1-i]) + } + return h & (obsStringPoolShards - 1) } // retainedBytes is what the pool itself holds. It only grows, and no eviction @@ -106,40 +145,38 @@ func (p *obsStringPool) entries() int { if p == nil { return 0 } - p.mu.RLock() - defer p.mu.RUnlock() - return len(p.table) + return int(p.count.Load()) } // intern returns the canonical copy of s and whether it is shared. A hit -// takes only the read lock and allocates nothing. The empty string is never -// interned: it holds no bytes and costs nothing to repeat. +// takes only its shard's read lock and allocates nothing. The empty string is +// never interned: it holds no bytes and costs nothing to repeat. func (p *obsStringPool) intern(s string) (string, bool) { if p == nil || s == "" { return s, false } - p.mu.RLock() - c, ok := p.table[s] - p.mu.RUnlock() + sh := &p.shards[obsStringShardOf(s)] + sh.mu.RLock() + c, ok := sh.table[s] + sh.mu.RUnlock() if ok { - p.interned.Add(1) return c, true } - p.mu.Lock() - defer p.mu.Unlock() - if c, ok := p.table[s]; ok { - p.interned.Add(1) + sh.mu.Lock() + defer sh.mu.Unlock() + if c, ok := sh.table[s]; ok { return c, true } - if len(p.table) >= obsStringPoolMax { + // Reserve the slot before taking it, so the cap is exact. + if p.count.Add(1) > obsStringPoolMax { + p.count.Add(-1) p.full.Add(1) return s, false } // Clone so a value scanned out of a larger buffer does not pin it. c = strings.Clone(s) - p.table[c] = c + sh.table[c] = c p.bytes.Add(int64(len(c)) + obsStringEntryBytes) - p.interned.Add(1) return c, true } diff --git a/cmd/server/obs_strings_pool_489_test.go b/cmd/server/obs_strings_pool_489_test.go index ed157b5a7..b71ce28fc 100644 --- a/cmd/server/obs_strings_pool_489_test.go +++ b/cmd/server/obs_strings_pool_489_test.go @@ -250,8 +250,8 @@ func TestObsStringPoolRetainedBytes_489(t *testing.T) { t.Fatalf("retainedBytes grew to %d on repeat interning, want %d", got, want) } - // Upper bound against the real thing: a map[string]string of the same - // shape, measured the way obsStringEntryBytes was. + // Upper bound against the real thing: the heap a second pool of the same + // shape really costs, measured the way obsStringEntryBytes was. keys := make([]string, 2000) for i := range keys { keys[i] = fmt.Sprintf("%064x", i) @@ -260,16 +260,15 @@ func TestObsStringPoolRetainedBytes_489(t *testing.T) { runtime.GC() runtime.GC() runtime.ReadMemStats(&m0) - table := make(map[string]string, 256) + measured := newObsStringPool() for _, k := range keys { - c := strings.Clone(k) - table[c] = c + measured.intern(k) } runtime.GC() runtime.GC() runtime.ReadMemStats(&m1) actual := int64(m1.HeapAlloc) - int64(m0.HeapAlloc) - runtime.KeepAlive(table) + runtime.KeepAlive(measured) runtime.KeepAlive(keys) if actual <= 0 { t.Skipf("heap measurement unusable (%d B)", actual) From 53682d4f7e559b602a6e88932f92409bd56697ba Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 12:02:47 +0200 Subject: [PATCH 4/4] test(store): give the pool lock-discipline test teeth Relates to #489. The lock test probed s.mu once, right after starting the ingest, so the prober usually won the lock before the ingest had reached it: a mutant that moved interning from the row scan into the locked merge loop survived. It now first asserts the ingest really is parked in the pool (it would otherwise prove nothing), then probes s.mu twenty times while it stays parked. Both "intern under s.mu" mutants, one per live path, now fail on probe 0. Also pins obsStringShard at 64 bytes, so the padding that keeps one shard's lock off the next one's cache line cannot silently rot. Co-Authored-By: Claude Opus 5 --- cmd/server/obs_strings_paths_489_test.go | 35 ++++++++++++++++-------- cmd/server/obs_strings_pool_489_test.go | 5 ++++ 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/cmd/server/obs_strings_paths_489_test.go b/cmd/server/obs_strings_paths_489_test.go index a75b3fd08..01e5746d5 100644 --- a/cmd/server/obs_strings_paths_489_test.go +++ b/cmd/server/obs_strings_paths_489_test.go @@ -687,19 +687,32 @@ func TestObsStringPoolNeverBlocksStoreWriters_489(t *testing.T) { done := make(chan struct{}) go func() { defer close(done); tc.run() }() - // The ingest is now either finished or stuck in the pool. Either - // way it must not be holding s.mu: a writer gets the lock. - locked := make(chan struct{}) - go func() { - s.mu.Lock() - s.mu.Unlock() - close(locked) - }() + // It must really be parked: the row it reads carries a value the + // pool does not hold, so it cannot get past interning. Without + // this the rest of the test would prove nothing. select { - case <-locked: - case <-time.After(10 * time.Second): + case <-done: unpark() - t.Fatal("a store writer waited on s.mu while an ingest was parked in the sharing pool: the pool is being used under the store lock") + t.Fatal("the ingest finished while every pool shard was locked, so it never needed the pool: this test proves nothing") + case <-time.After(500 * time.Millisecond): + } + + // Parked in the pool, it must not be holding s.mu: a writer gets + // the lock, over and over, for as long as the pool stays locked. + for i := 0; i < 20; i++ { + locked := make(chan struct{}) + go func() { + s.mu.Lock() + s.mu.Unlock() + close(locked) + }() + select { + case <-locked: + case <-time.After(5 * time.Second): + unpark() + t.Fatalf("probe %d: a store writer waited on s.mu while an ingest was parked in the sharing pool: the pool is being used under the store lock", i) + } + time.Sleep(5 * time.Millisecond) } unpark() select { diff --git a/cmd/server/obs_strings_pool_489_test.go b/cmd/server/obs_strings_pool_489_test.go index b71ce28fc..f7ea7b3bc 100644 --- a/cmd/server/obs_strings_pool_489_test.go +++ b/cmd/server/obs_strings_pool_489_test.go @@ -310,4 +310,9 @@ func TestStoreObsLayoutFitsSharedStrsInPadding_489(t *testing.T) { if got := unsafe.Sizeof(StoreObs{}); got != 200 { t.Fatalf("unsafe.Sizeof(StoreObs{}) = %d, want 200: sharedStrs must use existing padding", got) } + // One shard per cache line, so a write on one does not invalidate the + // next one's lock (see obsStringShard). + if got := unsafe.Sizeof(obsStringShard{}); got != 64 { + t.Fatalf("unsafe.Sizeof(obsStringShard{}) = %d, want 64: adjust the padding", got) + } }