From 7989ded08829d465927ce15fc4d97a5490156bce Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 18:50:24 +0000 Subject: [PATCH 1/2] test(regions): pin #488 M2 placement, cache and dry-run against stubs Red commit. Tests for the direct-evidence region placement aggregate behind the nodes-region cache, the operator dry-run report, the regionPlacement config block and the relayed hop count that M1's advertHearing dropped (review N2 on #609). Production files carry the types and signatures only, so every new test except the default-mode guard fails. The placement rule, aggregate and cache tests are ported from the research PR #601 under new names (rpm2*, TestRegionPlacement488_*), so the two never collide in package main. Relates to #488 Co-Authored-By: Claude Opus 5.5 --- cmd/server/advert_evidence.go | 26 +- cmd/server/advert_evidence_hops_488_test.go | 44 + cmd/server/config.go | 31 + cmd/server/db.go | 15 + cmd/server/node_home_regions_488_test.go | 17 +- cmd/server/region_placement.go | 116 +++ cmd/server/region_placement_488_test.go | 955 ++++++++++++++++++ cmd/server/region_placement_dryrun.go | 48 + .../region_placement_dryrun_488_test.go | 257 +++++ cmd/server/region_placement_perf_488_test.go | 296 ++++++ cmd/server/store.go | 8 +- 11 files changed, 1794 insertions(+), 19 deletions(-) create mode 100644 cmd/server/advert_evidence_hops_488_test.go create mode 100644 cmd/server/region_placement.go create mode 100644 cmd/server/region_placement_488_test.go create mode 100644 cmd/server/region_placement_dryrun.go create mode 100644 cmd/server/region_placement_dryrun_488_test.go create mode 100644 cmd/server/region_placement_perf_488_test.go diff --git a/cmd/server/advert_evidence.go b/cmd/server/advert_evidence.go index 57f985218..b098f5a48 100644 --- a/cmd/server/advert_evidence.go +++ b/cmd/server/advert_evidence.go @@ -2,22 +2,30 @@ package main import "github.com/meshcore-analyzer/packetpath" -// advertEvidenceFromRawHex encodes what an observation's own frame says about -// an ADVERT's originator for StoreObs.advertEv (#488): 1 + the -// packetpath.ClassifyAdvertFrame kind, or 0 when no frame was stored. The -// frame itself is not kept (#881), so this is classified once, on load. -func advertEvidenceFromRawHex(rawHex string) uint8 { +// advertEvidence is StoreObs.advertEv: what an observation's own frame says +// about an ADVERT's originator (#488). +type advertEvidence struct { + // code is 1 + the packetpath.AdvertKind, or 0 when no frame was stored. + code uint8 + // hops is the relay count of an AdvertRelayed frame. + hops uint8 +} + +// advertEvidenceFromRawHex classifies an observation's frame for +// StoreObs.advertEv. The frame itself is not kept (#881), so this is +// classified once, on load. +func advertEvidenceFromRawHex(rawHex string) advertEvidence { if rawHex == "" { - return 0 + return advertEvidence{} } - return uint8(packetpath.ClassifyAdvertFrame(rawHex).Kind) + 1 + return advertEvidence{code: uint8(packetpath.ClassifyAdvertFrame(rawHex).Kind) + 1} } // advertHearing classifies one ADVERT observation: by its own frame when one // was stored, else (a legacy row) by its path_json, unverified. func (o *StoreObs) advertHearing() packetpath.AdvertHearing { - if o.advertEv != 0 { - return packetpath.AdvertHearing{Kind: packetpath.AdvertKind(o.advertEv - 1), Verified: true} + if o.advertEv.code != 0 { + return packetpath.AdvertHearing{Kind: packetpath.AdvertKind(o.advertEv.code - 1), Verified: true} } return packetpath.ClassifyAdvertPathJSON(o.PathJSON) } diff --git a/cmd/server/advert_evidence_hops_488_test.go b/cmd/server/advert_evidence_hops_488_test.go new file mode 100644 index 000000000..be17c023c --- /dev/null +++ b/cmd/server/advert_evidence_hops_488_test.go @@ -0,0 +1,44 @@ +package main + +import ( + "strings" + "testing" + + "github.com/meshcore-analyzer/packetpath" +) + +// TestAdvertHearingKeepsRelayHops_488 (#488 M2, review N2 on #609): an +// observation's stored frame class must keep a relayed advert's hop count, +// which the capped fewest-hops fallback reads. M1 kept only the kind, so a +// relayed hearing came back as Hops 0, the same as a direct one. +func TestAdvertHearingKeepsRelayHops_488(t *testing.T) { + const payload = "aabbccddeeff0011" + for _, tc := range []struct { + name string + raw string + want packetpath.AdvertHearing + }{ + {"flood, 1 hop", "1101a1" + payload, packetpath.AdvertHearing{Kind: packetpath.AdvertRelayed, Hops: 1, Verified: true}}, + {"flood, 3 hops", "1103a1a2a3" + payload, packetpath.AdvertHearing{Kind: packetpath.AdvertRelayed, Hops: 3, Verified: true}}, + {"transport flood, 2 hops of 2 bytes", "101a2b3c4d42a1a1a2a2" + payload, packetpath.AdvertHearing{Kind: packetpath.AdvertRelayed, Hops: 2, Verified: true}}, + {"flood, 63 hops (the most a path byte holds)", "113f" + strings.Repeat("a1", 63) + payload, packetpath.AdvertHearing{Kind: packetpath.AdvertRelayed, Hops: 63, Verified: true}}, + {"flood, zero hops", "1100" + payload, packetpath.AdvertHearing{Kind: packetpath.AdvertDirect, Verified: true}}, + {"direct with a path stays unknown", "1202a1a2" + payload, packetpath.AdvertHearing{Kind: packetpath.AdvertUnknown, Verified: true}}, + } { + t.Run(tc.name, func(t *testing.T) { + // The hearing from the stored class equals classifying the frame. + if direct := packetpath.ClassifyAdvertFrame(tc.raw); direct != tc.want { + t.Fatalf("ClassifyAdvertFrame = %+v, want %+v", direct, tc.want) + } + o := &StoreObs{advertEv: advertEvidenceFromRawHex(tc.raw), PathJSON: "[]"} + if got := o.advertHearing(); got != tc.want { + t.Errorf("advertHearing() = %+v, want %+v", got, tc.want) + } + }) + } + // A legacy row (no frame) still reads its path_json, unverified. + o := &StoreObs{PathJSON: `["A1","B2"]`} + if got, want := o.advertHearing(), (packetpath.AdvertHearing{Kind: packetpath.AdvertRelayed, Hops: 2}); got != want { + t.Errorf("legacy advertHearing() = %+v, want %+v", got, want) + } +} diff --git a/cmd/server/config.go b/cmd/server/config.go index aa58ef5b9..1888f2987 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -266,6 +266,11 @@ type Config struct { // listed", nothing more. RegionGroups []RegionGroup `json:"regionGroups,omitempty"` + // RegionPlacement selects the rule behind /api/nodes?region= (#488 M2). + // Optional: absent keeps today's rule ("heard"). Read once at startup; + // see RegionPlacementSettings for the defaults and bounds. + RegionPlacement *RegionPlacementConfig `json:"regionPlacement,omitempty"` + Roles map[string]interface{} `json:"roles"` HealthThresholds *HealthThresholds `json:"healthThresholds"` Map map[string]interface{} `json:"map"` @@ -1722,3 +1727,29 @@ func (c *Config) AnalyticsRecomputeIntervals() AnalyticsRecomputeIntervals { out.AdvertIntervals = get("advertIntervals") return out } + +// RegionPlacementConfig is the optional regionPlacement block (#488 M2). +// Every field defaults to today's behaviour; the values are meant for the +// customizer in a later milestone (AGENTS.md rule 8). +type RegionPlacementConfig struct { + // Mode is "heard" (default: a node is in every region whose observers + // heard any of its adverts) or "placed" (direct evidence first, then + // the capped fewest-hops fallback, else no region). + Mode string `json:"mode,omitempty"` + // MaxHops caps the fewest-hops fallback. Default 3, bounded to + // 0..regionPlacementHopSlots-1. + MaxHops *int `json:"maxHops,omitempty"` + // LifetimeDays is the evidence window, anchored at the node's own + // newest hearing. Default 30; 0 = no window (all retained history). + LifetimeDays *int `json:"lifetimeDays,omitempty"` + // PersistSpanHours turns on the transient-region rule: direct regions + // whose evidence spans less than this are dropped when another direct + // region's evidence spans at least this. Default 0 = off (deferred). + PersistSpanHours *int `json:"persistSpanHours,omitempty"` +} + +// RegionPlacementSettings resolves the regionPlacement block, logging and +// replacing values it cannot use. +func (c *Config) RegionPlacementSettings() regionPlacementSettings { + return regionPlacementSettings{} +} diff --git a/cmd/server/db.go b/cmd/server/db.go index 3bcfed05c..9bd4c93fa 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -139,6 +139,21 @@ type DB struct { // membership scan, before the query; a non-nil error fails that refresh // the way a failing scan does. nodeRegionQueryHook func() error + + // Direct-evidence region placement (#488 M2, region_placement.go). + // regionPlacement is set once at startup, before the first request. + // regionAgg is the global evidence aggregate, owned by the refresh + // that holds regionAggMu; readers only ever see regionSnap, an + // immutable snapshot swapped in after each refresh. + regionPlacement regionPlacementSettings + regionAggMu sync.Mutex + regionAgg *regionEvidenceAgg + regionAggBuilt time.Time + regionSnap atomic.Pointer[regionPlacementSnapshot] + regionPlacementSF singleflight.Group + // regionPlacementScanHook (test seam, nil in production) runs before + // every aggregate scan, full or delta; a non-nil error fails that scan. + regionPlacementScanHook func(full bool) error // identityNamesQueryHook (#410, test seam, nil in production) runs once // per real hidden-name lookup query, right before it executes, with the // lower-case keys it looks up. diff --git a/cmd/server/node_home_regions_488_test.go b/cmd/server/node_home_regions_488_test.go index ca3698137..f14f16d76 100644 --- a/cmd/server/node_home_regions_488_test.go +++ b/cmd/server/node_home_regions_488_test.go @@ -184,12 +184,13 @@ func TestNodeHomeRegions488_VerifiedPluralityAndTie(t *testing.T) { // Every path that materialises an observation records its frame's advert // class: cold Load, chunked load, loadChunk, IngestNewFromDB, -// IngestNewObservations. advertEv is 0 without a frame, else 1+kind. +// IngestNewObservations. advertEv.code is 0 without a frame, else 1+kind; +// advertEv.hops keeps a relayed frame's hop count (#488 M2, review N2). func TestAdvertEvidence_SetOnEveryObservationPath_488(t *testing.T) { - snapshot := func(s *PacketStore) map[int]uint8 { + snapshot := func(s *PacketStore) map[int]advertEvidence { s.mu.RLock() defer s.mu.RUnlock() - out := map[int]uint8{} + out := map[int]advertEvidence{} for _, tx := range s.packets { for _, o := range tx.Observations { out[o.ID] = o.advertEv @@ -197,7 +198,7 @@ func TestAdvertEvidence_SetOnEveryObservationPath_488(t *testing.T) { } return out } - ev := func(k packetpath.AdvertKind) uint8 { return uint8(k) + 1 } + ev := func(k packetpath.AdvertKind) advertEvidence { return advertEvidence{code: uint8(k) + 1} } db := routeMaskServerDB(t) now := time.Now().UTC() fs := now.Add(-time.Hour).Format(time.RFC3339) @@ -209,7 +210,10 @@ func TestAdvertEvidence_SetOnEveryObservationPath_488(t *testing.T) { rmInsertObs(t, db, 3, 2, 1, `["A1"]`, "1101a1aabb", ts) rmInsertTx(t, db, 3, "legacy", 2, nil, fs) rmInsertObs(t, db, 4, 3, 2, `[]`, "", ts) - want := map[int]uint8{1: ev(packetpath.AdvertDirect), 2: ev(packetpath.AdvertShared), 3: ev(packetpath.AdvertRelayed), 4: 0} + relayed := func(hops uint8) advertEvidence { + return advertEvidence{code: uint8(packetpath.AdvertRelayed) + 1, hops: hops} + } + want := map[int]advertEvidence{1: ev(packetpath.AdvertDirect), 2: ev(packetpath.AdvertShared), 3: relayed(1), 4: {}} load := NewPacketStore(db, nil) if err := load.Load(); err != nil { @@ -234,8 +238,9 @@ func TestAdvertEvidence_SetOnEveryObservationPath_488(t *testing.T) { load.IngestNewFromDB(3, 100) rmInsertObs(t, db, 6, 2, 2, `["A1","A2"]`, "1202a1a2aabb", live.Unix()) rmInsertObs(t, db, 7, 2, 1, `[]`, "", live.Unix()) + rmInsertObs(t, db, 8, 2, 2, `["A1","A2","A3"]`, "1103a1a2a3aabb", live.Unix()) load.IngestNewObservations(5, 100) - want[5], want[6], want[7] = ev(packetpath.AdvertDirect), ev(packetpath.AdvertUnknown), 0 + want[5], want[6], want[7], want[8] = ev(packetpath.AdvertDirect), ev(packetpath.AdvertUnknown), advertEvidence{}, relayed(3) if got := snapshot(load); !reflect.DeepEqual(got, want) { t.Fatalf("live ingest: advertEv %v, want %v", got, want) } diff --git a/cmd/server/region_placement.go b/cmd/server/region_placement.go new file mode 100644 index 000000000..4c788f642 --- /dev/null +++ b/cmd/server/region_placement.go @@ -0,0 +1,116 @@ +package main + +import ( + "context" + "time" + + "github.com/meshcore-analyzer/packetpath" +) + +// Direct-evidence region placement for /api/nodes?region= (#488 M2). +// +// STUB (red commit): types and signatures only. + +const regionPlacementHopSlots = 8 + +const ( + regionModeHeard = "heard" + regionModePlaced = "placed" +) + +const ( + placementBasisDirect = "direct" + placementBasisFewest = "fewest_hops" + placementBasisUnknown = "unknown" +) + +type regionPlacementParams struct { + MaxHops int + Lifetime time.Duration + PersistSpan time.Duration +} + +type regionPlacementSettings struct { + Mode string + Params regionPlacementParams +} + +var defaultRegionPlacementParams = regionPlacementParams{ + MaxHops: 3, + Lifetime: 30 * 24 * time.Hour, +} + +type regionPlacement struct { + Basis string + Regions []string + Transient []string + Hops int +} + +type regionEvidenceStats struct { + LastAt [regionPlacementHopSlots]int64 + DirectFirst int64 + LegacyAt int64 +} + +func (s *regionEvidenceStats) add(ts int64, h packetpath.AdvertHearing) regionHearingKind { + return hearingUnknown +} + +func (s *regionEvidenceStats) merge(o *regionEvidenceStats) {} + +type regionHearingKind uint8 + +const ( + hearingUnknown regionHearingKind = iota + hearingDirect + hearingLegacy + hearingRelayed + hearingFar + hearingShared + hearingKinds +) + +type regionHearingCounts [hearingKinds]int64 + +type regionEvidenceAgg struct { + nodes map[string]map[string]*regionEvidenceStats + pairs int + watermark int64 + hearings regionHearingCounts +} + +func newRegionEvidenceAgg() *regionEvidenceAgg { + return ®ionEvidenceAgg{nodes: map[string]map[string]*regionEvidenceStats{}} +} + +func (a *regionEvidenceAgg) add(pk, region string, ts int64, h packetpath.AdvertHearing) {} + +func (a *regionEvidenceAgg) merge(d *regionEvidenceAgg) {} + +func (db *DB) scanRegionEvidence(ctx context.Context, lo, hi int64, delta bool, agg *regionEvidenceAgg) (int, error) { + return 0, nil +} + +func placeNode(perRegion map[string]*regionEvidenceStats, p regionPlacementParams) regionPlacement { + return regionPlacement{} +} + +type regionPlacementSnapshot struct { + byRegion map[string][]string + watermark int64 + refreshed time.Time + built time.Time +} + +func newRegionPlacementSnapshot(agg *regionEvidenceAgg, p regionPlacementParams, built time.Time) *regionPlacementSnapshot { + return ®ionPlacementSnapshot{} +} + +func (s *regionPlacementSnapshot) members(codes []string) map[string]struct{} { + return map[string]struct{}{} +} + +func (db *DB) currentRegionPlacement(ctx context.Context) (*regionPlacementSnapshot, error) { + return ®ionPlacementSnapshot{}, nil +} diff --git a/cmd/server/region_placement_488_test.go b/cmd/server/region_placement_488_test.go new file mode 100644 index 000000000..005e07cc5 --- /dev/null +++ b/cmd/server/region_placement_488_test.go @@ -0,0 +1,955 @@ +package main + +// Tests for the #488 M2 direct-evidence region placement +// (region_placement.go). The placement rule, the aggregate and the cache +// tests are ported from the research prototype in draft PR #601 +// (region_placement_488_test.go there), renamed so the two never collide +// in package main, and run against the production code. + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "sort" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + "unsafe" + + "github.com/meshcore-analyzer/packetpath" +) + +func rpm2Null() sql.NullString { return sql.NullString{} } +func rpm2Str(s string) sql.NullString { return sql.NullString{String: s, Valid: true} } + +var ( + rpm2Direct = packetpath.AdvertHearing{Kind: packetpath.AdvertDirect, Verified: true} + // rpm2Legacy is an explicit '[]' path on a row without a frame. + rpm2Legacy = packetpath.AdvertHearing{Kind: packetpath.AdvertDirect} + rpm2Shared = packetpath.AdvertHearing{Kind: packetpath.AdvertShared, Verified: true} +) + +func rpm2Hop(n int) packetpath.AdvertHearing { + return packetpath.AdvertHearing{Kind: packetpath.AdvertRelayed, Hops: n, Verified: true} +} + +const rpm2Day = int64(24 * 3600) + +type rpm2Hearing struct { + region string + ts int64 + h packetpath.AdvertHearing +} + +func rpm2Stats(hs ...rpm2Hearing) map[string]*regionEvidenceStats { + m := map[string]*regionEvidenceStats{} + for _, h := range hs { + if m[h.region] == nil { + m[h.region] = ®ionEvidenceStats{} + } + m[h.region].add(h.ts, h.h) + } + return m +} + +func rpm2Daily(region string, from, to int64, h packetpath.AdvertHearing) []rpm2Hearing { + var out []rpm2Hearing + for d := from; d <= to; d++ { + out = append(out, rpm2Hearing{region, 1_000*rpm2Day + d*rpm2Day, h}) + } + return out +} + +func rpm2Cat(parts ...[]rpm2Hearing) []rpm2Hearing { + var out []rpm2Hearing + for _, p := range parts { + out = append(out, p...) + } + return out +} + +func TestRegionPlacement488_PlaceRule(t *testing.T) { + def := defaultRegionPlacementParams + span := def + span.PersistSpan = 24 * time.Hour + cap0 := def + cap0.MaxHops = 0 + noWindow := def + noWindow.Lifetime = 0 + base := int64(1_000 * rpm2Day) + direct := func(rs ...string) regionPlacement { return regionPlacement{Basis: placementBasisDirect, Regions: rs} } + fewest := func(h int, rs ...string) regionPlacement { + return regionPlacement{Basis: placementBasisFewest, Regions: rs, Hops: h} + } + unknown := regionPlacement{Basis: placementBasisUnknown, Hops: -1} + for _, tc := range []struct { + name string + hs []rpm2Hearing + p regionPlacementParams + want regionPlacement + }{ + {"relayed-only region does not leak", + rpm2Cat(rpm2Daily("AAA", 0, 9, rpm2Direct), rpm2Daily("BBB", 0, 9, rpm2Hop(2))), def, direct("AAA")}, + {"two direct regions are a set", + rpm2Cat(rpm2Daily("AAA", 0, 6, rpm2Direct), rpm2Daily("BBB", 0, 6, rpm2Direct)), def, direct("AAA", "BBB")}, + {"one-off long-distance direct hearing: kept while the transient rule is off (default)", + rpm2Cat(rpm2Daily("AAA", 0, 9, rpm2Direct), []rpm2Hearing{{"DDD", base + 5*rpm2Day, rpm2Direct}}), def, direct("AAA", "DDD")}, + {"one-off long-distance direct hearing: transient with persistSpan", + rpm2Cat(rpm2Daily("AAA", 0, 9, rpm2Direct), []rpm2Hearing{{"DDD", base + 5*rpm2Day, rpm2Direct}}), span, + regionPlacement{Basis: placementBasisDirect, Regions: []string{"AAA"}, Transient: []string{"DDD"}}}, + {"sustained second region is not transient", + rpm2Cat(rpm2Daily("AAA", 0, 9, rpm2Direct), []rpm2Hearing{{"DDD", base + 2*rpm2Day, rpm2Direct}, {"DDD", base + 5*rpm2Day, rpm2Direct}}), span, + direct("AAA", "DDD")}, + {"no persistent region: all direct regions kept", + []rpm2Hearing{{"AAA", base, rpm2Direct}, {"BBB", base + 3600, rpm2Direct}}, span, direct("AAA", "BBB")}, + {"fewest hops wins, ties keep both", + rpm2Cat(rpm2Daily("AAA", 0, 3, rpm2Hop(2)), rpm2Daily("BBB", 0, 3, rpm2Hop(2)), rpm2Daily("CCC", 0, 3, rpm2Hop(3))), def, + fewest(2, "AAA", "BBB")}, + {"exactly the cap is placed", rpm2Daily("AAA", 0, 3, rpm2Hop(3)), def, fewest(3, "AAA")}, + {"past the cap is unknown", + rpm2Cat(rpm2Daily("AAA", 0, 3, rpm2Hop(4)), rpm2Daily("BBB", 0, 3, rpm2Hop(9))), def, unknown}, + {"cap 0: no relayed fallback", rpm2Daily("AAA", 0, 3, rpm2Hop(1)), cap0, unknown}, + {"hops past the stored slots never place, whatever the cap", + rpm2Daily("AAA", 0, 3, rpm2Hop(regionPlacementHopSlots)), regionPlacementParams{MaxHops: 63, Lifetime: def.Lifetime}, unknown}, + {"shared contacts and unknown paths never place", + []rpm2Hearing{{"AAA", base, rpm2Shared}, {"BBB", base, packetpath.AdvertHearing{Kind: packetpath.AdvertUnknown}}}, def, unknown}, + {"undated hearing never places", []rpm2Hearing{{"AAA", 0, rpm2Direct}}, def, unknown}, + {"quiet node keeps its placement (window anchored at its newest hearing)", + rpm2Daily("AAA", -200, -190, rpm2Direct), def, direct("AAA")}, + {"moved node: evidence older than the lifetime before its newest is dropped", + rpm2Cat(rpm2Daily("AAA", -100, -60, rpm2Direct), rpm2Daily("BBB", -19, 0, rpm2Direct)), def, direct("BBB")}, + {"lifetime 0 keeps all history", + rpm2Cat(rpm2Daily("AAA", -100, -60, rpm2Direct), rpm2Daily("BBB", -19, 0, rpm2Direct)), noWindow, direct("AAA", "BBB")}, + {"stale fallback does not outrank a fresher, longer one", + rpm2Cat(rpm2Daily("AAA", -100, -60, rpm2Hop(1)), rpm2Daily("BBB", -10, 0, rpm2Hop(3))), def, fewest(3, "BBB")}, + {"direct anywhere beats a closer-looking fallback here", + rpm2Cat(rpm2Daily("AAA", 0, 9, rpm2Hop(1)), rpm2Daily("BBB", 0, 9, rpm2Direct)), def, direct("BBB")}, + // Owner decision on #488: an explicit '[]' without a frame cannot be + // told apart from a contact re-share, so it is fallback only. + {"legacy '[]' never outranks verified direct evidence", + rpm2Cat(rpm2Daily("AAA", 0, 9, rpm2Legacy), []rpm2Hearing{{"BBB", base, rpm2Direct}}), def, direct("BBB")}, + {"legacy-only node is a 0-hop fallback", rpm2Daily("AAA", 0, 3, rpm2Legacy), def, fewest(0, "AAA")}, + {"legacy '[]' beats a relayed hearing elsewhere in the fallback", + rpm2Cat(rpm2Daily("AAA", 0, 3, rpm2Legacy), rpm2Daily("BBB", 0, 3, rpm2Hop(1))), def, fewest(0, "AAA")}, + {"legacy '[]' places under cap 0", rpm2Daily("AAA", 0, 3, rpm2Legacy), cap0, fewest(0, "AAA")}, + } { + t.Run(tc.name, func(t *testing.T) { + got := placeNode(rpm2Stats(tc.hs...), tc.p) + if !reflect.DeepEqual(got, tc.want) { + t.Errorf("placeNode = %+v, want %+v", got, tc.want) + } + }) + } +} + +// TestRegionPlacement488_StatsAreBounded: one (node, region) pair costs a +// fixed number of bytes however many hearings it folds in. +func TestRegionPlacement488_StatsAreBounded(t *testing.T) { + if got := unsafe.Sizeof(regionEvidenceStats{}); got != 80 { + t.Fatalf("regionEvidenceStats is %d bytes, want a fixed 80", got) + } + s := ®ionEvidenceStats{} + far := 0 + for i := 1; i <= 10_000; i++ { + if s.add(int64(i), rpm2Hop(1+i%70)) == hearingFar { + far++ + } + } + if far == 0 || s.LastAt[regionPlacementHopSlots-1] == 0 { + t.Fatalf("hops past the slots must be counted, not stored: far %d, %+v", far, s) + } + // Slot 0 is verified-direct only: a malformed "relayed at 0 hops" is unknown. + z := ®ionEvidenceStats{} + if k := z.add(42, rpm2Hop(0)); k != hearingUnknown || z.LastAt[0] != 0 { + t.Fatalf("relayed at 0 hops reached the direct slot: %v %+v", k, z) + } + // A legacy '[]' goes to its own slot, never to the direct one. + if k := z.add(42, rpm2Legacy); k != hearingLegacy || z.LastAt[0] != 0 || z.LegacyAt != 42 { + t.Fatalf("legacy hearing: %v %+v", k, z) + } +} + +// ---- DB-backed: the aggregate behind the cache ---- + +const rpm2PK = "aaaa000000000001" + +// rpm2T0 is fixed so the tests are deterministic (2026-09-21). +const rpm2T0 = int64(1_790_000_000) + +func rpm2EmptyTestDB(t *testing.T) *DB { + t.Helper() + db := setupTestDB(t) + db.hasObsRawHexFlag.forceTrue() + t.Cleanup(func() { db.Close() }) + return db +} + +// rpm2OpenTestDB adds one observer per region AAA, BBB, CCC (rowids 1-3). +func rpm2OpenTestDB(t *testing.T) *DB { + t.Helper() + db := rpm2EmptyTestDB(t) + for i, iata := range []string{"AAA", "BBB", "CCC"} { + if _, err := db.conn.Exec(`INSERT INTO observers (rowid, id, name, iata, last_seen, first_seen) + VALUES (?, ?, ?, ?, '2026-10-01T00:00:00Z', '2026-01-01T00:00:00Z')`, + i+1, "obs-"+iata, "Obs "+iata, iata); err != nil { + t.Fatal(err) + } + } + return db +} + +func rpm2SetPlaced(db *DB, p regionPlacementParams) { + db.regionPlacement = regionPlacementSettings{Mode: regionModePlaced, Params: p} +} + +// rpm2Insert writes one ADVERT transmission by pk heard once by observer +// rowid obs, with the given per-observation frame and path. +func rpm2Insert(t *testing.T, db *DB, pk string, seq, obs int, ts int64, raw, path sql.NullString) { + t.Helper() + decoded := `{"type":"ADVERT","pubKey":"` + pk + `"}` + res, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('1100aabb', ?, '2026-10-01T00:00:00Z', 1, 4, ?, ?)`, fmt.Sprintf("h-%s-%d", pk, seq), decoded, pk) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + if _, err = db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, raw_hex) + VALUES (?, ?, ?, ?, ?)`, txID, obs, path, ts, raw); err != nil { + t.Fatal(err) + } +} + +func rpm2MaxID(t testing.TB, db *DB) int64 { + t.Helper() + wm, err := db.maxObservationID(context.Background()) + if err != nil { + t.Fatal(err) + } + return wm +} + +func rpm2Scan(t testing.TB, db *DB, lo, hi int64, agg *regionEvidenceAgg) *regionEvidenceAgg { + t.Helper() + if agg == nil { + agg = newRegionEvidenceAgg() + } + if _, err := db.scanRegionEvidence(context.Background(), lo, hi, lo > 0, agg); err != nil { + t.Fatal(err) + } + return agg +} + +func rpm2Placements(agg *regionEvidenceAgg, p regionPlacementParams) map[string]regionPlacement { + out := make(map[string]regionPlacement, len(agg.nodes)) + for pk, per := range agg.nodes { + out[pk] = placeNode(per, p) + } + return out +} + +// rpm2Members is what /api/nodes?region= filters on: the real cache. +func rpm2Members(t testing.TB, db *DB, codes ...string) []string { + t.Helper() + js, err := db.nodeRegionKeysJSON(codes) + if err != nil { + t.Fatal(err) + } + var keys []string + if err := json.Unmarshal([]byte(js), &keys); err != nil { + t.Fatal(err) + } + sort.Strings(keys) + return keys +} + +// rpm2Refresh ages the snapshot and the entry for codes past the fresh TTL +// and runs that entry's refresh synchronously, as the stale-while-revalidate +// path does in the background. +func rpm2Refresh(t testing.TB, db *DB, codes ...string) []string { + t.Helper() + if s := db.regionSnap.Load(); s != nil { + c := *s + c.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) + db.regionSnap.Store(&c) + } + key := nodeRegionKey(codes) + if e := db.getNodeRegionEntry(key); e != nil { + c := *e + c.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) + db.setNodeRegionEntry(key, &c) + } + e, err := db.refreshNodeRegion(key, codes) + if err != nil { + t.Fatal(err) + } + keys := append([]string(nil), e.keys...) + sort.Strings(keys) + return keys +} + +// rpm2RebuildDue makes the aggregate's next refresh a full rebuild. +func rpm2RebuildDue(db *DB) { + db.regionAggMu.Lock() + db.regionAggBuilt = time.Now().Add(-2 * nodeRegionRebuildInterval) + db.regionAggMu.Unlock() +} + +// rpm2CountScans installs the scan hook and returns the full/delta counters. +func rpm2CountScans(db *DB) (full, delta *atomic.Int32) { + full, delta = &atomic.Int32{}, &atomic.Int32{} + db.regionPlacementScanHook = func(isFull bool) error { + if isFull { + full.Add(1) + } else { + delta.Add(1) + } + return nil + } + return full, delta +} + +func TestRegionPlacement488_DeltaEqualsFullBuild(t *testing.T) { + db := rpm2EmptyTestDB(t) + truth, _ := rpm2SeedScenario(t, db.conn, 1, rpm2T0) + wm := rpm2MaxID(t, db) + full := rpm2Scan(t, db, 0, wm, nil) + if len(full.nodes) != len(truth) { + t.Fatalf("full build saw %d nodes, scenario has %d", len(full.nodes), len(truth)) + } + params := defaultRegionPlacementParams + params.PersistSpan = 24 * time.Hour + for _, mid := range []int64{1, wm / 3, wm / 2, wm - 1} { + inc := rpm2Scan(t, db, 0, mid, nil) + inc.merge(rpm2Scan(t, db, mid, wm, nil)) + if !reflect.DeepEqual(rpm2Placements(inc, params), rpm2Placements(full, params)) { + t.Fatalf("delta from %d: placements differ from the full build", mid) + } + if inc.watermark != wm || inc.pairs != full.pairs || inc.hearings != full.hearings { + t.Fatalf("delta from %d: watermark %d/%d, pairs %d/%d, hearings %v/%v", + mid, inc.watermark, wm, inc.pairs, full.pairs, inc.hearings, full.hearings) + } + } +} + +// TestRegionPlacement488_DefaultModeIsHeard: with no regionPlacement config +// the filter is today's rule (a relayed-only node is a member), and the +// aggregate is never built. +func TestRegionPlacement488_DefaultModeIsHeard(t *testing.T) { + db := rpm2OpenTestDB(t) + db.regionPlacement = (&Config{}).RegionPlacementSettings() + full, delta := rpm2CountScans(db) + rpm2Insert(t, db, rpm2PK, 1, 1, rpm2T0, rpm2Str("1200aabbccdd"), rpm2Str("[]")) + rpm2Insert(t, db, rpm2PK, 2, 2, rpm2T0, rpm2Str("1105a1a2a3a4a5aabbccdd"), rpm2Str(`["A1","A2","A3","A4","A5"]`)) + if got := rpm2Members(t, db, "BBB"); !reflect.DeepEqual(got, []string{rpm2PK}) { + t.Fatalf("default mode: BBB members %v, want the relayed-only node (heard rule)", got) + } + if db.regionSnap.Load() != nil || full.Load()+delta.Load() != 0 { + t.Fatalf("default mode built the placement aggregate (%d full, %d delta scans)", full.Load(), delta.Load()) + } +} + +// TestRegionPlacement488_PlacedModeDropsRelayedOnly: the leak #488 is +// about, closed in placed mode. +func TestRegionPlacement488_PlacedModeDropsRelayedOnly(t *testing.T) { + db := rpm2OpenTestDB(t) + rpm2SetPlaced(db, defaultRegionPlacementParams) + rpm2Insert(t, db, rpm2PK, 1, 1, rpm2T0, rpm2Str("1200aabbccdd"), rpm2Str("[]")) + rpm2Insert(t, db, rpm2PK, 2, 2, rpm2T0, rpm2Str("1105a1a2a3a4a5aabbccdd"), rpm2Str(`["A1","A2","A3","A4","A5"]`)) + if got := rpm2Members(t, db, "BBB"); len(got) != 0 { + t.Fatalf("placed mode: BBB members %v, want none", got) + } + if got := rpm2Members(t, db, "AAA"); !reflect.DeepEqual(got, []string{rpm2PK}) { + t.Fatalf("placed mode: AAA members %v", got) + } + if got := rpm2Members(t, db, "aaa", "BBB"); !reflect.DeepEqual(got, []string{rpm2PK}) { + t.Fatalf("placed mode: AAA,BBB members %v (union of codes)", got) + } +} + +// TestRegionPlacement488_NewObservationCanRemoveRegion: a new direct +// hearing takes a node OUT of its fallback region, so a placed-mode refresh +// re-derives membership instead of extending it, and it does so from a +// delta scan, not a rebuild. +func TestRegionPlacement488_NewObservationCanRemoveRegion(t *testing.T) { + db := rpm2OpenTestDB(t) + rpm2SetPlaced(db, defaultRegionPlacementParams) + full, delta := rpm2CountScans(db) + rpm2Insert(t, db, rpm2PK, 1, 1, rpm2T0, rpm2Str("1101a1aabbccdd"), rpm2Str(`["A1"]`)) // AAA, 1 hop + if got := rpm2Members(t, db, "AAA"); !reflect.DeepEqual(got, []string{rpm2PK}) { + t.Fatalf("before: AAA members %v", got) + } + rpm2Members(t, db, "BBB") + rpm2Insert(t, db, rpm2PK, 2, 2, rpm2T0+3600, rpm2Str("1200aabbccdd"), rpm2Str("[]")) // BBB, direct + if got := rpm2Refresh(t, db, "AAA"); len(got) != 0 { + t.Errorf("after: AAA members %v, want none (direct evidence elsewhere)", got) + } + if got := rpm2Refresh(t, db, "BBB"); !reflect.DeepEqual(got, []string{rpm2PK}) { + t.Errorf("after: BBB members %v", got) + } + if full.Load() != 1 || delta.Load() != 1 { + t.Errorf("scans: %d full, %d delta; want 1 full (cold) and 1 delta (the refresh)", full.Load(), delta.Load()) + } + if s := db.regionSnap.Load(); s == nil || s.watermark != rpm2MaxID(t, db) { + t.Errorf("snapshot watermark not advanced: %+v", s) + } +} + +// TestRegionPlacement488_RebuildNeededForEditsPruneBackfill: observer region +// edits, retention pruning and late from_pubkey backfills change rows BELOW +// the watermark. A delta refresh cannot see them; the periodic full rebuild +// does, exactly as for the heard cache. +func TestRegionPlacement488_RebuildNeededForEditsPruneBackfill(t *testing.T) { + const late = "bbbb000000000002" + for _, tc := range []struct { + name string + mutate func(t *testing.T, db *DB) + after []string // AAA members after the rebuild + }{ + {"observer region edit", func(t *testing.T, db *DB) { + if _, err := db.conn.Exec(`UPDATE observers SET iata = 'CCC' WHERE id = 'obs-AAA'`); err != nil { + t.Fatal(err) + } + }, nil}, + {"retention prune", func(t *testing.T, db *DB) { + if _, err := db.conn.Exec(`DELETE FROM observations`); err != nil { + t.Fatal(err) + } + }, nil}, + {"late from_pubkey backfill", func(t *testing.T, db *DB) { + if _, err := db.conn.Exec(`UPDATE transmissions SET from_pubkey = ? WHERE from_pubkey IS NULL`, late); err != nil { + t.Fatal(err) + } + }, []string{rpm2PK, late}}, + } { + t.Run(tc.name, func(t *testing.T) { + db := rpm2OpenTestDB(t) + rpm2SetPlaced(db, defaultRegionPlacementParams) + full, _ := rpm2CountScans(db) + rpm2Insert(t, db, rpm2PK, 1, 1, rpm2T0, rpm2Str("1200aabbccdd"), rpm2Str("[]")) + // An advert whose from_pubkey the ingestor has not filled yet. + if _, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES ('1200ccdd', 'h-late', '2026-10-01T00:00:00Z', 2, 4, '{}')`); err != nil { + t.Fatal(err) + } + if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, raw_hex) + VALUES (2, 1, '[]', ?, '1200ccddeeff')`, rpm2T0); err != nil { + t.Fatal(err) + } + before := []string{rpm2PK} + if got := rpm2Members(t, db, "AAA"); !reflect.DeepEqual(got, before) { + t.Fatalf("before: %v", got) + } + tc.mutate(t, db) + if got := rpm2Refresh(t, db, "AAA"); !reflect.DeepEqual(got, before) { + t.Fatalf("a delta refresh saw a change below the watermark: %v", got) + } + rpm2RebuildDue(db) + if got := rpm2Refresh(t, db, "AAA"); !reflect.DeepEqual(got, tc.after) { + t.Fatalf("after the rebuild: %v, want %v", got, tc.after) + } + if full.Load() != 2 { + t.Fatalf("%d full scans, want 2 (cold + rebuild)", full.Load()) + } + }) + } +} + +// TestRegionPlacement488_QuietNodePlacedFromSQLHistory: a node last heard +// long before any RAM window is placed by a cold build, which reads SQLite +// only (no PacketStore exists here at all). +func TestRegionPlacement488_QuietNodePlacedFromSQLHistory(t *testing.T) { + db := rpm2OpenTestDB(t) + rpm2SetPlaced(db, defaultRegionPlacementParams) + old := rpm2T0 - 120*rpm2Day + rpm2Insert(t, db, rpm2PK, 1, 2, old, rpm2Str("1200aabbccdd"), rpm2Str("[]")) + rpm2Insert(t, db, rpm2PK, 2, 2, old+2*rpm2Day, rpm2Str("1100aabbccdd"), rpm2Str("[]")) + rpm2Insert(t, db, rpm2PK, 3, 1, old+2*rpm2Day, rpm2Str("1102a1a2aabbccdd"), rpm2Str(`["A1","A2"]`)) + if got := rpm2Members(t, db, "BBB"); !reflect.DeepEqual(got, []string{rpm2PK}) { + t.Fatalf("quiet node: BBB members %v", got) + } + if got := rpm2Members(t, db, "AAA"); len(got) != 0 { + t.Fatalf("quiet node: AAA members %v, want none", got) + } +} + +// TestRegionPlacement488_StartupSingleflight: concurrent first requests for +// different region sets share ONE full scan of the history, and all of them +// get the right membership. +func TestRegionPlacement488_StartupSingleflight(t *testing.T) { + db := rpm2OpenTestDB(t) + rpm2SetPlaced(db, defaultRegionPlacementParams) + rpm2Insert(t, db, rpm2PK, 1, 1, rpm2T0, rpm2Str("1200aabbccdd"), rpm2Str("[]")) + rpm2Insert(t, db, "bbbb000000000002", 2, 2, rpm2T0, rpm2Str("1100aabbccdd"), rpm2Str("[]")) + release := make(chan struct{}) + var full, delta atomic.Int32 + db.regionPlacementScanHook = func(isFull bool) error { + if isFull { + full.Add(1) + } else { + delta.Add(1) + } + <-release + return nil + } + sets := [][]string{{"AAA"}, {"BBB"}, {"AAA", "BBB"}, {"CCC"}, {"AAA"}, {"BBB"}} + want := [][]string{{rpm2PK}, {"bbbb000000000002"}, {rpm2PK, "bbbb000000000002"}, nil, {rpm2PK}, {"bbbb000000000002"}} + got := make([][]string, len(sets)) + var wg sync.WaitGroup + for i, codes := range sets { + wg.Add(1) + go func() { + defer wg.Done() + got[i] = rpm2Members(t, db, codes...) + }() + } + time.Sleep(50 * time.Millisecond) // let every request reach the flight + close(release) + wg.Wait() + for i := range sets { + if len(got[i]) == 0 && len(want[i]) == 0 { + continue + } + if !reflect.DeepEqual(got[i], want[i]) { + t.Errorf("%v: members %v, want %v", sets[i], got[i], want[i]) + } + } + if full.Load() != 1 || delta.Load() != 0 { + t.Fatalf("startup: %d full and %d delta scans, want exactly 1 full", full.Load(), delta.Load()) + } + // Warm: a fresh snapshot serves a new region set without scanning. + rpm2Members(t, db, "CCC", "AAA") + if full.Load() != 1 || delta.Load() != 0 { + t.Fatalf("warm request scanned again: %d full, %d delta", full.Load(), delta.Load()) + } +} + +// TestRegionPlacement488_FailedDeltaLeavesAggregateIntact: a delta scan +// that fails part-way must not fold half its rows in; the next refresh then +// equals a full build. +func TestRegionPlacement488_FailedDeltaLeavesAggregateIntact(t *testing.T) { + db := rpm2OpenTestDB(t) + rpm2SetPlaced(db, defaultRegionPlacementParams) + rpm2Insert(t, db, rpm2PK, 1, 1, rpm2T0, rpm2Str("1101a1aabbccdd"), rpm2Str(`["A1"]`)) + rpm2Members(t, db, "AAA") + rpm2Insert(t, db, rpm2PK, 2, 2, rpm2T0+60, rpm2Str("1200aabbccdd"), rpm2Str("[]")) + db.regionPlacementScanHook = func(bool) error { return errors.New("injected scan failure") } + key := nodeRegionKey([]string{"AAA"}) + if e := db.getNodeRegionEntry(key); e != nil { + c := *e + c.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) + db.setNodeRegionEntry(key, &c) + } + s := *db.regionSnap.Load() + s.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) + db.regionSnap.Store(&s) + if _, err := db.refreshNodeRegion(key, []string{"AAA"}); err == nil { + t.Fatal("refresh with a failing scan reported success") + } + db.regionAggMu.Lock() + wm, pairs := db.regionAgg.watermark, db.regionAgg.pairs + db.regionAggMu.Unlock() + if wm != 1 || pairs != 1 { + t.Fatalf("failed delta changed the aggregate: watermark %d, pairs %d", wm, pairs) + } + db.regionPlacementScanHook = nil + if got := rpm2Refresh(t, db, "AAA"); len(got) != 0 { + t.Fatalf("after recovery: AAA members %v, want none", got) + } + db.regionAggMu.Lock() + inc := db.regionAgg + db.regionAggMu.Unlock() + fresh := rpm2Scan(t, db, 0, rpm2MaxID(t, db), nil) + if inc.pairs != fresh.pairs || inc.hearings != fresh.hearings || + !reflect.DeepEqual(rpm2Placements(inc, defaultRegionPlacementParams), rpm2Placements(fresh, defaultRegionPlacementParams)) { + t.Fatalf("recovered aggregate differs from a full build: %v vs %v", inc.hearings, fresh.hearings) + } +} + +// TestRegionPlacement488_BoundedMemory: the aggregate grows with distinct +// (node, region) pairs, never with hearings or refreshes; the snapshot holds +// at most one key per placed membership; the per-set cache stays capped. +func TestRegionPlacement488_BoundedMemory(t *testing.T) { + db := rpm2OpenTestDB(t) + rpm2SetPlaced(db, defaultRegionPlacementParams) + seq := 0 + hearAll := func(ts int64) { + for n := range 5 { + pk := fmt.Sprintf("%04x000000000000", n+1) + for obs := 1; obs <= 3; obs++ { + seq++ + rpm2Insert(t, db, pk, seq, obs, ts, rpm2Str(fmt.Sprintf("11%02xa1a2a3aabbccdd", obs%3)), rpm2Str("[]")) + } + } + } + hearAll(rpm2T0) + rpm2Members(t, db, "AAA") + for round := 1; round <= 20; round++ { + hearAll(rpm2T0 + int64(round)*3600) + rpm2Refresh(t, db, "AAA") + } + db.regionAggMu.Lock() + agg := db.regionAgg + db.regionAggMu.Unlock() + if agg.pairs != 15 || len(agg.nodes) != 5 { + t.Fatalf("aggregate grew with hearings: %d pairs, %d nodes; want 15 and 5", agg.pairs, len(agg.nodes)) + } + if got := agg.hearings[hearingDirect] + agg.hearings[hearingRelayed] + agg.hearings[hearingFar]; got != int64(seq) { + t.Fatalf("hearing counters %v, want %d hearings in all", agg.hearings, seq) + } + members := 0 + for _, keys := range db.regionSnap.Load().byRegion { + members += len(keys) + } + if members > agg.pairs { + t.Fatalf("snapshot holds %d keys for %d pairs", members, agg.pairs) + } + for i := range 2 * nodeRegionMaxEntries { + rpm2Members(t, db, fmt.Sprintf("Z%02d", i)) + } + db.nodeRegionCacheMu.Lock() + n := len(db.nodeRegionCache) + db.nodeRegionCacheMu.Unlock() + if n > nodeRegionMaxEntries { + t.Fatalf("placed mode cached %d region sets, cap %d", n, nodeRegionMaxEntries) + } +} + +// TestRegionPlacement488_V2Schema: the scan also runs on the legacy schema +// (observations.observer_id, ISO timestamps). +func TestRegionPlacement488_V2Schema(t *testing.T) { + db := setupTestDBv2(t) + t.Cleanup(func() { db.Close() }) + if db.isV3() { + t.Fatal("setupTestDBv2 produced a v3 schema") + } + db.hasObsRawHexFlag.forceTrue() + rpm2SetPlaced(db, defaultRegionPlacementParams) + for _, o := range []string{"AAA", "BBB"} { + if _, err := db.conn.Exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen) VALUES (?, ?, ?, '2026-10-01T00:00:00Z', '2026-01-01T00:00:00Z')`, + "v2-"+o, "V2 "+o, o); err != nil { + t.Fatal(err) + } + } + if _, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('1100aabb', 'h-v2', '2026-10-01T00:00:00Z', 1, 4, '{}', ?)`, rpm2PK); err != nil { + t.Fatal(err) + } + for _, o := range []struct{ obs, path, raw, ts string }{ + {"v2-AAA", "[]", "1100aabbccdd", "2026-10-01T00:00:00Z"}, + {"v2-BBB", `["A1","A2"]`, "1102a1a2aabbccdd", "2026-10-01T00:05:00Z"}, + } { + if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_id, observer_name, path_json, timestamp, raw_hex) + VALUES (1, ?, ?, ?, ?, ?)`, o.obs, o.obs, o.path, o.ts, o.raw); err != nil { + t.Fatal(err) + } + } + if got := rpm2Members(t, db, "AAA"); !reflect.DeepEqual(got, []string{rpm2PK}) { + t.Fatalf("v2: AAA members %v", got) + } + if got := rpm2Members(t, db, "BBB"); len(got) != 0 { + t.Fatalf("v2: BBB members %v, want none", got) + } +} + +// ---- synthetic scenario with known ground truth ---- + +func TestRegionPlacement488_SyntheticAccuracy(t *testing.T) { + db := rpm2EmptyTestDB(t) + truth, transient := rpm2SeedScenario(t, db.conn, 1, rpm2T0) + heard := rpm2RegionSets(func(r string) []string { return rpm2Members(t, db, r) }) + db.nodeRegionCacheMu.Lock() + db.nodeRegionCache, db.nodeRegionUsed = nil, nil + db.nodeRegionCacheMu.Unlock() + + params := defaultRegionPlacementParams + params.PersistSpan = 24 * time.Hour + rpm2SetPlaced(db, params) + placed := rpm2RegionSets(func(r string) []string { return rpm2Members(t, db, r) }) + + placedExact, placedWrong := rpm2Accuracy(truth, placed) + heardExact, _ := rpm2Accuracy(truth, heard) + t.Logf("synthetic: %d nodes; exact region set: placed %d, heard %d", len(truth), placedExact, heardExact) + for _, pk := range placedWrong { + t.Errorf("placed %s: got %v, want %v", pk, placed[pk], truth[pk]) + } + db.regionAggMu.Lock() + pl := rpm2Placements(db.regionAgg, params) + db.regionAggMu.Unlock() + for pk, want := range transient { + if !reflect.DeepEqual(pl[pk].Transient, want) { + t.Errorf("placed %s transient %v, want %v", pk, pl[pk].Transient, want) + } + } + // Heard is right only for the 3 boundary nodes and the 4 scoped + // repeaters, which are heard at home only. + if len(truth) != 39 || heardExact != 7 { + t.Errorf("pinned counts moved: nodes %d (want 39), heard exact %d (want 7)", len(truth), heardExact) + } + for pk, rs := range placed { + for _, r := range rs { + if !contains(heard[pk], r) { + t.Errorf("placed %s in %s, where it was never heard", pk, r) + } + } + } +} + +// rpm2Regions lie on a line: AAA–BBB–CCC–DDD. A flood advert reaches a +// region d steps away after about 2d more relays. +var rpm2Regions = []string{"AAA", "BBB", "CCC", "DDD"} + +const rpm2ObsPerRegion = 3 + +func rpm2RegionSets(members func(region string) []string) map[string][]string { + out := map[string][]string{} + for _, r := range rpm2Regions { + for _, pk := range members(r) { + out[pk] = append(out[pk], r) + } + } + for pk := range out { + sort.Strings(out[pk]) + } + return out +} + +func rpm2Accuracy(truth, got map[string][]string) (exact int, wrong []string) { + for pk, want := range truth { + g := got[pk] + if len(g) == 0 && len(want) == 0 || reflect.DeepEqual(g, want) { + exact++ + } else { + wrong = append(wrong, pk) + } + } + sort.Strings(wrong) + return exact, wrong +} + +// rpm2Exec is what the seeder writes through: a test connection or a +// transaction (the server's own handle is read-only in production). +type rpm2Exec interface { + Exec(string, ...any) (sql.Result, error) +} + +type rpm2Seeder struct { + t testing.TB + exec rpm2Exec + seq int +} + +func (s *rpm2Seeder) obs(region, i int) int { return region*rpm2ObsPerRegion + i + 1 } + +func rpm2Payload(pk string, seq int) string { return pk + fmt.Sprintf("%08x", seq) + "00112233" } + +func rpm2FloodRaw(header byte, codes string, pk string, seq, hops int) (raw, path string) { + hs := make([]string, hops) + b := strings.Builder{} + fmt.Fprintf(&b, "%02x%s%02x", header, codes, hops) + for i := range hops { + hs[i] = fmt.Sprintf("%02X", 0xA0+i) + b.WriteString(hs[i]) + } + b.WriteString(rpm2Payload(pk, seq)) + if hops == 0 { + return b.String(), "[]" + } + return b.String(), `["` + strings.Join(hs, `","`) + `"]` +} + +func (s *rpm2Seeder) advert(pk string, ts int64, raw string) int64 { + s.seq++ + decoded := `{"type":"ADVERT","pubKey":"` + pk + `"}` + res, err := s.exec.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES (?, ?, ?, ?, 4, ?, ?)`, raw, fmt.Sprintf("rpm2-%d", s.seq), + time.Unix(ts, 0).UTC().Format(time.RFC3339), int(raw[1]-'0')&3, decoded, pk) + if err != nil { + s.t.Fatal(err) + } + id, _ := res.LastInsertId() + return id +} + +func (s *rpm2Seeder) hear(txID int64, obs int, ts int64, raw, path sql.NullString) { + if _, err := s.exec.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, raw_hex, snr, rssi) + VALUES (?, ?, ?, ?, ?, 5, -100)`, txID, obs, path, ts, raw); err != nil { + s.t.Fatal(err) + } +} + +// directAdvert: a zero-hop DIRECT advert heard by observers obs. +func (s *rpm2Seeder) directAdvert(pk string, ts int64, obs ...int) int64 { + raw := "1200" + rpm2Payload(pk, s.seq+1) + id := s.advert(pk, ts, raw) + for _, o := range obs { + s.hear(id, o, ts, rpm2Str(raw), rpm2Str("[]")) + } + return id +} + +// floodAdvert: one flood advert; byRegion maps a region index to the hop +// count its observer 0 hears it at (header 0x11, or 0x10 with codes). +func (s *rpm2Seeder) floodAdvert(pk string, ts int64, header byte, codes string, byRegion map[int]int) { + regions := make([]int, 0, len(byRegion)) + for r := range byRegion { + regions = append(regions, r) + } + sort.Ints(regions) + var id int64 + for i, r := range regions { + raw, path := rpm2FloodRaw(header, codes, pk, s.seq+1, byRegion[r]) + if i == 0 { + id = s.advert(pk, ts, raw) + } + s.hear(id, s.obs(r, 0), ts, rpm2Str(raw), rpm2Str(path)) + } +} + +// rpm2Leak returns hop counts for a flood heard at h0 hops at home: +2 per +// region step away, as relays carry it down the line. +func rpm2Leak(home, h0 int) map[int]int { + m := map[int]int{} + for r := range rpm2Regions { + d := r - home + if d < 0 { + d = -d + } + m[r] = h0 + 2*d + } + return m +} + +// rpm2SeedScenario writes scale copies of each archetype per region and +// returns the expected home regions (nil = should be unknown) and the +// expected transient regions per node (with the transient rule on). +func rpm2SeedScenario(t testing.TB, exec rpm2Exec, scale int, now int64) (truth, transient map[string][]string) { + t.Helper() + s := &rpm2Seeder{t: t, exec: exec} + for r, iata := range rpm2Regions { + for i := range rpm2ObsPerRegion { + if _, err := exec.Exec(`INSERT INTO observers (rowid, id, name, iata, last_seen, first_seen) + VALUES (?, ?, ?, ?, '2026-10-01T00:00:00Z', '2026-01-01T00:00:00Z')`, + s.obs(r, i), fmt.Sprintf("obs-%s-%d", iata, i), fmt.Sprintf("Obs %s %d", iata, i), iata); err != nil { + t.Fatal(err) + } + } + } + truth, transient = map[string][]string{}, map[string][]string{} + at := func(d int64) int64 { return now - d*rpm2Day - 3600 } + set := func(rs ...int) []string { + out := make([]string, len(rs)) + for i, r := range rs { + out[i] = rpm2Regions[r] + } + sort.Strings(out) + return out + } + for c := range scale { + for r := range rpm2Regions { + pk := func(arch int) string { return fmt.Sprintf("%02x%02x%012x", r, arch, c) } + resident := func(p string, from, to int64) { + for d := from; d <= to; d++ { + s.directAdvert(p, at(d), s.obs(r, 0), s.obs(r, 1)) + s.floodAdvert(p, at(d)+60, 0x11, "", rpm2Leak(r, int(d%2))) + } + } + // 1 resident repeater: zero-hop adverts at home, floods leak down the line. + resident(pk(1), 0, 13) + truth[pk(1)] = set(r) + // 2 flood-only companion: never heard zero-hop; 1 hop at home. + for d := int64(0); d < 5; d++ { + s.floodAdvert(pk(2), at(d), 0x11, "", rpm2Leak(r, 1)) + } + truth[pk(2)] = set(r) + // 3 boundary node between r and r+1, heard directly in both. + if r+1 < len(rpm2Regions) { + for d := int64(0); d < 7; d++ { + s.directAdvert(pk(3), at(d), s.obs(r, 0), s.obs(r+1, 0)) + } + truth[pk(3)] = set(r, r+1) + } + // 4 resident with one long-distance direct hearing two regions away. + far := (r + 2) % len(rpm2Regions) + for d := int64(0); d < 10; d++ { + id := s.directAdvert(pk(4), at(d), s.obs(r, 0)) + if d == 5 { + raw := "1200" + rpm2Payload(pk(4), int(id)) + s.hear(id, s.obs(far, 1), at(d), rpm2Str(raw), rpm2Str("[]")) + } + } + truth[pk(4)], transient[pk(4)] = set(r), set(far) + // 5 resident whose advert a companion re-shares into another + // region (TRANSPORT_DIRECT, codes {0,0}), as an extra observation + // of the same transmission, on three days. + other := (r + 3) % len(rpm2Regions) + for d := int64(0); d < 10; d++ { + id := s.directAdvert(pk(5), at(d), s.obs(r, 0)) + if d%3 == 1 { + raw := "130000000000" + rpm2Payload(pk(5), int(id)) + s.hear(id, s.obs(other, 2), at(d)+600, rpm2Str(raw), rpm2Str("[]")) + } + } + truth[pk(5)] = set(r) + // 6 far-only: heard only after 5 relays. + for d := int64(0); d < 5; d++ { + s.floodAdvert(pk(6), at(d), 0x11, "", map[int]int{r: 5}) + } + truth[pk(6)] = nil + // 7 quiet resident: nothing in the last 90 days. + resident(pk(7), 90, 99) + truth[pk(7)] = set(r) + // 8 moved from r to r+1 (mod 4) 60 days ago. + next := (r + 1) % len(rpm2Regions) + for d := int64(60); d < 100; d++ { + s.directAdvert(pk(8), at(d), s.obs(r, 0)) + } + for d := int64(0); d < 20; d++ { + s.directAdvert(pk(8), at(d), s.obs(next, 0)) + } + truth[pk(8)] = set(next) + // 9 legacy rows only: NULL path_json, no frame. + for d := int64(0); d < 5; d++ { + id := s.advert(pk(9), at(d), "1100"+rpm2Payload(pk(9), int(d))) + s.hear(id, s.obs(r, 2), at(d), rpm2Null(), rpm2Null()) + } + truth[pk(9)] = nil + // 10 scoped repeater: zero-hop TRANSPORT_FLOOD adverts at home. + for d := int64(0); d < 10; d++ { + s.floodAdvert(pk(10), at(d), 0x10, "1a2b3c4d", map[int]int{r: 0}) + } + truth[pk(10)] = set(r) + } + } + return truth, transient +} + +// rpm2FileDB writes the scenario to a SQLite file the way the generator +// does (VACUUM INTO), for code that opens a path read-only. +func rpm2FileDB(t testing.TB, scale int, now int64) (string, map[string][]string) { + t.Helper() + db := setupTestDB(t) + defer db.Close() + tx, err := db.conn.Begin() + if err != nil { + t.Fatal(err) + } + truth, _ := rpm2SeedScenario(t, tx, scale, now) + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "rpm2.db") + if _, err := db.conn.Exec(`VACUUM INTO ?`, path); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path); err != nil { + t.Fatal(err) + } + return path, truth +} diff --git a/cmd/server/region_placement_dryrun.go b/cmd/server/region_placement_dryrun.go new file mode 100644 index 000000000..917017436 --- /dev/null +++ b/cmd/server/region_placement_dryrun.go @@ -0,0 +1,48 @@ +package main + +import ( + "context" + "io" +) + +// STUB (red commit): the operator dry-run report for #488 M2. + +var regionDryRunSweep = []int{0, 1, 2, 3, 4, 5, 7} + +type regionBasisCounts struct { + Direct, Fewest, Unknown, Transient int + FewestByHops [regionPlacementHopSlots]int +} + +type regionDryRunRegion struct { + Region string + Heard, Placed, Direct, Fewest, Removed, Transient int +} + +type regionDryRunSweepRow struct { + MaxHops, Direct, Fewest, Unknown int +} + +type regionPlacementDryRun struct { + Settings regionPlacementSettings + Watermark int64 + ObsRawHex bool + Nodes int + Pairs int + Hearings regionHearingCounts + Bases regionBasisCounts + Regions []regionDryRunRegion + Sweep []regionDryRunSweepRow +} + +func buildRegionPlacementDryRun(ctx context.Context, db *DB, s regionPlacementSettings, sweep []int) (*regionPlacementDryRun, error) { + return ®ionPlacementDryRun{}, nil +} + +func runRegionPlacementDryRun(dbPath string, s regionPlacementSettings, w io.Writer) error { + return nil +} + +func (r *regionPlacementDryRun) write(w io.Writer) error { + return nil +} diff --git a/cmd/server/region_placement_dryrun_488_test.go b/cmd/server/region_placement_dryrun_488_test.go new file mode 100644 index 000000000..355f1e13e --- /dev/null +++ b/cmd/server/region_placement_dryrun_488_test.go @@ -0,0 +1,257 @@ +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "reflect" + "regexp" + "strings" + "testing" + "time" +) + +func rpm2FileHash(t *testing.T, path string) [32]byte { + t.Helper() + b, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return sha256.Sum256(b) +} + +// rpm2LongHex matches anything shaped like a node key or a key prefix. +var rpm2LongHex = regexp.MustCompile(`(?i)[0-9a-f]{12,}`) + +// TestRegionPlacement488_DryRunReport: the operator report counts nodes per +// basis and region and sweeps maxHops, on the synthetic scenario whose +// ground truth is known. It must name no node and change no byte of the +// database it reads. +func TestRegionPlacement488_DryRunReport(t *testing.T) { + path, truth := rpm2FileDB(t, 1, time.Now().Unix()) + before := rpm2FileHash(t, path) + + settings := regionPlacementSettings{Mode: regionModeHeard, Params: defaultRegionPlacementParams} + var out bytes.Buffer + if err := runRegionPlacementDryRun(path, settings, &out); err != nil { + t.Fatal(err) + } + report := out.String() + t.Logf("\n%s", report) + + if rpm2FileHash(t, path) != before { + t.Error("the dry-run changed the database file") + } + for pk := range truth { + if strings.Contains(strings.ToLower(report), pk) { + t.Fatalf("report names node %s", pk) + } + } + if m := rpm2LongHex.FindString(report); m != "" { + t.Fatalf("report contains a key-shaped hex string %q", m) + } + for _, want := range []string{ + "# Region placement dry-run (#488)", + "maxHops 3", "lifetimeDays 30", "persistSpanHours 0", + "| AAA |", "| DDD |", + "| 0 |", "| 7 |", + } { + if !strings.Contains(report, want) { + t.Errorf("report lacks %q", want) + } + } + + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + r, err := buildRegionPlacementDryRun(context.Background(), db, settings, regionDryRunSweep) + if err != nil { + t.Fatal(err) + } + // Scenario under the default rule (maxHops 3, transient rule off): + // direct = residents, boundary, one-off far, re-shared, quiet, moved + // and scoped (27); fewest_hops = the four 1-hop companions; unknown = + // the four 5-hop nodes and the four legacy NULL-path nodes. + if r.Nodes != 39 || r.Bases.Direct != 27 || r.Bases.Fewest != 4 || r.Bases.Unknown != 8 { + t.Errorf("bases: nodes %d direct %d fewest %d unknown %d; want 39/27/4/8", r.Nodes, r.Bases.Direct, r.Bases.Fewest, r.Bases.Unknown) + } + if r.Bases.FewestByHops[1] != 4 { + t.Errorf("fewest by hops %v, want 4 at 1 hop", r.Bases.FewestByHops) + } + wantUnknown := map[int]int{0: 12, 1: 8, 2: 8, 3: 8, 4: 8, 5: 4, 7: 4} + if len(r.Sweep) != len(regionDryRunSweep) { + t.Fatalf("sweep has %d rows, want %d", len(r.Sweep), len(regionDryRunSweep)) + } + for _, row := range r.Sweep { + if row.Direct+row.Fewest+row.Unknown != r.Nodes { + t.Errorf("maxHops %d: bases do not partition the nodes: %+v", row.MaxHops, row) + } + if w, ok := wantUnknown[row.MaxHops]; ok && row.Unknown != w { + t.Errorf("maxHops %d: unknown %d, want %d", row.MaxHops, row.Unknown, w) + } + } + // Heard per region is exactly what today's filter serves. + cur := map[string]int{} + for _, reg := range rpm2Regions { + cur[reg] = len(rpm2Members(t, db, reg)) + } + got := map[string]int{} + for _, row := range r.Regions { + got[row.Region] = row.Heard + if row.Placed > row.Heard || row.Removed != row.Heard-row.Placed || row.Direct+row.Fewest != row.Placed { + t.Errorf("region %s row inconsistent: %+v", row.Region, row) + } + } + if !reflect.DeepEqual(got, cur) || !reflect.DeepEqual(cur, map[string]int{"AAA": 22, "BBB": 23, "CCC": 23, "DDD": 22}) { + t.Errorf("heard per region %v, today's filter %v, pinned AAA 22 BBB 23 CCC 23 DDD 22", got, cur) + } + if r.Hearings[hearingShared] != 12 || r.Hearings[hearingUnknown] != 20 { + t.Errorf("hearings %v: want 12 shared (3 re-shares × 4) and 20 unknown (5 NULL paths × 4)", r.Hearings) + } + // With the transient rule on, the four one-off far hearings are counted. + span := settings + span.Params.PersistSpan = 24 * time.Hour + rs, err := buildRegionPlacementDryRun(context.Background(), db, span, regionDryRunSweep) + if err != nil { + t.Fatal(err) + } + if rs.Bases.Transient != 4 { + t.Errorf("transient nodes %d, want 4", rs.Bases.Transient) + } +} + +// TestRegionPlacement488_DryRunEmptyDB: a database without advert evidence +// reports zeros instead of failing or dividing by zero. +func TestRegionPlacement488_DryRunEmptyDB(t *testing.T) { + db := setupTestDB(t) + path := filepath.Join(t.TempDir(), "empty.db") + if _, err := db.conn.Exec(`VACUUM INTO ?`, path); err != nil { + t.Fatal(err) + } + db.Close() + var out bytes.Buffer + if err := runRegionPlacementDryRun(path, regionPlacementSettings{Mode: regionModeHeard, Params: defaultRegionPlacementParams}, &out); err != nil { + t.Fatal(err) + } + if !strings.Contains(out.String(), "Nodes with advert evidence in a region: 0") || strings.Contains(out.String(), "NaN") { + t.Fatalf("empty report:\n%s", out.String()) + } +} + +const rpm2DryRunHelperEnv = "CORESCOPE_488_DRYRUN_HELPER" + +// TestRegionPlacement488_DryRunProcess is the child of the process test +// below: it runs the real main() with the dry-run flag. +func TestRegionPlacement488_DryRunProcess(t *testing.T) { + if os.Getenv(rpm2DryRunHelperEnv) != "1" { + t.Skip("helper process for TestRegionPlacement488_DryRunFlag") + } + os.Args = []string{"corescope-server", + "-config-dir", os.Getenv("CORESCOPE_488_DRYRUN_CFG"), + "-db", os.Getenv("CORESCOPE_488_DRYRUN_DB"), + "-region-placement-dryrun"} + main() +} + +// TestRegionPlacement488_DryRunFlag: `corescope-server -region-placement-dryrun` +// prints the report for the configured settings and exits 0 without +// serving. +func TestRegionPlacement488_DryRunFlag(t *testing.T) { + path, _ := rpm2FileDB(t, 1, time.Now().Unix()) + cfgDir := t.TempDir() + cfg := `{"regionPlacement": {"maxHops": 2, "lifetimeDays": 0}}` + if err := os.WriteFile(filepath.Join(cfgDir, "config.json"), []byte(cfg), 0o600); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestRegionPlacement488_DryRunProcess$") + cmd.Env = append(os.Environ(), rpm2DryRunHelperEnv+"=1", "CORESCOPE_488_DRYRUN_CFG="+cfgDir, "CORESCOPE_488_DRYRUN_DB="+path) + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + if err := cmd.Run(); err != nil { + t.Fatalf("dry-run process: %v\nstdout:\n%s\nstderr:\n%s", err, stdout.String(), stderr.String()) + } + for _, want := range []string{"# Region placement dry-run (#488)", "maxHops 2", "lifetimeDays 0 (no window)"} { + if !strings.Contains(stdout.String(), want) { + t.Errorf("stdout lacks %q:\n%s", want, stdout.String()) + } + } + if strings.Contains(stderr.String(), "listening") || strings.Contains(stdout.String(), "listening") { + t.Errorf("dry-run started serving:\n%s", stderr.String()) + } +} + +// TestRegionPlacement488_Config: every field defaults to today's behaviour, +// explicit values are honoured, and values the rule cannot use are replaced. +func TestRegionPlacement488_Config(t *testing.T) { + load := func(t *testing.T, js string) regionPlacementSettings { + t.Helper() + dir := t.TempDir() + if js != "" { + if err := os.WriteFile(filepath.Join(dir, "config.json"), []byte(js), 0o600); err != nil { + t.Fatal(err) + } + } + cfg, err := LoadConfig(dir) + if err != nil { + t.Fatal(err) + } + return cfg.RegionPlacementSettings() + } + def := regionPlacementSettings{Mode: regionModeHeard, Params: regionPlacementParams{MaxHops: 3, Lifetime: 30 * 24 * time.Hour}} + for _, tc := range []struct { + name, js string + want regionPlacementSettings + }{ + {"no config file", "", def}, + {"no block", `{"port": 3000}`, def}, + {"empty block", `{"regionPlacement": {}}`, def}, + {"explicit", `{"regionPlacement": {"mode": " Placed ", "maxHops": 0, "lifetimeDays": 0, "persistSpanHours": 24}}`, + regionPlacementSettings{Mode: regionModePlaced, Params: regionPlacementParams{MaxHops: 0, Lifetime: 0, PersistSpan: 24 * time.Hour}}}, + {"maxHops above the stored slots", `{"regionPlacement": {"maxHops": 99}}`, + regionPlacementSettings{Mode: regionModeHeard, Params: regionPlacementParams{MaxHops: regionPlacementHopSlots - 1, Lifetime: def.Params.Lifetime}}}, + {"negative values", `{"regionPlacement": {"maxHops": -1, "lifetimeDays": -5, "persistSpanHours": -2}}`, + regionPlacementSettings{Mode: regionModeHeard, Params: regionPlacementParams{MaxHops: 0, Lifetime: def.Params.Lifetime}}}, + {"unknown mode keeps heard", `{"regionPlacement": {"mode": "guess"}}`, def}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := load(t, tc.js); got != tc.want { + t.Errorf("settings %+v, want %+v", got, tc.want) + } + }) + } +} + +// TestRegionPlacement488_ConfigExample: config.example.json documents the +// block, with values that equal the defaults. +func TestRegionPlacement488_ConfigExample(t *testing.T) { + raw, err := os.ReadFile("../../config.example.json") + if err != nil { + t.Fatal(err) + } + var ex struct { + RegionPlacement map[string]any `json:"regionPlacement"` + } + if err := json.Unmarshal(raw, &ex); err != nil { + t.Fatal(err) + } + for _, k := range []string{"_comment", "mode", "maxHops", "lifetimeDays", "persistSpanHours"} { + if _, ok := ex.RegionPlacement[k]; !ok { + t.Errorf("config.example.json regionPlacement lacks %q", k) + } + } + var cfg Config + if err := json.Unmarshal(raw, &cfg); err != nil { + t.Fatal(err) + } + if got := cfg.RegionPlacementSettings(); got != (&Config{}).RegionPlacementSettings() { + t.Errorf("config.example.json settings %+v differ from the defaults", got) + } +} diff --git a/cmd/server/region_placement_perf_488_test.go b/cmd/server/region_placement_perf_488_test.go new file mode 100644 index 000000000..45af8dd1b --- /dev/null +++ b/cmd/server/region_placement_perf_488_test.go @@ -0,0 +1,296 @@ +package main + +// Cost of the #488 M2 placement aggregate against the heard cache, and the +// dry-run on the committed e2e fixture. Ported from the measurement file of +// the research PR #601, run against the production code. +// +// The large-database runs are env-gated, like +// nodes_region_cache_perf_176_test.go: +// +// CORESCOPE_PERF_488_GEN=/path/big.db CORESCOPE_PERF_488_SCALE=60 \ +// go test -run '^TestPerf_RegionPlacementM2Generate_488$' -timeout 1h . +// CORESCOPE_PERF_488_DB=/path/big.db \ +// go test -run '^TestPerf_RegionPlacementM2_488$' -v -timeout 1h . + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "runtime" + "sort" + "strconv" + "strings" + "testing" + "time" +) + +func rpm2CopyFile(t testing.TB, src string) string { + t.Helper() + in, err := os.Open(src) + if err != nil { + t.Skipf("fixture not available: %v", err) + } + defer in.Close() + dst := filepath.Join(t.TempDir(), filepath.Base(src)) + out, err := os.Create(dst) + if err != nil { + t.Fatal(err) + } + if _, err := io.Copy(out, in); err != nil { + t.Fatal(err) + } + if err := out.Close(); err != nil { + t.Fatal(err) + } + return dst +} + +// TestRegionPlacement488_FixtureDryRun runs the dry-run on the committed e2e +// fixture, opened read-only through OpenDB, and checks the invariants (heard +// equals today's filter, placed never adds a membership, bases partition the +// nodes), not the counts, so refreshing the fixture does not break it. With +// -v it logs the report. +func TestRegionPlacement488_FixtureDryRun(t *testing.T) { + path := rpm2CopyFile(t, filepath.Join("..", "..", "test-fixtures", "e2e-fixture.db")) + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + settings := regionPlacementSettings{Mode: regionModeHeard, Params: defaultRegionPlacementParams} + r, err := buildRegionPlacementDryRun(context.Background(), db, settings, regionDryRunSweep) + if err != nil { + t.Fatal(err) + } + var b strings.Builder + if err := r.write(&b); err != nil { + t.Fatal(err) + } + t.Logf("e2e fixture:\n%s", b.String()) + if len(r.Regions) == 0 { + t.Fatal("the fixture's observers have regions and hear adverts, but the dry-run reports no region") + } + if r.Bases.Direct+r.Bases.Fewest+r.Bases.Unknown != r.Nodes { + t.Errorf("bases do not partition the nodes: %+v", r.Bases) + } + all := map[string]bool{} + for _, row := range r.Regions { + cur := rpm2Members(t, db, row.Region) + for _, pk := range cur { + all[pk] = true + } + if row.Heard != len(cur) { + t.Errorf("%s: heard %d, today's filter %d", row.Region, row.Heard, len(cur)) + } + if row.Placed > row.Heard { + t.Errorf("%s: placed %d > heard %d", row.Region, row.Placed, row.Heard) + } + } + if len(all) != r.Nodes { + t.Errorf("dry-run saw %d nodes, today's filter %d", r.Nodes, len(all)) + } +} + +// ---- large-database cost (env-gated) ---- + +func TestPerf_RegionPlacementM2Generate_488(t *testing.T) { + path := os.Getenv("CORESCOPE_PERF_488_GEN") + if path == "" { + t.Skip("set CORESCOPE_PERF_488_GEN= to generate the #488 perf database") + } + scale, _ := strconv.Atoi(os.Getenv("CORESCOPE_PERF_488_SCALE")) + if scale <= 0 { + scale = 60 + } + filler, _ := strconv.Atoi(os.Getenv("CORESCOPE_PERF_488_FILLER")) + if filler <= 0 { + filler = 1_000_000 + } + db := setupTestDB(t) + defer db.Close() + tx, err := db.conn.Begin() + if err != nil { + t.Fatal(err) + } + start := time.Now() + truth, _ := rpm2SeedScenario(t, tx, scale, time.Now().Unix()) + // Non-advert traffic around the adverts, as on a live instance, so the + // scans pay for skipping it. + for i := range filler { + raw := fmt.Sprintf("1502a1a2%064x", i) + res, err := tx.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES (?, ?, '2026-10-01T00:00:00Z', 1, 5, '{"type":"GRP_TXT"}')`, raw, fmt.Sprintf("fill-%d", i)) + if err != nil { + t.Fatal(err) + } + id, _ := res.LastInsertId() + if _, err := tx.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, raw_hex) + VALUES (?, ?, '["A1","A2"]', ?, ?)`, id, 1+i%(len(rpm2Regions)*rpm2ObsPerRegion), time.Now().Unix()-int64(i), raw); err != nil { + t.Fatal(err) + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + for _, q := range []string{ + `CREATE INDEX IF NOT EXISTS idx_transmissions_payload_type ON transmissions(payload_type)`, + `CREATE INDEX IF NOT EXISTS idx_observations_observer_idx ON observations(observer_idx)`, + `ANALYZE`, + } { + if _, err := db.conn.Exec(q); err != nil { + t.Fatal(err) + } + } + _ = os.Remove(path) + if _, err := db.conn.Exec(`VACUUM INTO ?`, path); err != nil { + t.Fatal(err) + } + var obs int + db.conn.QueryRow(`SELECT COUNT(*) FROM observations`).Scan(&obs) + t.Logf("generated %s: %d nodes, %d observations in %v", path, len(truth), obs, time.Since(start).Round(time.Second)) +} + +// TestPerf_RegionPlacementM2_488 times, on one database, medians of 7: +// - heard: cold build of every region (each set scans separately), warm +// hit, delta refresh over the last N observations; +// - placed: cold global aggregate + placements serving every region, +// placement-only recompute, delta scan + merge + recompute, warm hit; +// - heap of the four heard entries vs the aggregate and its snapshot. +func TestPerf_RegionPlacementM2_488(t *testing.T) { + path := os.Getenv("CORESCOPE_PERF_488_DB") + if path == "" { + t.Skip("set CORESCOPE_PERF_488_DB= (see TestPerf_RegionPlacementM2Generate_488)") + } + rpm2PerfReport(t, path, 5000) +} + +// TestRegionPlacement488_FixtureCost runs the same timing on the e2e +// fixture, always (logs only). +func TestRegionPlacement488_FixtureCost(t *testing.T) { + rpm2PerfReport(t, rpm2CopyFile(t, filepath.Join("..", "..", "test-fixtures", "e2e-fixture.db")), 100) +} + +func rpm2Median(f func()) time.Duration { + const runs = 7 + d := make([]time.Duration, runs) + for i := range d { + start := time.Now() + f() + d[i] = time.Since(start) + } + sort.Slice(d, func(i, j int) bool { return d[i] < d[j] }) + return d[runs/2] +} + +func rpm2Heap() uint64 { + runtime.GC() + runtime.GC() + var m runtime.MemStats + runtime.ReadMemStats(&m) + return m.HeapAlloc +} + +func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { + t.Helper() + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + ctx := context.Background() + var regions []string + rows, err := db.conn.Query(`SELECT DISTINCT UPPER(TRIM(iata)) FROM observers WHERE iata IS NOT NULL AND TRIM(iata) != '' ORDER BY 1`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var r string + if err := rows.Scan(&r); err != nil { + t.Fatal(err) + } + regions = append(regions, r) + } + rows.Close() + wm := rpm2MaxID(t, db) + lo := max(wm-deltaRows, 0) + + clearEntries := func() { + db.nodeRegionCacheMu.Lock() + db.nodeRegionCache, db.nodeRegionUsed = nil, nil + db.nodeRegionCacheMu.Unlock() + } + clearPlaced := func() { + clearEntries() + db.regionAggMu.Lock() + db.regionAgg = nil + db.regionAggMu.Unlock() + db.regionSnap.Store(nil) + } + serveAll := func() { + for _, r := range regions { + if _, err := db.nodeRegionKeysJSON([]string{r}); err != nil { + t.Fatal(err) + } + } + } + + // heard (today) + db.regionPlacement = regionPlacementSettings{Mode: regionModeHeard, Params: defaultRegionPlacementParams} + heardCold := rpm2Median(func() { clearEntries(); serveAll() }) + heardWarm := rpm2Median(serveAll) + heardDelta := rpm2Median(func() { + for _, r := range regions { + key := nodeRegionKey([]string{r}) + e := *db.getNodeRegionEntry(key) + e.watermark, e.refreshed = lo, time.Now().Add(-2*nodeRegionFreshTTL) + db.setNodeRegionEntry(key, &e) + if _, err := db.refreshNodeRegion(key, []string{r}); err != nil { + t.Fatal(err) + } + } + }) + clearEntries() + h0 := rpm2Heap() + serveAll() + heardHeap := int64(rpm2Heap()) - int64(h0) + clearEntries() + + // placed + rpm2SetPlaced(db, defaultRegionPlacementParams) + placedCold := rpm2Median(func() { clearPlaced(); serveAll() }) + placedWarm := rpm2Median(serveAll) + db.regionAggMu.Lock() + agg := db.regionAgg + db.regionAggMu.Unlock() + placeOnly := rpm2Median(func() { newRegionPlacementSnapshot(agg, defaultRegionPlacementParams, time.Now()) }) + // The delta re-folds the same tail rows each run: the stats merge by + // min/max, so the placement work is the same as for new rows. + placedDelta := rpm2Median(func() { + d := newRegionEvidenceAgg() + if _, err := db.scanRegionEvidence(ctx, lo, wm, true, d); err != nil { + t.Fatal(err) + } + agg.merge(d) + snap := newRegionPlacementSnapshot(agg, defaultRegionPlacementParams, time.Now()) + for _, r := range regions { + if _, err := newNodeRegionEntry(snap.members([]string{r}), wm, time.Now()); err != nil { + t.Fatal(err) + } + } + }) + pairs, nodes := agg.pairs, len(agg.nodes) + agg = nil + clearPlaced() + h1 := rpm2Heap() + serveAll() + placedHeap := int64(rpm2Heap()) - int64(h1) + + t.Logf("%s: %d observations, %d regions, %d nodes, %d (node, region) pairs", filepath.Base(path), wm, len(regions), nodes, pairs) + t.Logf("heard cold %v | warm %v | delta refresh (%d obs) %v | heap %d KiB", + heardCold.Round(time.Microsecond), heardWarm, deltaRows, heardDelta.Round(time.Microsecond), heardHeap/1024) + t.Logf("placed cold %v | warm %v | delta scan+merge+placement+entries (%d obs) %v | placement only %v | heap %d KiB", + placedCold.Round(time.Microsecond), placedWarm, deltaRows, placedDelta.Round(time.Microsecond), + placeOnly.Round(time.Microsecond), placedHeap/1024) +} diff --git a/cmd/server/store.go b/cmd/server/store.go index f8bc388d3..c4297b6ee 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -121,10 +121,10 @@ type StoreObs struct { // wireRoute, so StoreObs stays 168 bytes (#490). sharedStrs uint8 // advertEv is what this observation's own frame says about an ADVERT's - // originator (packetpath.ClassifyAdvertFrame, #488): 1 + AdvertKind, or - // 0 when no frame was stored (a legacy row, see advertHearing). It - // takes one more padding byte, so StoreObs stays 168 bytes. - advertEv uint8 + // originator (packetpath.ClassifyAdvertFrame, #488), see + // advertEvidence. It takes two more padding bytes, so StoreObs stays + // 168 bytes. + advertEv advertEvidence } // tsCache holds tsCacheNotParsed until the first ParsedTime call, then one From e6222c8d20f2cacb5da98762f0b701a3da313671 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 25 Oct 2026 19:30:48 +0000 Subject: [PATCH 2/2] feat(regions): direct-evidence placement aggregate + operator dry-run (#488 M2) Green commit for #488 milestone M2. - region_placement.go: one global (node, region) evidence aggregate, classified per observation by M1's packetpath.ClassifyAdvert, behind the nodes-region cache machinery (watermark, delta scans merged only on success, full rebuild every 30 min, per-set singleflight, fresh/stale ages). Rule: verified direct evidence first, then the fewest-hops fallback capped at maxHops (a legacy '[]' without a frame is a 0-hop fallback only), else unknown. Membership is re-derived on every refresh, never extended. - regionPlacement config (mode/maxHops/lifetimeDays/persistSpanHours); every default keeps today's behaviour: mode "heard" never builds the aggregate. Noted for the customizer later. - corescope-server -region-placement-dryrun: read-only report (counts per evidence kind, basis and region, unknown share for several maxHops values), no node keys. - Review N2 on #609: StoreObs.advertEv keeps a relayed frame's hop count, so advertHearing no longer returns Hops 0 for relayed hearings. - Review N7 on #609: the computeNodeHomeRegions caller comment describes the M1 rule. Relates to #488 Co-Authored-By: Claude Opus 5.5 --- cmd/server/advert_evidence.go | 9 +- cmd/server/config.go | 31 +- cmd/server/db.go | 17 +- cmd/server/main.go | 14 + cmd/server/nodes_region_cache.go | 9 +- cmd/server/region_placement.go | 414 ++++++++++++++++-- cmd/server/region_placement_488_test.go | 5 +- cmd/server/region_placement_dryrun.go | 194 +++++++- .../region_placement_dryrun_488_test.go | 66 ++- cmd/server/region_placement_perf_488_test.go | 50 ++- cmd/server/store.go | 11 +- config.example.json | 7 + 12 files changed, 751 insertions(+), 76 deletions(-) diff --git a/cmd/server/advert_evidence.go b/cmd/server/advert_evidence.go index b098f5a48..3ecaeea91 100644 --- a/cmd/server/advert_evidence.go +++ b/cmd/server/advert_evidence.go @@ -7,7 +7,9 @@ import "github.com/meshcore-analyzer/packetpath" type advertEvidence struct { // code is 1 + the packetpath.AdvertKind, or 0 when no frame was stored. code uint8 - // hops is the relay count of an AdvertRelayed frame. + // hops is the relay count of an AdvertRelayed frame: at most 63, the + // largest count a path byte holds, so it fits a byte (review N2 on + // #609: the fewest-hops fallback needs it). hops uint8 } @@ -18,14 +20,15 @@ func advertEvidenceFromRawHex(rawHex string) advertEvidence { if rawHex == "" { return advertEvidence{} } - return advertEvidence{code: uint8(packetpath.ClassifyAdvertFrame(rawHex).Kind) + 1} + h := packetpath.ClassifyAdvertFrame(rawHex) + return advertEvidence{code: uint8(h.Kind) + 1, hops: uint8(h.Hops)} } // advertHearing classifies one ADVERT observation: by its own frame when one // was stored, else (a legacy row) by its path_json, unverified. func (o *StoreObs) advertHearing() packetpath.AdvertHearing { if o.advertEv.code != 0 { - return packetpath.AdvertHearing{Kind: packetpath.AdvertKind(o.advertEv.code - 1), Verified: true} + return packetpath.AdvertHearing{Kind: packetpath.AdvertKind(o.advertEv.code - 1), Hops: int(o.advertEv.hops), Verified: true} } return packetpath.ClassifyAdvertPathJSON(o.PathJSON) } diff --git a/cmd/server/config.go b/cmd/server/config.go index 1888f2987..3ca5310c4 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -1751,5 +1751,34 @@ type RegionPlacementConfig struct { // RegionPlacementSettings resolves the regionPlacement block, logging and // replacing values it cannot use. func (c *Config) RegionPlacementSettings() regionPlacementSettings { - return regionPlacementSettings{} + out := regionPlacementSettings{Mode: regionModeHeard, Params: defaultRegionPlacementParams} + rp := c.RegionPlacement + if rp == nil { + return out + } + switch mode := strings.ToLower(strings.TrimSpace(rp.Mode)); mode { + case "", regionModeHeard: + case regionModePlaced: + out.Mode = regionModePlaced + default: + log.Printf("[config] regionPlacement.mode %q is not heard or placed; using heard", rp.Mode) + } + if rp.MaxHops != nil { + out.Params.MaxHops = min(max(*rp.MaxHops, 0), regionPlacementHopSlots-1) + if out.Params.MaxHops != *rp.MaxHops { + log.Printf("[config] regionPlacement.maxHops %d is outside 0..%d; using %d", + *rp.MaxHops, regionPlacementHopSlots-1, out.Params.MaxHops) + } + } + if rp.LifetimeDays != nil { + if *rp.LifetimeDays >= 0 { + out.Params.Lifetime = time.Duration(*rp.LifetimeDays) * 24 * time.Hour + } else { + log.Printf("[config] regionPlacement.lifetimeDays %d is negative; using the default", *rp.LifetimeDays) + } + } + if rp.PersistSpanHours != nil && *rp.PersistSpanHours > 0 { + out.Params.PersistSpan = time.Duration(*rp.PersistSpanHours) * time.Hour + } + return out } diff --git a/cmd/server/db.go b/cmd/server/db.go index 9bd4c93fa..21d98e292 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -145,14 +145,15 @@ type DB struct { // regionAgg is the global evidence aggregate, owned by the refresh // that holds regionAggMu; readers only ever see regionSnap, an // immutable snapshot swapped in after each refresh. - regionPlacement regionPlacementSettings - regionAggMu sync.Mutex - regionAgg *regionEvidenceAgg - regionAggBuilt time.Time - regionSnap atomic.Pointer[regionPlacementSnapshot] - regionPlacementSF singleflight.Group - // regionPlacementScanHook (test seam, nil in production) runs before - // every aggregate scan, full or delta; a non-nil error fails that scan. + regionPlacement regionPlacementSettings + regionAggMu sync.Mutex + regionAgg *regionEvidenceAgg + regionAggBuilt time.Time + regionSnap atomic.Pointer[regionPlacementSnapshot] + // regionPlacementScanHook (test seam, nil in production) runs at the + // end of every aggregate scan, full or delta, after its rows were + // folded in; a non-nil error fails that scan the way an error part-way + // through the rows does. regionPlacementScanHook func(full bool) error // identityNamesQueryHook (#410, test seam, nil in production) runs once // per real hidden-name lookup query, right before it executes, with the diff --git a/cmd/server/main.go b/cmd/server/main.go index 61be778db..6dc8cdf2c 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -91,6 +91,7 @@ func main() { publicDir string pollMs int wsHistoryFile string + regionDryRun bool ) flag.StringVar(&configDir, "config-dir", ".", "Directory containing config.json") @@ -99,6 +100,7 @@ func main() { flag.StringVar(&publicDir, "public", "public", "Directory to serve static files from") flag.IntVar(&pollMs, "poll-ms", 1000, "SQLite poll interval for WebSocket broadcast (ms)") flag.StringVar(&wsHistoryFile, "ws-history-file", "", "WebSocket history file (default: beside packet database; off disables persistence)") + flag.BoolVar(®ionDryRun, "region-placement-dryrun", false, "Print the read-only #488 region placement report for -db and the regionPlacement config, then exit") flag.Parse() // Load config @@ -164,6 +166,12 @@ func main() { // Resolve DB path resolvedDB := cfg.ResolveDBPath(configDir) + if regionDryRun { + if err := runRegionPlacementDryRun(resolvedDB, cfg.RegionPlacementSettings(), os.Stdout); err != nil { + log.Fatalf("[region-placement] dry-run: %v", err) + } + return + } log.Printf("[config] port=%d db=%s public=%s", cfg.Port, resolvedDB, publicDir) if len(cfg.NodeBlacklist) > 0 { log.Printf("[config] nodeBlacklist: %d node(s) will be hidden from API", len(cfg.NodeBlacklist)) @@ -179,6 +187,12 @@ func main() { if err != nil { log.Fatalf("[db] failed to open %s: %v", resolvedDB, err) } + database.regionPlacement = cfg.RegionPlacementSettings() + if database.regionPlacement.Mode != regionModeHeard { + log.Printf("[config] regionPlacement: mode=%s maxHops=%d lifetime=%v persistSpan=%v", + database.regionPlacement.Mode, database.regionPlacement.Params.MaxHops, + database.regionPlacement.Params.Lifetime, database.regionPlacement.Params.PersistSpan) + } var dbCloseOnce sync.Once dbClose := func() error { var err error diff --git a/cmd/server/nodes_region_cache.go b/cmd/server/nodes_region_cache.go index d7f5bd719..13085254a 100644 --- a/cmd/server/nodes_region_cache.go +++ b/cmd/server/nodes_region_cache.go @@ -268,9 +268,14 @@ func (db *DB) refreshNodeRegion(key string, codes []string) (*nodeRegionEntry, e var e *nodeRegionEntry var err error - if prev == nil || prev.rebuildDue() { + switch { + case db.regionPlacement.Mode == regionModePlaced: + // #488 M2: cut from the global placement aggregate, which keeps its + // own watermark and rebuild clock (region_placement.go). + e, err = db.placedNodeRegion(ctx, codes) + case prev == nil || prev.rebuildDue(): e, err = db.buildNodeRegion(ctx, key, codes) - } else { + default: e, err = db.extendNodeRegion(ctx, key, codes, prev) } if err != nil { diff --git a/cmd/server/region_placement.go b/cmd/server/region_placement.go index 4c788f642..9874e916d 100644 --- a/cmd/server/region_placement.go +++ b/cmd/server/region_placement.go @@ -2,6 +2,11 @@ package main import ( "context" + "database/sql" + "fmt" + "log" + "math" + "sort" "time" "github.com/meshcore-analyzer/packetpath" @@ -9,8 +14,45 @@ import ( // Direct-evidence region placement for /api/nodes?region= (#488 M2). // -// STUB (red commit): types and signatures only. +// The heard rule (nodes_region_cache.go) puts a node in every region whose +// observers heard any of its adverts, so a flooded advert drags a distant +// node into every region it reaches. The placed rule decides from what each +// ADVERT observation says about its originator (packetpath.ClassifyAdvert, +// M1): +// +// 1. direct: every region that heard the node's own transmission (hash +// count 0 on its own frame) is a home region. Ties are a set. +// 2. fewest_hops: with no direct evidence anywhere, the regions that heard +// it in the fewest hops, up to MaxHops. A legacy row's explicit '[]' +// (no frame of its own, so a contact re-share cannot be excluded) is a +// 0-hop fallback, never direct evidence (owner decision on #488). +// 3. unknown: in no region filter; still listed unfiltered. +// +// Evidence older than Lifetime before the node's own newest hearing is +// ignored, so a quiet node keeps the placement its last active period +// earned and a node that moved drops its old region. With PersistSpan set, +// direct regions whose evidence spans less than it are dropped when another +// region's spans at least it (one-off long-distance reception). +// +// Placement needs every region's evidence for a node at once (direct +// evidence anywhere decides whether the fallback applies here), so it is +// computed from ONE global aggregate rather than per region set. The +// aggregate keeps the heard cache's machinery: an observations.id +// watermark, delta scans past it, a full rebuild every +// nodeRegionRebuildInterval (retention pruning, observer region edits and +// late from_pubkey backfills change rows below the watermark), the per-set +// singleflight and the same fresh/stale ages. Unlike the heard rule, +// placement is not monotone — a new direct hearing removes fallback +// regions — so every refresh re-derives the placements instead of +// extending them. +// +// Everything is read from SQLite on the server's read-only handle; nothing +// touches the PacketStore or its lock. +// regionPlacementHopSlots bounds the per-pair state: hop counts +// 1..regionPlacementHopSlots-1 keep their newest timestamp. Higher counts +// can never place a node (MaxHops is capped below this) and are only +// counted. const regionPlacementHopSlots = 8 const ( @@ -25,8 +67,12 @@ const ( ) type regionPlacementParams struct { - MaxHops int - Lifetime time.Duration + // MaxHops caps the relayed fallback (0 = none). + MaxHops int + // Lifetime is the evidence window before the node's own newest + // hearing; 0 = no window. + Lifetime time.Duration + // PersistSpan turns on the transient-region rule; 0 = off. PersistSpan time.Duration } @@ -35,82 +81,388 @@ type regionPlacementSettings struct { Params regionPlacementParams } +// defaultRegionPlacementParams: MaxHops 3 is a working default to be +// checked with the dry-run on a real instance; 30 days matches the default +// retention.packetDays; the transient rule stays off (deferred on #488). var defaultRegionPlacementParams = regionPlacementParams{ MaxHops: 3, Lifetime: 30 * 24 * time.Hour, } +// regionPlacement says where a node is placed and on what basis. type regionPlacement struct { - Basis string - Regions []string + Basis string + // Regions are the inferred home regions, sorted; empty for unknown. + Regions []string + // Transient are direct regions outranked by persistent direct + // evidence elsewhere (PersistSpan only). Transient []string - Hops int + // Hops is 0 for direct, the fallback hop count for fewest_hops (0 for + // a legacy '[]'), -1 for unknown. + Hops int } +// regionEvidenceStats is the bounded evidence one region holds about one +// node. Every field merges by min/max, so folding a delta scan into a full +// build equals a full build over the union. type regionEvidenceStats struct { - LastAt [regionPlacementHopSlots]int64 + // LastAt[0] is the newest verified direct hearing, LastAt[h] the newest + // hearing relayed h times; 0 = never. + LastAt [regionPlacementHopSlots]int64 + // DirectFirst is the oldest verified direct hearing; 0 = never. DirectFirst int64 - LegacyAt int64 -} - -func (s *regionEvidenceStats) add(ts int64, h packetpath.AdvertHearing) regionHearingKind { - return hearingUnknown + // LegacyAt is the newest explicit '[]' on a row without a frame. + LegacyAt int64 } -func (s *regionEvidenceStats) merge(o *regionEvidenceStats) {} - +// regionHearingKind is how one observation was folded into the aggregate. type regionHearingKind uint8 const ( - hearingUnknown regionHearingKind = iota - hearingDirect - hearingLegacy - hearingRelayed - hearingFar - hearingShared + hearingUnknown regionHearingKind = iota // no usable hop count, or undated + hearingDirect // verified, own transmission + hearingLegacy // '[]' without a frame: fallback only + hearingRelayed // relayed, within the stored slots + hearingFar // relayed past the stored slots + hearingShared // a contact re-share: no evidence hearingKinds ) +// regionHearingCounts counts folded observations per kind, for the dry-run. type regionHearingCounts [hearingKinds]int64 +// add folds one hearing at unix time ts and reports how it was counted. +func (s *regionEvidenceStats) add(ts int64, h packetpath.AdvertHearing) regionHearingKind { + if ts <= 0 { + return hearingUnknown + } + switch h.Kind { + case packetpath.AdvertDirect: + if !h.Verified { + s.LegacyAt = max(s.LegacyAt, ts) + return hearingLegacy + } + s.LastAt[0] = max(s.LastAt[0], ts) + s.DirectFirst = minNonZero(s.DirectFirst, ts) + return hearingDirect + case packetpath.AdvertRelayed: + switch { + case h.Hops < 1: + return hearingUnknown // the classifiers never yield this; keep slot 0 direct-only + case h.Hops < regionPlacementHopSlots: + s.LastAt[h.Hops] = max(s.LastAt[h.Hops], ts) + return hearingRelayed + } + return hearingFar + case packetpath.AdvertShared: + return hearingShared + } + return hearingUnknown +} + +// merge folds o into s. +func (s *regionEvidenceStats) merge(o *regionEvidenceStats) { + for i, ts := range o.LastAt { + s.LastAt[i] = max(s.LastAt[i], ts) + } + s.DirectFirst = minNonZero(s.DirectFirst, o.DirectFirst) + s.LegacyAt = max(s.LegacyAt, o.LegacyAt) +} + +// newest is the newest dated hearing that can place a node. +func (s *regionEvidenceStats) newest() int64 { + n := s.LegacyAt + for _, ts := range s.LastAt { + n = max(n, ts) + } + return n +} + +// minNonZero is the smaller of a and b, where 0 means "none". +func minNonZero(a, b int64) int64 { + if a == 0 || (b != 0 && b < a) { + return b + } + return a +} + +// regionEvidenceAgg is the global evidence table: one entry per (node, +// region) pair that heard the node's adverts, keyed by node. Its size is +// bounded by the nodes in the retained history times the regions each was +// heard in; repeated hearings only update the fixed-size stats. type regionEvidenceAgg struct { nodes map[string]map[string]*regionEvidenceStats pairs int - watermark int64 + watermark int64 // max observations.id folded in hearings regionHearingCounts } func newRegionEvidenceAgg() *regionEvidenceAgg { - return ®ionEvidenceAgg{nodes: map[string]map[string]*regionEvidenceStats{}} + return ®ionEvidenceAgg{nodes: make(map[string]map[string]*regionEvidenceStats)} +} + +// pair returns the stats for (pk, region), creating them. +func (a *regionEvidenceAgg) pair(pk, region string) *regionEvidenceStats { + m := a.nodes[pk] + if m == nil { + m = make(map[string]*regionEvidenceStats, 2) + a.nodes[pk] = m + } + s := m[region] + if s == nil { + s = ®ionEvidenceStats{} + m[region] = s + a.pairs++ + } + return s } -func (a *regionEvidenceAgg) add(pk, region string, ts int64, h packetpath.AdvertHearing) {} +func (a *regionEvidenceAgg) add(pk, region string, ts int64, h packetpath.AdvertHearing) { + a.hearings[a.pair(pk, region).add(ts, h)]++ +} -func (a *regionEvidenceAgg) merge(d *regionEvidenceAgg) {} +// merge folds the delta d (scanned past a's watermark) into a. +func (a *regionEvidenceAgg) merge(d *regionEvidenceAgg) { + for pk, per := range d.nodes { + for region, s := range per { + a.pair(pk, region).merge(s) + } + } + for i, n := range d.hearings { + a.hearings[i] += n + } + a.watermark = max(a.watermark, d.watermark) +} +// scanRegionEvidence folds every ADVERT observation with lo < id <= hi, +// heard by an observer with a region, into agg, and raises agg's watermark +// to hi. Same scope as scanNodeRegionKeys (so a placed region is always a +// heard one), plus the per-observation frame and path the classifier needs. +// A delta scan forces the join order to drive from the rowid range, as +// scanNodeRegionKeys explains. func (db *DB) scanRegionEvidence(ctx context.Context, lo, hi int64, delta bool, agg *regionEvidenceAgg) (int, error) { - return 0, nil + joinCond, tsCol := "obs.rowid = o.observer_idx", "o.timestamp" + if !db.isV3() { + // The legacy schema stores ISO-8601 text timestamps. + joinCond = "obs.id = o.observer_id" + tsCol = "CASE WHEN typeof(o.timestamp) = 'integer' THEN o.timestamp ELSE CAST(strftime('%s', o.timestamp) AS INTEGER) END" + } + join := "JOIN" + if delta { + join = "CROSS JOIN" + } + rawCol := "NULL" + if db.hasObsRawHex() { + rawCol = "o.raw_hex" + } + // Only fixed fragments are formatted in; every value is bound. See + // scanNodeRegionKeys for the from_pubkey contract. + q := fmt.Sprintf(`SELECT t.from_pubkey, UPPER(TRIM(obs.iata)), %[3]s, o.path_json, %[4]s + FROM observations o + %[1]s transmissions t ON t.id = o.transmission_id + %[1]s observers obs ON %[2]s + WHERE t.payload_type = ? + AND t.from_pubkey IS NOT NULL + AND obs.iata IS NOT NULL AND TRIM(obs.iata) != '' + AND o.id > ? AND o.id <= ?`, join, joinCond, tsCol, rawCol) + rows, err := db.conn.QueryContext(ctx, q, payloadTypeAdvert, lo, hi) + if err != nil { + return 0, fmt.Errorf("scan region evidence: %w", err) + } + defer rows.Close() + n := 0 + for rows.Next() { + var pk, region string + var ts sql.NullInt64 + var path, raw sql.NullString + if err := rows.Scan(&pk, ®ion, &ts, &path, &raw); err != nil { + return n, fmt.Errorf("scan region evidence: %w", err) + } + agg.add(pk, region, ts.Int64, packetpath.ClassifyAdvert(raw.String, path.String)) + n++ + } + if db.regionPlacementScanHook != nil { + if err := db.regionPlacementScanHook(!delta); err != nil { + return n, err + } + } + if err := rows.Err(); err != nil { + return n, fmt.Errorf("scan region evidence: %w", err) + } + agg.watermark = max(agg.watermark, hi) + return n, nil } +// placeNode applies the rule to one node's per-region evidence. func placeNode(perRegion map[string]*regionEvidenceStats, p regionPlacementParams) regionPlacement { - return regionPlacement{} + unknown := regionPlacement{Basis: placementBasisUnknown, Hops: -1} + var nodeLast int64 + for _, s := range perRegion { + nodeLast = max(nodeLast, s.newest()) + } + if nodeLast == 0 { + return unknown + } + ws := int64(math.MinInt64) + if p.Lifetime > 0 { + ws = nodeLast - int64(p.Lifetime/time.Second) + } + within := func(ts int64) bool { return ts != 0 && ts >= ws } + + var direct, persistent, transient []string + span := int64(p.PersistSpan / time.Second) + for r, s := range perRegion { + if !within(s.LastAt[0]) { + continue + } + direct = append(direct, r) + if span > 0 { + if s.LastAt[0]-max(s.DirectFirst, ws) >= span { + persistent = append(persistent, r) + } else { + transient = append(transient, r) + } + } + } + if len(direct) > 0 { + out := regionPlacement{Basis: placementBasisDirect, Regions: direct} + if len(persistent) > 0 && len(transient) > 0 { + out.Regions, out.Transient = persistent, transient + sort.Strings(out.Transient) + } + sort.Strings(out.Regions) + return out + } + // h == 0 is the legacy '[]' tier, so MaxHops 0 still lets it place. + for h := 0; h == 0 || h <= p.MaxHops && h < regionPlacementHopSlots; h++ { + var tied []string + for r, s := range perRegion { + ts := s.LastAt[h] + if h == 0 { + ts = s.LegacyAt // slot 0 is verified direct, handled above + } + if within(ts) { + tied = append(tied, r) + } + } + if len(tied) > 0 { + sort.Strings(tied) + return regionPlacement{Basis: placementBasisFewest, Regions: tied, Hops: h} + } + } + return unknown } +// regionPlacementSnapshot is the immutable result of one aggregate refresh: +// the region → members index the per-set cache entries are cut from. type regionPlacementSnapshot struct { - byRegion map[string][]string + byRegion map[string][]string // sorted pubkeys placed in each region watermark int64 refreshed time.Time - built time.Time + built time.Time // the aggregate's last full build } +// newRegionPlacementSnapshot places every node in agg. It shares the +// pubkey strings with agg, so the index costs one slice slot per placed +// membership (at most agg.pairs). func newRegionPlacementSnapshot(agg *regionEvidenceAgg, p regionPlacementParams, built time.Time) *regionPlacementSnapshot { - return ®ionPlacementSnapshot{} + byRegion := make(map[string][]string) + for pk, per := range agg.nodes { + for _, r := range placeNode(per, p).Regions { + byRegion[r] = append(byRegion[r], pk) + } + } + for _, keys := range byRegion { + sort.Strings(keys) + } + return ®ionPlacementSnapshot{byRegion: byRegion, watermark: agg.watermark, refreshed: time.Now(), built: built} } +// members is the set of nodes placed in any of codes. func (s *regionPlacementSnapshot) members(codes []string) map[string]struct{} { - return map[string]struct{}{} + set := make(map[string]struct{}) + for _, c := range codes { + for _, pk := range s.byRegion[c] { + set[pk] = struct{}{} + } + } + return set } +// currentRegionPlacement returns a snapshot no older than +// nodeRegionFreshTTL, refreshing the aggregate when needed. It is called +// from a per-set refresh (already deduplicated per region set by +// nodeRegionSF), so it never runs on a request a cached entry can serve. +// Refreshes for different region sets coalesce on regionAggMu: the first +// builds, the ones queued behind it find no rows past its watermark and +// only re-stamp its snapshot. func (db *DB) currentRegionPlacement(ctx context.Context) (*regionPlacementSnapshot, error) { - return ®ionPlacementSnapshot{}, nil + if s := db.regionSnap.Load(); s != nil && time.Since(s.refreshed) < nodeRegionFreshTTL { + return s, nil + } + return db.refreshRegionPlacement(ctx) +} + +// refreshRegionPlacement brings the aggregate up to date — a full build when +// there is none or one is due, else a delta past the watermark scanned into +// a separate aggregate and merged only on success — and publishes a new +// snapshot. A failed refresh leaves the aggregate and the published +// snapshot as they were. +func (db *DB) refreshRegionPlacement(ctx context.Context) (*regionPlacementSnapshot, error) { + db.regionAggMu.Lock() + defer db.regionAggMu.Unlock() + params := db.regionPlacement.Params + if db.regionAgg == nil || time.Since(db.regionAggBuilt) >= nodeRegionRebuildInterval { + // Full builds share nodeRegionFullMu with the heard cache's, so at + // most one full observation scan holds a pooled connection. + db.nodeRegionFullMu.Lock() + defer db.nodeRegionFullMu.Unlock() + wm, err := db.maxObservationID(ctx) + if err != nil { + return nil, err + } + start := time.Now() + agg := newRegionEvidenceAgg() + if _, err := db.scanRegionEvidence(ctx, 0, wm, false, agg); err != nil { + return nil, err + } + db.regionAgg, db.regionAggBuilt = agg, start + s := newRegionPlacementSnapshot(agg, params, start) + db.regionSnap.Store(s) + log.Printf("[nodes-region] placement: full build, %d nodes, %d (node, region) pairs in %v (obs id <= %d)", + len(agg.nodes), agg.pairs, time.Since(start).Round(time.Millisecond), wm) + return s, nil + } + wm, err := db.maxObservationID(ctx) + if err != nil { + return nil, err + } + agg := db.regionAgg + if wm <= agg.watermark { + prev := db.regionSnap.Load() + s := *prev + s.refreshed = time.Now() + db.regionSnap.Store(&s) + return &s, nil + } + delta := newRegionEvidenceAgg() + if _, err := db.scanRegionEvidence(ctx, agg.watermark, wm, true, delta); err != nil { + return nil, err + } + agg.merge(delta) + s := newRegionPlacementSnapshot(agg, params, db.regionAggBuilt) + db.regionSnap.Store(s) + return s, nil +} + +// placedNodeRegion cuts the entry for one region set from the current +// placement snapshot. The entry's age and stale-while-revalidate handling +// stay those of nodeRegionKeysJSON; its membership is re-derived on every +// refresh, never extended. +func (db *DB) placedNodeRegion(ctx context.Context, codes []string) (*nodeRegionEntry, error) { + s, err := db.currentRegionPlacement(ctx) + if err != nil { + return nil, err + } + return newNodeRegionEntry(s.members(codes), s.watermark, s.built) } diff --git a/cmd/server/region_placement_488_test.go b/cmd/server/region_placement_488_test.go index 005e07cc5..310affef4 100644 --- a/cmd/server/region_placement_488_test.go +++ b/cmd/server/region_placement_488_test.go @@ -331,6 +331,9 @@ func TestRegionPlacement488_DeltaEqualsFullBuild(t *testing.T) { if !reflect.DeepEqual(rpm2Placements(inc, params), rpm2Placements(full, params)) { t.Fatalf("delta from %d: placements differ from the full build", mid) } + if !reflect.DeepEqual(inc.nodes, full.nodes) { + t.Fatalf("delta from %d: per-pair stats differ from the full build", mid) + } if inc.watermark != wm || inc.pairs != full.pairs || inc.hearings != full.hearings { t.Fatalf("delta from %d: watermark %d/%d, pairs %d/%d, hearings %v/%v", mid, inc.watermark, wm, inc.pairs, full.pairs, inc.hearings, full.hearings) @@ -368,7 +371,7 @@ func TestRegionPlacement488_PlacedModeDropsRelayedOnly(t *testing.T) { if got := rpm2Members(t, db, "AAA"); !reflect.DeepEqual(got, []string{rpm2PK}) { t.Fatalf("placed mode: AAA members %v", got) } - if got := rpm2Members(t, db, "aaa", "BBB"); !reflect.DeepEqual(got, []string{rpm2PK}) { + if got := rpm2Members(t, db, "AAA", "BBB"); !reflect.DeepEqual(got, []string{rpm2PK}) { t.Fatalf("placed mode: AAA,BBB members %v (union of codes)", got) } } diff --git a/cmd/server/region_placement_dryrun.go b/cmd/server/region_placement_dryrun.go index 917017436..7d5430dab 100644 --- a/cmd/server/region_placement_dryrun.go +++ b/cmd/server/region_placement_dryrun.go @@ -2,27 +2,68 @@ package main import ( "context" + "fmt" "io" + "sort" + "strings" + "time" ) -// STUB (red commit): the operator dry-run report for #488 M2. +// Read-only dry-run of the #488 placement rule for operators +// (`corescope-server -region-placement-dryrun`). It builds the same global +// aggregate the placed mode serves from, on the database the server would +// open, and prints counts only — per evidence kind, per basis, per region, +// and the unplaced share for several maxHops values — so the maxHops +// default can be chosen from a real instance before placed mode is turned +// on. It names no node: no public key, no name, no observer. +// regionDryRunSweep are the maxHops values the report compares. var regionDryRunSweep = []int{0, 1, 2, 3, 4, 5, 7} +// regionBasisCounts counts nodes per placement basis. type regionBasisCounts struct { - Direct, Fewest, Unknown, Transient int - FewestByHops [regionPlacementHopSlots]int + Direct, Fewest, Unknown int + // Transient counts nodes with at least one transient region. + Transient int + // FewestByHops counts fewest_hops nodes by the hop count that placed + // them (0 = a legacy '[]'). + FewestByHops [regionPlacementHopSlots]int } +func (c *regionBasisCounts) add(p regionPlacement) { + switch p.Basis { + case placementBasisDirect: + c.Direct++ + case placementBasisFewest: + c.Fewest++ + c.FewestByHops[p.Hops]++ + default: + c.Unknown++ + } + if len(p.Transient) > 0 { + c.Transient++ + } +} + +// regionDryRunRegion is one region under the configured settings. type regionDryRunRegion struct { - Region string - Heard, Placed, Direct, Fewest, Removed, Transient int + Region string + // Heard is what today's filter returns for the region. + Heard int + // Placed = Direct + Fewest nodes placed in the region. + Placed, Direct, Fewest int + // Removed = Heard - Placed: heard here, placed elsewhere or nowhere. + Removed int + // Transient counts nodes for which this region is transient. + Transient int } +// regionDryRunSweepRow is the basis split at one maxHops value. type regionDryRunSweepRow struct { MaxHops, Direct, Fewest, Unknown int } +// regionPlacementDryRun is the whole report. type regionPlacementDryRun struct { Settings regionPlacementSettings Watermark int64 @@ -33,16 +74,151 @@ type regionPlacementDryRun struct { Bases regionBasisCounts Regions []regionDryRunRegion Sweep []regionDryRunSweepRow + Elapsed time.Duration } +// buildRegionPlacementDryRun scans the full history once and evaluates the +// rule under s and under each maxHops in sweep. func buildRegionPlacementDryRun(ctx context.Context, db *DB, s regionPlacementSettings, sweep []int) (*regionPlacementDryRun, error) { - return ®ionPlacementDryRun{}, nil + start := time.Now() + wm, err := db.maxObservationID(ctx) + if err != nil { + return nil, err + } + agg := newRegionEvidenceAgg() + if _, err := db.scanRegionEvidence(ctx, 0, wm, false, agg); err != nil { + return nil, err + } + r := ®ionPlacementDryRun{ + Settings: s, Watermark: wm, ObsRawHex: db.hasObsRawHex(), + Nodes: len(agg.nodes), Pairs: agg.pairs, Hearings: agg.hearings, + } + rows := map[string]*regionDryRunRegion{} + row := func(region string) *regionDryRunRegion { + if rows[region] == nil { + rows[region] = ®ionDryRunRegion{Region: region} + } + return rows[region] + } + for _, per := range agg.nodes { + for region := range per { + row(region).Heard++ + } + p := placeNode(per, s.Params) + r.Bases.add(p) + for _, region := range p.Regions { + x := row(region) + x.Placed++ + if p.Basis == placementBasisDirect { + x.Direct++ + } else { + x.Fewest++ + } + } + for _, region := range p.Transient { + row(region).Transient++ + } + } + for _, x := range rows { + x.Removed = x.Heard - x.Placed + r.Regions = append(r.Regions, *x) + } + sort.Slice(r.Regions, func(i, j int) bool { return r.Regions[i].Region < r.Regions[j].Region }) + for _, h := range sweep { + p := s.Params + p.MaxHops = h + var c regionBasisCounts + for _, per := range agg.nodes { + c.add(placeNode(per, p)) + } + r.Sweep = append(r.Sweep, regionDryRunSweepRow{MaxHops: h, Direct: c.Direct, Fewest: c.Fewest, Unknown: c.Unknown}) + } + r.Elapsed = time.Since(start) + return r, nil } -func runRegionPlacementDryRun(dbPath string, s regionPlacementSettings, w io.Writer) error { - return nil +func dryRunShare(n, of int) string { + if of == 0 { + return "–" + } + return fmt.Sprintf("%.1f%%", 100*float64(n)/float64(of)) } +// write prints the report as Markdown, ready to paste into an issue. func (r *regionPlacementDryRun) write(w io.Writer) error { - return nil + b := &strings.Builder{} + p := r.Settings.Params + lifetime := fmt.Sprintf("lifetimeDays %d", int(p.Lifetime/(24*time.Hour))) + if p.Lifetime == 0 { + lifetime += " (no window)" + } + span := fmt.Sprintf("persistSpanHours %d", int(p.PersistSpan/time.Hour)) + if p.PersistSpan == 0 { + span += " (transient rule off)" + } + raw := "no (every hearing is classified by path_json; a legacy '[]' is fallback only)" + if r.ObsRawHex { + raw = "yes" + } + fmt.Fprintf(b, "# Region placement dry-run (#488)\n\n") + fmt.Fprintf(b, "Read-only, counts only: no node is named. Placement is inferred from reception, not a verified location.\n\n") + fmt.Fprintf(b, "- Settings: mode %s (the filter in use), maxHops %d, %s, %s\n", r.Settings.Mode, p.MaxHops, lifetime, span) + fmt.Fprintf(b, "- Observations scanned: ids up to %d; per-observation frames (observations.raw_hex): %s\n", r.Watermark, raw) + fmt.Fprintf(b, "- Nodes with advert evidence in a region: %d; (node, region) pairs: %d; built in %v\n\n", + r.Nodes, r.Pairs, r.Elapsed.Round(time.Millisecond)) + + fmt.Fprintf(b, "## Advert observations by evidence\n\n| Evidence | Observations |\n|---|---:|\n") + for _, e := range []struct { + label string + kind regionHearingKind + }{ + {"direct: own transmission, hash count 0 on the observation's frame", hearingDirect}, + {"legacy '[]' without a frame (fallback only)", hearingLegacy}, + {fmt.Sprintf("relayed 1-%d hops", regionPlacementHopSlots-1), hearingRelayed}, + {fmt.Sprintf("relayed %d+ hops (never places)", regionPlacementHopSlots), hearingFar}, + {"contact re-share (no evidence)", hearingShared}, + {"unknown path or undated (no evidence)", hearingUnknown}, + } { + fmt.Fprintf(b, "| %s | %d |\n", e.label, r.Hearings[e.kind]) + } + + fmt.Fprintf(b, "\n## Nodes by basis (maxHops %d)\n\n| Basis | Nodes | Share |\n|---|---:|---:|\n", p.MaxHops) + fmt.Fprintf(b, "| direct | %d | %s |\n", r.Bases.Direct, dryRunShare(r.Bases.Direct, r.Nodes)) + for h, n := range r.Bases.FewestByHops { + if n > 0 { + fmt.Fprintf(b, "| fewest_hops, hops = %d | %d | %s |\n", h, n, dryRunShare(n, r.Nodes)) + } + } + fmt.Fprintf(b, "| unknown (not placed) | %d | %s |\n", r.Bases.Unknown, dryRunShare(r.Bases.Unknown, r.Nodes)) + if p.PersistSpan > 0 { + fmt.Fprintf(b, "| with a transient region | %d | %s |\n", r.Bases.Transient, dryRunShare(r.Bases.Transient, r.Nodes)) + } + + fmt.Fprintf(b, "\n## Per region (maxHops %d)\n\n", p.MaxHops) + fmt.Fprintf(b, "| Region | Heard (today's filter) | Placed | Direct | Fewest hops | Heard, not placed here | Transient |\n|---|---:|---:|---:|---:|---:|---:|\n") + for _, x := range r.Regions { + fmt.Fprintf(b, "| %s | %d | %d | %d | %d | %d | %d |\n", x.Region, x.Heard, x.Placed, x.Direct, x.Fewest, x.Removed, x.Transient) + } + + fmt.Fprintf(b, "\n## Unknown share by maxHops\n\n| maxHops | Direct | Fewest hops | Unknown | Unknown share |\n|---:|---:|---:|---:|---:|\n") + for _, x := range r.Sweep { + fmt.Fprintf(b, "| %d | %d | %d | %d | %s |\n", x.MaxHops, x.Direct, x.Fewest, x.Unknown, dryRunShare(x.Unknown, r.Nodes)) + } + _, err := io.WriteString(w, b.String()) + return err +} + +// runRegionPlacementDryRun opens dbPath read-only (OpenDB, mode=ro), writes +// the report to w and closes the database. +func runRegionPlacementDryRun(dbPath string, s regionPlacementSettings, w io.Writer) error { + db, err := OpenDB(dbPath) + if err != nil { + return fmt.Errorf("open %s: %w", dbPath, err) + } + defer db.Close() + r, err := buildRegionPlacementDryRun(context.Background(), db, s, regionDryRunSweep) + if err != nil { + return err + } + return r.write(w) } diff --git a/cmd/server/region_placement_dryrun_488_test.go b/cmd/server/region_placement_dryrun_488_test.go index 355f1e13e..99dcc5c26 100644 --- a/cmd/server/region_placement_dryrun_488_test.go +++ b/cmd/server/region_placement_dryrun_488_test.go @@ -5,12 +5,15 @@ import ( "context" "crypto/sha256" "encoding/json" + "fmt" + "net/http" "os" "os/exec" "path/filepath" "reflect" "regexp" "strings" + "syscall" "testing" "time" ) @@ -247,11 +250,70 @@ func TestRegionPlacement488_ConfigExample(t *testing.T) { t.Errorf("config.example.json regionPlacement lacks %q", k) } } - var cfg Config - if err := json.Unmarshal(raw, &cfg); err != nil { + var block struct { + RegionPlacement *RegionPlacementConfig `json:"regionPlacement"` + } + if err := json.Unmarshal(raw, &block); err != nil { t.Fatal(err) } + cfg := Config{RegionPlacement: block.RegionPlacement} if got := cfg.RegionPlacementSettings(); got != (&Config{}).RegionPlacementSettings() { t.Errorf("config.example.json settings %+v differ from the defaults", got) } } + +// TestRegionPlacement488_ServerUsesConfiguredMode: the real server main() +// reads regionPlacement from config.json and serves /api/nodes?region= by +// it. Expected totals come from GetNodes in-process on the same database. +func TestRegionPlacement488_ServerUsesConfiguredMode(t *testing.T) { + path := fullSchemaDB(t) + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + _, heard, _, err := db.GetNodes(1, 0, "", "", "", "", "", "OAK") + if err != nil { + t.Fatal(err) + } + db.nodeRegionCacheMu.Lock() + db.nodeRegionCache, db.nodeRegionUsed = nil, nil + db.nodeRegionCacheMu.Unlock() + rpm2SetPlaced(db, defaultRegionPlacementParams) + _, placed, _, err := db.GetNodes(1, 0, "", "", "", "", "", "OAK") + if err != nil { + t.Fatal(err) + } + db.Close() + if placed >= heard { + t.Fatalf("fixture does not tell the modes apart: OAK heard %d, placed %d", heard, placed) + } + + cfgDir := t.TempDir() + if err := os.WriteFile(filepath.Join(cfgDir, "config.json"), []byte(`{"regionPlacement": {"mode": "placed"}}`), 0o600); err != nil { + t.Fatal(err) + } + p := startServerProc(t, path, time.Minute, "CORESCOPE_SCHEMA_WAIT_CFG="+cfgDir) + p.waitReady(t) + c := http.Client{Timeout: 10 * time.Second} + resp, err := c.Get(fmt.Sprintf("http://127.0.0.1:%d/api/nodes?region=OAK&limit=1", p.port)) + if err != nil { + t.Fatal(err) + } + var body struct { + Total int `json:"total"` + } + err = json.NewDecoder(resp.Body).Decode(&body) + resp.Body.Close() + if err != nil { + t.Fatal(err) + } + t.Logf("OAK: heard %d, placed %d, server %d", heard, placed, body.Total) + if body.Total != placed { + t.Errorf("server with mode placed: OAK total %d, want %d (heard would be %d)", body.Total, placed, heard) + } + if !strings.Contains(p.out.String(), "regionPlacement: mode=placed") { + t.Errorf("server did not log the placed mode:\n%s", p.out.String()) + } + p.cmd.Process.Signal(syscall.SIGTERM) + p.waitExit(t, 30*time.Second) +} diff --git a/cmd/server/region_placement_perf_488_test.go b/cmd/server/region_placement_perf_488_test.go index 45af8dd1b..607d9f34b 100644 --- a/cmd/server/region_placement_perf_488_test.go +++ b/cmd/server/region_placement_perf_488_test.go @@ -215,6 +215,14 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { rows.Close() wm := rpm2MaxID(t, db) lo := max(wm-deltaRows, 0) + // Second delta window: deltaRows ending at the newest ADVERT + // observation, so every scanned row is classified (the worst case). + var advHi int64 + if err := db.conn.QueryRow(`SELECT COALESCE(MAX(o.id), 0) FROM observations o + JOIN transmissions t ON t.id = o.transmission_id WHERE t.payload_type = ?`, payloadTypeAdvert).Scan(&advHi); err != nil { + t.Fatal(err) + } + advLo := max(advHi-deltaRows, 0) clearEntries := func() { db.nodeRegionCacheMu.Lock() @@ -251,6 +259,13 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { } } }) + heardAdvScan := rpm2Median(func() { + for _, r := range regions { + if _, err := db.scanNodeRegionKeys(ctx, []string{r}, advLo, advHi, true, map[string]struct{}{}); err != nil { + t.Fatal(err) + } + } + }) clearEntries() h0 := rpm2Heap() serveAll() @@ -267,19 +282,23 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { placeOnly := rpm2Median(func() { newRegionPlacementSnapshot(agg, defaultRegionPlacementParams, time.Now()) }) // The delta re-folds the same tail rows each run: the stats merge by // min/max, so the placement work is the same as for new rows. - placedDelta := rpm2Median(func() { - d := newRegionEvidenceAgg() - if _, err := db.scanRegionEvidence(ctx, lo, wm, true, d); err != nil { - t.Fatal(err) - } - agg.merge(d) - snap := newRegionPlacementSnapshot(agg, defaultRegionPlacementParams, time.Now()) - for _, r := range regions { - if _, err := newNodeRegionEntry(snap.members([]string{r}), wm, time.Now()); err != nil { + placedDeltaOver := func(lo, hi int64) time.Duration { + return rpm2Median(func() { + d := newRegionEvidenceAgg() + if _, err := db.scanRegionEvidence(ctx, lo, hi, true, d); err != nil { t.Fatal(err) } - } - }) + agg.merge(d) + snap := newRegionPlacementSnapshot(agg, defaultRegionPlacementParams, time.Now()) + for _, r := range regions { + if _, err := newNodeRegionEntry(snap.members([]string{r}), hi, time.Now()); err != nil { + t.Fatal(err) + } + } + }) + } + placedDelta := placedDeltaOver(lo, wm) + placedAdvDelta := placedDeltaOver(advLo, advHi) pairs, nodes := agg.pairs, len(agg.nodes) agg = nil clearPlaced() @@ -288,9 +307,10 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { placedHeap := int64(rpm2Heap()) - int64(h1) t.Logf("%s: %d observations, %d regions, %d nodes, %d (node, region) pairs", filepath.Base(path), wm, len(regions), nodes, pairs) - t.Logf("heard cold %v | warm %v | delta refresh (%d obs) %v | heap %d KiB", - heardCold.Round(time.Microsecond), heardWarm, deltaRows, heardDelta.Round(time.Microsecond), heardHeap/1024) - t.Logf("placed cold %v | warm %v | delta scan+merge+placement+entries (%d obs) %v | placement only %v | heap %d KiB", + t.Logf("heard cold %v | warm %v | delta refresh, newest %d obs %v | delta scan, %d advert obs %v | heap %d KiB", + heardCold.Round(time.Microsecond), heardWarm, deltaRows, heardDelta.Round(time.Microsecond), + deltaRows, heardAdvScan.Round(time.Microsecond), heardHeap/1024) + t.Logf("placed cold %v | warm %v | delta scan+merge+placement+entries, newest %d obs %v, %d advert obs %v | placement only %v | heap %d KiB", placedCold.Round(time.Microsecond), placedWarm, deltaRows, placedDelta.Round(time.Microsecond), - placeOnly.Round(time.Microsecond), placedHeap/1024) + deltaRows, placedAdvDelta.Round(time.Microsecond), placeOnly.Round(time.Microsecond), placedHeap/1024) } diff --git a/cmd/server/store.go b/cmd/server/store.go index c4297b6ee..e412200b0 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -9505,10 +9505,13 @@ func (s *PacketStore) computeAnalyticsHashSizes(region, area string) map[string] allNodes, pm := s.getCachedNodesAndPM() s.mu.RLock() defer s.mu.RUnlock() - // #804: derive each node's HOME region from zero-hop direct adverts (the - // most authoritative location signal — those packets cannot have been - // relayed). When non-empty, multi-byte node attribution prefers this - // over observer-region. Falls back to observer-region when unknown. + // #804: derive each node's HOME region from ADVERT observations heard + // directly from the originator's own transmission, classified by each + // observation's own frame (#488): zero-hop FLOOD, TRANSPORT_FLOOD and + // DIRECT adverts count; TRANSPORT_DIRECT {0,0} contact re-shares never + // do; a legacy '[]' row without a frame is fallback only. When + // non-empty, multi-byte node attribution prefers this over + // observer-region. Falls back to observer-region when unknown. nodeHomeRegion := s.computeNodeHomeRegions(obsIATA) attributionMethod := "observer" if region != "" && len(nodeHomeRegion) > 0 { diff --git a/config.example.json b/config.example.json index de88f8ee9..dd1b6d794 100644 --- a/config.example.json +++ b/config.example.json @@ -330,6 +330,13 @@ { "name": "Bay Area", "codes": ["SFO", "SJC", "OAK"] }, { "name": "Central Coast", "codes": ["MRY"] } ], + "regionPlacement": { + "_comment": "Optional (server, #488). Which nodes /api/nodes?region= returns. mode: \"heard\" (default, today's rule) puts a node in every region whose observers heard any of its adverts, so flooded adverts list distant nodes too. \"placed\" uses direct evidence first: a node is in every region that heard its own transmission (hash count 0 on the observation's own frame; contact re-shares never count), else in the regions that heard it in the fewest relays up to maxHops, else in no region (it stays in the unfiltered list). A legacy row without a stored frame counts an explicit [] path only as a 0-hop fallback. This is an inference from reception, not a verified location. maxHops: cap of the relayed fallback, 0-7, default 3. lifetimeDays: evidence older than this before the node's own newest hearing is ignored, default 30, 0 = no window. persistSpanHours: when set, a region whose direct hearings span less than this is dropped if another region's span at least this (one-off long-distance reception); default 0 = off. Read at startup. Before switching to placed, run the read-only dry-run on a copy of the database: corescope-server -db -config-dir -region-placement-dryrun prints counts per basis and region and the unplaced share for several maxHops values, and exits. These values are meant for the customizer in a later release.", + "mode": "heard", + "maxHops": 3, + "lifetimeDays": 30, + "persistSpanHours": 0 + }, "cacheTTL": { "stats": 10, "nodeDetail": 300,