perf(store): avoid eager per-transmission observation dedup maps

cmd/server/chunked_load.go

@@ -637,8 +637,6 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 				PayloadType: nullIntPtr(payloadType),
 				DecodedJSON: nullStrVal(decodedJSON),
 				ScopeName:   nullStrVal(scopeName),
-				obsKeys:     make(map[string]bool),
-				observerSet: make(map[string]bool),
 			}
 			s.byHash[hashStr] = tx
 			s.mergeRouteMaskOrQueue(tx, routeMask)
@@ -661,8 +659,8 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 			obsIDStr := nullStrVal(observerID)
 			obsPJ := nullStrVal(pathJSON)
 
-			dk := obsIDStr + "|" + obsPJ
-			if tx.obsKeys[dk] {
+			dup, knownObserver := tx.lookupObservation(obsIDStr, obsPJ)
+			if dup {
 				continue
 			}
 
@@ -692,14 +690,8 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 			s.indexObservationRelayHops(tx, decodePersistedRelayPath(nullStrVal(resolvedPathStr)), obsPJ, obsIDStr, relayPM, hopsSeen, coldLoadAmbiguousHopsSkipped)
 
 			tx.mergeObservedPathHashSize(obsPJ)
-			tx.Observations = append(tx.Observations, obs)
-			tx.obsKeys[dk] = true
+			s.trackedBytes += tx.addObservation(obs, knownObserver)
 			s.analyticsHistory.observedFrom = earlierNodeAnalyticsEvidence(s.analyticsHistory.observedFrom, obs.Timestamp)
-			if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] {
-				tx.observerSet[obs.ObserverID] = true
-				tx.UniqueObserverCount++
-			}
-			tx.ObservationCount++
 			if obs.Timestamp > tx.LatestSeen {
 				tx.LatestSeen = obs.Timestamp
 			}

cmd/server/eviction_test.go

@@ -565,11 +565,10 @@ func TestEstimateStoreTxBytes(t *testing.T) {
 		PathJSON:    `["aa","bb"]`,
 	}
 	est := estimateStoreTxBytes(tx)
-	// Manual calculation: base + string lengths + index entries + perTxMaps + path hops + subpaths
+	// Manual calculation: base + string lengths + index entries + 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 {
 		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)
-	// + the dedup key: obsKeyEntryBytes + len(ObserverID) + len(PathJSON)
-	expected := int64(192+6+6+2*48) + int64(obsKeyEntryBytes+6+6)
+	// storeObsBaseBytes(192) + len(ObserverID=6) + len(PathJSON=6) + 2*48(96);
+	// no dedup key is allocated or charged per observation (#411)
+	expected := int64(192 + 6 + 6 + 2*48)
 	if est != expected {
 		t.Fatalf("estimateStoreObsBytes = %d, want %d", est, expected)
 	}

cmd/server/hash_migrate.go

@@ -346,13 +346,9 @@ func (s *PacketStore) rekeyNodeHashesSweep(renames map[string]string) {
 // from the path-hop index, which reads it; the index entries of loser itself
 // are removed there.
 func (s *PacketStore) moveObservations(winner, loser *StoreTx, m *hashMerge) {
-	if winner.obsKeys == nil {
-		winner.obsKeys = make(map[string]bool)
-		winner.observerSet = make(map[string]bool)
-	}
 	for _, o := range loser.Observations {
-		dk := o.ObserverID + "|" + o.PathJSON
-		if winner.obsKeys[dk] {
+		dup, knownObserver := winner.lookupObservation(o.ObserverID, o.PathJSON)
+		if dup {
 			delete(s.byObsID, o.ID)
 			s.totalObs--
 			s.trackedBytes -= estimateStoreObsBytes(o)
@@ -363,13 +359,7 @@ func (s *PacketStore) moveObservations(winner, loser *StoreTx, m *hashMerge) {
 			continue
 		}
 		o.TransmissionID = winner.ID
-		winner.Observations = append(winner.Observations, o)
-		winner.obsKeys[dk] = true
-		if o.ObserverID != "" && !winner.observerSet[o.ObserverID] {
-			winner.observerSet[o.ObserverID] = true
-			winner.UniqueObserverCount++
-		}
-		winner.ObservationCount++
+		s.trackedBytes += winner.addObservation(o, knownObserver)
 		if o.Timestamp > winner.LatestSeen {
 			winner.LatestSeen = o.Timestamp
 		}
@@ -473,6 +463,7 @@ func (s *PacketStore) finishHashMerge(m *hashMerge, metadata pathNodeSnapshot) i
 	for _, l := range m.losers {
 		l.Observations = nil
 		l.ObservationCount = 0
+		l.obsKeys, l.observerSet = nil, nil
 	}
 	s.removeTxsFromByNode(m.loserSet, nodePks)
 	for pt := range affectedPayloadTypes {

cmd/server/hash_migrate_readonly_215_test.go

@@ -431,8 +431,12 @@ func TestHashMigrate_MergesDuplicatesInMemory_215(t *testing.T) {
 						t.Errorf("byObsID[%d] is not the observation of tx %d", o.ID, tx.ID)
 					}
 				}
-				if int(tx.UniqueObserverCount) != len(tx.observerSet) || len(tx.observerSet) != 4 {
-					t.Errorf("group %d survivor has %d unique observers (%d in the set), want 4", g, tx.UniqueObserverCount, len(tx.observerSet))
+				observers := map[string]bool{}
+				for _, o := range tx.Observations {
+					observers[o.ObserverID] = true
+				}
+				if int(tx.UniqueObserverCount) != len(observers) || len(observers) != 4 {
+					t.Errorf("group %d survivor has %d unique observers (%d in its observations), want 4", g, tx.UniqueObserverCount, len(observers))
 				}
 				for k := 1; k <= 3; k++ {
 					if loser := s.byTxID[g*10+k]; k != 1 && loser != nil {

cmd/server/obs_dedup.go

@@ -0,0 +1,98 @@
+package main
+
+// Observation dedup per transmission (#411).
+//
+// The store keeps one observation per (observer ID, path_json) pair of a
+// transmission and counts the distinct non-empty observer IDs. Most
+// transmissions are heard a handful of times, so a scan of tx.Observations
+// answers both questions without any per-transmission allocation. A
+// transmission that reaches obsIndexThreshold observations builds two maps
+// (obsKeys, observerSet) keyed by the strings the observations already hold,
+// so a busy transmission keeps O(1) checks and no key is ever concatenated.
+//
+// The threshold comes from the distributions measured for #411. On a light
+// synthetic 30K-transmission set (median 4 distinct pairs, p99 21, p99.9 36)
+// about 0.2% of transmissions reach 32. With the production mean (about 18
+// observations per transmission: median 15, p90 35, p99 69) about 13% build
+// the maps, and the heap still drops by about 19%. A threshold of 64 measured
+// about 5% less heap on top of that and is a possible follow-up.
+// BenchmarkObsDedup: a missing check scans
+// 31 observations in 100-130 ns (180 ns when one observer holds them all),
+// against 55-95 ns for the old lookup, which also allocated an 80-100 byte
+// key on every check; from 32 on it is a 25 ns map lookup.
+const obsIndexThreshold = 32
+
+// obsDedupKey identifies an observation within its transmission.
+type obsDedupKey struct {
+	observerID, pathJSON string
+}
+
+// Self-accounted cost of the maps of a transmission past obsIndexThreshold,
+// measured for #411 (Go swiss maps, 32 to 600 entries): about 86 B per
+// obsKeys entry and 40 B per observerSet entry including growth slack, plus
+// the two map headers.
+const (
+	obsIndexRecordBytes   = 96
+	obsIndexKeyBytes      = 88
+	obsIndexObserverBytes = 40
+)
+
+// obsIndexBytes is what trackedBytes holds for tx's dedup maps. It is a
+// function of the maps alone, so eviction re-derives the charge (see
+// txChargedBytes); the maps only grow while tx is in the store.
+func obsIndexBytes(tx *StoreTx) int64 {
+	if tx.obsKeys == nil {
+		return 0
+	}
+	return obsIndexRecordBytes + int64(len(tx.obsKeys))*obsIndexKeyBytes + int64(len(tx.observerSet))*obsIndexObserverBytes
+}
+
+// lookupObservation reports whether tx already holds an observation by
+// observerID on pathJSON, and whether observerID (when not empty) already
+// observed tx.
+func (tx *StoreTx) lookupObservation(observerID, pathJSON string) (dup, knownObserver bool) {
+	if tx.obsKeys != nil {
+		_, dup = tx.obsKeys[obsDedupKey{observerID, pathJSON}]
+		_, knownObserver = tx.observerSet[observerID]
+		return dup, knownObserver
+	}
+	for _, o := range tx.Observations {
+		if o.ObserverID != observerID {
+			continue
+		}
+		if o.PathJSON == pathJSON {
+			return true, observerID != ""
+		}
+		knownObserver = observerID != ""
+	}
+	return false, knownObserver
+}
+
+// addObservation appends obs, for which lookupObservation reported no
+// duplicate and knownObserver, and keeps ObservationCount,
+// UniqueObserverCount and the dedup maps in step. It returns the change in
+// obsIndexBytes(tx), for the caller to add to trackedBytes.
+func (tx *StoreTx) addObservation(obs *StoreObs, knownObserver bool) int64 {
+	before := obsIndexBytes(tx)
+	if obs.ObserverID != "" && !knownObserver {
+		tx.UniqueObserverCount++
+		if tx.observerSet != nil {
+			tx.observerSet[obs.ObserverID] = struct{}{}
+		}
+	}
+	tx.Observations = append(tx.Observations, obs)
+	tx.ObservationCount++
+	if tx.obsKeys != nil {
+		tx.obsKeys[obsDedupKey{obs.ObserverID, obs.PathJSON}] = struct{}{}
+	} else if len(tx.Observations) >= obsIndexThreshold {
+		tx.obsKeys = make(map[obsDedupKey]struct{}, len(tx.Observations))
+		tx.observerSet = make(map[string]struct{})
+		for _, o := range tx.Observations {
+			tx.obsKeys[obsDedupKey{o.ObserverID, o.PathJSON}] = struct{}{}
+			if o.ObserverID != "" {
+				tx.observerSet[o.ObserverID] = struct{}{}
+			}
+		}
+	}
+	return obsIndexBytes(tx) - before
+}