feat(regions): direct-evidence placement aggregate + operator dry-run (#488 M2)

cmd/server/advert_evidence.go

@@ -2,22 +2,33 @@ 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: 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
+}
+
+// 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
+	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 != 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), Hops: int(o.advertEv.hops), Verified: true}
 	}
 	return packetpath.ClassifyAdvertPathJSON(o.PathJSON)
 }

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)
+	}
+}

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,58 @@ 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 {
+	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
+}

cmd/server/db.go

@@ -139,6 +139,22 @@ 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]
+	// 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
 	// lower-case keys it looks up.

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(&regionDryRun, "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

cmd/server/node_home_regions_488_test.go

@@ -184,20 +184,21 @@ 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
 			}
 		}
 		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)
 	}

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 {