diff --git a/cmd/server/config.go b/cmd/server/config.go index acc3315bc..355a41783 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -109,6 +109,10 @@ type Config struct { // NormalizedRegionQuickPicks and config.example.json. RegionQuickPicks []RegionQuickPick `json:"regionQuickPicks,omitempty"` + // RegionFallback tunes how the /api/nodes region filter places nodes + // never heard zero-hop (#1879). See RegionFallbackMaxHops. + RegionFallback *RegionFallbackConfig `json:"regionFallback,omitempty"` + Roles map[string]interface{} `json:"roles"` HealthThresholds *HealthThresholds `json:"healthThresholds"` PathTrust *PathTrustConfig `json:"pathTrust,omitempty"` @@ -1284,3 +1288,20 @@ func (c *Config) configuredChannelNames() map[string]bool { } return names } + +// RegionFallbackConfig is the regionFallback block (#1879). +type RegionFallbackConfig struct { + // MaxHops caps the fallback for nodes never heard zero-hop: a node whose + // closest hearing took more hops than this is in no region. Default 3. + MaxHops *int `json:"maxHops,omitempty"` +} + +// RegionFallbackMaxHops returns regionFallback.maxHops, or the default +// (regionFallbackMaxHopsDefault, 3) when it is unset or negative. 0 is valid: +// only nodes some observer heard over an empty path are placed. +func (c *Config) RegionFallbackMaxHops() int { + if c != nil && c.RegionFallback != nil && c.RegionFallback.MaxHops != nil && *c.RegionFallback.MaxHops >= 0 { + return *c.RegionFallback.MaxHops + } + return regionFallbackMaxHopsDefault +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 58a679825..887ead0a8 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 new file mode 100644 index 000000000..4753adde9 --- /dev/null +++ b/cmd/server/node_home_regions.go @@ -0,0 +1,263 @@ +// 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. +// +// A node never heard zero-hop goes in the regions that heard it in the fewest +// hops, and in no region when even that is over regionFallback.maxHops (it still +// shows unfiltered). Every region that heard it at all, the old fallback, put +// most nodes in every region on a statewide deployment: adverts flood the mesh, +// so observers in each metro hear far-off nodes many hops out. The observer- +// based check now decides only for nodes newer than the last recompute. +// +// Each node's outcome is a NodeRegionPlacement, which RegionNodePubkeys reads +// and /api/nodes returns as region_placement, so an operator can see which of +// the four rules placed a node. +// +// 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, plus one recompute when the startup load is +// done. Each recompute makes two passes over the in-memory adverts (zero-hop +// counts, then fewest hops; decoded JSON cached on the tx) under s.mu read, +// off the request path: BenchmarkRecomputeNodeHomeRegionSets. RegionNodePubkeys +// and handleNodes read the snapshot with a single atomic load. +package main + +import ( + "sort" + "strings" + "sync" + "time" +) + +const nodeHomeRegionsRecomputerDefaultInterval = 5 * time.Minute + +// regionFallbackMaxHopsDefault bounds the fallback for nodes never heard +// zero-hop: a node whose closest hearing took more hops than this is in no +// region. Config regionFallback.maxHops overrides it (Config.RegionFallbackMaxHops). +const regionFallbackMaxHopsDefault = 3 + +var ( + nodeHomeRegionsRecompStartedMu sync.Mutex + nodeHomeRegionsRecompStarted bool +) + +// StartNodeHomeRegionsRecomputer computes the region placements once, again +// when the startup load is done, 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() + + // main.go starts this at the first load chunk, so the prewarm saw only + // part of the store. Recompute once the whole startup load (hot window + // and background fill) is done, as the analytics recomputers do. + loaded := s.StartupLoadDone() + + var stopOnce sync.Once + go func() { + defer close(done) + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-loaded: + loaded = nil // once + s.recomputeNodeHomeRegionSets() + 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() + var counts, hops map[string]map[string]int + func() { + s.mu.RLock() + defer s.mu.RUnlock() + counts = s.nodeZeroHopRegionCounts(obsIATA) + hops = s.nodeFewestHopsByRegion(obsIATA) + }() + placements := regionPlacements(counts, hops, s.config.RegionFallbackMaxHops()) + s.nodeRegionPlacements.Store(&placements) +} + +// nodeFewestHopsByRegion returns, per advert pubkey, the fewest hops any +// observer in each region heard it over. Caller must hold s.mu (read). +func (s *PacketStore) nodeFewestHopsByRegion(obsIATA map[string]string) map[string]map[string]int { + per := make(map[string]map[string]int, 256) + if len(obsIATA) == 0 { + return per + } + for _, tx := range s.byPayloadType[PayloadADVERT] { + var m map[string]int + for _, obs := range tx.Observations { + iata := obsIATA[obs.ObserverID] + if iata == "" { + continue + } + h, known := pathHopCount(obs.PathJSON) + if !known { + continue + } + if m == nil { + // Parsed only for adverts heard in some region, and cached on the tx. + pk := advertPubkey(tx) + if pk == "" { + break + } + if m = per[pk]; m == nil { + m = map[string]int{} + per[pk] = m + } + } + if cur, ok := m[iata]; !ok || h < cur { + m[iata] = h + } + } + } + return per +} + +// pathHopCount counts the hops in an observation's path_json without decoding +// it: the elements are hop-hash hex strings, so they hold no commas. Run over +// every advert observation per recompute, under s.mu, where pathLen's +// json.Unmarshal would allocate per observation. known is false for a missing +// path (legacy rows with NULL path_json): "[]" is a direct hearing, "" is not. +func pathHopCount(pathJSON string) (hops int, known bool) { + p := strings.TrimSpace(pathJSON) + switch p { + case "", "null": + return 0, false + case "[]": + return 0, true + } + return strings.Count(p, ",") + 1, true +} + +// RegionBasis names the rule that decided a node's regions for ?region=. +type RegionBasis string + +const ( + // RegionBasisZeroHop: observers heard the node zero-hop; Regions are + // theirs. + RegionBasisZeroHop RegionBasis = "zero_hop" + // RegionBasisFewestHops: never heard zero-hop; Regions heard it in the + // fewest hops (Hops), at most regionFallback.maxHops. + RegionBasisFewestHops RegionBasis = "fewest_hops" + // RegionBasisNone: never heard zero-hop, and its closest hearing (Hops) + // is over regionFallback.maxHops, so it is in no region. + RegionBasisNone RegionBasis = "none" + // RegionBasisObserverScan: not in the last recompute (its first advert + // is newer, the store holds none, or every hearing lacks a path); the + // filter puts it in each region whose observers heard any of its adverts. + RegionBasisObserverScan RegionBasis = "observer_scan" +) + +// NodeRegionPlacement is which rule placed a node for the /api/nodes region +// filter, and where (#1879). Returned on each node as region_placement. +type NodeRegionPlacement struct { + Basis RegionBasis `json:"basis"` + // Regions is sorted; absent for none and observer_scan. + Regions []string `json:"regions,omitempty"` + // Hops is the fewest hops any region heard the node over; set for + // fewest_hops and none. + Hops *int `json:"hops,omitempty"` +} + +// regionPlacements places every node in zeroHop (per-region zero-hop counts) +// by those regions, and every other node in fewest (per-region fewest hops) +// by its fewest-hop regions, or in none past maxHops. +func regionPlacements(zeroHop, fewest map[string]map[string]int, maxHops int) map[string]NodeRegionPlacement { + out := make(map[string]NodeRegionPlacement, len(fewest)) + for pk, perIATA := range zeroHop { + regions := make([]string, 0, len(perIATA)) + for iata := range perIATA { + regions = append(regions, iata) + } + sort.Strings(regions) + out[pk] = NodeRegionPlacement{Basis: RegionBasisZeroHop, Regions: regions} + } + for pk, perIATA := range fewest { + if _, placed := out[pk]; placed || len(perIATA) == 0 { + continue + } + hops := -1 + for _, h := range perIATA { + if hops < 0 || h < hops { + hops = h + } + } + if hops > maxHops { + out[pk] = NodeRegionPlacement{Basis: RegionBasisNone, Hops: &hops} + continue + } + regions := []string{} + for iata, h := range perIATA { + if h == hops { + regions = append(regions, iata) + } + } + sort.Strings(regions) + out[pk] = NodeRegionPlacement{Basis: RegionBasisFewestHops, Regions: regions, Hops: &hops} + } + return out +} + +// GetNodeRegionPlacements returns the current snapshot (pubkey → placement), +// or an empty map before the first compute. A node absent from it is placed by +// the observer scan. Read-only by convention: the map is shared by all readers. +func (s *PacketStore) GetNodeRegionPlacements() map[string]NodeRegionPlacement { + snap := s.nodeRegionPlacements.Load() + if snap == nil { + return map[string]NodeRegionPlacement{} + } + return *snap +} + +// RegionPlacementFor returns a node's placement from a snapshot, observer_scan +// when the snapshot does not cover it. +func RegionPlacementFor(placements map[string]NodeRegionPlacement, pubkey string) NodeRegionPlacement { + if p, ok := placements[pubkey]; ok { + return p + } + return NodeRegionPlacement{Basis: RegionBasisObserverScan} +} diff --git a/cmd/server/node_home_regions_test.go b/cmd/server/node_home_regions_test.go new file mode 100644 index 000000000..b86c2c768 --- /dev/null +++ b/cmd/server/node_home_regions_test.go @@ -0,0 +1,433 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "sort" + "testing" + "time" +) + +// #1879: the region filter put a node in every region where any observer heard +// any of its adverts, relayed ones included, so a node leaked into each region +// its flood adverts were relayed into. A node now belongs to the regions whose +// observers heard it zero-hop (#804's rule, as a set); a node never heard +// zero-hop goes by its fewest hops (TestRegionFallbackUsesFewestHops). + +const ( + homeZeroHopDirect = "1240aabbccdd" // route direct, path byte 0x40: 2-byte hashes, 0 hops + homeRelayedFlood = "1141aabbccdd" // route flood, path byte 0x41: 1 hop +) + +func mkHomeRegionAdvert(id int, pubkey, rawHex string, observerIDs ...string) *StoreTx { + tx := mkRegionAdvert(id, pubkey, observerIDs...) + tx.RawHex = rawHex + return tx +} + +func newHomeRegionStore(t *testing.T) *PacketStore { + t.Helper() + ps := newRegionTestStore(t, map[string]string{"o-aus": "AUS", "o-sat": "SAT"}) + addToStore(ps, + // Home in AUS; a relayed advert also reached a SAT observer. + mkHomeRegionAdvert(1, "pk_aus", homeZeroHopDirect, "o-aus"), + mkHomeRegionAdvert(2, "pk_aus", homeRelayedFlood, "o-sat"), + // On the boundary: heard zero-hop in both regions. + mkHomeRegionAdvert(3, "pk_edge", homeZeroHopDirect, "o-aus", "o-sat"), + // Never heard zero-hop: only the fallback can place it. + mkPathAdvert(4, "pk_far", map[string]string{"o-sat": `["aa"]`}), + ) + ps.recomputeNodeHomeRegionSets() + return ps +} + +func TestRegionNodePubkeysUsesZeroHopHomeRegions(t *testing.T) { + ps := newHomeRegionStore(t) + + cases := []struct { + region string + want []string + }{ + // pk_aus is out of SAT despite its relayed advert there; pk_far stays + // through the fallback. + {"SAT", []string{"pk_edge", "pk_far"}}, + {"AUS", []string{"pk_aus", "pk_edge"}}, + // The boundary node is in both regions, and counted once. + {"AUS,SAT", []string{"pk_aus", "pk_edge", "pk_far"}}, + } + for _, c := range cases { + got, ok := ps.RegionNodePubkeys(c.region) + if !ok { + t.Fatalf("region %q: ok=false", c.region) + } + if g := sortedKeys(got); !reflect.DeepEqual(g, c.want) { + t.Errorf("region %q: got %v, want %v", c.region, g, c.want) + } + } +} + +// Selecting one region must not drop a boundary node from the other: per-region +// counts no longer sum to the node total, by design. +func TestBoundaryNodeStaysInEachHomeRegion(t *testing.T) { + ps := newHomeRegionStore(t) + for _, region := range []string{"AUS", "SAT"} { + got, _ := ps.RegionNodePubkeys(region) + found := false + for _, pk := range got { + found = found || pk == "pk_edge" + } + if !found { + t.Errorf("boundary node missing from %s: %v", region, sortedKeys(got)) + } + } +} + +func TestNodeRegionPlacementsZeroHop(t *testing.T) { + ps := newHomeRegionStore(t) + got := ps.GetNodeRegionPlacements() + want := map[string]NodeRegionPlacement{ + "pk_aus": {Basis: RegionBasisZeroHop, Regions: []string{"AUS"}}, + "pk_edge": {Basis: RegionBasisZeroHop, Regions: []string{"AUS", "SAT"}}, + "pk_far": {Basis: RegionBasisFewestHops, Regions: []string{"SAT"}, Hops: intPtr(1)}, + } + if !reflect.DeepEqual(got, want) { + t.Errorf("placements = %s, want %s", placementsString(got), placementsString(want)) + } +} + +// /api/nodes says which rule placed each node, for every rule (#2190 review): +// zero-hop home regions, fewest hops, no region past the cap, and the +// observer scan for a node the last recompute did not cover. +func TestHandleNodesRegionPlacement(t *testing.T) { + srv, router := setupTestServer(t) + // obs1 is SJC in seedTestData. The repeater gets a zero-hop advert, the + // room is heard only five hops out, and the companion has no advert. + addToStore(srv.store, + mkHomeRegionAdvert(7101, "aabbccdd11223344", homeZeroHopDirect, "obs1"), + mkPathAdvert(7102, "1122334455667788", map[string]string{"obs1": `["aa","bb","cc","dd","ee"]`}), + ) + srv.store.recomputeNodeHomeRegionSets() + srv.invalidateNodesCache() + + req := httptest.NewRequest(http.MethodGet, "/api/nodes", nil) + rr := httptest.NewRecorder() + router.ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status %d: %s", rr.Code, rr.Body.String()) + } + var body struct { + Nodes []struct { + PublicKey string `json:"public_key"` + RegionPlacement *NodeRegionPlacement `json:"region_placement"` + } `json:"nodes"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + want := map[string]NodeRegionPlacement{ + "aabbccdd11223344": {Basis: RegionBasisZeroHop, Regions: []string{"SJC"}}, + "1122334455667788": {Basis: RegionBasisNone, Hops: intPtr(5)}, + "eeff00112233aabb": {Basis: RegionBasisObserverScan}, + } + seen := 0 + for _, n := range body.Nodes { + w, ok := want[n.PublicKey] + if !ok { + continue + } + seen++ + if n.RegionPlacement == nil { + t.Errorf("%s: region_placement missing", n.PublicKey) + continue + } + if !reflect.DeepEqual(*n.RegionPlacement, w) { + t.Errorf("%s: region_placement = %s, want %s", n.PublicKey, + placementString(*n.RegionPlacement), placementString(w)) + } + } + if seen != len(want) { + t.Fatalf("expected all %d seeded nodes in the response, saw %d", len(want), seen) + } +} + +func placementString(p NodeRegionPlacement) string { + hops := "-" + if p.Hops != nil { + hops = fmt.Sprint(*p.Hops) + } + return fmt.Sprintf("{%s %v hops=%s}", p.Basis, p.Regions, hops) +} + +func placementsString(m map[string]NodeRegionPlacement) string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + out := "" + for _, k := range keys { + out += k + ":" + placementString(m[k]) + " " + } + return out +} + +// mkPathAdvert is a relayed flood advert heard by each observer over the given +// path (JSON array of hop hashes, "[]" for a direct hearing). +func mkPathAdvert(id int, pubkey string, hearings map[string]string) *StoreTx { + tx := mkHomeRegionAdvert(id, pubkey, homeRelayedFlood) + i := 0 + for oid, path := range hearings { + tx.Observations = append(tx.Observations, &StoreObs{ + ID: id*10 + i, TransmissionID: id, ObserverID: oid, PathJSON: path, + }) + i++ + } + return tx +} + +// A node never heard zero-hop used to be in every region that heard any of its +// adverts. Adverts flood the whole mesh, so on a statewide deployment that put +// most nodes in every region. It now goes in the regions that heard it in the +// fewest hops, and in none when even that is over the cap (default 3). +func TestRegionFallbackUsesFewestHops(t *testing.T) { + ps := newRegionTestStore(t, map[string]string{"o-aus": "AUS", "o-sat": "SAT", "o-hou": "HOU"}) + addToStore(ps, + // One hop from SAT, relayed four hops to AUS: SAT only. + mkPathAdvert(1, "pk_near", map[string]string{"o-sat": `["aa"]`, "o-aus": `["aa","bb","cc","dd"]`}), + // Tied fewest hops: both. + mkPathAdvert(2, "pk_tie", map[string]string{"o-sat": `["aa","bb"]`, "o-hou": `["cc","dd"]`, "o-aus": `["aa","bb","cc"]`}), + // Exactly the cap is still placed. + mkPathAdvert(3, "pk_cap", map[string]string{"o-aus": `["aa","bb","cc"]`}), + // Closest hearing beyond the cap: no region. + mkPathAdvert(4, "pk_gulf", map[string]string{"o-aus": `["aa","bb","cc","dd"]`, "o-hou": `["aa","bb","cc","dd","ee","ff"]`}), + // A later advert heard closer counts. + mkPathAdvert(5, "pk_gulf2", map[string]string{"o-hou": `["aa","bb","cc","dd","ee"]`}), + mkPathAdvert(6, "pk_gulf2", map[string]string{"o-hou": `["aa"]`}), + ) + ps.recomputeNodeHomeRegionSets() + + cases := []struct { + region string + want []string + }{ + {"SAT", []string{"pk_near", "pk_tie"}}, + {"AUS", []string{"pk_cap"}}, + {"HOU", []string{"pk_gulf2", "pk_tie"}}, + } + for _, c := range cases { + got, ok := ps.RegionNodePubkeys(c.region) + if !ok { + t.Fatalf("region %q: ok=false", c.region) + } + if g := sortedKeys(got); !reflect.DeepEqual(g, c.want) { + t.Errorf("region %q: got %v, want %v", c.region, g, c.want) + } + } + + want := map[string]NodeRegionPlacement{ + "pk_near": {Basis: RegionBasisFewestHops, Regions: []string{"SAT"}, Hops: intPtr(1)}, + "pk_tie": {Basis: RegionBasisFewestHops, Regions: []string{"HOU", "SAT"}, Hops: intPtr(2)}, + "pk_cap": {Basis: RegionBasisFewestHops, Regions: []string{"AUS"}, Hops: intPtr(3)}, + "pk_gulf": {Basis: RegionBasisNone, Hops: intPtr(4)}, + "pk_gulf2": {Basis: RegionBasisFewestHops, Regions: []string{"HOU"}, Hops: intPtr(1)}, + } + if got := ps.GetNodeRegionPlacements(); !reflect.DeepEqual(got, want) { + t.Errorf("placements = %s, want %s", placementsString(got), placementsString(want)) + } +} + +// regionFallback.maxHops sets the cap (#2190 review); unset or negative +// keeps the default of 3, and 0 places only nodes heard over an empty path. +func TestRegionFallbackMaxHopsConfig(t *testing.T) { + var parsed Config + if err := json.Unmarshal([]byte(`{"regionFallback":{"maxHops":4}}`), &parsed); err != nil { + t.Fatal(err) + } + cases := []struct { + name string + cfg *Config + want int + }{ + {"nil config", nil, 3}, + {"unset", &Config{}, 3}, + {"empty block", &Config{RegionFallback: &RegionFallbackConfig{}}, 3}, + {"negative", &Config{RegionFallback: &RegionFallbackConfig{MaxHops: intPtr(-1)}}, 3}, + {"zero", &Config{RegionFallback: &RegionFallbackConfig{MaxHops: intPtr(0)}}, 0}, + {"from JSON", &parsed, 4}, + } + for _, c := range cases { + if got := c.cfg.RegionFallbackMaxHops(); got != c.want { + t.Errorf("%s: RegionFallbackMaxHops() = %d, want %d", c.name, got, c.want) + } + } + + ps := newRegionTestStore(t, map[string]string{"o-aus": "AUS"}) + ps.config = &Config{RegionFallback: &RegionFallbackConfig{MaxHops: intPtr(4)}} + addToStore(ps, + mkPathAdvert(1, "pk_four", map[string]string{"o-aus": `["aa","bb","cc","dd"]`}), + mkPathAdvert(2, "pk_five", map[string]string{"o-aus": `["aa","bb","cc","dd","ee"]`}), + ) + ps.recomputeNodeHomeRegionSets() + want := map[string]NodeRegionPlacement{ + "pk_four": {Basis: RegionBasisFewestHops, Regions: []string{"AUS"}, Hops: intPtr(4)}, + "pk_five": {Basis: RegionBasisNone, Hops: intPtr(5)}, + } + if got := ps.GetNodeRegionPlacements(); !reflect.DeepEqual(got, want) { + t.Errorf("maxHops 4: placements = %s, want %s", placementsString(got), placementsString(want)) + } +} + +// A legacy observation with no path_json says nothing about hops, so it must +// not count as a direct hearing (#2190 review). "[]" is still zero hops. +func TestRegionFallbackIgnoresUnknownPaths(t *testing.T) { + ps := newRegionTestStore(t, map[string]string{"o-aus": "AUS", "o-sat": "SAT"}) + addToStore(ps, + // The empty AUS path would have been 0 hops and won. + mkPathAdvert(1, "pk_mixed", map[string]string{"o-aus": "", "o-sat": `["aa","bb"]`}), + // Only unknown paths: no placement, so the observer scan decides. + mkPathAdvert(2, "pk_legacy", map[string]string{"o-aus": ""}), + mkPathAdvert(3, "pk_direct", map[string]string{"o-aus": `[]`}), + ) + ps.recomputeNodeHomeRegionSets() + want := map[string]NodeRegionPlacement{ + "pk_mixed": {Basis: RegionBasisFewestHops, Regions: []string{"SAT"}, Hops: intPtr(2)}, + "pk_direct": {Basis: RegionBasisFewestHops, Regions: []string{"AUS"}, Hops: intPtr(0)}, + } + if got := ps.GetNodeRegionPlacements(); !reflect.DeepEqual(got, want) { + t.Errorf("placements = %s, want %s", placementsString(got), placementsString(want)) + } +} + +// A node with no placement yet (its first advert arrived after the last +// recompute) is decided by the observer scan: in each region whose observers +// heard any of its adverts, however many hops out. +func TestRegionFilterObserverScanForNodesNewerThanRecompute(t *testing.T) { + ps := newHomeRegionStore(t) + addToStore(ps, mkPathAdvert(9, "pk_new", map[string]string{"o-aus": `["aa","bb","cc","dd","ee"]`})) + if _, placed := ps.GetNodeRegionPlacements()["pk_new"]; placed { + t.Fatal("pk_new placed before any recompute") + } + got, _ := ps.RegionNodePubkeys("AUS") + if want := []string{"pk_aus", "pk_edge", "pk_new"}; !reflect.DeepEqual(sortedKeys(got), want) { + t.Errorf("AUS before recompute: got %v, want %v", sortedKeys(got), want) + } +} + +// Naming every region explicitly is still a region filter: a node in no +// region drops out, unlike "All regions", which sends no ?region= at all. +func TestRegionFilterListingEveryRegionOmitsUnplacedNodes(t *testing.T) { + ps := newRegionTestStore(t, map[string]string{"o-aus": "AUS", "o-sat": "SAT"}) + addToStore(ps, + mkPathAdvert(1, "pk_near", map[string]string{"o-sat": `["aa"]`}), + mkPathAdvert(2, "pk_gulf", map[string]string{"o-aus": `["aa","bb","cc","dd"]`, "o-sat": `["aa","bb","cc","dd","ee"]`}), + ) + ps.recomputeNodeHomeRegionSets() + got, ok := ps.RegionNodePubkeys("AUS,SAT") + if !ok { + t.Fatal("ok=false") + } + if want := []string{"pk_near"}; !reflect.DeepEqual(sortedKeys(got), want) { + t.Errorf("AUS,SAT: got %v, want %v", sortedKeys(got), want) + } +} + +// The recomputer starts at the first load chunk, so its prewarm misses +// adverts in later chunks. It recomputes once the startup load is done +// instead of up to one interval later (#2190 review). +func TestNodeRegionsRecomputeWhenStartupLoadDone(t *testing.T) { + resetStarted := func() { + nodeHomeRegionsRecompStartedMu.Lock() + nodeHomeRegionsRecompStarted = false + nodeHomeRegionsRecompStartedMu.Unlock() + } + resetStarted() + t.Cleanup(resetStarted) + + ps := newRegionTestStore(t, map[string]string{"o-aus": "AUS"}) + stop := ps.StartNodeHomeRegionsRecomputer(time.Hour) + defer stop() + // A later chunk lands, then the background fill finishes. + addToStore(ps, mkHomeRegionAdvert(1, "pk_late", homeZeroHopDirect, "o-aus")) + ps.signalStartupLoadDone() + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if _, ok := ps.GetNodeRegionPlacements()["pk_late"]; ok { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("pk_late not placed after the startup load finished") +} + +// 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 { + observers["o-"+r] = r + } + 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+nonAdverts) + for i := 0; i < 220000; i++ { + 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, + ObserverID: "o-" + regions[(i+j*3)%len(regions)], PathJSON: paths[(i+j)%len(paths)], + }) + } + 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...) + 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() + hops := ps.nodeFewestHopsByRegion(obsIATA) + ps.mu.RUnlock() + if len(hops) == 0 { + b.Fatal("no nodes") + } + } +} + +// 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.GetNodeRegionPlacements()) == 0 { + b.Fatal("no placements") + } +} diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index 401a9d61e..9d4b8a264 100644 --- a/cmd/server/openapi.go +++ b/cmd/server/openapi.go @@ -215,6 +215,17 @@ func routeDescriptions() map[string]routeMeta { } } +// openAPISchema is a typed OpenAPI schema object, for component schemas that +// would otherwise be nested untyped map literals (AGENTS.md). +type openAPISchema struct { + Type string `json:"type"` + Description string `json:"description,omitempty"` + Enum []string `json:"enum,omitempty"` + Items *openAPISchema `json:"items,omitempty"` + Properties map[string]openAPISchema `json:"properties,omitempty"` + Required []string `json:"required,omitempty"` +} + // schemaRef returns an OpenAPI $ref pointing at a named component schema. func schemaRef(name string) map[string]interface{} { return map[string]interface{}{"$ref": "#/components/schemas/" + name} @@ -270,6 +281,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."}, + "region_placement": schemaRef("NodeRegionPlacement"), "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."), @@ -283,6 +295,23 @@ func componentSchemas() map[string]interface{} { "declared_regions_truncated": map[string]interface{}{"type": "boolean", "description": "Repeater/room only (#1862): present, and true, only when the answer behind declared_regions was flagged as truncated, so that list is partial. GET /api/scope-audit shows the same flag as truncated. Absent otherwise, including for a source that does not record truncation, so absence does not mean the list is complete."}, }, }, + "NodeRegionPlacement": openAPISchema{ + Type: "object", + Description: "#1879: which rule puts this node in regions for the /api/nodes ?region= filter, and which regions. Present on every node row.", + Required: []string{"basis"}, + Properties: map[string]openAPISchema{ + "basis": { + Type: "string", + Enum: []string{string(RegionBasisZeroHop), string(RegionBasisFewestHops), string(RegionBasisNone), string(RegionBasisObserverScan)}, + Description: "zero_hop: observers heard the node zero-hop (a direct advert, not relayed); it is in their regions, and a node on a boundary can have several. " + + "fewest_hops: never heard zero-hop; it is in the regions that heard it in the fewest hops. " + + "none: never heard zero-hop, and even its closest hearing is more than regionFallback.maxHops (default 3) hops out, so no ?region= filter includes it, even one listing every region; it still shows unfiltered. " + + "observer_scan: not in the last placement recompute (its first advert is newer, the packet store holds none, or no hearing of it recorded a path), so the filter puts it in each region whose observers heard any of its adverts.", + }, + "regions": {Type: "array", Items: &openAPISchema{Type: "string"}, Description: "IATA codes, sorted. Present for zero_hop and fewest_hops."}, + "hops": {Type: "integer", Description: "Fewest hops any region heard the node over. Present for fewest_hops and none."}, + }, + }, "NodeListResponse": map[string]interface{}{ "type": "object", "properties": map[string]interface{}{ diff --git a/cmd/server/region_nodes.go b/cmd/server/region_nodes.go index ca5fc3b70..9a882f771 100644 --- a/cmd/server/region_nodes.go +++ b/cmd/server/region_nodes.go @@ -36,11 +36,15 @@ 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 the regions that heard it in the +// fewest hops, or none past regionFallback.maxHops. Only nodes newer than the +// last recompute fall back to the observer-based check: 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 +74,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 in the last recompute are placed by its snapshot; the + // observer-based scan below decides only for nodes newer than that. + // Atomic load, under no lock. + placements := s.GetNodeRegionPlacements() seen := make(map[string]bool, 256) keys = make([]string, 0, 256) @@ -86,13 +94,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 := placements[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, p := range placements { + for _, r := range p.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 608e95d5b..641093529 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 rule places each node for ?region=. Atomic snapshot. + placements := s.store.GetNodeRegionPlacements() for _, node := range nodes { if pk, ok := node["public_key"].(string); ok { EnrichNodeWithHashSize(node, hashInfo[pk]) mbEntry, _ := s.store.GetMultibyteCapFor(pk) EnrichNodeWithMultiByte(node, mbEntry) + node["region_placement"] = RegionPlacementFor(placements, 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) + node["region_placement"] = RegionPlacementFor(s.store.GetNodeRegionPlacements(), 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 3c8ffbc18..734006cd0 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 → which rule places it for the /api/nodes region filter, + // and in which regions. Written by the node-home-regions recomputer. + nodeRegionPlacements atomic.Pointer[map[string]NodeRegionPlacement] // Coverage + Redundancy axes (issue #672 axes 3 & 4 of 4): atomic // snapshots of pubkey → 0..1 score over the current neighbor graph. @@ -4024,23 +4027,38 @@ 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 { - if len(obsIATA) == 0 { - return 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 +} - // Tally zero-hop direct ADVERT region observations per pubkey. - type tally struct { - counts map[string]int +// 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 per } - per := make(map[string]*tally, 256) - - 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 } @@ -4076,30 +4094,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, diff --git a/config.example.json b/config.example.json index 74809dee6..f7e81c4ab 100644 --- a/config.example.json +++ b/config.example.json @@ -297,6 +297,10 @@ "OAK": "Oakland, US", "MRY": "Monterey, US" }, + "_comment_regionFallback": "#1879: how the /api/nodes ?region= filter places a node no observer has heard zero-hop (a direct, unrelayed advert). It goes in the regions that heard it in the fewest hops, or in no region when even that is more than maxHops (default 3; 0 places only nodes heard over an empty path). A node in no region still shows with All regions. Each node's region_placement says which rule placed it.", + "regionFallback": { + "maxHops": 3 + }, "cacheTTL": { "stats": 10, "nodeDetail": 300,