perf(store): key byNode's de-dup sets by transmission ID

cmd/server/area_filter_test.go

@@ -85,7 +85,7 @@ func newTestStoreWithDB(t testing.TB, db *DB, cfg *Config) *PacketStore {
 		byObserver:         make(map[string][]*StoreObs),
 		byHash:             make(map[string]*StoreTx),
 		byPayloadType:      make(map[int][]*StoreTx),
-		nodeHashes:         make(map[string]map[string]bool),
+		nodeHashes:         newNodeSets(),
 		byPathHop:          make(map[string][]*StoreTx),
 		advertPubkeys:      make(map[string]int),
 		rfCache:            make(map[string]*cachedResult),

cmd/server/bulk_health_region_test.go

@@ -0,0 +1,84 @@
+package main
+
+import (
+	"sort"
+	"testing"
+	"time"
+)
+
+// GetBulkHealth's region filter keeps a node when one of its transmissions
+// (an advert from it, or a packet whose path it relayed) was heard by an
+// observer in the region. It matches the region's observations to the node
+// sets by transmission ID, so this fixture gives observations IDs that differ
+// from their transmissions' IDs: a filter keyed by the wrong ID fails here.
+func TestGetBulkHealthRegionKeepsNodesHeardThere(t *testing.T) {
+	db := setupTestDB(t)
+	defer db.Close()
+	now := time.Now().UTC()
+	ts, epoch := now.Add(-time.Hour).Format(time.RFC3339), now.Add(-time.Hour).Unix()
+
+	const (
+		a = "aaaa000000000001" // heard only in EDI
+		b = "bbbb000000000002" // heard only in GLA
+		c = "cccc000000000003" // adverts in GLA, and relays a packet heard in EDI
+	)
+	exec := func(q string, args ...interface{}) {
+		t.Helper()
+		if _, err := db.conn.Exec(q, args...); err != nil {
+			t.Fatalf("%v\n%s", err, q)
+		}
+	}
+	exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) VALUES ('obsE', 'Edinburgh', 'EDI', ?, ?, 1)`, ts, ts)
+	exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) VALUES ('obsG', 'Glasgow', 'GLA', ?, ?, 1)`, ts, ts)
+	for _, pk := range []string{a, b, c} {
+		exec(`INSERT INTO nodes (public_key, name, role, lat, lon, last_seen, first_seen, advert_count) VALUES (?, ?, 'repeater', 56.0, -3.5, ?, ?, 1)`, pk, "R-"+pk[:4], ts, ts)
+	}
+	advert := func(id int, pk string) {
+		exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) VALUES (?, 'AA00', ?, ?, 1, 4, ?, ?)`,
+			id, "h"+pk, ts, `{"pubKey":"`+pk+`","name":"R","type":"ADVERT"}`, pk)
+	}
+	advert(10, a)
+	advert(20, b)
+	advert(30, c)
+	// A channel message heard in EDI that c relayed.
+	exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES (40, 'CC00', 'hchan', ?, 1, 5, '{"type":"CHAN","text":"x"}')`, ts)
+
+	// Observation IDs 1-4 never equal a transmission ID (10-40). observer_idx
+	// is the observers rowid: 1 = obsE (EDI), 2 = obsG (GLA).
+	exec(`INSERT INTO observations (id, transmission_id, observer_idx, snr, rssi, path_json, timestamp) VALUES (1, 10, 1, 5, -90, '[]', ?)`, epoch)
+	exec(`INSERT INTO observations (id, transmission_id, observer_idx, snr, rssi, path_json, timestamp) VALUES (2, 20, 2, 5, -90, '[]', ?)`, epoch)
+	exec(`INSERT INTO observations (id, transmission_id, observer_idx, snr, rssi, path_json, timestamp) VALUES (3, 30, 2, 5, -90, '[]', ?)`, epoch)
+	exec(`INSERT INTO observations (id, transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path) VALUES (4, 40, 1, 5, -90, '["cc"]', ?, ?)`, epoch, `["`+c+`"]`)
+
+	store := NewPacketStore(db, nil)
+	if err := store.Load(); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	keys := func(region string) []string {
+		var out []string
+		for _, r := range store.GetBulkHealth(50, region, "") {
+			out = append(out, r["public_key"].(string))
+		}
+		sort.Strings(out)
+		return out
+	}
+	for _, tc := range []struct {
+		region string
+		want   []string
+	}{
+		{"EDI", []string{a, c}},
+		{"GLA", []string{b, c}},
+	} {
+		got := keys(tc.region)
+		if len(got) != len(tc.want) {
+			t.Errorf("region %s: got %v, want %v", tc.region, got, tc.want)
+			continue
+		}
+		for i := range got {
+			if got[i] != tc.want[i] {
+				t.Errorf("region %s: got %v, want %v", tc.region, got, tc.want)
+				break
+			}
+		}
+	}
+}

cmd/server/channel_analytics_test.go

@@ -18,7 +18,7 @@ func newChannelTestStore(packets []*StoreTx) *PacketStore {
 		byObserver:     make(map[string][]*StoreObs),
 		byNode:         make(map[string][]*StoreTx),
 		byPathHop:      make(map[string][]*StoreTx),
-		nodeHashes:     make(map[string]map[string]bool),
+		nodeHashes:     newNodeSets(),
 		byPayloadType:  make(map[int][]*StoreTx),
 		rfCache:        make(map[string]*cachedResult),
 		topoCache:      make(map[string]*cachedResult),

cmd/server/coverage_test.go

@@ -4326,7 +4326,7 @@ func TestResolveRegionObserversCacheMissNewRegion(t *testing.T) {
 func TestIndexByNodePreCheck(t *testing.T) {
 	store := &PacketStore{
 		byNode:     make(map[string][]*StoreTx),
-		nodeHashes: make(map[string]map[string]bool),
+		nodeHashes: newNodeSets(),
 	}
 
 	t.Run("indexes ADVERT with pubKey", func(t *testing.T) {
@@ -4385,7 +4385,7 @@ func TestIndexByNodePreCheck(t *testing.T) {
 func TestIndexByNodeResolvedPath(t *testing.T) {
 	store := &PacketStore{
 		byNode:     make(map[string][]*StoreTx),
-		nodeHashes: make(map[string]map[string]bool),
+		nodeHashes: newNodeSets(),
 	}
 
 	t.Run("decoded JSON pubkeys still indexed", func(t *testing.T) {
@@ -4435,7 +4435,7 @@ func BenchmarkIndexByNode(b *testing.B) {
 	b.Run("no_pubkey_skip", func(b *testing.B) {
 		store := &PacketStore{
 			byNode:     make(map[string][]*StoreTx),
-			nodeHashes: make(map[string]map[string]bool),
+			nodeHashes: newNodeSets(),
 		}
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
@@ -4450,7 +4450,7 @@ func BenchmarkIndexByNode(b *testing.B) {
 	b.Run("with_pubkey_parse", func(b *testing.B) {
 		store := &PacketStore{
 			byNode:     make(map[string][]*StoreTx),
-			nodeHashes: make(map[string]map[string]bool),
+			nodeHashes: newNodeSets(),
 		}
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {

cmd/server/eviction_test.go

@@ -17,7 +17,7 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore
 		byObsID:       make(map[int]*StoreObs, count*2),
 		byObserver:    make(map[string][]*StoreObs),
 		byNode:        make(map[string][]*StoreTx),
-		nodeHashes:    make(map[string]map[string]bool),
+		nodeHashes:    newNodeSets(),
 		byPayloadType: make(map[int][]*StoreTx),
 		distHops:      make([]distHopRecord, 0),
 		distPaths:     make([]distPathRecord, 0),
@@ -73,9 +73,9 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore
 		// Index by node
 		pk := fmt.Sprintf("pk%04d", i)
 		if store.nodeHashes[pk] == nil {
-			store.nodeHashes[pk] = make(map[string]bool)
+			store.nodeHashes[pk] = make(map[int]struct{})
 		}
-		store.nodeHashes[pk][hash] = true
+		store.nodeHashes[pk][txID] = struct{}{}
 		store.byNode[pk] = append(store.byNode[pk], tx)
 
 		// Add to distance index
@@ -255,7 +255,7 @@ func TestEvictStale_CleansResolvedPathNodeIndexes(t *testing.T) {
 		byObsID:              make(map[int]*StoreObs),
 		byObserver:           make(map[string][]*StoreObs),
 		byNode:               make(map[string][]*StoreTx),
-		nodeHashes:           make(map[string]map[string]bool),
+		nodeHashes:           newNodeSets(),
 		byPayloadType:        make(map[int][]*StoreTx),
 		distHops:             make([]distHopRecord, 0),
 		distPaths:            make([]distPathRecord, 0),
@@ -309,7 +309,7 @@ func TestEvictStale_CleansResolvedPathNodeIndexes(t *testing.T) {
 	if len(store.byNode[relayPK]) != 1 {
 		t.Fatalf("expected 1 entry in byNode[%s], got %d", relayPK, len(store.byNode[relayPK]))
 	}
-	if !store.nodeHashes[relayPK][tx.Hash] {
+	if !store.nodeHasTx(relayPK, tx) {
 		t.Fatalf("expected nodeHashes[%s] to contain %s", relayPK, tx.Hash)
 	}
 

cmd/server/neighbor_graph_test.go

@@ -26,7 +26,7 @@ func ngTestStore(nodes []nodeInfo, packets []*StoreTx) *PacketStore {
 		byObsID:        make(map[int]*StoreObs),
 		byObserver:     make(map[string][]*StoreObs),
 		byNode:         make(map[string][]*StoreTx),
-		nodeHashes:     make(map[string]map[string]bool),
+		nodeHashes:     newNodeSets(),
 		byPayloadType:  make(map[int][]*StoreTx),
 		rfCache:        make(map[string]*cachedResult),
 		topoCache:      make(map[string]*cachedResult),

cmd/server/node_set_memory_test.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+)
+
+// nodeHashes is the set behind byNode's de-dup: for every node, the packets
+// already listed under it. Keyed by packet ID rather than by the packets'
+// 16-character hashes, the same membership takes markedly less memory; the
+// limit below sits between the two.
+func TestNodeSetRetainedHeap(t *testing.T) {
+	const nodes, perNode = 2000, 250 // 500k (node, packet) pairs
+	txs := make([]*StoreTx, perNode*4)
+	for i := range txs {
+		txs[i] = &StoreTx{ID: i + 1, Hash: fmt.Sprintf("%016x", i+1)}
+	}
+	pks := make([]string, nodes)
+	for i := range pks {
+		pks[i] = fmt.Sprintf("%064x", i)
+	}
+	mb := retainedHeapMB(func() interface{} {
+		s := &PacketStore{byNode: map[string][]*StoreTx{}, nodeHashes: newNodeSets()}
+		for n, pk := range pks {
+			for j := 0; j < perNode; j++ {
+				s.addToByNode(txs[(n+j)%len(txs)], pk)
+			}
+		}
+		s.byNode = nil // measure the set alone
+		return s
+	})
+	t.Logf("node sets for 500k (node, packet) pairs keep %.1f MB", mb)
+	const limitMB = 21
+	if mb > limitMB {
+		t.Fatalf("node sets keep %.1f MB for 500k pairs, limit %d MB", mb, limitMB)
+	}
+}
+
+// Adding the same packet under a node twice lists it once.
+func TestAddToByNodeDedup(t *testing.T) {
+	s := &PacketStore{byNode: map[string][]*StoreTx{}, nodeHashes: newNodeSets()}
+	tx := &StoreTx{ID: 7, Hash: "abc"}
+	if !s.addToByNode(tx, "pk") {
+		t.Fatal("first add should report a new node")
+	}
+	if s.addToByNode(tx, "pk") || len(s.byNode["pk"]) != 1 {
+		t.Fatalf("second add listed it again: %d", len(s.byNode["pk"]))
+	}
+	if !s.nodeHasTx("pk", tx) || s.nodeHasTx("pk", &StoreTx{ID: 8, Hash: "def"}) {
+		t.Fatal("nodeHasTx disagrees with what was added")
+	}
+}

cmd/server/relay_airtime_share_test.go

@@ -19,7 +19,7 @@ func newRelayAirtimeShareTestStore(packets []*StoreTx) *PacketStore {
 		byObserver:     make(map[string][]*StoreObs),
 		byNode:         make(map[string][]*StoreTx),
 		byPathHop:      make(map[string][]*StoreTx),
-		nodeHashes:     make(map[string]map[string]bool),
+		nodeHashes:     newNodeSets(),
 		byPayloadType:  make(map[int][]*StoreTx),
 		rfCache:        make(map[string]*cachedResult),
 		topoCache:      make(map[string]*cachedResult),