perf(nodes): time the whole region recompute, and cut it ~23% (#1879)

BenchmarkRecomputeNodeHomeRegionSets times recomputeNodeHomeRegionSets
itself (zero-hop count + fewest-hops pass + snapshots) at 220k adverts x 8
observations plus 1M channel messages, the scale from the #2190 review.

- nodeZeroHopRegionCounts walks byPayloadType[ADVERT] instead of every
  packet; it already skipped everything but adverts. Same counts, so #804
  analytics is unchanged.
- nodeFewestHopsByRegion looks up a node's per-region map once per advert
  instead of once per observation.

Apple M5 (Docker, linux/arm64), -count 3:
  RecomputeNodeHomeRegionSets  72 ms -> 55 ms/op
  NodeFewestHopsByRegion       52 ms -> 42 ms/op

cmd/server/node_home_regions.go

@@ -117,24 +117,24 @@ func (s *PacketStore) nodeFewestHopsByRegion(obsIATA map[string]string) map[stri
 		return per
 	}
 	for _, tx := range s.byPayloadType[PayloadADVERT] {
-		pk := ""
+		var m map[string]int
 		for _, obs := range tx.Observations {
 			iata := obsIATA[obs.ObserverID]
 			if iata == "" {
 				continue
 			}
-			if pk == "" {
+			if m == nil {
 				// Parsed only for adverts heard in some region, and cached on the tx.
-				if pk = advertPubkey(tx); pk == "" {
+				pk := advertPubkey(tx)
+				if pk == "" {
 					break
 				}
+				if m = per[pk]; m == nil {
+					m = map[string]int{}
+					per[pk] = m
+				}
 			}
 			h := pathHopCount(obs.PathJSON)
-			m := per[pk]
-			if m == nil {
-				m = map[string]int{}
-				per[pk] = m
-			}
 			if cur, ok := m[iata]; !ok || h < cur {
 				m[iata] = h
 			}

cmd/server/node_home_regions_test.go

@@ -214,10 +214,11 @@ func TestRegionFallbackSkipsZeroHopNodes(t *testing.T) {
 	}
 }
 
-// BenchmarkNodeFewestHopsByRegion measures the fallback's added recompute work
-// at BenchmarkRegionNodePubkeys' scale: 220k adverts from 4,000 nodes, each
-// heard by 8 observers across 10 regions over 0-7 hop paths.
-func BenchmarkNodeFewestHopsByRegion(b *testing.B) {
+// newRegionRecomputeBenchStore builds BenchmarkRegionNodePubkeys' scale: 220k
+// adverts from 4,000 nodes, each heard by 8 observers across 10 regions over
+// 0-7 hop paths (one in eight sent zero-hop direct), plus nonAdverts channel
+// messages, which the zero-hop count also walks past.
+func newRegionRecomputeBenchStore(b *testing.B, nonAdverts int) *PacketStore {
 	regions := []string{"ANR", "BRU", "GNE", "HEP", "KJK", "LGG", "MST", "NRW", "OBL", "OST"}
 	observers := make(map[string]string, len(regions))
 	for _, r := range regions {
@@ -226,9 +227,13 @@ func BenchmarkNodeFewestHopsByRegion(b *testing.B) {
 	ps := newRegionTestStore(b, observers)
 	paths := []string{`[]`, `["aa"]`, `["aa","bb"]`, `["aa","bb","cc"]`, `["aa","bb","cc","dd"]`,
 		`["aa","bb","cc","dd","ee"]`, `["aa","bb","cc","dd","ee","ff"]`, `["aa","bb","cc","dd","ee","ff","11"]`}
-	txs := make([]*StoreTx, 0, 220000)
+	txs := make([]*StoreTx, 0, 220000+nonAdverts)
 	for i := 0; i < 220000; i++ {
-		tx := mkRegionAdvert(i+1, fmt.Sprintf("pk%04d", i%4000))
+		raw := homeRelayedFlood
+		if i%8 == 0 {
+			raw = homeZeroHopDirect
+		}
+		tx := mkHomeRegionAdvert(i+1, fmt.Sprintf("pk%04d", i%4000), raw)
 		for j := 0; j < 8; j++ {
 			tx.Observations = append(tx.Observations, &StoreObs{
 				ID: (i+1)*10 + j, TransmissionID: i + 1,
@@ -237,9 +242,21 @@ func BenchmarkNodeFewestHopsByRegion(b *testing.B) {
 		}
 		txs = append(txs, tx)
 	}
+	chanType := PayloadGRP_TXT
+	for i := 0; i < nonAdverts; i++ {
+		id := 220000 + i + 1
+		txs = append(txs, &StoreTx{ID: id, Hash: fmt.Sprintf("rchan%d", id), PayloadType: &chanType, RawHex: "1541aabbccdd",
+			Observations: []*StoreObs{{ID: id * 10, TransmissionID: id, ObserverID: "o-" + regions[i%len(regions)], PathJSON: `["aa"]`}}})
+	}
 	addToStore(ps, txs...)
-	obsIATA := ps.observerIATAs()
 	ps.recomputeNodeHomeRegionSets() // parse each advert's JSON once, as a running server has
+	return ps
+}
+
+// BenchmarkNodeFewestHopsByRegion measures the fallback's own pass.
+func BenchmarkNodeFewestHopsByRegion(b *testing.B) {
+	ps := newRegionRecomputeBenchStore(b, 0)
+	obsIATA := ps.observerIATAs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		ps.mu.RLock()
@@ -250,3 +267,18 @@ func BenchmarkNodeFewestHopsByRegion(b *testing.B) {
 		}
 	}
 }
+
+// BenchmarkRecomputeNodeHomeRegionSets measures the whole recompute, the cost
+// a running server pays once per interval: the zero-hop count, the fewest-hops
+// pass, and building both snapshots, here with 1M channel messages in the
+// store besides the adverts.
+func BenchmarkRecomputeNodeHomeRegionSets(b *testing.B) {
+	ps := newRegionRecomputeBenchStore(b, 1000000)
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		ps.recomputeNodeHomeRegionSets()
+	}
+	if len(ps.GetNodeHomeRegionSets()) == 0 {
+		b.Fatal("no home regions")
+	}
+}

cmd/server/store.go

@@ -4057,13 +4057,11 @@ func (s *PacketStore) nodeZeroHopRegionCounts(obsIATA map[string]string) map[str
 	if len(obsIATA) == 0 {
 		return per
 	}
-	for _, tx := range s.packets {
+	// Adverts only: the full packet list is mostly other traffic.
+	for _, tx := range s.byPayloadType[PayloadADVERT] {
 		if tx.RawHex == "" || len(tx.RawHex) < 4 {
 			continue
 		}
-		if tx.PayloadType == nil || *tx.PayloadType != PayloadADVERT {
-			continue
-		}
 		if tx.DecodedJSON == "" {
 			continue
 		}