From 80bad407409f897b054e1edd26dc27e27f8839dd Mon Sep 17 00:00:00 2001 From: Adam Nieto Date: Sun, 11 Oct 2026 19:49:40 -0500 Subject: [PATCH] fix(nodes): place nodes in regions by zero-hop home regions (#1879) The region filter put a node in every region where any observer heard any of its adverts, so flooded adverts leaked nodes into neighbouring regions. A recomputer now keeps each node's set of home regions from zero-hop DIRECT adverts (#804's rule, counted by the shared nodeZeroHopRegionCounts); RegionNodePubkeys places those nodes by their set and keeps the observer-based check only for nodes never heard zero-hop. /api/nodes reports home_regions when zero-hop data placed a node. --- cmd/server/main.go | 9 +++ cmd/server/node_home_regions.go | 133 ++++++++++++++++++++++++++++++-- cmd/server/openapi.go | 1 + cmd/server/region_nodes.go | 41 ++++++++-- cmd/server/routes.go | 4 + cmd/server/store.go | 60 +++++++------- 6 files changed, 206 insertions(+), 42 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 58a67982..887ead0a 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -471,6 +471,15 @@ func main() { log.Printf("[bridge-recompute] background recompute enabled (interval=%s)", cfg.AnalyticsDefaultRecomputeInterval()) + // Zero-hop home regions for the /api/nodes region filter (#1879). Read + // by RegionNodePubkeys and handleNodes via a single atomic load. + stopNodeHomeRegionsRecomp := store.StartNodeHomeRegionsRecomputer( + cfg.AnalyticsDefaultRecomputeInterval(), + ) + defer stopNodeHomeRegionsRecomp() + log.Printf("[node-home-regions-recompute] background recompute enabled (interval=%s)", + cfg.AnalyticsDefaultRecomputeInterval()) + // Steady-state coverage + redundancy recomputer (issue #672 axes 3 & 4). // Computes harmonic-reach coverage and articulation-point redundancy over // the same neighbor graph as the bridge axis, storing both per-pubkey diff --git a/cmd/server/node_home_regions.go b/cmd/server/node_home_regions.go index 4963de0b..2569109d 100644 --- a/cmd/server/node_home_regions.go +++ b/cmd/server/node_home_regions.go @@ -1,10 +1,131 @@ +// Package main: zero-hop home regions for the /api/nodes region filter (#1879). +// +// The region filter used to put a node in every region where any observer +// heard any of its adverts. Adverts are flooded, so a node leaked into each +// region its adverts were relayed into. A zero-hop DIRECT advert is not +// relayed, so an observer that hears one is within RF range of the node (#804). +// This recomputer turns those hearings into a set of home regions per node; +// RegionNodePubkeys uses the set for nodes that have one, and keeps the +// observer-based check only for nodes never heard zero-hop. +// +// A set rather than #804's plurality winner: a node on a region boundary, +// heard zero-hop in two regions, belongs to both, so per-region node counts +// no longer sum to the node total. +// +// Same shape as bridge_recomputer.go: background ticker, atomic snapshot, +// synchronous prewarm at start. The cost is the computeNodeHomeRegions scan +// (one pass over the in-memory packets, decoded JSON cached on the tx) once +// per interval, off the request path; RegionNodePubkeys and handleNodes read +// the snapshot with a single atomic load. package main -// recomputeNodeHomeRegionSets will rebuild the zero-hop home region sets -// (#1879). Stub. -func (s *PacketStore) recomputeNodeHomeRegionSets() {} +import ( + "sort" + "sync" + "time" +) -// GetNodeHomeRegionSets will return pubkey → home regions (#1879). Stub. -func (s *PacketStore) GetNodeHomeRegionSets() map[string][]string { - return map[string][]string{} +const nodeHomeRegionsRecomputerDefaultInterval = 5 * time.Minute + +var ( + nodeHomeRegionsRecompStartedMu sync.Mutex + nodeHomeRegionsRecompStarted bool +) + +// StartNodeHomeRegionsRecomputer computes the home region sets once, then +// every interval (default 5min if <= 0). Idempotent: later calls are no-ops +// and return a no-op stop closure. +func (s *PacketStore) StartNodeHomeRegionsRecomputer(interval time.Duration) func() { + if interval <= 0 { + interval = nodeHomeRegionsRecomputerDefaultInterval + } + + nodeHomeRegionsRecompStartedMu.Lock() + if nodeHomeRegionsRecompStarted { + nodeHomeRegionsRecompStartedMu.Unlock() + return func() {} + } + nodeHomeRegionsRecompStarted = true + stop := make(chan struct{}) + done := make(chan struct{}) + nodeHomeRegionsRecompStartedMu.Unlock() + + s.recomputeNodeHomeRegionSets() + + var stopOnce sync.Once + go func() { + defer close(done) + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-t.C: + s.recomputeNodeHomeRegionSets() + case <-stop: + return + } + } + }() + + return func() { + stopOnce.Do(func() { + close(stop) + }) + select { + case <-done: + case <-time.After(5 * time.Second): + } + } +} + +// recomputeNodeHomeRegionSets rebuilds and installs the snapshot. A panic in +// compute is swallowed so the goroutine never dies; the previous snapshot +// stays valid. +func (s *PacketStore) recomputeNodeHomeRegionSets() { + defer func() { _ = recover() }() + // The observers table is read before s.mu, as for computeNodeHomeRegions (#2146). + obsIATA := s.observerIATAs() + counts := func() map[string]map[string]int { + s.mu.RLock() + defer s.mu.RUnlock() + return s.nodeZeroHopRegionCounts(obsIATA) + }() + sets := homeRegionSets(counts) + s.nodeHomeRegionSets.Store(&sets) +} + +// homeRegionSets turns per-region zero-hop counts into each node's sorted set +// of home regions. +func homeRegionSets(counts map[string]map[string]int) map[string][]string { + out := make(map[string][]string, len(counts)) + for pk, perIATA := range counts { + regions := make([]string, 0, len(perIATA)) + for iata := range perIATA { + regions = append(regions, iata) + } + sort.Strings(regions) + out[pk] = regions + } + return out +} + +// GetNodeHomeRegionSets returns the current snapshot (pubkey → sorted home +// regions), or an empty map before the first compute. Read-only by convention: +// the map is shared by all readers. +func (s *PacketStore) GetNodeHomeRegionSets() map[string][]string { + snap := s.nodeHomeRegionSets.Load() + if snap == nil { + return map[string][]string{} + } + return *snap +} + +// enrichNodeHomeRegions sets home_regions on a node row when zero-hop data +// placed it, and leaves the field absent otherwise, so an operator can tell +// which rule put a node in a region (#1879). +func enrichNodeHomeRegions(node map[string]interface{}, regions []string) { + if node == nil || len(regions) == 0 { + return + } + node["home_regions"] = regions } diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index 401a9d61..fcebbb6c 100644 --- a/cmd/server/openapi.go +++ b/cmd/server/openapi.go @@ -270,6 +270,7 @@ func componentSchemas() map[string]interface{} { "unscoped_relay_count_24h": map[string]interface{}{"type": "integer", "description": "Repeater/room only: subset of relay_count_24h that were unscoped floods (route_type FLOOD). A well-configured repeater sets flood.max.unscoped 0, so a non-trivial count flags a base-config problem."}, "last_relayed": str("Repeater/room only: RFC3339 time this node last appeared as a relay hop."), "declared_regions": map[string]interface{}{"type": "array", "items": map[string]interface{}{"type": "string"}, "description": "Repeater/room only (#1862): named regions from this node's newest declared-regions answer, spelled as GET /api/scope-audit declaredRegions spells them (leading '#' stripped, '*' wildcard excluded). Empty array: it answered and named no region. Absent: it never answered, this database has no declared-regions source, or the declared-regions lookup failed. Absence is not evidence the node lacks a region."}, + "home_regions": map[string]interface{}{"type": "array", "items": map[string]interface{}{"type": "string"}, "description": "#1879: IATA codes of the regions whose observers heard this node zero-hop (a direct advert, not relayed), sorted. Present only when zero-hop data placed the node, and then it decides which regions the ?region= filter puts the node in; a node can have several. Absent: no zero-hop advert in the packet store, so the region filter falls back to every region where an observer heard any of its adverts."}, "relay_window_hours": map[string]interface{}{"type": "integer", "description": "Repeater/room only, /api/nodes/{pubkey} detail endpoint only: width (hours) of the relay-activity window the relay_count_* values cover."}, "traffic_share_score": score01("#672 Traffic axis: share of non-advert traffic relayed through this repeater. Repeater/room only."), "bridge_score": score01("#672 Bridge axis: normalized betweenness centrality (chokepoint importance). Repeater/room only."), diff --git a/cmd/server/region_nodes.go b/cmd/server/region_nodes.go index ca5fc3b7..df756133 100644 --- a/cmd/server/region_nodes.go +++ b/cmd/server/region_nodes.go @@ -36,11 +36,13 @@ func advertPubkey(tx *StoreTx) string { return v } -// RegionNodePubkeys returns the pubkeys of nodes with at least one advert heard -// by an observer in region (comma-separated IATA codes, case- and -// space-insensitive), over the adverts the store holds. ok is false when the -// region names no code, or there is no database to resolve observers from, so -// the caller applies the SQL filter or none. +// RegionNodePubkeys returns the pubkeys of nodes in region (comma-separated +// IATA codes, case- and space-insensitive), over the adverts the store holds. +// A node heard zero-hop is in its home regions (#1879, see +// node_home_regions.go); any other node is in a region when at least one of +// its adverts was heard by an observer there. ok is false when the region +// names no code, or there is no database to resolve observers from, so the +// caller applies the SQL filter or none. // // Observers are matched by ID through resolveRegionObservers, i.e. against // the observers table as it is now, like every other store region filter. @@ -70,6 +72,10 @@ func (s *PacketStore) RegionNodePubkeys(region string) (keys []string, ok bool) // Its own mutex and a 30s cache; taken before s.mu, never under it. inRegion := s.resolveRegionObservers(cacheKey) + // #1879: nodes heard zero-hop are placed by their home regions; the + // observer-based scan below decides only for nodes never heard zero-hop. + // An atomic load, taken under no lock. + homes := s.GetNodeHomeRegionSets() seen := make(map[string]bool, 256) keys = make([]string, 0, 256) @@ -86,13 +92,32 @@ func (s *PacketStore) RegionNodePubkeys(region string) (keys []string, ok bool) continue } // Parsed only for adverts heard in the region, and cached on the tx. - if pk := advertPubkey(tx); pk != "" && !seen[pk] { - seen[pk] = true - keys = append(keys, pk) + pk := advertPubkey(tx) + if pk == "" || seen[pk] { + continue } + if _, placed := homes[pk]; placed { + continue + } + seen[pk] = true + keys = append(keys, pk) } s.mu.RUnlock() + wanted := make(map[string]bool, len(codes)) + for _, c := range codes { + wanted[c] = true + } + for pk, regions := range homes { + for _, r := range regions { + if wanted[r] { + seen[pk] = true + keys = append(keys, pk) + break + } + } + } + s.regionNodesMu.Lock() if len(s.regionNodesCache) >= regionNodesCacheMax { for k, e := range s.regionNodesCache { diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 608e95d5..5262b499 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -1617,11 +1617,14 @@ func (s *Server) buildNodesResponse(r *http.Request) (NodeListResponse, error) { // traffic into the composite (the other three axes are already // max-normalized; #1762 review). maxUseful := maxFloat(usefulMap) + // #1879: which regions heard each node zero-hop. Atomic snapshot. + homeRegions := s.store.GetNodeHomeRegionSets() for _, node := range nodes { if pk, ok := node["public_key"].(string); ok { EnrichNodeWithHashSize(node, hashInfo[pk]) mbEntry, _ := s.store.GetMultibyteCapFor(pk) EnrichNodeWithMultiByte(node, mbEntry) + enrichNodeHomeRegions(node, homeRegions[pk]) if role, _ := node["role"].(string); role == "repeater" || role == "room" { info, _ := lookupRelayInfo(relayMap, pk) info.WindowHours = relayWindow @@ -1828,6 +1831,7 @@ func (s *Server) handleNodeDetail(w http.ResponseWriter, r *http.Request) { EnrichNodeWithHashSize(node, hashInfo[pubkey]) mbEntry, _ := s.store.GetMultibyteCapFor(pubkey) EnrichNodeWithMultiByte(node, mbEntry) + enrichNodeHomeRegions(node, s.store.GetNodeHomeRegionSets()[pubkey]) if role, _ := node["role"].(string); role == "repeater" || role == "room" { ht := s.cfg.GetHealthThresholds() info := s.store.GetRepeaterRelayInfo(pubkey, ht.RelayActiveHours) diff --git a/cmd/server/store.go b/cmd/server/store.go index 3c8ffbc1..1b1e1177 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -445,6 +445,9 @@ type PacketStore struct { // pointer load — no lock contention with the per-request enrichment // path in handleNodes (same discipline as #1248). bridgeScoreMap atomic.Pointer[map[string]float64] + // #1879: pubkey → regions whose observers heard it zero-hop, for the + // /api/nodes region filter. Written by the node-home-regions recomputer. + nodeHomeRegionSets atomic.Pointer[map[string][]string] // Coverage + Redundancy axes (issue #672 axes 3 & 4 of 4): atomic // snapshots of pubkey → 0..1 score over the current neighbor graph. @@ -4024,16 +4027,33 @@ func (s *PacketStore) observerIATAs() map[string]string { // #804: feeds analytics region-attribution so a multi-byte repeater whose // flood adverts get relayed across regions is still attributed to its home. func (s *PacketStore) computeNodeHomeRegions(obsIATA map[string]string) map[string]string { + counts := s.nodeZeroHopRegionCounts(obsIATA) + out := make(map[string]string, len(counts)) + for pk, perIATA := range counts { + var bestIATA string + bestCount := 0 + for iata, n := range perIATA { + if n > bestCount || (n == bestCount && iata < bestIATA) { + bestCount = n + bestIATA = iata + } + } + if bestIATA != "" { + out[pk] = bestIATA + } + } + return out +} + +// nodeZeroHopRegionCounts tallies, per pubkey, how many observations of its +// zero-hop DIRECT adverts each region's observers made (the #804 rule; see +// computeNodeHomeRegions). Nodes without such an advert are absent. Caller +// must hold s.mu (read or write); obsIATA is loaded before it, as above. +func (s *PacketStore) nodeZeroHopRegionCounts(obsIATA map[string]string) map[string]map[string]int { + per := make(map[string]map[string]int, 256) if len(obsIATA) == 0 { - return map[string]string{} + return per } - - // Tally zero-hop direct ADVERT region observations per pubkey. - type tally struct { - counts map[string]int - } - per := make(map[string]*tally, 256) - for _, tx := range s.packets { if tx.RawHex == "" || len(tx.RawHex) < 4 { continue @@ -4076,30 +4096,14 @@ func (s *PacketStore) computeNodeHomeRegions(obsIATA map[string]string) map[stri if iata == "" { continue } - t := per[pk] - if t == nil { - t = &tally{counts: map[string]int{}} - per[pk] = t + if per[pk] == nil { + per[pk] = map[string]int{} } - t.counts[iata]++ + per[pk][iata]++ } } - out := make(map[string]string, len(per)) - for pk, t := range per { - var bestIATA string - bestCount := 0 - for iata, n := range t.counts { - if n > bestCount || (n == bestCount && iata < bestIATA) { - bestCount = n - bestIATA = iata - } - } - if bestIATA != "" { - out[pk] = bestIATA - } - } - return out + return per } // enrichObsRPJob returns a map with observation fields + transmission fields,