From ae2e3933dde7ceb91ddbc056645def3dccc3822b Mon Sep 17 00:00:00 2001 From: "Michael J. Arcan" Date: Sun, 28 Jun 2026 22:48:47 +0200 Subject: [PATCH] feat(server): store memory diagnostics + drop redundant obs.RawHex (#1773) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drops the redundant per-observation RawHex (~98MB on a live store; reader already falls back to tx.RawHex #881) and adds an opt-in /api/perf?mem=1 memory breakdown (flood-forward share + per-component bytes). Profiled against a live instance. **Savings substantiation:** live-instance profiling shows ~1.66M observations in the store, each previously carrying its own per-observation `raw_hex` (avg ≈118 hex chars ≈59 bytes) that exactly duplicates the parent transmission's `raw_hex`. Dropping the duplicate on every load/ingest path eliminates ≈98 MB of redundant in-memory storage plus ~1.66M string allocations, with no data loss — the read path (`enrichObs`) already falls back to `tx.RawHex` when `obs.RawHex` is empty (verified by the new safety-gate test). The patched build cannot be run against the live instance here; instead the new opt-in `/api/perf?mem=1` diagnostic lets operators measure the real before/after (`trackedMB` and the per-component breakdown) directly after deploy. --- cmd/server/chunked_load.go | 7 +- cmd/server/routes.go | 31 +++- cmd/server/store.go | 87 +++++++++- cmd/server/store_memory_diagnostics_test.go | 174 ++++++++++++++++++++ cmd/server/types.go | 26 +++ 5 files changed, 309 insertions(+), 16 deletions(-) create mode 100644 cmd/server/store_memory_diagnostics_test.go diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index 65fadf05..99916cc0 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -456,8 +456,11 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold RSSI: nullFloatPtr(rssi), Score: nullIntPtr(score), PathJSON: obsPJ, - RawHex: nullStrVal(obsRawHex), - Timestamp: normalizeTimestamp(nullStrVal(obsTimestamp)), + // obs.RawHex deliberately NOT stored: it duplicates the parent + // tx.RawHex (same content hash ⇒ same frame) and enrichObs falls + // back to tx.RawHex when obs.RawHex == "". obsRawHex is still + // scanned to keep scanArgs aligned with the o.raw_hex column. + Timestamp: normalizeTimestamp(nullStrVal(obsTimestamp)), } rpStr := nullStrVal(resolvedPathStr) diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 1b5f8054..6f8af2ef 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -22,6 +22,10 @@ import ( "github.com/meshcore-analyzer/prunequeue" ) +// memBreakdownNote is the static accounting caveat attached to the opt-in +// /api/perf?mem=1 store memory breakdown (PerfResponse.MemoryBreakdownNote). +const memBreakdownNote = "Per-component *MB count string content + one Go string header each and are a deliberate upper bound (the header is also in the *EstimatedMB base figures). struct/index/map overhead, the neighbor graph and analytics caches are excluded, so the totals sit below goHeapInuseMB. floodTxSharePct sizes the flood-forward multiplication: each flood hop is stored as its own transmission." + // Server holds shared state for route handlers. type Server struct { db *DB @@ -905,6 +909,15 @@ func (s *Server) handlePerf(w http.ResponseWriter, r *http.Request) { pktStoreStats = &ps } + // Opt-in store memory diagnostic: an O(tx+obs) walk, only when explicitly + // requested so the hot /api/perf path stays cheap. + var memBreakdown *StoreMemoryBreakdown + var breakdownNote string + if s.store != nil && r.URL.Query().Get("mem") == "1" { + memBreakdown = s.store.GetStoreMemoryBreakdown() + breakdownNote = memBreakdownNote + } + // SQLite stats var sqliteStats *SqliteStats if s.db != nil { @@ -913,14 +926,16 @@ func (s *Server) handlePerf(w http.ResponseWriter, r *http.Request) { } writeJSON(w, PerfResponse{ - Uptime: uptimeSec, - TotalRequests: totalRequests, - AvgMs: safeAvg(totalMs, float64(totalRequests)), - Endpoints: summary, - SlowQueries: slowQueries, - Cache: perfCS, - PacketStore: pktStoreStats, - Sqlite: sqliteStats, + Uptime: uptimeSec, + TotalRequests: totalRequests, + AvgMs: safeAvg(totalMs, float64(totalRequests)), + Endpoints: summary, + SlowQueries: slowQueries, + Cache: perfCS, + PacketStore: pktStoreStats, + Sqlite: sqliteStats, + MemoryBreakdown: memBreakdown, + MemoryBreakdownNote: breakdownNote, GoRuntime: func() *GoRuntimeStats { ms := s.getMemStats() return &GoRuntimeStats{ diff --git a/cmd/server/store.go b/cmd/server/store.go index 97333ba5..4151ed94 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -895,7 +895,6 @@ func (s *PacketStore) Load() error { RSSI: nullFloatPtr(rssi), Score: nullIntPtr(score), PathJSON: obsPJ, - RawHex: nullStrVal(obsRawHex), Timestamp: normalizeTimestamp(nullStrVal(obsTimestamp)), } @@ -1216,7 +1215,6 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { RSSI: nullFloatPtr(rssi), Score: nullIntPtr(score), PathJSON: obsPJ, - RawHex: nullStrVal(obsRawHex), Timestamp: normalizeTimestamp(nullStrVal(obsTimestamp)), } @@ -2310,6 +2308,77 @@ func (s *PacketStore) GetPerfStoreStatsTyped() PerfPacketStoreStats { } } +// GetStoreMemoryBreakdown walks the whole store ONCE (under RLock) and returns +// the flood-forward (route_type 0/1) share of stored transmissions plus a +// per-component breakdown of the string bytes held in memory. O(tx + obs) — it +// touches every observation, so it is opt-in (/api/perf?mem=1) and must not be +// on the hot path. +// +// LOCK CONTENTION: this holds s.mu.RLock for the entire scan of the store. On a +// large store (millions of observations) the walk takes long enough to stall +// concurrent writers (ingest/eviction take the write lock) for the duration. +// Operators should poll this infrequently (e.g. on demand, not on a tight +// dashboard refresh) — do not wire it into a high-frequency scrape. +// +// The per-component *MB figures count string CONTENT plus one Go string header +// 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 +// graph and analytics caches are excluded entirely — which is why the total +// here sits below goHeapInuseMB. +func (s *PacketStore) GetStoreMemoryBreakdown() *StoreMemoryBreakdown { + s.mu.RLock() + defer s.mu.RUnlock() + + out := &StoreMemoryBreakdown{} + var txRawHex, txDecodedJSON, txPathJSON int64 + var obsPathJSON, obsStrings int64 + var floodTxBytes, totalTxBytes int64 + + for _, tx := range s.packets { + if tx == nil { + continue + } + out.TotalTx++ + isFlood := tx.RouteType != nil && (*tx.RouteType == 0 || *tx.RouteType == 1) + if isFlood { + out.FloodTx++ + } + txRawHex += int64(len(tx.RawHex) + strHdr) + txDecodedJSON += int64(len(tx.DecodedJSON) + strHdr) + txPathJSON += int64(len(tx.PathJSON) + strHdr) + b := estimateStoreTxBytes(tx) + totalTxBytes += b + if isFlood { + floodTxBytes += b + } + for _, o := range tx.Observations { + if o == nil { + continue + } + out.Observations++ + obsPathJSON += int64(len(o.PathJSON) + strHdr) + obsStrings += int64(len(o.ObserverID)+len(o.ObserverName)+len(o.ObserverIATA)+len(o.Direction)+len(o.Timestamp)) + 5*strHdr + } + } + + // 2 decimal places so sub-0.1MB components don't round away to 0. + mb := func(b int64) float64 { return math.Round(float64(b)/1048576*100) / 100 } + out.TxRawHexMB = mb(txRawHex) + out.TxDecodedJsonMB = mb(txDecodedJSON) + out.TxPathJsonMB = mb(txPathJSON) + out.ObsPathJsonMB = mb(obsPathJSON) + out.ObsStringsMB = mb(obsStrings) + out.FloodTxEstimatedMB = mb(floodTxBytes) + out.TotalTxEstimatedMB = mb(totalTxBytes) + if out.TotalTx > 0 { + out.FloodTxSharePct = math.Round(float64(out.FloodTx)/float64(out.TotalTx)*1000) / 10 + out.ObsPerTx = math.Round(float64(out.Observations)/float64(out.TotalTx)*10) / 10 + } + return out +} + // GetTransmissionByID returns a transmission by its DB ID, formatted as a map. func (s *PacketStore) GetTransmissionByID(id int) map[string]interface{} { s.mu.RLock() @@ -2674,7 +2743,6 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac RSSI: r.rssi, Score: r.score, PathJSON: r.pathJSON, - RawHex: r.obsRawHex, Timestamp: normalizeTimestamp(r.obsTS), } @@ -3006,8 +3074,10 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] RSSI: r.rssi, Score: r.score, PathJSON: r.pathJSON, - RawHex: r.rawHex, - Timestamp: normalizeTimestamp(r.timestamp), + // obs.RawHex deliberately NOT stored: it duplicates the parent + // tx.RawHex and enrichObs falls back to tx.RawHex when obs.RawHex + // == "". Live-polled observations must not re-introduce the dup. + Timestamp: normalizeTimestamp(r.timestamp), } // Resolve path at ingest time for late-arriving observations (review item #2). @@ -3672,7 +3742,11 @@ func (s *PacketStore) enrichObs(obs *StoreObs) map[string]interface{} { if tx != nil { m["hash"] = strOrNil(tx.Hash) - // Prefer per-observation raw_hex; fall back to transmission-level (#881) + // raw_hex comes from the transmission (#881). obs.RawHex is no longer + // retained when loading/ingesting the store — it duplicated this exact + // value (same content hash ⇒ same frame) and at ~10.6 observations/tx + // wasted ~98MB on a live ~1.7M-observation store. The obs.RawHex branch + // stays for callers that build a StoreObs from a response map. if obs.RawHex != "" { m["raw_hex"] = obs.RawHex } else { @@ -4268,6 +4342,7 @@ func (s *PacketStore) buildDistanceIndex() { const ( storeTxBaseBytes = 384 // StoreTx struct fields + map headers + sync.Once + string headers storeObsBaseBytes = 192 // StoreObs struct fields + string headers + strHdr = 16 // Go string header (ptr + len) on a 64-bit build; used by GetStoreMemoryBreakdown indexEntryBytes = 48 // average cost of one index map entry (key + pointer + bucket overhead) numIndexesPerTx = 5 // byHash, byTxID, byNode, byPayloadType, nodeHashes entries numIndexesPerObs = 2 // byObsID, byObserver entries diff --git a/cmd/server/store_memory_diagnostics_test.go b/cmd/server/store_memory_diagnostics_test.go new file mode 100644 index 00000000..989a76cc --- /dev/null +++ b/cmd/server/store_memory_diagnostics_test.go @@ -0,0 +1,174 @@ +package main + +import ( + "database/sql" + "math" + "path/filepath" + "strings" + "testing" + "time" +) + +// decodedJSONFixtureBytes is the size of the synthetic decoded_json payload used +// below: ~0.3 MB, chosen so the per-component byte breakdown registers a +// measurable, predictable TxDecodedJsonMB rather than rounding to zero. +const decodedJSONFixtureBytes = 300_000 + +// TestStoreMemoryBreakdown exercises the opt-in /api/perf?mem=1 diagnostic: the +// flood-forward (route_type 0/1) share and the per-component byte breakdown, +// over a hand-built store. +func TestStoreMemoryBreakdown(t *testing.T) { + rt0, rt1, rt2 := 0, 1, 2 + obs := func(id, pj string) *StoreObs { + return &StoreObs{ObserverID: id, ObserverName: "obs-" + id, PathJSON: pj} + } + bigJSON := strings.Repeat("x", decodedJSONFixtureBytes) + s := &PacketStore{} + s.packets = []*StoreTx{ + {RouteType: &rt0, RawHex: "aabbccdd", DecodedJSON: bigJSON, PathJSON: "[]", + Observations: []*StoreObs{obs("a", "[1]"), obs("b", "[2]"), obs("c", "[3]")}}, + {RouteType: &rt1, RawHex: "ee", PathJSON: "[]", + Observations: []*StoreObs{obs("a", "[1]")}}, + {RouteType: &rt2, RawHex: "ff0011", PathJSON: "[]", + Observations: []*StoreObs{obs("a", "[1]"), obs("b", "[2]")}}, + {RouteType: nil, RawHex: "22", PathJSON: "[]"}, // unknown route_type, no obs + } + + b := s.GetStoreMemoryBreakdown() + if b.TotalTx != 4 { + t.Errorf("TotalTx: want 4, got %d", b.TotalTx) + } + if b.FloodTx != 2 { + t.Errorf("FloodTx (route_type 0/1): want 2, got %d", b.FloodTx) + } + if b.FloodTxSharePct != 50 { + t.Errorf("FloodTxSharePct: want 50, got %v", b.FloodTxSharePct) + } + if b.Observations != 6 { + t.Errorf("Observations: want 6, got %d", b.Observations) + } + if b.ObsPerTx != 1.5 { + t.Errorf("ObsPerTx: want 1.5, got %v", b.ObsPerTx) + } + // decodedJSONFixtureBytes (+ a few string headers) over 1 MiB is ~0.286 MB, + // which rounds to 0.29 at 2 dp. Assert the real magnitude, not merely > 0, + // so a units/rounding regression in the byte accounting is caught. + wantDecodedMB := float64(decodedJSONFixtureBytes) / (1024 * 1024) + if math.Abs(b.TxDecodedJsonMB-wantDecodedMB) > 0.02 { + t.Errorf("TxDecodedJsonMB: want ≈%.2f, got %v", wantDecodedMB, b.TxDecodedJsonMB) + } + if b.TotalTxEstimatedMB <= 0 { + t.Errorf("TotalTxEstimatedMB should be > 0, got %v", b.TotalTxEstimatedMB) + } + // The route_type 0 flood tx carries the ~0.3 MB decoded_json, so the flood + // share of estimated bytes must register above the 2-dp MB rounding floor. + if b.FloodTxEstimatedMB == 0 { + t.Errorf("expected FloodTxEstimatedMB > 0, got %v", b.FloodTxEstimatedMB) + } + if b.FloodTxEstimatedMB > b.TotalTxEstimatedMB { + t.Error("flood estimated bytes cannot exceed total") + } + if b.ObsStringsMB < 0 || b.ObsPathJsonMB < 0 || b.TxRawHexMB < 0 { + t.Error("byte breakdown components must be >= 0") + } +} + +// TestStoreMemoryBreakdown_Empty: an empty store yields all-zero, no divide. +func TestStoreMemoryBreakdown_Empty(t *testing.T) { + s := &PacketStore{} + b := s.GetStoreMemoryBreakdown() + if b.TotalTx != 0 || b.FloodTx != 0 || b.FloodTxSharePct != 0 || b.Observations != 0 || b.ObsPerTx != 0 { + t.Errorf("empty store: want all zero, got %+v", b) + } +} + +// TestObsRawHexNotRetainedOnLoad locks in the behavior change: even when the +// DB carries a non-empty observations.raw_hex, the production load path +// (LoadChunked) must NOT retain it on the in-memory StoreObs (it duplicates the +// parent tx.RawHex), and the read/enrich path must still serve raw_hex by +// falling back to the parent transmission. If this regresses, obs raw_hex would +// be silently lost — so this is the safety gate for dropping obs.RawHex. +func TestObsRawHexNotRetainedOnLoad(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "obsrawhex.db") + + const txHex = "deadbeefcafe" + const obsHex = "c0ffee0102" // distinct from txHex: proves we DON'T keep it + + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + stmts := []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_id TEXT, + observer_name TEXT, direction TEXT, snr REAL, rssi REAL, score INTEGER, + path_json TEXT, timestamp TEXT, raw_hex 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 INDEX idx_tx_first_seen ON transmissions(first_seen)`, + } + for _, st := range stmts { + if _, err := conn.Exec(st); err != nil { + t.Fatalf("schema exec: %v\nSQL: %s", err, st) + } + } + now := time.Now().UTC().Truncate(time.Second) + if _, err := conn.Exec( + `INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + 1, txHex, "hashobs", now.Add(-time.Minute).Format(time.RFC3339), 1, 5, 0, `{"type":"CHAN"}`); err != nil { + t.Fatalf("insert tx: %v", err) + } + // Observation carries its OWN non-empty raw_hex in the DB. + if _, err := conn.Exec( + `INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp, raw_hex) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + 1, 1, "obs1", "Obs1", "rx", 5.0, -95.0, 0, `["AA"]`, now.Add(-time.Minute).Unix(), obsHex); err != nil { + t.Fatalf("insert obs: %v", err) + } + conn.Close() + + db, err := OpenDB(dbPath) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + if !db.hasObsRawHex { + t.Fatal("fixture precondition: observations.raw_hex must be detected (hasObsRawHex)") + } + store := NewPacketStore(db, &PacketStoreConfig{}) + defer store.db.conn.Close() + + if err := store.LoadChunked(5); err != nil { + t.Fatalf("LoadChunked: %v", err) + } + + tx := store.byHash["hashobs"] + if tx == nil { + t.Fatal("transmission not loaded") + } + if len(tx.Observations) != 1 { + t.Fatalf("expected 1 observation, got %d", len(tx.Observations)) + } + obs := tx.Observations[0] + + // The dup must NOT be retained, even though the DB column held obsHex. + if obs.RawHex != "" { + t.Errorf("obs.RawHex should be empty after load (dropped as redundant dup), got %q", obs.RawHex) + } + + // The read path must still serve raw_hex via the parent-tx fallback. + store.mu.RLock() + m := store.enrichObs(obs) + store.mu.RUnlock() + rh, _ := m["raw_hex"].(string) + if rh != txHex { + t.Errorf("enrichObs raw_hex: want parent tx fallback %q, got %q", txHex, rh) + } +} diff --git a/cmd/server/types.go b/cmd/server/types.go index 29b3b5dd..88b36f31 100644 --- a/cmd/server/types.go +++ b/cmd/server/types.go @@ -270,6 +270,32 @@ type PerfResponse struct { PacketStore *PerfPacketStoreStats `json:"packetStore"` Sqlite *SqliteStats `json:"sqlite"` GoRuntime *GoRuntimeStats `json:"goRuntime,omitempty"` + // MemoryBreakdown is populated only for /api/perf?mem=1 (an O(tx+obs) + // walk, opt-in so the normal hot endpoint stays cheap). It sizes the + // flood-forward multiplication (store memory diagnostics) and where the + // store's string bytes go. + MemoryBreakdown *StoreMemoryBreakdown `json:"memoryBreakdown,omitempty"` + // MemoryBreakdownNote documents the accounting scope of MemoryBreakdown. + // It lives here (one occurrence) rather than repeating in every breakdown. + MemoryBreakdownNote string `json:"memoryBreakdownNote,omitempty"` +} + +// StoreMemoryBreakdown is the opt-in /api/perf?mem=1 diagnostic: the +// flood-forward (route_type 0/1) share of stored transmissions and a +// per-component breakdown of the string bytes held in the packet store. +type StoreMemoryBreakdown struct { + TotalTx int `json:"totalTx"` + FloodTx int `json:"floodTx"` // route_type 0 or 1 + FloodTxSharePct float64 `json:"floodTxSharePct"` // flood share of stored tx + Observations int `json:"observations"` + ObsPerTx float64 `json:"obsPerTx"` + TxRawHexMB float64 `json:"txRawHexMB"` + TxDecodedJsonMB float64 `json:"txDecodedJsonMB"` + TxPathJsonMB float64 `json:"txPathJsonMB"` + ObsPathJsonMB float64 `json:"obsPathJsonMB"` + ObsStringsMB float64 `json:"obsStringsMB"` // observerID/name/iata/direction/timestamp + FloodTxEstimatedMB float64 `json:"floodTxEstimatedMB"` + TotalTxEstimatedMB float64 `json:"totalTxEstimatedMB"` } // GoRuntimeStats holds Go runtime metrics for the perf endpoint.