diff --git a/cmd/server/obs_dedup_test.go b/cmd/server/obs_dedup_test.go new file mode 100644 index 00000000..82068562 --- /dev/null +++ b/cmd/server/obs_dedup_test.go @@ -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 + } + } + } + }) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 346d2e2a..5723fd3d 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -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