Regions: path and step budgets for real data

cmd/server/openapi.go

@@ -63,8 +63,8 @@ func routeDescriptions() map[string]routeMeta {
 		// Admin / system
 		"GET /api/health": {Summary: "Health check", Description: "Returns server health, uptime, and memory stats.", Tag: "admin"},
 		"GET /api/stats":  {Summary: "Network statistics", Description: "Returns aggregate stats (node counts, packet counts, observer counts). Cached for 10s.", Tag: "admin"},
-		"GET /api/perf": {Summary: "Performance statistics", Description: "Returns per-endpoint request timing and slow query log. mem=1 adds the store memory breakdown, a full store walk: one scan at a time per server plus a 5 s cooldown after it, otherwise 429 with Retry-After.", Tag: "admin",
-			QueryParams: []paramMeta{{Name: "mem", Description: "1 = include memoryBreakdown (rate-limited, see description)", Type: "string"}}},
+		"GET /api/perf": {Summary: "Performance statistics", Description: "Returns per-endpoint request timing and slow query log. mem=1 adds the store memory breakdown, a full store walk: one scan at a time per server plus a 5 s cooldown after it, otherwise 429 with Retry-After. mem=1 also adds regionsSnapshot, the retained size of the published Regions route/membership/history/evidence snapshot (#508); it is O(published cells), takes no packet lock and is absent before the first background relay snapshot. regionsSnapshot.capture adds the transient capture state's fill against its own budgets (retained paths and interned paths, retained references with the per-hour high-water mark, retained transmissions, aggregated membership cells and identities, each against its budget, the per-transmission path and path-byte high-water marks against theirs, and per-reason capped counters -- cappedPathBytes, cappedTransmissionPaths, cappedHourReferences and cappedOverlongPaths account for cappedObservations, while cappedMembershipCells and cappedMembershipNodes are dropped memberships), since those decide whether evidence is reported as capped; regionsSnapshot.truncated is budget truncation only and regionsSnapshot.evidenceIncomplete carries evidence qualification separately (#510).", Tag: "admin",
+			QueryParams: []paramMeta{{Name: "mem", Description: "1 = include memoryBreakdown and regionsSnapshot (rate-limited, see description)", Type: "string"}}},
 		"GET /api/mqtt/status": {Summary: "MQTT source status", Description: "Returns per-MQTT-source connection state and counters (lastConnectUnix, lastPacketUnix, packetsTotal, etc.). Broker URL credentials (user-info, query, fragment) are masked in broker and lastError, and in a name that is a raw broker URL. Sourced from the ingestor stats file; empty list when unavailable. stale is true when the file's sampleAt is older than the /api/perf/io freshness threshold (5s) or unreadable, i.e. the rows are frozen; sampleAgeSec gives its age. (#1043, #160)", Tag: "admin"},
 		"GET /api/perf/ws-clients": {Summary: "WebSocket client history", Description: "Returns open WebSocket connection counts, oldest first. range accepts 24h (default), 7d or 30d; other values return 400. Samples are one minute for 24h, five minutes for 7d and fifteen minutes for 30d, with the last clients count and maximum peak in each bucket. gapBefore marks interrupted sampling; missing periods are not zero-filled. persistence.enabled and persistence.healthy indicate whether the separate history journal is saving successfully. At most 30 days are retained; history survives server/container replacement when its data directory is retained and writable. No packet database writes.", Tag: "admin",
 			QueryParams: []paramMeta{{Name: "range", Description: "24h (default), 7d or 30d", Type: "string"}}},

cmd/server/regions_evidence.go

@@ -6,8 +6,14 @@ import (
 )
 
 const (
-	maxRegionsMembershipEntries = 1000000
-	maxRegionsMembershipNodes   = 10000
+	// Aggregated membership cells, one per (relay, scope, hour) (#508). The
+	// retired budget retained one entry per (relay, transmission), so it grew
+	// with observations: 48 h of real traffic already reached a third of it,
+	// and 168 h would exhaust it. Cells grow with populated hours instead —
+	// about 134,000 for 400 relays x 2 scopes x 168 hours — and repeated
+	// transmissions in one hour reuse their cell.
+	maxRegionsMembershipCells = 500000
+	maxRegionsMembershipNodes = 10000
 )
 
 // regionsObservationClass is the single provenance gate for Regions
@@ -221,13 +227,42 @@ type regionsMembershipSnapshot struct {
 	evidence *regionsEvidenceSnapshot
 }
 
+// regionsMembershipCellKey identifies one relay's membership evidence for one
+// scope in one hourly bucket, 0 through regionsHistoryBucketCount; the last is
+// the current unfinished hour, which only Latest summarises.
+type regionsMembershipCellKey struct {
+	key, scope string
+	hour       int32
+}
+
+type regionsMembershipCell struct {
+	count int
+	last  time.Time
+}
+
+// regionsMembershipNodeCell is one relay's aggregated cell, handed to the
+// hourly history builder in deterministic (scope, hour) order.
+type regionsMembershipNodeCell struct {
+	scope string
+	hour  int32
+	count int
+	last  time.Time
+}
+
 type regionsMembershipBuilder struct {
-	entries   map[string][]relayEntry
+	cells     map[regionsMembershipCellKey]regionsMembershipCell
+	nodes     map[string]struct{}
 	remaining int
+	nodeBound int
+	// Memberships each ceiling dropped, for the capture diagnostic (#510):
+	// without them a withheld rate cannot be told apart from a capture
+	// refusal, because both reach the bucket as `capped`.
+	cellDrops, nodeDrops int
 }
 
-func newRegionsMembershipBuilder() *regionsMembershipBuilder {
-	return &regionsMembershipBuilder{entries: make(map[string][]relayEntry), remaining: maxRegionsMembershipEntries}
+func newRegionsMembershipBuilder(cells, nodes int) *regionsMembershipBuilder {
+	return &regionsMembershipBuilder{cells: make(map[regionsMembershipCellKey]regionsMembershipCell),
+		nodes: make(map[string]struct{}), remaining: cells, nodeBound: nodes}
 }
 
 // retainLatestRegionScope keeps the newest 32 names with bounded scratch
@@ -258,42 +293,71 @@ func retainLatestRegionScope(latest map[string]time.Time, name string, when time
 }
 
 // Called once per resolved node per distinct transmission, outside the packet
-// lock. This independent budget is never spent by directed edges or edge cells.
-// It reports false when a safety limit omitted the membership.
-func (b *regionsMembershipBuilder) add(key string, tx regionsRouteTx) bool {
-	if b.remaining == 0 {
-		return false
+// lock, and aggregated straight into the node's (scope, hour) cell: nothing
+// per-transmission is retained, so this budget no longer grows with the number
+// of observations. It is never spent by directed edges or edge cells, and
+// reports false when a safety limit omitted the membership.
+//
+// Only hour/scope/time are consumed; payload and provenance exclusion already
+// happened before capture, and no canonical/direct relay fields are reused.
+func (b *regionsMembershipBuilder) add(key, scope string, hour int, when time.Time) bool {
+	id := regionsMembershipCellKey{key: key, scope: scope, hour: int32(hour)}
+	cell, exists := b.cells[id]
+	if !exists {
+		if b.remaining <= 0 {
+			b.cellDrops++
+			return false
+		}
+		if _, known := b.nodes[key]; !known {
+			if len(b.nodes) >= b.nodeBound {
+				b.nodeDrops++
+				return false
+			}
+			b.nodes[key] = struct{}{}
+		}
+		b.remaining--
 	}
-	if _, exists := b.entries[key]; !exists && len(b.entries) >= maxRegionsMembershipNodes {
-		return false
+	cell.count++
+	if when.After(cell.last) {
+		cell.last = when
 	}
-	// Only time/scope are consumed below; payload exclusion was already made
-	// before capture. No canonical/direct relay-entry fields are reused.
-	b.entries[key] = append(b.entries[key], relayEntry{t: tx.when, parsed: true, valid: true, scope: tx.scope})
-	b.remaining--
+	b.cells[id] = cell
 	return true
 }
 
 func (b *regionsMembershipBuilder) build(now time.Time, evidence *regionsEvidenceSnapshot) (*regionsMembershipSnapshot, *regionsRelayHistory) {
-	out := &regionsMembershipSnapshot{nodes: make(map[string]regionsNodeMembership, len(b.entries)), evidence: evidence}
+	byNode := make(map[string][]regionsMembershipNodeCell, len(b.nodes))
+	for id, cell := range b.cells {
+		byNode[id.key] = append(byNode[id.key], regionsMembershipNodeCell{scope: id.scope, hour: id.hour, count: cell.count, last: cell.last})
+	}
+	out := &regionsMembershipSnapshot{nodes: make(map[string]regionsNodeMembership, len(byNode)), evidence: evidence}
 	history := newRegionsHistoryBuilder(now)
-	keys := make([]string, 0, len(b.entries))
-	for key := range b.entries {
+	keys := make([]string, 0, len(byNode))
+	for key := range byNode {
 		keys = append(keys, key)
 	}
 	sort.Strings(keys)
 	for _, key := range keys {
-		entries := b.entries[key]
+		cells := byNode[key]
+		// Keys and cells are both visited in a deterministic order, so neither
+		// the 32-scope selection nor the shared publication budget depends on
+		// map iteration.
+		sort.Slice(cells, func(i, j int) bool {
+			if cells[i].scope != cells[j].scope {
+				return cells[i].scope < cells[j].scope
+			}
+			return cells[i].hour < cells[j].hour
+		})
 		// At most 32 newest scopes per key, with bounded scratch space even
 		// if an adversary creates a new scope for every retained transmission.
 		latest := make(map[string]time.Time, maxTransportedScopes)
 		truncated := false
-		for _, e := range entries {
-			truncated = retainLatestRegionScope(latest, e.scope, e.t) || truncated
+		for _, cell := range cells {
+			truncated = retainLatestRegionScope(latest, cell.scope, cell.last) || truncated
 		}
 		scopes, _ := newestRegionScopeRelays(latest)
 		out.nodes[key] = regionsNodeMembership{scopes: scopes, scopesTruncated: truncated}
-		history.add(key, entries)
+		history.add(key, cells)
 	}
 	return out, history.history
 }

cmd/server/regions_evidence_test.go

@@ -147,23 +147,29 @@ func TestRegionsEvidenceMembershipBudgetIndependentOfSteps(t *testing.T) {
 	now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC)
 	collector := newRegionsRoutesCollector(now)
 	a := strings.Repeat("ab", 32)
-	for i := 0; i < maxRegionsRouteSteps+1; i++ {
+	// One row plus its first bucket fits; the second row does not.
+	collector.stepBytes = regionsRouteStepRowBytes + regionsRouteStepCellBytes
+	for i := 0; i < 2; i++ {
 		key := fmt.Sprintf("%064x", i+1)
 		collector.capture(routeTransmission(i+1, now.Add(-time.Hour), `["`+a+`","`+key+`"]`))
 	}
 	got := collector.build(nil)
-	last := fmt.Sprintf("%064x", maxRegionsRouteSteps+1)
-	if len(got.steps) != maxRegionsRouteSteps || !got.stepsTruncated || len(got.memberships.nodes[last].scopes) != 1 ||
+	last := fmt.Sprintf("%064x", 2)
+	if len(got.steps) != 1 || !got.stepsTruncated || len(got.memberships.nodes[last].scopes) != 1 ||
 		got.memberships.evidence.history.Capped != 0 || got.memberships.evidence.truncated || got.evidence.CappedStepAppearances != 1 {
-		t.Fatal("directed-step cap suppressed independent valid membership or polluted its evidence")
+		t.Fatal("directed-step payload budget suppressed independent valid membership or polluted its evidence")
 	}
-	b := newRegionsMembershipBuilder()
+	b := newRegionsMembershipBuilder(maxRegionsMembershipCells, maxRegionsMembershipNodes)
 	b.remaining = 1
-	if !b.add(a, regionsRouteTx{when: now, scope: "#one"}) || b.add(a, regionsRouteTx{when: now, scope: "#two"}) {
-		t.Fatal("independent membership entry cap was silent")
+	if !b.add(a, "#one", 0, now) || b.add(a, "#two", 0, now) {
+		t.Fatal("independent membership cell cap was silent")
 	}
-	if bounded, _ := b.build(now, nil); len(b.entries[a]) != 1 || len(bounded.nodes[a].scopes) != 1 {
-		t.Fatal("independent membership entry cap was not applied")
+	// Aggregation: a further transmission in an admitted cell costs no budget.
+	if !b.add(a, "#one", 0, now.Add(time.Minute)) {
+		t.Fatal("an aggregated cell spent the membership budget twice")
+	}
+	if bounded, _ := b.build(now, nil); len(b.cells) != 1 || len(bounded.nodes[a].scopes) != 1 {
+		t.Fatal("independent membership cell cap was not applied")
 	}
 	// The node cap is counted as capped evidence in the omitted membership's
 	// own hour and scope only.
@@ -310,20 +316,26 @@ func TestRegionsEvidenceExcludedOnlyIsComplete(t *testing.T) {
 // appearances in stepEvidence, separately from observation-level evidence, and
 // never makes membership/history coverage incomplete.
 func TestRegionsEvidenceStepCapCountedSeparately(t *testing.T) {
-	srv, now, _ := regionsObservedFixture(t)
+	now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC)
 	a := strings.Repeat("ab", 32)
-	when := now.Truncate(time.Hour).Add(-time.Hour)
-	for i := 0; i < maxRegionsRouteSteps+1; i++ {
-		srv.store.byPathHop[a] = append(srv.store.byPathHop[a], routeTransmission(i+1, when, fmt.Sprintf(`["%s","%064x"]`, a, i+1)))
+	collector := newRegionsRoutesCollector(now)
+	// Budget for one row and its first bucket only, so a second directed step
+	// is omitted by the published payload budget rather than by evidence.
+	collector.stepBytes = regionsRouteStepRowBytes + regionsRouteStepCellBytes
+	for i := 0; i < 2; i++ {
+		collector.capture(routeTransmission(i+1, now.Add(-time.Hour), fmt.Sprintf(`["%s","%064x"]`, a, i+1)))
 	}
-	recomputeRepeaterEnrichmentSafe(srv.store, 24)
-	body := regionsEvidenceResponse(t, srv, "/api/regions/routes")
-	if got := *body.Coverage.Evidence; got.Excluded != 0 || got.Unprovable != 0 || got.Capped != 0 || got.Truncated || len(got.Scopes) != 0 {
-		t.Fatalf("directed-step cap leaked into membership coverage evidence: %+v", got)
+	got := collector.build(nil)
+	if e := got.memberships.evidence.history; e.Excluded != 0 || e.Unprovable != 0 || e.Capped != 0 || e.Truncated || len(e.Scopes) != 0 {
+		t.Fatalf("directed-step payload budget leaked into membership coverage evidence: %+v", e)
+	}
+	steps := got.evidence
+	if !got.stepsTruncated || steps.CappedStepAppearances != 1 || steps.CappedObservations != 0 || steps.UnprovableObservations != 0 || !steps.Incomplete {
+		t.Fatalf("the step payload budget must be counted as capped step appearances only: truncated=%v %+v", got.stepsTruncated, steps)
 	}
-	steps := body.StepEvidence
-	if !body.StepsTruncated || steps == nil || steps.CappedStepAppearances != 1 || steps.CappedObservations != 0 || steps.UnprovableObservations != 0 || !steps.Incomplete {
-		t.Fatalf("step cap must be counted as capped step appearances only: truncated=%v %+v", body.StepsTruncated, steps)
+	// Both relays keep their membership; only the publication was truncated.
+	if len(got.memberships.nodes) != 3 || len(got.history.nodes) != 3 {
+		t.Fatalf("payload truncation dropped memberships: %d/%d", len(got.memberships.nodes), len(got.history.nodes))
 	}
 }
 

cmd/server/regions_history.go

@@ -79,34 +79,33 @@ func newRegionsHistoryBuilder(now time.Time) *regionsHistoryBuilder {
 	}, remainingCells: maxRegionsHistoryCells}
 }
 
-func historyRelayTime(e relayEntry, start, end time.Time) (time.Time, bool) {
-	if e.fromPrefix || e.pt == payloadTypeAdvert || regions.ValidateCandidate(e.scope) != nil {
-		return time.Time{}, false
-	}
-	t, valid := e.t, e.valid
-	if !e.parsed {
-		t, valid = parseRelayTS(e.ts)
-	}
-	return t, valid && !t.Before(start) && t.Before(end)
+// historyCellEligible keeps completed hours of valid scope candidates only.
+// Adverts, TRACE, planned direct paths, prefix-only evidence and out-of-range
+// timestamps were already excluded before capture, so the current unfinished
+// hour (bucket regionsHistoryBucketCount, which only Latest summarises) and a
+// malformed scope name are all that can still reach here.
+func historyCellEligible(cell regionsMembershipNodeCell) bool {
+	return cell.hour >= 0 && int(cell.hour) < regionsHistoryBucketCount && regions.ValidateCandidate(cell.scope) == nil
 }
 
-// add receives deduplicated own-observation memberships from the strict
-// Regions collector, after the packet lock has been released. Callers
-// visit keys in sorted order so the shared cell budget is deterministic.
-func (b *regionsHistoryBuilder) add(key string, entries []relayEntry) {
+// add receives own-observation memberships from the strict Regions collector,
+// already aggregated per (scope, hour) and with each transmission counted once
+// per relay, after the packet lock has been released. Callers visit keys in
+// sorted order, and cells in (scope, hour) order, so the shared cell budget is
+// deterministic.
+func (b *regionsHistoryBuilder) add(key string, cells []regionsMembershipNodeCell) {
 	if len(key) != 64 {
 		return
 	}
 	h := b.history
 	// Select the 32 newest valid scopes with bounded scratch space, even if a
 	// sender introduces arbitrarily many names in the retained packet store.
 	latest := make(map[string]time.Time, maxTransportedScopes)
-	for _, e := range entries {
-		t, valid := historyRelayTime(e, h.start, h.end)
-		if !valid {
+	for _, cell := range cells {
+		if !historyCellEligible(cell) {
 			continue
 		}
-		h.scopesTruncated = retainLatestRegionScope(latest, e.scope, t) || h.scopesTruncated
+		h.scopesTruncated = retainLatestRegionScope(latest, cell.scope, cell.last) || h.scopesTruncated
 	}
 	if len(latest) == 0 {
 		return
@@ -126,38 +125,35 @@ func (b *regionsHistoryBuilder) add(key string, entries []relayEntry) {
 		count  int
 		latest time.Time
 	}
-	cells := make(map[string]map[int]historyCell, len(latest))
-	for _, e := range entries {
-		t, valid := historyRelayTime(e, h.start, h.end)
-		if !valid {
+	byScope := make(map[string]map[int]historyCell, len(latest))
+	for _, cell := range cells {
+		if !historyCellEligible(cell) {
 			continue
 		}
-		if _, selected := latest[e.scope]; !selected {
+		if _, selected := latest[cell.scope]; !selected {
 			continue
 		}
-		index := int(t.Sub(h.start) / time.Hour)
-		byIndex := cells[e.scope]
-		bucket, exists := byIndex[index]
-		if !exists {
-			if byIndex == nil {
-				byIndex = make(map[int]historyCell)
-				cells[e.scope] = byIndex
-			}
+		index := int(cell.hour)
+		byIndex := byScope[cell.scope]
+		if byIndex == nil {
+			byIndex = make(map[int]historyCell)
+			byScope[cell.scope] = byIndex
 		}
-		bucket.count++
-		if t.After(bucket.latest) {
-			bucket.latest = t
+		bucket := byIndex[index]
+		bucket.count += cell.count
+		if cell.last.After(bucket.latest) {
+			bucket.latest = cell.last
 		}
 		byIndex[index] = bucket
 	}
-	scopes := make([]RegionsHistoryScope, 0, len(cells))
-	names := make([]string, 0, len(cells))
-	for name := range cells {
+	scopes := make([]RegionsHistoryScope, 0, len(byScope))
+	names := make([]string, 0, len(byScope))
+	for name := range byScope {
 		names = append(names, name)
 	}
 	sort.Strings(names)
 	for _, name := range names {
-		byIndex := cells[name]
+		byIndex := byScope[name]
 		indexes := make([]int, 0, len(byIndex))
 		for index := range byIndex {
 			indexes = append(indexes, index)
@@ -232,8 +228,15 @@ func (s *Server) regionsHistoryAndStepNodes(ctx context.Context, snapshot *repea
 		}
 	}
 	stepNodes := []RegionsRouteNode{}
+	endpointLimit := maxRegionsRouteEndpoints
+	if s.stepEndpointLimit > 0 {
+		endpointLimit = s.stepEndpointLimit
+	}
 	err := s.forEachVisibleRegionNode(ctx, keys, true, func(n observedRegionNodeMetadata) bool {
-		if routeKeys[n.publicKey] {
+		// Endpoint metadata has its own publication ceiling, so the directed-step
+		// payload budget cannot be multiplied by the node table. Steps whose
+		// endpoints are omitted become missingEndpointSteps in the handler.
+		if routeKeys[n.publicKey] && len(stepNodes) < endpointLimit {
 			stepNodes = append(stepNodes, regionRouteNodeMetadata(n))
 		}
 		scopes := h.nodes[n.publicKey]