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..01e5746d5 --- /dev/null +++ b/cmd/server/obs_strings_paths_489_test.go @@ -0,0 +1,791 @@ +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. 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) + } + 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. + 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() }() + + // 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 <-done: + unpark() + 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 { + 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..fd281b75b --- /dev/null +++ b/cmd/server/obs_strings_pool.go @@ -0,0 +1,241 @@ +package main + +import ( + "strings" + "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 + + // 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 + // (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 +) + +// 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 { + 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. + bytes 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 { + 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 +// 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 + } + return int(p.count.Load()) +} + +// intern returns the canonical copy of s and whether it is shared. A hit +// 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 + } + sh := &p.shards[obsStringShardOf(s)] + sh.mu.RLock() + c, ok := sh.table[s] + sh.mu.RUnlock() + if ok { + return c, true + } + sh.mu.Lock() + defer sh.mu.Unlock() + if c, ok := sh.table[s]; ok { + return c, true + } + // 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) + sh.table[c] = c + p.bytes.Add(int64(len(c)) + obsStringEntryBytes) + return c, true +} + +// 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..f7ea7b3bc --- /dev/null +++ b/cmd/server/obs_strings_pool_489_test.go @@ -0,0 +1,318 @@ +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: 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) + } + var m0, m1 runtime.MemStats + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m0) + measured := newObsStringPool() + for _, k := range keys { + measured.intern(k) + } + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m1) + actual := int64(m1.HeapAlloc) - int64(m0.HeapAlloc) + runtime.KeepAlive(measured) + 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) + } + // 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) + } +} 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 {