perf(store): check observation duplicates without per-transmission maps (#2181)

## Problem
Every `StoreTx` carried two maps for observation de-duplication:
- `obsKeys`, keyed by `observerID + "|" + pathJSON` (a concatenated
string allocated for every observation);
- `observerSet`, for `UniqueObserverCount`.

On a busy regional network's 7-day window (64k transmissions, 400k
observations), these maps and their concatenated keys held about 60 MB
of heap. Most transmissions are heard a handful of times; the busiest in
that week was heard 40 times. The maps cost a few hundred bytes per
transmission to de-duplicate lists that are usually under ten long.

## Change
- `hasObservation` and `hasObserver` scan `tx.Observations`, comparing
the observer ID and path strings the observations already hold. These
are the same two values the old map key was built from, so the answers
are identical.
- Once a transmission passes 64 observations (`obsIndexThreshold`), it
builds the maps. They're keyed by a struct that shares the observations'
strings, so no key bytes are allocated, and busy transmissions keep O(1)
checks (the case #543 / #355 addressed).
- The five places that appended an observation now call
`addObservation`, which keeps `UniqueObserverCount` and
`ObservationCount` in step. Those three lines were previously copied at
each site.
- The memory estimate no longer charges for the per-transmission maps or
the concatenated keys.

## Perf justification (AGENTS.md rule 0)
- **Complexity.**
- Below the threshold, a duplicate check is O(k) for a transmission with
k < 64 observations, so building it costs at most 63·64/2 = 2016
string-header comparisons.
- At and above 64 observations, checks are O(1) map lookups, as before.
- Nothing is O(n²) in the number of transmissions or observations
overall: the scan is bounded per transmission.
- **Scale.** Real data peaks at 40 observations per transmission. The
scan compares string headers (length first, then bytes), and
64-character observer IDs differ early.

`BenchmarkObsDedup` (a miss, the common case on ingest):

| observations on tx | check used | time |
|---|---|---|
| 4 | scan | 15 ns |
| 16 | scan | 70 ns |
| 40 | scan | 151 ns |
| 63 | scan | 233 ns |
| 64+ | map (struct key) | 15 ns |

For comparison, the old map lookup was about 15 ns too, plus building
and allocating the `observerID|path` key string on every check.

At 40 observations that's about 3 µs extra to build a whole
transmission, against roughly 2 KB of maps no longer allocated for it.

**Measured on a copy of that 7-day window** (`GOMEMLIMIT=750MiB`, same
queries):

| | before | after |
|---|---|---|
| in-use heap after load | 612 MB | 568 MB |
| HeapAlloc after load | 660 MB | 564 MB |
| peak RSS | 1077 MB | 1009 MB |
| startup to ready | 31 s | 21 s |

It has also run on a staging instance (live MQTT ingest) for about nine
hours alongside other store changes, with no change in observation or
observer counts.

## Tests
- `obs_dedup_test.go`:
- duplicates (same observer and path) are rejected, while a new path or
a new observer is kept;
  - a transmission under the threshold carries no maps;
  - across the threshold the maps take over with the same answers;
  - empty observer IDs aren't counted as observers;
  - `BenchmarkObsDedup`.
- Red commit first (stubs, failing on assertions), then green.
- `cd cmd/server && go test ./...` passes.
- SonarQube: quality gate passed, no issues on added lines.

Backend only: no UI, API shape or config change.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

https://claude.ai/code/session_01KaR4xu8gVykQBkH8yG8m6H

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

cmd/server/chunked_load.go

@@ -583,8 +583,6 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 				PayloadType: nullIntPtr(payloadType),
 				DecodedJSON: nullStrVal(decodedJSON),
 				ScopeName:   nullStrPtr(scopeName),
-				obsKeys:     make(map[string]bool),
-				observerSet: make(map[string]bool),
 			}
 			s.byHash[hashStr] = tx
 			s.packets = append(s.packets, tx)
@@ -606,8 +604,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 			obsIDStr := nullStrVal(observerID)
 			obsPJ := nullStrVal(pathJSON)
 
-			dk := obsIDStr + "|" + obsPJ
-			if tx.obsKeys[dk] {
+			if tx.hasObservation(obsIDStr, obsPJ) {
 				continue
 			}
 
@@ -649,13 +646,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 				}
 			}
 
-			tx.Observations = append(tx.Observations, obs)
-			tx.obsKeys[dk] = true
-			if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] {
-				tx.observerSet[obs.ObserverID] = true
-				tx.UniqueObserverCount++
-			}
-			tx.ObservationCount++
+			tx.addObservation(obs)
 			if obs.Timestamp > tx.LatestSeen {
 				tx.LatestSeen = obs.Timestamp
 			}

cmd/server/eviction_test.go

@@ -569,7 +569,6 @@ func TestEstimateStoreTxBytes(t *testing.T) {
 	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 {
 		manualCalc += (hops * (hops - 1) / 2) * perSubpathEntryBytes
@@ -588,9 +587,9 @@ func TestEstimateStoreObsBytes(t *testing.T) {
 		PathJSON:   `["aa"]`,
 	}
 	est := estimateStoreObsBytes(obs)
-	// 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)
+	// storeObsBaseBytes(192) + len(ObserverID=6) + len(PathJSON=6) + 2*48(96) = 300.
+	// The dedup check scans tx.Observations, so it adds no per-obs bytes.
+	expected := int64(192 + 6 + 6 + 2*48)
 	if est != expected {
 		t.Fatalf("estimateStoreObsBytes = %d, want %d", est, expected)
 	}

cmd/server/obs_dedup.go

@@ -0,0 +1,70 @@
+package main
+
+// obsDedupKey identifies an observation within its transmission: the same
+// observer reporting the same path is a duplicate.
+type obsDedupKey struct{ observerID, pathJSON string }
+
+// obsIndexThreshold is the observation count at which a transmission starts
+// keeping maps for its duplicate and distinct-observer checks. Below it the
+// checks scan tx.Observations, which compares string headers and costs
+// nothing to keep; the maps cost a few hundred bytes per transmission, and
+// most transmissions are heard a handful of times (a week of a busy regional
+// network peaked at 40). A miss scans in about 150ns at 40 observations and
+// 240ns at 63; past the threshold a map lookup is about 15ns. The maps' keys
+// share the observations' strings, so building them allocates no key bytes.
+const obsIndexThreshold = 64
+
+// hasObservation reports whether tx already holds an observation from
+// observerID with pathJSON.
+func (tx *StoreTx) hasObservation(observerID, pathJSON string) bool {
+	if tx.obsKeys != nil {
+		_, ok := tx.obsKeys[obsDedupKey{observerID, pathJSON}]
+		return ok
+	}
+	for _, o := range tx.Observations {
+		if o.ObserverID == observerID && o.PathJSON == pathJSON {
+			return true
+		}
+	}
+	return false
+}
+
+// hasObserver reports whether any observation on tx came from observerID.
+func (tx *StoreTx) hasObserver(observerID string) bool {
+	if tx.observerSet != nil {
+		_, ok := tx.observerSet[observerID]
+		return ok
+	}
+	for _, o := range tx.Observations {
+		if o.ObserverID == observerID {
+			return true
+		}
+	}
+	return false
+}
+
+// addObservation appends obs to tx and keeps UniqueObserverCount and
+// ObservationCount in step. Callers check hasObservation first.
+func (tx *StoreTx) addObservation(obs *StoreObs) {
+	if obs.ObserverID != "" && !tx.hasObserver(obs.ObserverID) {
+		tx.UniqueObserverCount++
+	}
+	tx.Observations = append(tx.Observations, obs)
+	tx.ObservationCount++
+	switch {
+	case tx.obsKeys != nil:
+		tx.obsKeys[obsDedupKey{obs.ObserverID, obs.PathJSON}] = struct{}{}
+		if obs.ObserverID != "" {
+			tx.observerSet[obs.ObserverID] = struct{}{}
+		}
+	case len(tx.Observations) >= obsIndexThreshold:
+		tx.obsKeys = make(map[obsDedupKey]struct{}, 2*len(tx.Observations))
+		tx.observerSet = make(map[string]struct{}, len(tx.Observations))
+		for _, o := range tx.Observations {
+			tx.obsKeys[obsDedupKey{o.ObserverID, o.PathJSON}] = struct{}{}
+			if o.ObserverID != "" {
+				tx.observerSet[o.ObserverID] = struct{}{}
+			}
+		}
+	}
+}

cmd/server/obs_dedup_test.go

@@ -5,129 +5,108 @@ import (
 	"testing"
 )
 
-// TestObsDedupCorrectness verifies that the map-based dedup produces correct
-// results: no duplicate observations (same observerID + pathJSON) on a single
-// transmission.
-func TestObsDedupCorrectness(t *testing.T) {
-	tx := &StoreTx{
-		ID:      1,
-		Hash:    "abc123",
-		obsKeys: make(map[string]bool),
+func addTestObs(tx *StoreTx, observerID, pathJSON string) bool {
+	if tx.hasObservation(observerID, pathJSON) {
+		return false
 	}
+	tx.addObservation(&StoreObs{ObserverID: observerID, PathJSON: pathJSON})
+	return true
+}
 
-	// Add 5 unique observations
+// TestObsDedupCorrectness: the same observer with the same path is a
+// duplicate; a new path or a new observer is not.
+func TestObsDedupCorrectness(t *testing.T) {
+	tx := &StoreTx{ID: 1, Hash: "abc123"}
 	for i := 0; i < 5; i++ {
-		obsID := fmt.Sprintf("obs-%d", i)
-		pathJSON := fmt.Sprintf(`["path-%d"]`, i)
-		dk := obsID + "|" + pathJSON
-		if tx.obsKeys[dk] {
+		if !addTestObs(tx, fmt.Sprintf("obs-%d", i), fmt.Sprintf(`["path-%d"]`, i)) {
 			t.Fatalf("observation %d should not be a duplicate", i)
 		}
-		tx.Observations = append(tx.Observations, &StoreObs{
-			ID:         i,
-			ObserverID: obsID,
-			PathJSON:   pathJSON,
-		})
-		tx.obsKeys[dk] = true
-		tx.ObservationCount++
 	}
-
-	if tx.ObservationCount != 5 {
-		t.Fatalf("expected 5 observations, got %d", tx.ObservationCount)
+	if tx.ObservationCount != 5 || len(tx.Observations) != 5 {
+		t.Fatalf("expected 5 observations, got count=%d len=%d", tx.ObservationCount, len(tx.Observations))
 	}
-
-	// Try to add duplicates of each — all should be rejected
 	for i := 0; i < 5; i++ {
-		obsID := fmt.Sprintf("obs-%d", i)
-		pathJSON := fmt.Sprintf(`["path-%d"]`, i)
-		dk := obsID + "|" + pathJSON
-		if !tx.obsKeys[dk] {
+		if !tx.hasObservation(fmt.Sprintf("obs-%d", i), fmt.Sprintf(`["path-%d"]`, i)) {
 			t.Fatalf("observation %d should be detected as duplicate", i)
 		}
 	}
-
-	// Same observer, different path — should NOT be a duplicate
-	dk := "obs-0" + "|" + `["different-path"]`
-	if tx.obsKeys[dk] {
+	if tx.hasObservation("obs-0", `["different-path"]`) {
 		t.Fatal("different path should not be a duplicate")
 	}
-
-	// Different observer, same path — should NOT be a duplicate
-	dk = "obs-new" + "|" + `["path-0"]`
-	if tx.obsKeys[dk] {
+	if tx.hasObservation("obs-new", `["path-0"]`) {
 		t.Fatal("different observer should not be a duplicate")
 	}
 }
 
-// TestObsDedupNilMapSafety ensures obsKeys lazy init works for pre-existing
-// transmissions that may not have the map initialized.
-func TestObsDedupNilMapSafety(t *testing.T) {
-	tx := &StoreTx{ID: 1, Hash: "abc"}
-	// obsKeys is nil — the lazy init pattern used in IngestNewFromDB/IngestNewObservations
-	if tx.obsKeys == nil {
-		tx.obsKeys = make(map[string]bool)
+// TestObsDedupSmallTxCarriesNoMaps: most transmissions have a handful of
+// observations, and two maps each cost them a few hundred bytes. Below the
+// threshold the checks scan tx.Observations instead.
+func TestObsDedupSmallTxCarriesNoMaps(t *testing.T) {
+	tx := &StoreTx{ID: 1}
+	for i := 0; i < obsIndexThreshold-1; i++ {
+		addTestObs(tx, fmt.Sprintf("obs-%d", i%4), fmt.Sprintf(`["p%d"]`, i))
 	}
-	dk := "obs1|path1"
-	if tx.obsKeys[dk] {
-		t.Fatal("should not be duplicate on empty map")
+	if len(tx.Observations) != obsIndexThreshold-1 {
+		t.Fatalf("len = %d, want %d", len(tx.Observations), obsIndexThreshold-1)
 	}
-	tx.obsKeys[dk] = true
-	if !tx.obsKeys[dk] {
-		t.Fatal("should be duplicate after insert")
+	if tx.obsKeys != nil || tx.observerSet != nil {
+		t.Fatal("a transmission under the threshold should not carry dedup maps")
+	}
+	if tx.UniqueObserverCount != 4 {
+		t.Fatalf("UniqueObserverCount = %d, want 4", tx.UniqueObserverCount)
 	}
 }
 
-// BenchmarkObsDedupMap benchmarks the map-based O(1) dedup approach.
-func BenchmarkObsDedupMap(b *testing.B) {
-	for _, obsCount := range []int{10, 50, 100, 500} {
-		b.Run(fmt.Sprintf("obs=%d", obsCount), func(b *testing.B) {
-			// Pre-populate a tx with obsCount observations
-			tx := &StoreTx{
-				ID:      1,
-				obsKeys: make(map[string]bool),
-			}
-			for i := 0; i < obsCount; i++ {
-				obsID := fmt.Sprintf("obs-%d", i)
-				pathJSON := fmt.Sprintf(`["hop-%d"]`, i)
-				dk := obsID + "|" + pathJSON
-				tx.Observations = append(tx.Observations, &StoreObs{
-					ObserverID: obsID,
-					PathJSON:   pathJSON,
-				})
-				tx.obsKeys[dk] = true
-			}
-
-			// Benchmark: check dedup for a new observation (not duplicate)
-			newDK := "new-obs|new-path"
-			b.ResetTimer()
-			for i := 0; i < b.N; i++ {
-				_ = tx.obsKeys[newDK]
-			}
-		})
+// TestObsDedupAcrossTheThreshold: past the threshold the maps take over and
+// give the same answers as the scan did.
+func TestObsDedupAcrossTheThreshold(t *testing.T) {
+	tx := &StoreTx{ID: 1}
+	n := obsIndexThreshold * 3
+	for i := 0; i < n; i++ {
+		if !addTestObs(tx, fmt.Sprintf("obs-%d", i%10), fmt.Sprintf(`["p%d"]`, i)) {
+			t.Fatalf("observation %d rejected", i)
+		}
+	}
+	if tx.obsKeys == nil || tx.observerSet == nil {
+		t.Fatal("a busy transmission should index its observations")
+	}
+	if tx.UniqueObserverCount != 10 || tx.ObservationCount != n {
+		t.Fatalf("unique=%d count=%d, want 10 and %d", tx.UniqueObserverCount, tx.ObservationCount, n)
+	}
+	for i := 0; i < n; i++ {
+		if addTestObs(tx, fmt.Sprintf("obs-%d", i%10), fmt.Sprintf(`["p%d"]`, i)) {
+			t.Fatalf("duplicate %d accepted after the threshold", i)
+		}
+	}
+	if !addTestObs(tx, "obs-new", `["p0"]`) || tx.UniqueObserverCount != 11 {
+		t.Fatalf("new observer: unique=%d, want 11", tx.UniqueObserverCount)
 	}
 }
 
-// BenchmarkObsDedupLinear benchmarks the old O(n) linear scan for comparison.
-func BenchmarkObsDedupLinear(b *testing.B) {
-	for _, obsCount := range []int{10, 50, 100, 500} {
+// TestObsDedupEmptyObserverNotCounted: an observation without an observer
+// ID is kept but does not count as a distinct observer.
+func TestObsDedupEmptyObserverNotCounted(t *testing.T) {
+	tx := &StoreTx{ID: 1}
+	addTestObs(tx, "", `["a"]`)
+	addTestObs(tx, "", `["b"]`)
+	if tx.UniqueObserverCount != 0 || tx.ObservationCount != 2 {
+		t.Fatalf("unique=%d count=%d, want 0 and 2", tx.UniqueObserverCount, tx.ObservationCount)
+	}
+}
+
+// BenchmarkObsDedup measures a miss (the common case on ingest) at sizes
+// either side of the threshold.
+func BenchmarkObsDedup(b *testing.B) {
+	for _, obsCount := range []int{4, 16, 40, 63, 64, 500} {
 		b.Run(fmt.Sprintf("obs=%d", obsCount), func(b *testing.B) {
 			tx := &StoreTx{ID: 1}
 			for i := 0; i < obsCount; i++ {
-				tx.Observations = append(tx.Observations, &StoreObs{
-					ObserverID: fmt.Sprintf("obs-%d", i),
-					PathJSON:   fmt.Sprintf(`["hop-%d"]`, i),
-				})
+				addTestObs(tx, fmt.Sprintf("%064d", i), fmt.Sprintf(`["hop-%d"]`, i))
 			}
-
-			newObsID := "new-obs"
-			newPath := "new-path"
+			newID := fmt.Sprintf("%064d", obsCount+1)
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
-				for _, existing := range tx.Observations {
-					if existing.ObserverID == newObsID && existing.PathJSON == newPath {
-						break
-					}
-				}
+				_ = tx.hasObservation(newID, `["new-path"]`)
 			}
 		})
 	}

cmd/server/store.go

@@ -64,9 +64,10 @@ type StoreTx struct {
 	AdvertRouteEvidence uint8                  // union of known flood/direct-empty-path evidence
 	decodedOnce         sync.Once              // guards parsedDecoded
 	parsedDecoded       map[string]interface{} // cached json.Unmarshal of DecodedJSON
-	// Dedup map: "observerID|pathJSON" → true for O(1) duplicate checks
-	obsKeys     map[string]bool
-	observerSet map[string]bool // unique observer IDs (for UniqueObserverCount)
+	// Dedup maps, built only once a transmission passes obsIndexThreshold
+	// observations (see obs_dedup.go); nil below it.
+	obsKeys     map[obsDedupKey]struct{}
+	observerSet map[string]struct{} // distinct observer IDs (for UniqueObserverCount)
 	// accountedBytes is what trackedBytes was last charged for this tx
 	// (observations excluded). See rechargeTx.
 	accountedBytes int64
@@ -970,8 +971,6 @@ func (s *PacketStore) Load() error {
 				PayloadType: nullIntPtr(payloadType),
 				DecodedJSON: nullStrVal(decodedJSON),
 				ScopeName:   nullStrPtr(scopeName),
-				obsKeys:     make(map[string]bool),
-				observerSet: make(map[string]bool),
 			}
 			s.byHash[hashStr] = tx
 			s.packets = append(s.packets, tx)
@@ -994,8 +993,7 @@ func (s *PacketStore) Load() error {
 			obsPJ := nullStrVal(pathJSON)
 
 			// Dedup: skip if same observer + same path already loaded (O(1) map lookup)
-			dk := obsIDStr + "|" + obsPJ
-			if tx.obsKeys[dk] {
+			if tx.hasObservation(obsIDStr, obsPJ) {
 				continue
 			}
 
@@ -1037,13 +1035,7 @@ func (s *PacketStore) Load() error {
 				}
 			}
 
-			tx.Observations = append(tx.Observations, obs)
-			tx.obsKeys[dk] = true
-			if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] {
-				tx.observerSet[obs.ObserverID] = true
-				tx.UniqueObserverCount++
-			}
-			tx.ObservationCount++
+			tx.addObservation(obs)
 			if obs.Timestamp > tx.LatestSeen {
 				tx.LatestSeen = obs.Timestamp
 			}
@@ -1314,8 +1306,6 @@ func (s *PacketStore) loadChunk(from, to time.Time) error {
 				PayloadType: nullIntPtr(payloadType),
 				DecodedJSON: nullStrVal(decodedJSON),
 				ScopeName:   nullStrPtr(scopeName),
-				obsKeys:     make(map[string]bool),
-				observerSet: make(map[string]bool),
 			}
 			localByHash[hashStr] = tx
 			localPackets = append(localPackets, tx)
@@ -1331,8 +1321,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error {
 			obsIDStr := nullStrVal(observerID)
 			obsPJ := nullStrVal(pathJSON)
 
-			dk := obsIDStr + "|" + obsPJ
-			if tx.obsKeys[dk] {
+			if tx.hasObservation(obsIDStr, obsPJ) {
 				continue
 			}
 
@@ -1349,13 +1338,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error {
 				Timestamp:      normalizeTimestamp(nullStrVal(obsTimestamp)),
 			}
 
-			tx.Observations = append(tx.Observations, obs)
-			tx.obsKeys[dk] = true
-			if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] {
-				tx.observerSet[obs.ObserverID] = true
-				tx.UniqueObserverCount++
-			}
-			tx.ObservationCount++
+			tx.addObservation(obs)
 			if obs.Timestamp > tx.LatestSeen {
 				tx.LatestSeen = obs.Timestamp
 			}
@@ -2944,8 +2927,6 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 				PayloadType: r.payloadType,
 				DecodedJSON: r.decodedJSON,
 				ScopeName:   r.scopeName,
-				obsKeys:     make(map[string]bool),
-				observerSet: make(map[string]bool),
 			}
 			s.byHash[r.hash] = tx
 			s.packets = append(s.packets, tx) // oldest-first; new items go to tail
@@ -2977,12 +2958,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 		if r.obsID != nil {
 			oid := *r.obsID
 			// Dedup (O(1) map lookup)
-			dk := r.observerID + "|" + r.pathJSON
-			if tx.obsKeys == nil {
-				tx.obsKeys = make(map[string]bool)
-				tx.observerSet = make(map[string]bool)
-			}
-			if tx.obsKeys[dk] {
+			if tx.hasObservation(r.observerID, r.pathJSON) {
 				continue
 			}
 
@@ -3017,13 +2993,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
 				}
 				broadcastRP[*r.obsID] = rpForBroadcast
 			}
-			tx.Observations = append(tx.Observations, obs)
-			tx.obsKeys[dk] = true
-			if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] {
-				tx.observerSet[obs.ObserverID] = true
-				tx.UniqueObserverCount++
-			}
-			tx.ObservationCount++
+			tx.addObservation(obs)
 			if obs.Timestamp > tx.LatestSeen {
 				tx.LatestSeen = obs.Timestamp
 			}
@@ -3307,12 +3277,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 		}
 
 		// Dedup by observer + path (O(1) map lookup)
-		dk := r.observerID + "|" + r.pathJSON
-		if tx.obsKeys == nil {
-			tx.obsKeys = make(map[string]bool)
-			tx.observerSet = make(map[string]bool)
-		}
-		if tx.obsKeys[dk] {
+		if tx.hasObservation(r.observerID, r.pathJSON) {
 			continue
 		}
 
@@ -3352,13 +3317,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
 			obsRPMap[r.obsID] = obsResolvedPath
 		}
 
-		tx.Observations = append(tx.Observations, obs)
-		tx.obsKeys[dk] = true
-		if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] {
-			tx.observerSet[obs.ObserverID] = true
-			tx.UniqueObserverCount++
-		}
-		tx.ObservationCount++
+		tx.addObservation(obs)
 		newObs = append(newObs, obs)
 		if obs.Timestamp > tx.LatestSeen {
 			tx.LatestSeen = obs.Timestamp
@@ -4796,7 +4755,7 @@ func (s *PacketStore) buildDistanceIndex() {
 //
 // Issue #743: Previous estimates missed major per-packet allocations:
 // - spTxIndex: O(path²) entries per tx (50-150MB at scale)
-// - Per-tx maps: obsKeys, observerSet (~11MB at scale)
+// - Per-tx dedup maps (now only past obsIndexThreshold observations)
 // - byPathHop index entries (20-40MB at scale)
 // Note: ResolvedPath per-obs overhead eliminated by #800 refactor.
 const (
@@ -4807,9 +4766,6 @@ const (
 	numIndexesPerTx   = 5   // byHash, byTxID, byNode, byPayloadType, nodeHashes entries
 	numIndexesPerObs  = 2   // byObsID, byObserver entries
 
-	// Per-tx map overhead (obsKeys + observerSet): map header + initial buckets
-	perTxMapsBytes = 200
-
 	// Per path hop: byPathHop index entry (pointer + map bucket)
 	perPathHopBytes = 50
 
@@ -4821,24 +4777,16 @@ const (
 	// 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).
-// Includes per-tx maps (obsKeys, observerSet), byPathHop entries, and spTxIndex subpath entries.
+// Includes byPathHop entries and spTxIndex subpath entries.
 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
-
 	// Path-dependent costs
 	hops := int64(len(txGetParsedPath(tx)))
 	base += hops * perPathHopBytes
@@ -4860,14 +4808,12 @@ func estimateStoreTxBytesTypical(numObs int) int64 {
 	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 + 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
@@ -4880,7 +4826,6 @@ func estimateStoreObsBytes(obs *StoreObs) int64 {
 	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
 }

cmd/server/tracked_bytes_test.go

@@ -173,8 +173,7 @@ 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{}}
+	tx := &StoreTx{ID: 1, Hash: "aabb", FirstSeen: time.Now().UTC().Add(-2 * time.Hour).Format(time.RFC3339)}
 	store.trackedBytes += rechargeTx(tx)
 	bare := store.trackedBytes