perf: replace O(n²) observation dedup with map-based O(n) (#355) (#543)

## Summary

Fixes #355 — replaces O(n²) observation dedup in `Load()`,
`IngestNewFromDB()`, and `IngestNewObservations()` with an O(1)
map-based lookup.

## Changes

- Added `obsKeys map[string]bool` field to `StoreTx` for O(1) dedup
keyed on `observerID + "|" + pathJSON`
- Replaced all 3 linear-scan dedup sites in `store.go` with map lookups
- Lazy-init `obsKeys` for transmissions created before this change (in
`IngestNewFromDB` and `IngestNewObservations`)
- Added regression test (`TestObsDedupCorrectness`) verifying dedup
correctness
- Added nil-map safety test (`TestObsDedupNilMapSafety`)
- Added benchmark comparing map vs linear scan

## Benchmark Results (ARM64, 16 cores)

| Observations | Map (O(1)) | Linear (O(n)) | Speedup |
|---|---|---|---|
| 10 | 34 ns/op | 41 ns/op | 1.2x |
| 50 | 34 ns/op | 186 ns/op | 5.5x |
| 100 | 34 ns/op | 361 ns/op | 10.6x |
| 500 | 34 ns/op | 4,903 ns/op | **146x** |

Map lookup is constant time regardless of observation count. The linear
scan degrades quadratically — at 500 observations per transmission
(realistic for popular packets seen by many observers), the old code is
146x slower per dedup check.

All existing tests pass.

---------

Co-authored-by: you <you@example.com>
This commit is contained in:
Kpa-clawbot
2026-04-03 13:33:26 -07:00
committed by GitHub
co-authored by you
parent b472c8de30
commit 9b9f396af5
2 changed files with 154 additions and 25 deletions
+134
View File
@@ -0,0 +1,134 @@
package main
import (
"fmt"
"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),
}
// Add 5 unique observations
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] {
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)
}
// 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] {
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] {
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] {
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)
}
dk := "obs1|path1"
if tx.obsKeys[dk] {
t.Fatal("should not be duplicate on empty map")
}
tx.obsKeys[dk] = true
if !tx.obsKeys[dk] {
t.Fatal("should be duplicate after insert")
}
}
// 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]
}
})
}
}
// BenchmarkObsDedupLinear benchmarks the old O(n) linear scan for comparison.
func BenchmarkObsDedupLinear(b *testing.B) {
for _, obsCount := range []int{10, 50, 100, 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),
})
}
newObsID := "new-obs"
newPath := "new-path"
b.ResetTimer()
for i := 0; i < b.N; i++ {
for _, existing := range tx.Observations {
if existing.ObserverID == newObsID && existing.PathJSON == newPath {
break
}
}
}
})
}
}
+20 -25
View File
@@ -43,6 +43,8 @@ type StoreTx struct {
// Cached parsed fields (set once, read many)
parsedPath []string // cached parsePathJSON result
pathParsed bool // whether parsedPath has been set
// Dedup map: "observerID|pathJSON" → true for O(1) duplicate checks
obsKeys map[string]bool
}
// StoreObs is a lean in-memory observation (no duplication of transmission fields).
@@ -253,6 +255,7 @@ func (s *PacketStore) Load() error {
RouteType: nullIntPtr(routeType),
PayloadType: nullIntPtr(payloadType),
DecodedJSON: nullStrVal(decodedJSON),
obsKeys: make(map[string]bool),
}
s.byHash[hashStr] = tx
s.packets = append(s.packets, tx)
@@ -269,15 +272,9 @@ func (s *PacketStore) Load() error {
obsIDStr := nullStrVal(observerID)
obsPJ := nullStrVal(pathJSON)
// Dedup: skip if same observer + same path already loaded
isDupe := false
for _, existing := range tx.Observations {
if existing.ObserverID == obsIDStr && existing.PathJSON == obsPJ {
isDupe = true
break
}
}
if isDupe {
// Dedup: skip if same observer + same path already loaded (O(1) map lookup)
dk := obsIDStr + "|" + obsPJ
if tx.obsKeys[dk] {
continue
}
@@ -295,6 +292,7 @@ func (s *PacketStore) Load() error {
}
tx.Observations = append(tx.Observations, obs)
tx.obsKeys[dk] = true
tx.ObservationCount++
if obs.Timestamp > tx.LatestSeen {
tx.LatestSeen = obs.Timestamp
@@ -1061,6 +1059,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
RouteType: r.routeType,
PayloadType: r.payloadType,
DecodedJSON: r.decodedJSON,
obsKeys: make(map[string]bool),
}
s.byHash[r.hash] = tx
s.packets = append(s.packets, tx) // oldest-first; new items go to tail
@@ -1081,15 +1080,12 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
if r.obsID != nil {
oid := *r.obsID
// Dedup
isDupe := false
for _, existing := range tx.Observations {
if existing.ObserverID == r.observerID && existing.PathJSON == r.pathJSON {
isDupe = true
break
}
// Dedup (O(1) map lookup)
dk := r.observerID + "|" + r.pathJSON
if tx.obsKeys == nil {
tx.obsKeys = make(map[string]bool)
}
if isDupe {
if tx.obsKeys[dk] {
continue
}
@@ -1106,6 +1102,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac
Timestamp: normalizeTimestamp(r.obsTS),
}
tx.Observations = append(tx.Observations, obs)
tx.obsKeys[dk] = true
tx.ObservationCount++
if obs.Timestamp > tx.LatestSeen {
tx.LatestSeen = obs.Timestamp
@@ -1326,15 +1323,12 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
continue // transmission not yet in store
}
// Dedup by observer + path
isDupe := false
for _, existing := range tx.Observations {
if existing.ObserverID == r.observerID && existing.PathJSON == r.pathJSON {
isDupe = true
break
}
// Dedup by observer + path (O(1) map lookup)
dk := r.observerID + "|" + r.pathJSON
if tx.obsKeys == nil {
tx.obsKeys = make(map[string]bool)
}
if isDupe {
if tx.obsKeys[dk] {
continue
}
@@ -1351,6 +1345,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string]
Timestamp: normalizeTimestamp(r.timestamp),
}
tx.Observations = append(tx.Observations, obs)
tx.obsKeys[dk] = true
tx.ObservationCount++
if obs.Timestamp > tx.LatestSeen {
tx.LatestSeen = obs.Timestamp