fix(store): account path, decode-cache and dedup-key bytes so maxMemoryMB eviction triggers (#2035)

trackedBytes undercounted the packet store by about 2.5x, so packetStore.maxMemoryMB never triggered: a tx was charged at creation, before pickBestObservation set its path, so the byPathHop and spTxIndex costs were never added, and eviction then re-estimated with the path known and subtracted more than had been added, drifting the total downwards. The ParsedDecoded cache, the obsKeys dedup key and several per-observation strings were not estimated at all.

StoreTx.accountedBytes now records what was charged, rechargeTx returns the delta after every pickBestObservation, and eviction subtracts accountedBytes instead of re-estimating. Measured by the author on a production database copy: trackedMB 151 against 402 MB of heap in use before, 351 against 396 MB after, with GC cycles dropping from ~0.71/s to ~0.011/s over 12 h on their instance.

Reviewed by auditing the accounting rather than the arithmetic: all six production pickBestObservation sites recharge, all three subtraction sites read accountedBytes, every recharge site holds s.mu (Load from :857, the two ingest paths at :2778 and :3142 with deferred unlocks), observations are charged only after acceptance, and no charged tx is discarded during the chunk merge. That lock audit is the independent check, because CI's race job covers the ingestor only.

Operator impact, both from the estimate growing rather than any limit moving: where maxMemoryMB is set, eviction now caps the real store size, and the cold load clamp drops about a third of the boot walk (124420 to 84374 packets at 650 MB). Where it is unset, which is the default, the change is inert. docs/go-migration.md claimed the Go server ignored the setting, which was never true, and is corrected here.

Merged by the interim maintainer without a second human reviewer: CI (run 35258839322) plus the review above are the independent checks.

cmd/server/chunked_load.go

@@ -468,6 +468,7 @@ func (s *PacketStore) LoadChunked(chunkSize int) error {
 	s.mu.Lock()
 	for _, tx := range s.packets {
 		pickBestObservation(tx)
+		s.trackedBytes += rechargeTx(tx)
 		s.indexByNode(tx)
 	}
 	// Restore the "s.packets sorted oldest-first by FirstSeen" invariant
@@ -592,7 +593,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 				s.byPayloadType[pt] = append(s.byPayloadType[pt], tx)
 			}
 			s.trackAdvertPubkey(tx)
-			s.trackedBytes += estimateStoreTxBytes(tx)
+			s.trackedBytes += rechargeTx(tx)
 		}
 
 		if obsID.Valid {

cmd/server/eviction_test.go

@@ -87,7 +87,7 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore
 		addTxToSubpathIndex(store.spIndex, tx)
 
 		// Track bytes for self-accounting
-		store.trackedBytes += estimateStoreTxBytes(tx)
+		store.trackedBytes += rechargeTx(tx)
 		for _, obs := range tx.Observations {
 			store.trackedBytes += estimateStoreObsBytes(obs)
 		}
@@ -460,7 +460,7 @@ func TestTrackedBytes_MatchesExpectedAfterMixedInsertEvict(t *testing.T) {
 	var evictedBytes int64
 	for i := 0; i < 50; i++ {
 		tx := store.packets[i]
-		evictedBytes += estimateStoreTxBytes(tx)
+		evictedBytes += tx.accountedBytes
 		for _, obs := range tx.Observations {
 			evictedBytes += estimateStoreObsBytes(obs)
 		}
@@ -568,6 +568,7 @@ func TestEstimateStoreTxBytes(t *testing.T) {
 	// Manual calculation: base + string lengths + index entries + perTxMaps + path hops + subpaths
 	hops := int64(len(txGetParsedPath(tx)))
 	manualCalc := int64(storeTxBaseBytes) + int64(len(tx.RawHex)+len(tx.Hash)+len(tx.DecodedJSON)+len(tx.PathJSON)) + int64(numIndexesPerTx*indexEntryBytes)
+	manualCalc += int64(decodedCacheFactor * len(tx.DecodedJSON))
 	manualCalc += perTxMapsBytes
 	manualCalc += hops * perPathHopBytes
 	if hops > 1 {
@@ -576,8 +577,8 @@ func TestEstimateStoreTxBytes(t *testing.T) {
 	if est != manualCalc {
 		t.Fatalf("estimateStoreTxBytes = %d, want %d (manual calc)", est, manualCalc)
 	}
-	if est < 600 || est > 1200 {
-		t.Fatalf("estimateStoreTxBytes = %d, expected in range [600, 1200]", est)
+	if est < 600 || est > 1300 {
+		t.Fatalf("estimateStoreTxBytes = %d, expected in range [600, 1300]", est)
 	}
 }
 
@@ -587,8 +588,9 @@ func TestEstimateStoreObsBytes(t *testing.T) {
 		PathJSON:   `["aa"]`,
 	}
 	est := estimateStoreObsBytes(obs)
-	// storeObsBaseBytes(192) + len(ObserverID=6) + len(PathJSON=6) + 2*48(96) = 300
-	expected := int64(192 + 6 + 6 + 2*48)
+	// storeObsBaseBytes(192) + len(ObserverID=6) + len(PathJSON=6) + 2*48(96) = 300,
+	// plus the obsKeys dedup entry: obsKeyEntryBytes(65) + 6 + 6 = 77
+	expected := int64(192 + 6 + 6 + 2*48 + 77)
 	if est != expected {
 		t.Fatalf("estimateStoreObsBytes = %d, want %d", est, expected)
 	}

cmd/server/main.go

@@ -339,6 +339,7 @@ func main() {
 			store.mu.Lock()
 			for j := i; j < end && j < len(store.packets); j++ {
 				pickBestObservation(store.packets[j])
+				store.trackedBytes += rechargeTx(store.packets[j])
 			}
 			store.mu.Unlock()
 			if end < totalPackets {

cmd/server/store.go

@@ -65,6 +65,9 @@ type StoreTx struct {
 	// Dedup map: "observerID|pathJSON" → true for O(1) duplicate checks
 	obsKeys     map[string]bool
 	observerSet map[string]bool // unique observer IDs (for UniqueObserverCount)
+	// accountedBytes is what trackedBytes was last charged for this tx
+	// (observations excluded). See rechargeTx.
+	accountedBytes int64
 }
 
 // StoreObs is a lean in-memory observation (no duplication of transmission fields).
@@ -922,7 +925,7 @@ func (s *PacketStore) Load() error {
 				s.byPayloadType[pt] = append(s.byPayloadType[pt], tx)
 			}
 			s.trackAdvertPubkey(tx)
-			s.trackedBytes += estimateStoreTxBytes(tx)
+			s.trackedBytes += rechargeTx(tx)
 		}
 
 		if obsID.Valid {
@@ -1006,6 +1009,7 @@ func (s *PacketStore) Load() error {
 	// now that pickBestObservation has propagated the best path.
 	for _, tx := range s.packets {
 		pickBestObservation(tx)
+		s.trackedBytes += rechargeTx(tx)
 		s.indexByNode(tx)
 	}
 
@@ -1244,7 +1248,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error {
 			if txID > localMaxTxID {
 				localMaxTxID = txID
 			}
-			localTrackedBytes += estimateStoreTxBytes(tx)
+			localTrackedBytes += rechargeTx(tx)
 		}
 
 		if obsID.Valid {
@@ -1325,6 +1329,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error {
 	// Pick best observation for each local packet before merging.
 	for _, tx := range localPackets {
 		pickBestObservation(tx)
+		localTrackedBytes += rechargeTx(tx)
 	}
 
 	if len(localPackets) == 0 {
@@ -1685,6 +1690,19 @@ func pickBestObservation(tx *StoreTx) {
 	tx.pathParsed = false // invalidate cached parsed path
 }
 
+// rechargeTx re-estimates tx and returns the change since it was last charged,
+// for the caller to add to trackedBytes. A tx is first charged when it is
+// created, before its observations are merged, so its path costs (byPathHop,
+// spTxIndex) are unknown then: call this again once pickBestObservation has
+// set the path. Eviction subtracts accountedBytes, so the running total stays
+// consistent however often the path changed in between.
+func rechargeTx(tx *StoreTx) int64 {
+	est := estimateStoreTxBytes(tx)
+	d := est - tx.accountedBytes
+	tx.accountedBytes = est
+	return d
+}
+
 func pathLen(pathJSON string) int {
 	if pathJSON == "" {
 		return 0
@@ -2813,7 +2831,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 				s.byPayloadType[pt] = append(s.byPayloadType[pt], tx)
 			}
 			s.trackAdvertPubkey(tx)
-			s.trackedBytes += estimateStoreTxBytes(tx)
+			s.trackedBytes += rechargeTx(tx)
 
 			if _, exists := broadcastTxs[r.txID]; !exists {
 				broadcastTxs[r.txID] = tx
@@ -2889,6 +2907,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 	// Pick best observation for new transmissions
 	for _, tx := range broadcastTxs {
 		pickBestObservation(tx)
+		s.trackedBytes += rechargeTx(tx)
 	}
 
 	// Incrementally update precomputed subpath index with new transmissions
@@ -3280,6 +3299,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 	}
 	for _, tx := range updatedTxs {
 		pickBestObservation(tx)
+		s.trackedBytes += rechargeTx(tx)
 	}
 	pathHopMutated := false
 	for txID, tx := range updatedTxs {
@@ -4559,6 +4579,17 @@ const (
 
 	// Per subpath entry in spTxIndex: string key + slice append + pointer
 	perSubpathEntryBytes = 40
+
+	// ParsedDecoded caches json.Unmarshal of DecodedJSON as a
+	// map[string]interface{}: measured at about 4x the JSON length on a
+	// production heap profile. Analytics touch every tx, so it is charged
+	// up front rather than when the cache fills.
+	decodedCacheFactor = 4
+
+	// Per obs: the tx.obsKeys dedup entry ("observerID|pathJSON" key string
+	// plus bucket slot); the key is built by concatenation, so its bytes are
+	// a separate allocation from the obs fields.
+	obsKeyEntryBytes = 16 + indexEntryBytes + 1
 )
 
 // estimateStoreTxBytes returns the estimated memory cost of a StoreTx (excluding observations).
@@ -4567,6 +4598,7 @@ func estimateStoreTxBytes(tx *StoreTx) int64 {
 	base := int64(storeTxBaseBytes)
 	base += int64(len(tx.RawHex) + len(tx.Hash) + len(tx.DecodedJSON) + len(tx.PathJSON))
 	base += int64(numIndexesPerTx * indexEntryBytes)
+	base += int64(decodedCacheFactor * len(tx.DecodedJSON))
 
 	// Per-tx maps: obsKeys + observerSet
 	base += perTxMapsBytes
@@ -4591,13 +4623,15 @@ func estimateStoreTxBytesTypical(numObs int) int64 {
 	// Typical tx: ~64 byte hash, ~200 byte decoded JSON, ~40 byte path, 3 hops
 	base := int64(storeTxBaseBytes) + 64 + 200 + 40
 	base += int64(numIndexesPerTx * indexEntryBytes)
+	base += decodedCacheFactor * 200
 	base += perTxMapsBytes
 	hops := int64(3)
 	base += hops * perPathHopBytes
 	base += (hops * (hops - 1) / 2) * perSubpathEntryBytes
 	// Add observation costs
-	obsBase := int64(storeObsBaseBytes) + 30 + 30 + 60 // observer ID + name + path
+	obsBase := int64(storeObsBaseBytes) + 30 + 30 + 60 + 25 // observer ID + name + path + timestamp
 	obsBase += int64(numIndexesPerObs * indexEntryBytes)
+	obsBase += obsKeyEntryBytes + 30 + 60 // dedup key: observer ID + path
 	// No per-obs ResolvedPath overhead (#800)
 	base += int64(numObs) * obsBase
 	return base
@@ -4607,8 +4641,10 @@ 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))
+	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(numIndexesPerObs * indexEntryBytes)
+	base += int64(obsKeyEntryBytes + len(obs.ObserverID) + len(obs.PathJSON))
 	// ResolvedPath field removed (#800) — no per-obs RP overhead
 	return base
 }
@@ -4665,7 +4701,7 @@ func (s *PacketStore) evictionCandidateTxIDs() []int {
 			memCutoff := cutoffIdx
 			for memCutoff < len(s.packets) && (s.trackedBytes-bytesToEvict) > lowWatermark {
 				tx := s.packets[memCutoff]
-				bytesToEvict += estimateStoreTxBytes(tx)
+				bytesToEvict += tx.accountedBytes
 				for _, obs := range tx.Observations {
 					bytesToEvict += estimateStoreObsBytes(obs)
 				}
@@ -4734,7 +4770,7 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int {
 			memCutoff := cutoffIdx
 			for memCutoff < len(s.packets) && (s.trackedBytes-bytesToEvict) > lowWatermark {
 				tx := s.packets[memCutoff]
-				bytesToEvict += estimateStoreTxBytes(tx)
+				bytesToEvict += tx.accountedBytes
 				for _, obs := range tx.Observations {
 					bytesToEvict += estimateStoreObsBytes(obs)
 				}
@@ -4779,7 +4815,7 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int {
 		delete(s.byHash, tx.Hash)
 		delete(s.byTxID, tx.ID)
 		evictedTxIDs[tx.ID] = struct{}{}
-		evictedBytes += estimateStoreTxBytes(tx)
+		evictedBytes += tx.accountedBytes
 
 		for _, obs := range tx.Observations {
 			delete(s.byObsID, obs.ID)

cmd/server/tracked_bytes_test.go

@@ -164,3 +164,40 @@ func BenchmarkEstimateStoreObsBytes(b *testing.B) {
 		estimateStoreObsBytes(obs)
 	}
 }
+
+// TestTrackedBytes_PathKnownAfterCreate reproduces the load/ingest order: a tx
+// is charged when it is created, before its observations set the path. Once
+// pickBestObservation has run, recharging must pick up the path costs, and
+// eviction must subtract exactly what was charged.
+func TestTrackedBytes_PathKnownAfterCreate(t *testing.T) {
+	store := makeTestStore(0, time.Now().UTC(), 0)
+	store.retentionHours = 1
+
+	tx := &StoreTx{ID: 1, Hash: "aabb", FirstSeen: time.Now().UTC().Add(-2 * time.Hour).Format(time.RFC3339),
+		obsKeys: map[string]bool{}, observerSet: map[string]bool{}}
+	store.trackedBytes += rechargeTx(tx)
+	bare := store.trackedBytes
+
+	obs := &StoreObs{ID: 1, TransmissionID: 1, ObserverID: "o1", PathJSON: `["a1","b2","c3","d4","e5","f6"]`}
+	tx.Observations = append(tx.Observations, obs)
+	store.trackedBytes += estimateStoreObsBytes(obs)
+	pickBestObservation(tx)
+	store.trackedBytes += rechargeTx(tx)
+	store.packets = append(store.packets, tx)
+	store.byHash[tx.Hash] = tx
+	store.byTxID[tx.ID] = tx
+	store.byObsID[obs.ID] = obs
+
+	if want := estimateStoreTxBytes(tx) + estimateStoreObsBytes(obs); store.trackedBytes != want {
+		t.Fatalf("trackedBytes = %d, want %d (path costs missing?)", store.trackedBytes, want)
+	}
+	if tx.accountedBytes <= bare {
+		t.Fatalf("recharge did not add the path costs: %d <= %d", tx.accountedBytes, bare)
+	}
+	if n := store.EvictStale(); n != 1 {
+		t.Fatalf("evicted %d, want 1", n)
+	}
+	if store.trackedBytes != 0 {
+		t.Fatalf("trackedBytes after evicting everything = %d, want 0", store.trackedBytes)
+	}
+}

docs/deployment.md

@@ -625,6 +625,8 @@ The in-memory packet store grows with retained packets. Configure retention limi
 }
 ```
 
+`packetStore.maxMemoryMB` bounds the store **and the caches that belong to it** — the decoded-packet cache and per-packet index entries, not just the stored rows. It is enforced in two places: the startup load stops at the budget, and the store evicts oldest-first when it exceeds it. Leaving it unset means no limit. Actual usage is on `/api/perf` as `packetStore.trackedMB`, next to `maxMB`.
+
 ### Database locked errors
 
 SQLite doesn't support concurrent writers well. Ensure only one CoreScope instance accesses the database file. If running multiple containers, each needs its own database.

docs/go-migration.md

@@ -78,9 +78,15 @@ Node.js uses `mqtt://` and `mqtts://` scheme prefixes. The Go MQTT library (paho
 
 Both engines support `retention.nodeDays` (default: 7). Stale nodes are moved to the `inactive_nodes` table on the same schedule. No config change needed.
 
-### `packetStore.maxMemoryMB` (Go ignores this — it's Node-only)
+### `packetStore.maxMemoryMB` (both engines read it)
 
-The Node.js server has a configurable in-memory packet store limit (`packetStore.maxMemoryMB`). The Go server has its own in-memory store that loads all packets from SQLite on startup — it does not read this config value. This is safe to leave in your config; Go simply ignores it.
+Both engines take a memory budget for the in-memory packet store from `packetStore.maxMemoryMB`. In the Go server it does three things:
+
+- **Startup:** the cold load stops once the budget is reached, so the server starts with the newest packets that fit rather than everything in SQLite.
+- **While running:** the store evicts oldest-first when it passes the budget, down to 85% of it.
+- **Go heap:** when neither `GOMEMLIMIT` nor `runtime.maxMemoryMB` is set, `GOMEMLIMIT` is derived from this value plus 50% headroom.
+
+Unset (the default) means no limit: every packet within retention is loaded and none is evicted for memory. An old Node-era value left in a config file is therefore **not** inert — it caps the Go store too. Check it before upgrading, and read `/api/perf` (`packetStore.trackedMB` against `maxMB`) afterwards.
 
 ### `channelKeys` / `channel-rainbow.json` (compatible)
 
@@ -338,7 +344,7 @@ docker start corescope-prod
 |---------|--------|------------|
 | Companion bridge advertisements | `meshcore/advertisement` topic not handled by Go ingestor | Users relying on companion bridge adverts must stay on Node.js or wait for Go support |
 | Companion bridge `self_info` | `meshcore/self_info` topic not handled | Same as above — minimal impact (only affects local node identity) |
-| `packetStore.maxMemoryMB` config | Go doesn't read this setting | Go manages its own memory; no action needed |
+| `packetStore.maxMemoryMB` config | Read by both engines, but Go's store accounts memory differently | Review the value before upgrading — see [`packetStore.maxMemoryMB`](#packetstoremaxmemorymb-both-engines-read-it) |
 | Docker Hub images | Go images not published yet | Build locally with `docker build -f Dockerfile.go` |
 | `manage.sh --engine` flag | Can't toggle engines via manage.sh | Manual image swap required (see [Switch to Go](#switch-to-go)) |
 
@@ -349,7 +355,7 @@ docker start corescope-prod
 | `engine` field in `/api/health` | Not present or `"node"` | Always `"go"` |
 | MQTT URL scheme | Uses `mqtt://` / `mqtts://` natively | Auto-converts to `tcp://` / `ssl://` (transparent) |
 | Process model | Single Node.js process (server + ingestor) | Two binaries: `corescope-ingestor` + `corescope-server` (managed by supervisord) |
-| Memory management | Configurable via `packetStore.maxMemoryMB` | Loads all packets; no configurable limit |
+| Memory management | Configurable via `packetStore.maxMemoryMB` | Same setting: caps the cold load, evicts oldest-first above it, and sets `GOMEMLIMIT` when that is otherwise unset |
 | Startup time | Faster (no compilation) | Slightly slower (loads all packets from DB into memory) |
 
 ---