mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-11-12 09:59:47 +00:00
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.
This commit is contained in:
1 parent
cd546b23d5
commit
80bad40740
6 files changed
+206
-42
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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."),
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
+32
-28
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user