diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index 38813dbf9..e763a1ec0 100644 --- a/cmd/server/openapi.go +++ b/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"}}}, diff --git a/cmd/server/regions_budgets_508_test.go b/cmd/server/regions_budgets_508_test.go new file mode 100644 index 000000000..b883c7d9d --- /dev/null +++ b/cmd/server/regions_budgets_508_test.go @@ -0,0 +1,760 @@ +package main + +import ( + "encoding/json" + "fmt" + "math/rand" + "net/http/httptest" + "strings" + "testing" + "time" + "unsafe" +) + +// #508: every Regions capture and publication budget was a single global +// count, and real data (48 h retention, ~805k observations) exhausted two of +// them: the 250,000 path-variant capture budget, which reported `capped` +// membership evidence for 17 of 20 hours in #dk and withheld every rate, and +// the 5,000-row directed-step cap. These tests describe the redesigned +// budgets through the collector's public behaviour only, so they run on the +// pre-#508 tree as well. +// +// Shared fixture note: routeTransmission uses scope "#sample" and a canonical +// transport-FLOOD frame, so every observation below is eligible evidence and +// anything reported as excluded/unprovable/capped is a budget effect. + +func stepCells(snapshot *regionsRouteSnapshot) (cells, count int) { + for _, step := range snapshot.steps { + cells += len(step.Buckets) + for _, bucket := range step.Buckets { + count += bucket.Count + } + } + return cells, count +} + +// capturedEvidenceBucket returns the capped count attributed to one completed +// hour of the collector's own scope. +func capturedEvidenceBucket(t *testing.T, snapshot *regionsRouteSnapshot, index int) RegionsEvidenceCounts { + t.Helper() + if snapshot.memberships == nil || snapshot.memberships.evidence == nil { + t.Fatal("snapshot published no membership evidence") + } + for _, scope := range snapshot.memberships.evidence.history.Scopes { + if scope.Name != "#sample" { + continue + } + for _, bucket := range scope.Buckets { + if bucket.Index == index { + return bucket.RegionsEvidenceCounts + } + } + } + return RegionsEvidenceCounts{} +} + +// Point 1: the capture budget is per hourly bucket, so a 168 h retention is +// not capped. 262,080 retained path references spread evenly over the horizon +// exceed the old 250,000 global budget while staying far below any single +// hour's share of it. +func TestRegionsRoutePathCaptureBudgetIsPerHour(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + a := strings.Repeat("ab", 32) + paths := make([]string, 100) + for i := range paths { + paths[i] = fmt.Sprintf(`["%s","%064x"]`, a, i+1) + } + const perHour = 520 + id := 0 + for hour := 0; hour < regionsHistoryBucketCount; hour++ { + at := collector.start.Add(time.Duration(hour)*time.Hour + time.Minute) + for i := 0; i < perHour; i++ { + id++ + collector.capture(routeTransmission(id, at, paths[i%len(paths)])) + } + } + got := collector.build(nil) + if got.evidence.SourceTruncated || got.evidence.CappedObservations != 0 { + t.Fatalf("168 h of evenly spread capture was truncated: truncated=%v capped=%d", + got.evidence.SourceTruncated, got.evidence.CappedObservations) + } + if e := got.memberships.evidence.history; e.Capped != 0 || e.Unprovable != 0 || e.Truncated { + t.Fatalf("capture budget withheld membership rates: %+v", e) + } + cells, count := stepCells(got) + if len(got.steps) != len(paths) || cells != len(paths)*regionsHistoryBucketCount || count != perHour*regionsHistoryBucketCount { + t.Fatalf("steps=%d cells=%d appearances=%d", len(got.steps), cells, count) + } + if got.stepsTruncated || got.bucketsTruncated || got.evidence.CappedStepAppearances != 0 { + t.Fatalf("publication budget truncated a 168 h snapshot: %+v", got.evidence) + } +} + +// Point 1: one busy hour cannot starve the rest. 70,000 references in a single +// hour is above that hour's budget and far below the old global 250,000, so on +// the pre-#508 tree nothing was capped at all. +func TestRegionsRoutePathCaptureBudgetFairAcrossHours(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + a, b, c := strings.Repeat("ab", 32), fmt.Sprintf("%064x", 1), fmt.Sprintf("%064x", 2) + const overfull = 70000 + for i := 0; i < overfull; i++ { + collector.capture(routeTransmission(i+1, collector.start.Add(time.Minute), `["`+a+`","`+b+`"]`)) + } + collector.capture(routeTransmission(overfull+1, collector.start.Add(time.Hour+time.Minute), `["`+a+`","`+c+`"]`)) + got := collector.build(nil) + if !got.evidence.SourceTruncated { + t.Fatal("a single hour was allowed to spend more than its capture budget") + } + if hot := capturedEvidenceBucket(t, got, 0); hot.Capped == 0 { + t.Fatal("the overfull hour's omitted references were not reported as capped evidence") + } + if cold := capturedEvidenceBucket(t, got, 1); cold != (RegionsEvidenceCounts{}) { + t.Fatalf("a busy hour qualified another hour's evidence: %+v", cold) + } + found := false + for _, step := range got.steps { + if step.To == c { + found = len(step.Buckets) == 1 && step.Buckets[0].Index == 1 && step.Buckets[0].Count == 1 + } + } + if !found { + t.Fatalf("the quiet hour lost its own evidence: %+v", got.steps) + } +} + +// Point 2: memberships are aggregated per (relay, scope, hour) instead of +// retaining one entry per (relay, transmission), so the 1,000,000-entry +// budget no longer scales with observations. 20,400 transmissions x 64 relays +// is 1,305,600 memberships in 1,536 cells. +func TestRegionsMembershipAggregatesPerRelayScopeHour(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + keys := make([]string, 64) + for i := range keys { + keys[i] = fmt.Sprintf("%064x", i+1) + } + path := `["` + strings.Join(keys, `","`) + `"]` + const hours, perHour = 24, 850 + id := 0 + for hour := 0; hour < hours; hour++ { + at := collector.start.Add(time.Duration(hour)*time.Hour + time.Minute) + for i := 0; i < perHour; i++ { + id++ + collector.capture(routeTransmission(id, at, path)) + } + } + got := collector.build(nil) + if e := got.memberships.evidence.history; e.Capped != 0 || e.Unprovable != 0 || e.Truncated { + t.Fatalf("1.3M memberships exhausted the membership budget: %+v", e) + } + if len(got.memberships.nodes) != len(keys) { + t.Fatalf("retained %d of %d relay identities", len(got.memberships.nodes), len(keys)) + } + history := got.history + if len(history.nodes) != len(keys) || history.truncated || history.bucketsTruncated || history.scopesTruncated { + t.Fatalf("history dropped aggregated memberships: nodes=%d %+v", len(history.nodes), history) + } + for _, key := range keys { + scopes := history.nodes[key] + if len(scopes) != 1 { + t.Fatalf("%s retained %d scopes", key[:8], len(scopes)) + } + if len(scopes[0].Buckets) != hours { + t.Fatalf("%s retained %d hourly buckets, want %d", key[:8], len(scopes[0].Buckets), hours) + } + for _, bucket := range scopes[0].Buckets { + if bucket.Count != perHour { + t.Fatalf("%s bucket %d counted %d distinct transmissions, want %d", key[:8], bucket.Index, bucket.Count, perHour) + } + } + } +} + +// Point 3: directed steps are bounded by estimated published payload bytes +// instead of a fixed 5,000-row / 100,000-cell cap, both of which were reached +// on real data at 48 h retention. +func TestRegionsRouteStepsBudgetedByPayloadBytes(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + a := strings.Repeat("ab", 32) + const rows = 6000 + at := collector.start.Add(500 * time.Hour) + for i := 0; i < rows; i++ { + collector.capture(routeTransmission(i+1, at, fmt.Sprintf(`["%s","%064x"]`, a, i+1))) + } + got := collector.build(nil) + if len(got.steps) != rows || got.stepsTruncated || got.bucketsTruncated || got.evidence.CappedStepAppearances != 0 { + t.Fatalf("%d of %d directed step rows published (truncated=%v capped=%d)", + len(got.steps), rows, got.stepsTruncated, got.evidence.CappedStepAppearances) + } + + // A cell-heavy snapshot still has a payload guarantee, now expressed in + // bytes: 600 rows x 450 populated hours is 270,000 cells, above the byte + // budget and far above the retired 100,000-cell cap. + collector = newRegionsRoutesCollector(now) + paths := make([]string, 600) + for i := range paths { + paths[i] = fmt.Sprintf(`["%s","%064x"]`, a, i+1) + } + id := 0 + for hour := 0; hour < 450; hour++ { + when := collector.start.Add(time.Duration(hour)*time.Hour + time.Minute) + for i := range paths { + id++ + collector.capture(routeTransmission(id, when, paths[i])) + } + } + got = collector.build(nil) + cells, _ := stepCells(got) + if cells < 200000 { + t.Fatalf("published %d hourly step cells: the byte budget must admit more than the retired cell cap", cells) + } + // 180 bytes per published row and 75 per bucket, against a 16 MiB budget. + if payload := len(got.steps)*180 + cells*75; payload > 16<<20 { + t.Fatalf("estimated step payload %d bytes exceeds the published budget", payload) + } + if !got.stepsTruncated && !got.bucketsTruncated { + t.Fatal("the payload budget was not enforced") + } + if got.evidence.CappedStepAppearances == 0 || !got.evidence.Incomplete { + t.Fatalf("payload truncation was not disclosed: %+v", got.evidence) + } + // #507: directed-step caps are step appearances only. + if e := got.memberships.evidence.history; e.Capped != 0 || e.Truncated { + t.Fatalf("the step payload budget polluted membership coverage evidence: %+v", e) + } +} + +// Point 4: the retained route/membership snapshot is measurable, because +// publishing it raised staging RSS from ~1.55 to ~1.72 GiB and production +// hosts are smaller. +func TestPerfMemRegionsSnapshotMemory(t *testing.T) { + srv, now, key := regionsObservedFixture(t) + other := "cd" + strings.Repeat("34", 31) + at := now.Truncate(time.Hour).Add(-time.Hour) + srv.store.byPathHop[key] = []*StoreTx{routeTransmission(1, at, `["`+key+`","`+other+`"]`)} + recomputeRepeaterEnrichmentSafe(srv.store, 24) + + w := httptest.NewRecorder() + srv.handlePerf(w, httptest.NewRequest("GET", "/api/perf?mem=1", nil)) + if w.Code != 200 { + t.Fatalf("perf mem=1 status %d: %s", w.Code, w.Body.String()) + } + var body struct { + MemoryBreakdown *struct { + TotalTx int `json:"totalTx"` + } `json:"memoryBreakdown"` + RegionsSnapshot *struct { + Nodes int `json:"nodes"` + Scopes int `json:"scopes"` + HistoryNodes int `json:"historyNodes"` + HistoryCells int `json:"historyCells"` + Steps int `json:"steps"` + StepCells int `json:"stepCells"` + EvidenceCells int `json:"evidenceCells"` + EvidenceBytes int64 `json:"evidenceBytes"` + MembershipsBytes int64 `json:"membershipsBytes"` + HistoryBytes int64 `json:"historyBytes"` + StepsBytes int64 `json:"stepsBytes"` + TotalBytes int64 `json:"totalBytes"` + TotalMB float64 `json:"totalMB"` + EstimatedStepPayloadBytes int64 `json:"estimatedStepPayloadBytes"` + GeneratedAt string `json:"generatedAt"` + } `json:"regionsSnapshot"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if body.MemoryBreakdown == nil { + t.Fatal("mem=1 returned no store memory breakdown") + } + r := body.RegionsSnapshot + if r == nil { + t.Fatal("mem=1 returned no retained Regions route/membership snapshot size") + } + if r.Nodes != 2 || r.Scopes != 2 || r.HistoryNodes != 2 || r.HistoryCells != 2 || r.Steps != 1 || r.StepCells != 1 { + t.Fatalf("snapshot shape was not measured: %+v", *r) + } + if r.MembershipsBytes <= 0 || r.HistoryBytes <= 0 || r.StepsBytes <= 0 || r.EstimatedStepPayloadBytes <= 0 { + t.Fatalf("retained sizes were not measured: %+v", *r) + } + if r.TotalBytes != r.MembershipsBytes+r.HistoryBytes+r.StepsBytes+r.EvidenceBytes || r.TotalMB < 0 { + t.Fatalf("retained total does not account for its parts: %+v", *r) + } + if r.GeneratedAt == "" { + t.Fatal("the measured snapshot is not identified by its generation time") + } +} + +// --------------------------------------------------------------------------- +// #510 round 2. The first round replaced the global 250,000-reference budget +// with a per-hour one, but left the interning table (131,072 entries) as a +// single global ceiling over the whole 504 h horizon: once full, every new +// path string was refused and reported as capped evidence in whichever hour +// the transmission fell in. Real path diversity is high — e2e-fixture.db, real +// captured traffic, holds 376 distinct path_json values in 399 path-carrying +// observations (94 %) — so ordinary diversity reached that ceiling before the +// per-hour budget did, which is the same starvation #508 reported at a lower +// ceiling. These tests pin the redesign: the table is a dedup cache that falls +// back to un-interned retention, and the only refusal left is per hour. + +// shuffledDistinctPathTxs builds one transmission per distinct path, spread +// evenly over the last `hours` buckets, and shuffles them into the order +// production capture sees (s.byPathHop is walked by map iteration). Paths are +// distinct 5-hop chains drawn from a small relay pool, which is how real +// diversity looks: a few hundred relays, very many distinct chains. +func shuffledDistinctPathTxs(start time.Time, hours, perHour, relays int) []*StoreTx { + keys := make([]string, relays) + for i := range keys { + keys[i] = fmt.Sprintf("%064x", i+1) + } + txs := make([]*StoreTx, 0, hours*perHour) + id, n := 0, 0 + for hour := regionsHistoryBucketCount - hours; hour < regionsHistoryBucketCount; hour++ { + at := start.Add(time.Duration(hour)*time.Hour + time.Minute) + for i := 0; i < perHour; i++ { + id, n = id+1, n+1 + hops := make([]string, 5) + for h, d := 0, n; h < len(hops); h, d = h+1, d/relays { + hops[h] = keys[d%relays] + } + txs = append(txs, routeTransmission(id, at, `["`+strings.Join(hops, `","`)+`"]`)) + } + } + rand.New(rand.NewSource(1)).Shuffle(len(txs), func(i, j int) { txs[i], txs[j] = txs[j], txs[i] }) + return txs +} + +// F1/F2: 168 h at one distinct path per transmission must not be capped by any +// capture budget. 168,000 distinct chains is above the retired 131,072-entry +// interning table and is only 6 % of the rate real data showed (805k +// observations in 48 h is about 16,800 per hour), so this is a floor on real +// diversity, not a flood. +func TestRegionsRoutePathCaptureBudgetAtRealPathDiversity(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + const hours, perHour, relays = 168, 1000, 20 + for _, tx := range shuffledDistinctPathTxs(collector.start, hours, perHour, relays) { + collector.capture(tx) + } + got := collector.build(nil) + if got.evidence.SourceTruncated || got.evidence.CappedObservations != 0 { + t.Fatalf("%d distinct paths over %d h were capped: truncated=%v capped=%d", + hours*perHour, hours, got.evidence.SourceTruncated, got.evidence.CappedObservations) + } + if e := got.memberships.evidence.history; e.Capped != 0 || e.Unprovable != 0 || e.Truncated { + t.Fatalf("path diversity withheld membership rates: %+v", e) + } + // Every one of the 168 hours keeps its own clean evidence: a diverse hour + // must not qualify any other hour, whatever the capture order was. + for hour := regionsHistoryBucketCount - hours; hour < regionsHistoryBucketCount; hour++ { + if counts := capturedEvidenceBucket(t, got, hour); counts != (RegionsEvidenceCounts{}) { + t.Fatalf("hour %d reported %+v, want clean evidence", hour, counts) + } + } + if got.stepsTruncated || got.bucketsTruncated || got.evidence.CappedStepAppearances != 0 { + t.Fatalf("publication budget truncated the snapshot: %+v", got.evidence) + } + if got.capture.PathVariants != hours*perHour || got.capture.PathReferences != hours*perHour { + t.Fatalf("capture diagnostic lost the retained shape: %+v", got.capture) + } + if got.capture.MaxHourPathReferences != perHour { + t.Fatalf("per-hour reference high-water mark was %d, want %d", got.capture.MaxHourPathReferences, perHour) + } + // Each hour's 1,000 chains cover the whole relay pool, so every relay keeps + // one aggregated membership cell in each of the 168 hours. + for key, scopes := range got.history.nodes { + if len(scopes) != 1 || len(scopes[0].Buckets) != hours { + t.Fatalf("%s retained %d scopes and %d hourly buckets, want 1 and %d", + key[:8], len(scopes), len(scopes[0].Buckets), hours) + } + } + if len(got.history.nodes) != relays { + t.Fatalf("retained %d of %d relay identities", len(got.history.nodes), relays) + } +} + +// F1/F4 (M5): the interning table is a bounded dedup cache, not a budget. Past +// it, a path is retained un-interned — it costs a path slot and a reference +// instead of a map entry — and no observation is refused. +func TestRegionsRoutePathInterningFallsBackUnInterned(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + collector.internBound = 4 + a := strings.Repeat("ab", 32) + const paths = 40 + at := collector.start.Add(500 * time.Hour) + for i := 0; i < paths; i++ { + collector.capture(routeTransmission(i+1, at, fmt.Sprintf(`["%s","%064x"]`, a, i+1))) + } + // Repeating one of the interned paths must still share its entry. + collector.capture(routeTransmission(paths+1, at, fmt.Sprintf(`["%s","%064x"]`, a, 1))) + got := collector.build(nil) + if got.evidence.SourceTruncated || got.evidence.CappedObservations != 0 { + t.Fatalf("a full interning table refused paths: truncated=%v capped=%d", + got.evidence.SourceTruncated, got.evidence.CappedObservations) + } + if got.capture.PathVariantsInterned != collector.internBound { + t.Fatalf("interned %d paths against a bound of %d", got.capture.PathVariantsInterned, collector.internBound) + } + if got.capture.PathVariants != paths || got.capture.PathReferences != paths+1 { + t.Fatalf("un-interned retention lost paths: %+v", got.capture) + } + if len(got.steps) != paths { + t.Fatalf("published %d of %d directed step rows", len(got.steps), paths) + } + for _, step := range got.steps { + want := 1 + if step.To == fmt.Sprintf("%064x", 1) { + want = 2 + } + if len(step.Buckets) != 1 || step.Buckets[0].Count != want { + t.Fatalf("%s..%s published %+v, want one bucket counting %d", step.From[:8], step.To[:8], step.Buckets, want) + } + } +} + +// F2/F6: the per-hour reference budget must have headroom over the rate real +// data showed, so only a flood reaches it, and the cost model the disclosed +// transient ceiling rests on has to stay true. #508 reported about 805,000 +// observations in 48 h of retention — about 16,800 per hour across all scopes +// — and every one of them could be transport-scoped with a distinct path. +func TestRegionsRouteCaptureBudgetHeadroomOverMeasuredRate(t *testing.T) { + const measuredObservationsPerHour = 805000 / 48 + if maxRegionsRoutePathRefsPerHour < 2*measuredObservationsPerHour { + t.Fatalf("per-hour capture budget %d leaves no headroom over the measured %d observations/hour", + maxRegionsRoutePathRefsPerHour, measuredObservationsPerHour) + } + // The documented per-row capture cost: 8 bytes per retained reference, 24 + // for a retained path (header plus resolver slot) and 72 per retained + // transmission, against the ~100 bytes per observation and ~1,000 bytes + // per transmission the packet store already holds. + if got := int64(unsafe.Sizeof(regionsRoutePathRef{})); got != regionsRoutePathRefBytes { + t.Fatalf("a retained reference is %d bytes, documented as %d", got, regionsRoutePathRefBytes) + } + if got := int64(unsafe.Sizeof(regionsRouteTx{})); got != regionsRouteTxBytes { + t.Fatalf("a retained transmission header is %d bytes, documented as %d", got, regionsRouteTxBytes) + } + slot := int64(unsafe.Sizeof("")) + int64(unsafe.Sizeof((*regionsResolvedPath)(nil))) + if slot != regionsRoutePathSlotBytes { + t.Fatalf("a retained path slot is %d bytes, documented as %d", slot, regionsRoutePathSlotBytes) + } + // The reference ceiling this implies stays a disclosed figure (#510 F6). + refs := int64(maxRegionsRoutePathRefsPerHour) * (regionsHistoryBucketCount + 1) + if bytes := refs * regionsRoutePathRefBytes; bytes > 320<<20 { + t.Fatalf("transient reference ceiling grew to %d bytes; document or lower it", bytes) + } +} + +// F4 (M6): the published step-endpoint ceiling is enforced, and steps whose +// endpoints it omitted are disclosed instead of dropped silently. +func TestRegionsRouteStepEndpointCeilingDisclosed(t *testing.T) { + srv, _, _, _ := regionsRoutesFixture(t) + if full := getRegionsRoutes(t, srv); len(full.StepNodes) != 2 || len(full.Steps) != 1 || full.StepEvidence.MissingEndpointSteps != 0 { + t.Fatalf("fixture did not publish both endpoints: %+v", full.StepEvidence) + } + srv.stepEndpointLimit = 1 + got := getRegionsRoutes(t, srv) + if len(got.StepNodes) != 1 { + t.Fatalf("published %d step endpoints against a ceiling of 1", len(got.StepNodes)) + } + if len(got.Steps) != 0 || got.StepEvidence.MissingEndpointSteps != 1 || !got.StepEvidence.Incomplete { + t.Fatalf("the endpoint ceiling was not disclosed: steps=%d %+v", len(got.Steps), got.StepEvidence) + } +} + +// F3: the capture budgets that decide `capped` must be observable before they +// cap, so the diagnostic reports their fill against their ceilings. +func TestRegionsSnapshotCaptureHeadroomMeasured(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + a := strings.Repeat("ab", 32) + at := collector.start.Add(500 * time.Hour) + for i := 0; i < 3; i++ { + collector.capture(routeTransmission(i+1, at, fmt.Sprintf(`["%s","%064x"]`, a, i+1))) + } + snapshot := collector.build(nil) + measured := measureRegionsSnapshot(&repeaterRelaySnapshot{generatedAt: now, + regionMemberships: snapshot.memberships, history: snapshot.history, routes: snapshot}) + c := measured.Capture + if c.PathVariants != 3 || c.PathVariantsInterned != 3 || c.PathReferences != 3 || c.Transmissions != 3 { + t.Fatalf("capture fill was not reported: %+v", c) + } + if c.MaxHourPathReferences != 3 || c.HourPathReferenceBudget != maxRegionsRoutePathRefsPerHour { + t.Fatalf("per-hour reference headroom was not reported: %+v", c) + } + if c.PathVariantInternBudget != maxRegionsRoutePathVariants { + t.Fatalf("interning-table ceiling was not reported: %+v", c) + } + // Four relays in one hour and one scope: four aggregated membership cells. + if c.MembershipCells != 4 || c.MembershipCellBudget != maxRegionsMembershipCells || + c.MembershipNodes != 4 || c.MembershipNodeBudget != maxRegionsMembershipNodes { + t.Fatalf("membership headroom was not reported: %+v", c) + } + // N2: pin the sum, not just its sign. Three references, three retained and + // interned paths, three transmission headers -- dropping any term, the + // transmission header above all, has to show here. + want := int64(3*regionsRoutePathRefBytes + 3*regionsRoutePathSlotBytes + + 3*regionsRoutePathInternedBytes + 3*regionsRouteTxBytes) + if c.EstimatedBytes != want { + t.Fatalf("transient capture cost was %d, want %d: %+v", c.EstimatedBytes, want, c) + } +} + +// F5/F4 (M10): regionsSnapshot.truncated is budget truncation only. Unresolved +// hops and unprovable observations also set evidence.Incomplete, which is +// almost always true on real data, so it is reported separately. +func TestRegionsSnapshotTruncatedIsBudgetTruncationOnly(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + measure := func(snapshot *regionsRouteSnapshot) *RegionsSnapshotMemory { + return measureRegionsSnapshot(&repeaterRelaySnapshot{generatedAt: now, + regionMemberships: snapshot.memberships, history: snapshot.history, routes: snapshot}) + } + a, b := strings.Repeat("ab", 32), strings.Repeat("cd", 32) + // An unresolved hop token qualifies the evidence without spending any + // budget: nothing was truncated. + collector := newRegionsRoutesCollector(now) + collector.capture(routeTransmission(1, collector.start.Add(500*time.Hour), `["`+a+`","BE","`+b+`"]`)) + incomplete := collector.build(nil) + if !incomplete.evidence.Incomplete || incomplete.evidence.UnresolvedHops == 0 { + t.Fatalf("fixture did not produce an unresolved hop: %+v", incomplete.evidence) + } + if got := measure(incomplete); got.Truncated || !got.EvidenceIncomplete { + t.Fatalf("ordinary incompleteness was reported as budget truncation: truncated=%v evidenceIncomplete=%v", + got.Truncated, got.EvidenceIncomplete) + } + // A spent publication budget is budget truncation. + collector = newRegionsRoutesCollector(now) + collector.stepBytes = regionsRouteStepRowBytes + regionsRouteStepCellBytes + for i := 0; i < 3; i++ { + collector.capture(routeTransmission(i+1, collector.start.Add(500*time.Hour), fmt.Sprintf(`["%s","%064x"]`, a, i+1))) + } + truncated := collector.build(nil) + if !truncated.stepsTruncated { + t.Fatal("fixture did not spend the publication budget") + } + if got := measure(truncated); !got.Truncated || !got.EvidenceIncomplete { + t.Fatalf("publication truncation was not reported: truncated=%v evidenceIncomplete=%v", + got.Truncated, got.EvidenceIncomplete) + } + // So is a spent capture budget, even though the published steps fit. + collector = newRegionsRoutesCollector(now) + collector.hourRefs[500] = maxRegionsRoutePathRefsPerHour + collector.capture(routeTransmission(1, collector.start.Add(500*time.Hour), `["`+a+`","`+b+`"]`)) + collector.capture(routeTransmission(2, collector.start.Add(501*time.Hour), `["`+a+`","`+b+`"]`)) + capped := collector.build(nil) + if !capped.evidence.SourceTruncated { + t.Fatal("fixture did not spend the capture budget") + } + if got := measure(capped); !got.Truncated { + t.Fatalf("capture truncation was not reported: %+v", got) + } +} + +// Round 3 (#510). Staging on real data (master f56ef97c + #510, 48 h +// retention, three restarts) reported directed steps complete but +// coverage.evidence.capped still 1,068-1,096, all in #dk, so the Health and +// Timeline rates stayed withheld. The tests below cover the two capture +// ceilings a 65-observer mesh actually reaches, and make every refusal +// attributable so the next round reads the cause off the snapshot instead of +// guessing. + +// floodTransmission models one flood packet heard by observers observers, +// each over chains distinct relay chains, with one repeat of its first chain +// per observer. Hops are drawn from a fixed relay pool, as on a real mesh: +// the identities are bounded, the chains are not, and the pool is shifted by +// the transmission id so two transmissions do not reuse one another's chains. +// pool must be at least observers*chains for the chains to stay distinct. +func floodTransmission(id int, when time.Time, observers, chains, pool int) (*StoreTx, int) { + relay := func(n int) string { return fmt.Sprintf("%064x", n+1) } + paths := make([]string, 0, observers*chains+observers) + distinct := 0 + for o := 0; o < observers; o++ { + var first string + for c := 0; c < chains; c++ { + hops := []string{ + relay((id*7 + o*chains + c) % pool), // the chain's entry relay + relay((id/pool*31 + o*5 + c*3) % pool), // a relay shared with neighbours + relay(pool + o), // the observer's own uplink + } + path := `["` + strings.Join(hops, `","`) + `"]` + if c == 0 { + first = path + } + paths = append(paths, path) + distinct++ + } + paths = append(paths, first) // the same chain heard twice + } + return routeTransmission(id, when, paths...), distinct +} + +// N1: past the interning table, identical paths inside one transmission must +// still share one slot, one reference and one unit of the per-transmission +// budget. The table is full for most of a 48 h capture on real traffic, so +// this is the normal state, not an edge case. +func TestRegionsRouteCaptureDedupesRepeatedPathsUnInterned(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + collector.internBound = 0 // the table is already full + a, b := strings.Repeat("ab", 32), strings.Repeat("cd", 32) + const repeats = 200 + paths := make([]string, repeats) + for i := range paths { + paths[i] = `["` + a + `","` + b + `"]` + } + collector.capture(routeTransmission(1, collector.start.Add(500*time.Hour), paths...)) + got := collector.build(nil) + if got.evidence.SourceTruncated || got.evidence.CappedObservations != 0 { + t.Fatalf("repeated identical paths were capped: truncated=%v capped=%d", + got.evidence.SourceTruncated, got.evidence.CappedObservations) + } + if got.capture.PathVariants != 1 || got.capture.PathReferences != 1 { + t.Fatalf("un-interned duplicates were retained separately: %+v", got.capture) + } + if got.capture.MaxTransmissionPaths != 1 { + t.Fatalf("per-transmission high-water mark was %d, want 1", got.capture.MaxTransmissionPaths) + } + if e := got.memberships.evidence.history; e.Capped != 0 || e.Truncated { + t.Fatalf("duplicate observations withheld membership rates: %+v", e) + } +} + +// Point 1/2: a flood transmission heard by a 65-observer mesh over many +// distinct chains must not be capped, with the interning table full. 65 +// observers x 4 chains is 260 distinct paths in one transmission, twice the +// retired 128 per-transmission ceiling. +func TestRegionsRouteCaptureAdmitsFloodHeardByManyObservers(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + collector := newRegionsRoutesCollector(now) + collector.internBound = 0 // the table is already full + const observers, chains, pool = 65, 4, 512 + tx, distinct := floodTransmission(1, collector.start.Add(500*time.Hour), observers, chains, pool) + collector.capture(tx) + got := collector.build(nil) + if got.evidence.SourceTruncated || got.evidence.CappedObservations != 0 { + t.Fatalf("a flood heard by %d observers over %d chains was capped: truncated=%v capped=%d", + observers, distinct, got.evidence.SourceTruncated, got.evidence.CappedObservations) + } + if got.capture.PathVariants != distinct || got.capture.PathReferences != distinct { + t.Fatalf("capture lost chains of a single flood: %+v (want %d)", got.capture, distinct) + } + if got.capture.MaxTransmissionPaths != distinct || got.capture.TransmissionPathBudget != maxRegionsRoutePathsPerTx { + t.Fatalf("per-transmission headroom was not reported: %+v (want %d)", got.capture, distinct) + } + if e := got.memberships.evidence.history; e.Capped != 0 || e.Truncated { + t.Fatalf("a single flood withheld membership rates: %+v", e) + } + // The per-transmission pair budget used to be exactly 128 x 64, so the + // path ceiling always bound first. They are independent now, and a real + // flood must not reach the pair budget either. + if got.evidence.CappedStepAppearances != 0 || got.stepsTruncated || got.bucketsTruncated { + t.Fatalf("a single flood spent the directed-step budgets: %+v", got.evidence) + } +} + +// Point 2: the per-transmission ceiling has to clear the observer fan-out real +// data showed (65 observers on staging) with room for several chains each, +// while still stopping one transmission from spending most of an hour's +// reference budget. +func TestRegionsRouteTransmissionPathBudgetHeadroomOverObservers(t *testing.T) { + const stagingObservers = 65 + if maxRegionsRoutePathsPerTx < 16*stagingObservers { + t.Fatalf("per-transmission path budget %d leaves under 16 chains for each of %d observers", + maxRegionsRoutePathsPerTx, stagingObservers) + } + if maxRegionsRoutePathsPerTx > maxRegionsRoutePathRefsPerHour/8 { + t.Fatalf("one transmission may spend %d of an hour's %d references", + maxRegionsRoutePathsPerTx, maxRegionsRoutePathRefsPerHour) + } +} + +// Point 1: every capped observation is attributed to the ceiling that refused +// it, so an operator reads the cause off /api/perf?mem=1 instead of bisecting +// the budgets. The per-reason counters account for exactly the snapshot-wide +// capped observations; membership drops are reported separately because they +// are memberships, not observations. +func TestRegionsSnapshotCaptureAttributesCappedObservations(t *testing.T) { + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + a, b := strings.Repeat("ab", 32), strings.Repeat("cd", 32) + at := func(c *regionsRoutesCollector) time.Time { return c.start.Add(500 * time.Hour) } + ok := `["` + a + `","` + b + `"]` + + t.Run("path bytes", func(t *testing.T) { + c := newRegionsRoutesCollector(now) + long := `["` + strings.Join(make([]string, 200), a+`","`) + `"]` + if len(long) <= maxRegionsRoutePathBytes { + t.Fatalf("fixture path is %d bytes, under the %d ceiling", len(long), maxRegionsRoutePathBytes) + } + c.capture(routeTransmission(1, at(c), ok, long)) + got := c.build(nil) + if got.capture.CappedPathBytes != 1 || got.evidence.CappedObservations != 1 { + t.Fatalf("oversize path JSON was not attributed: %+v", got.capture) + } + if got.capture.MaxPathBytes != len(ok) || got.capture.PathByteBudget != maxRegionsRoutePathBytes { + t.Fatalf("path-byte headroom was not reported: %+v", got.capture) + } + }) + + t.Run("transmission paths", func(t *testing.T) { + c := newRegionsRoutesCollector(now) + c.txPathBound = 2 + c.capture(routeTransmission(1, at(c), ok, + `["`+a+`","`+strings.Repeat("ef", 32)+`"]`, `["`+b+`","`+a+`"]`)) + got := c.build(nil) + if got.capture.CappedTransmissionPaths != 1 || got.evidence.CappedObservations != 1 { + t.Fatalf("the per-transmission ceiling was not attributed: %+v", got.capture) + } + if !got.capture.SourceTruncated { + t.Fatal("a refused path did not set sourceTruncated") + } + }) + + t.Run("hour references", func(t *testing.T) { + c := newRegionsRoutesCollector(now) + c.hourRefs[500] = maxRegionsRoutePathRefsPerHour + c.capture(routeTransmission(1, at(c), ok)) + got := c.build(nil) + if got.capture.CappedHourReferences != 1 || got.evidence.CappedObservations != 1 { + t.Fatalf("the per-hour budget was not attributed: %+v", got.capture) + } + }) + + t.Run("overlong path", func(t *testing.T) { + c := newRegionsRoutesCollector(now) + hops := make([]string, 65) + for i := range hops { + hops[i] = fmt.Sprintf("%02x", i) + } + c.capture(routeTransmission(1, at(c), `["`+strings.Join(hops, `","`)+`"]`)) + got := c.build(nil) + if got.capture.CappedOverlongPaths != 1 || got.evidence.CappedObservations != 1 { + t.Fatalf("an over-length hop chain was not attributed: %+v", got.capture) + } + }) + + t.Run("membership nodes", func(t *testing.T) { + c := newRegionsRoutesCollector(now) + c.membershipNodeBound = 1 + c.capture(routeTransmission(1, at(c), ok)) + got := c.build(nil) + if got.capture.CappedMembershipNodes != 1 || got.capture.CappedMembershipCells != 0 { + t.Fatalf("the membership node ceiling was not attributed: %+v", got.capture) + } + if got.evidence.CappedObservations != 0 { + t.Fatalf("a dropped membership was counted as a capped observation: %+v", got.evidence) + } + }) + + t.Run("membership cells", func(t *testing.T) { + c := newRegionsRoutesCollector(now) + c.membershipCellBound = 1 + c.capture(routeTransmission(1, at(c), ok)) + got := c.build(nil) + if got.capture.CappedMembershipCells != 1 || got.capture.CappedMembershipNodes != 0 { + t.Fatalf("the membership cell ceiling was not attributed: %+v", got.capture) + } + }) +} diff --git a/cmd/server/regions_evidence.go b/cmd/server/regions_evidence.go index 77fd19c8d..d80456f25 100644 --- a/cmd/server/regions_evidence.go +++ b/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 ®ionsMembershipBuilder{entries: make(map[string][]relayEntry), remaining: maxRegionsMembershipEntries} +func newRegionsMembershipBuilder(cells, nodes int) *regionsMembershipBuilder { + return ®ionsMembershipBuilder{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 := ®ionsMembershipSnapshot{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 := ®ionsMembershipSnapshot{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 } diff --git a/cmd/server/regions_evidence_test.go b/cmd/server/regions_evidence_test.go index 86f12857c..b43b82263 100644 --- a/cmd/server/regions_evidence_test.go +++ b/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)) } } diff --git a/cmd/server/regions_history.go b/cmd/server/regions_history.go index 8505ee1f1..16aff01a2 100644 --- a/cmd/server/regions_history.go +++ b/cmd/server/regions_history.go @@ -79,21 +79,21 @@ 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 } @@ -101,12 +101,11 @@ func (b *regionsHistoryBuilder) add(key string, entries []relayEntry) { // 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] diff --git a/cmd/server/regions_history_test.go b/cmd/server/regions_history_test.go index d22e18fab..fc21a581d 100644 --- a/cmd/server/regions_history_test.go +++ b/cmd/server/regions_history_test.go @@ -10,25 +10,32 @@ import ( "time" ) -func historyEntry(at time.Time, scope string) relayEntry { - return relayEntry{t: at, parsed: true, valid: true, pt: 5, scope: scope} +// historyCell builds one aggregated membership cell the way the Regions +// collector does (#508): the hourly bucket index is derived from the +// transmission time, and a time before the horizon is out of range rather +// than truncated into bucket 0. +func historyCell(b *regionsHistoryBuilder, at time.Time, scope string) regionsMembershipNodeCell { + hour := int32(at.Sub(b.history.start) / time.Hour) + if at.Before(b.history.start) { + hour = -1 + } + return regionsMembershipNodeCell{scope: scope, hour: hour, count: 1, last: at} } func TestRegionsHistoryBucketsAndScopeProvenance(t *testing.T) { now := time.Date(2026, 10, 16, 12, 35, 0, 0, time.UTC) end := now.Truncate(time.Hour) start := end.Add(-21 * 24 * time.Hour) - entries := []relayEntry{historyEntry(start, "#sample"), historyEntry(end.Add(-time.Nanosecond), "#sample"), - historyEntry(end.Add(-time.Hour), "#sample"), historyEntry(start.Add(-time.Nanosecond), "#old"), - historyEntry(end, "#unfinished"), historyEntry(now.Add(time.Hour), "#future")} - entries = append(entries, relayEntry{ts: "invalid", scope: "#bad"}, historyEntry(end.Add(-time.Minute), "invalid")) - prefix := historyEntry(end.Add(-time.Minute), "#prefix") - prefix.fromPrefix = true - advert := historyEntry(end.Add(-time.Minute), "#advert") - advert.pt = payloadTypeAdvert - entries = append(entries, prefix, advert) + // Advert/TRACE payloads, planned direct paths, prefix-only evidence and + // unparsable timestamps are excluded before capture and can no longer + // reach this builder; TestRegionRoutesStrictObservationScopeAndBounds + // pins that gate. What remains here is the horizon and the scope name. b := newRegionsHistoryBuilder(now) - b.add(strings.Repeat("ab", 32), entries) + cells := []regionsMembershipNodeCell{historyCell(b, start, "#sample"), historyCell(b, end.Add(-time.Nanosecond), "#sample"), + historyCell(b, end.Add(-time.Hour), "#sample"), historyCell(b, start.Add(-time.Nanosecond), "#old"), + historyCell(b, end, "#unfinished"), historyCell(b, now.Add(time.Hour), "#future"), + historyCell(b, end.Add(-time.Minute), "invalid")} + b.add(strings.Repeat("ab", 32), cells) h := b.history if !h.start.Equal(start) || !h.end.Equal(end) || len(h.nodes) != 1 { t.Fatalf("history bounds/nodes: %+v", h) @@ -48,10 +55,11 @@ func TestRegionsHistoryBucketsAndScopeProvenance(t *testing.T) { func TestRegionsHistoryAdjacentSevenDayWindows(t *testing.T) { now := time.Date(2026, 10, 16, 12, 35, 0, 0, time.UTC) end := now.Truncate(time.Hour) - entries := []relayEntry{historyEntry(end.Add(-14*24*time.Hour), "#sample"), historyEntry(end.Add(-7*24*time.Hour), "#sample"), historyEntry(end.Add(-time.Second), "#sample")} b := newRegionsHistoryBuilder(now) + cells := []regionsMembershipNodeCell{historyCell(b, end.Add(-14*24*time.Hour), "#sample"), + historyCell(b, end.Add(-7*24*time.Hour), "#sample"), historyCell(b, end.Add(-time.Second), "#sample")} key := strings.Repeat("ab", 32) - b.add(key, entries) + b.add(key, cells) previous, current := 0, 0 for _, bucket := range b.history.nodes[key][0].Buckets { if bucket.Index >= 168 && bucket.Index < 336 { @@ -70,12 +78,12 @@ func TestRegionsHistoryBoundedScopesAndCells(t *testing.T) { now := time.Date(2026, 10, 16, 12, 35, 0, 0, time.UTC) b := newRegionsHistoryBuilder(now) b.remainingCells = 2 - var entries []relayEntry + var cells []regionsMembershipNodeCell for i := 0; i < 40; i++ { - entries = append(entries, historyEntry(b.history.end.Add(-time.Duration(i+1)*time.Hour), fmt.Sprintf("#s%02d", i))) + cells = append(cells, historyCell(b, b.history.end.Add(-time.Duration(i+1)*time.Hour), fmt.Sprintf("#s%02d", i))) } key := strings.Repeat("ab", 32) - b.add(key, entries) + b.add(key, cells) if !b.history.scopesTruncated || !b.history.bucketsTruncated || len(b.history.nodes[key]) > 32 || b.remainingCells != 0 { t.Fatalf("bounds/flags: %+v", b.history) } @@ -84,7 +92,7 @@ func TestRegionsHistoryBoundedScopesAndCells(t *testing.T) { t.Fatal("old scope displaced recent scope") } } - reversed := append([]relayEntry(nil), entries...) + reversed := append([]regionsMembershipNodeCell(nil), cells...) for i, j := 0, len(reversed)-1; i < j; i, j = i+1, j-1 { reversed[i], reversed[j] = reversed[j], reversed[i] } @@ -211,9 +219,9 @@ func TestRegionsHistoryArchivedMetadataAndVisibility(t *testing.T) { func TestRegionsHistoryNodeCapAndAvailability(t *testing.T) { now := time.Date(2026, 10, 16, 12, 35, 0, 0, time.UTC) b := newRegionsHistoryBuilder(now) - entries := []relayEntry{historyEntry(b.history.end.Add(-time.Minute), "#sample")} + cells := []regionsMembershipNodeCell{historyCell(b, b.history.end.Add(-time.Minute), "#sample")} for i := 0; i <= maxObservedRegionNodes; i++ { - b.add(fmt.Sprintf("%064x", i+1), entries) + b.add(fmt.Sprintf("%064x", i+1), cells) } if len(b.history.nodes) != maxObservedRegionNodes || !b.history.truncated { t.Fatal("snapshot node bound not enforced") @@ -238,12 +246,12 @@ func TestRegionsHistoryNodeCapAndAvailability(t *testing.T) { func TestRegionsHistoryActualGlobalCellBudget(t *testing.T) { now := time.Date(2026, 10, 16, 12, 35, 0, 0, time.UTC) b := newRegionsHistoryBuilder(now) - entries := make([]relayEntry, 501) - for i := range entries { - entries[i] = historyEntry(b.history.start.Add(time.Duration(i)*time.Hour), "#sample") + hours := make([]regionsMembershipNodeCell, 501) + for i := range hours { + hours[i] = historyCell(b, b.history.start.Add(time.Duration(i)*time.Hour), "#sample") } for i := 0; i < 500; i++ { - b.add(fmt.Sprintf("%064x", i+1), entries) + b.add(fmt.Sprintf("%064x", i+1), hours) } cells := 0 for _, scopes := range b.history.nodes { diff --git a/cmd/server/regions_memory.go b/cmd/server/regions_memory.go new file mode 100644 index 000000000..9c142e5e0 --- /dev/null +++ b/cmd/server/regions_memory.go @@ -0,0 +1,152 @@ +package main + +import ( + "math" + "time" + "unsafe" +) + +// Retained-heap accounting for the published Regions snapshot (#508). +// +// Publishing the route/membership snapshot raised staging container RSS from +// about 1.55 to about 1.72 GiB, and production hosts are smaller (one has +// 3.7 GiB shared with other services), so the retained size has to be an +// operator diagnostic rather than a guess. +// +// mapEntryBytes is a flat per-entry allowance for a Go map bucket slot and its +// load factor; sliceHeaderBytes covers one slice header. Like the store +// breakdown these are deliberate estimates: struct sizes plus owned string +// content plus a fixed per-entry allowance, not allocator measurements. Keys +// and scope names are shared with the packet store and the prefix map, so +// counting their content here is an upper bound. +const ( + mapEntryBytes = 48 + sliceHeaderBytes = 24 +) + +// RegionsSnapshotMemory sizes the retained Regions route, membership, hourly +// history and per-bucket evidence published with the shared relay snapshot. It +// walks only immutable published state: no SQL, no packet-store walk and no +// packet lock (#495). +type RegionsSnapshotMemory struct { + GeneratedAt string `json:"generatedAt"` + // Published counts, so a byte figure can be read against the budgets. + Nodes int `json:"nodes"` + Scopes int `json:"scopes"` + HistoryNodes int `json:"historyNodes"` + HistoryCells int `json:"historyCells"` + Steps int `json:"steps"` + StepCells int `json:"stepCells"` + EvidenceCells int `json:"evidenceCells"` + + MembershipsBytes int64 `json:"membershipsBytes"` + HistoryBytes int64 `json:"historyBytes"` + StepsBytes int64 `json:"stepsBytes"` + EvidenceBytes int64 `json:"evidenceBytes"` + TotalBytes int64 `json:"totalBytes"` + // TotalMB is TotalBytes in MiB, two decimals, matching the store breakdown. + TotalMB float64 `json:"totalMB"` + // EstimatedStepPayloadBytes is the published steps array measured with the + // same row/cell estimate the directed-step byte budget spends, so an + // operator can see how much of maxRegionsRouteStepBytes a snapshot used. + EstimatedStepPayloadBytes int64 `json:"estimatedStepPayloadBytes"` + StepPayloadBudgetBytes int64 `json:"stepPayloadBudgetBytes"` + // Truncated is budget truncation only: the hourly-history and directed-step + // publication flags plus the capture budget's own refusal. It deliberately + // excludes evidence.Incomplete, which is also raised by unresolved hops and + // unprovable observations and is therefore almost always true on real data + // (#510). + Truncated bool `json:"truncated"` + // EvidenceIncomplete repeats the snapshot's evidence qualification, so the + // distinction above costs no information. + EvidenceIncomplete bool `json:"evidenceIncomplete"` + // Capture is the transient capture state's fill against its own budgets. + // Publication sizes alone do not show how close the per-hour reference + // budget, the interning table or the membership cell budget are to their + // ceilings, and those are the budgets that decide `capped` (#510). + Capture RegionsRouteCapture `json:"capture"` +} + +func stringBytes(s string) int64 { return int64(len(s)) + strHdr } + +func regionsHistoryBucketsBytes(buckets []RegionsHistoryBucket) int64 { + total := sliceHeaderBytes + int64(len(buckets))*int64(unsafe.Sizeof(RegionsHistoryBucket{})) + for _, bucket := range buckets { + total += int64(len(bucket.LastRelayed)) + } + return total +} + +// measureRegionsSnapshot is O(published cells) and allocation-free. The +// snapshot is immutable once published, so the caller may release +// repeaterEnrichMu before calling. +func measureRegionsSnapshot(snapshot *repeaterRelaySnapshot) *RegionsSnapshotMemory { + out := &RegionsSnapshotMemory{GeneratedAt: snapshot.generatedAt.Format(time.RFC3339Nano), + StepPayloadBudgetBytes: maxRegionsRouteStepBytes} + if m := snapshot.regionMemberships; m != nil { + for key, info := range m.nodes { + out.Nodes++ + out.MembershipsBytes += stringBytes(key) + mapEntryBytes + + int64(unsafe.Sizeof(regionsNodeMembership{})) + sliceHeaderBytes + for _, scope := range info.scopes { + out.Scopes++ + out.MembershipsBytes += stringBytes(scope.name) + int64(unsafe.Sizeof(regionScopeRelay{})) + } + } + if e := m.evidence; e != nil { + out.EvidenceBytes += sliceHeaderBytes + for _, scope := range e.scopes { + out.EvidenceBytes += stringBytes(scope.Name) + int64(unsafe.Sizeof(RegionsEvidenceScope{})) + sliceHeaderBytes + out.EvidenceCells += len(scope.Buckets) + out.EvidenceBytes += int64(len(scope.Buckets)) * int64(unsafe.Sizeof(RegionsEvidenceBucket{})) + } + // The precomputed history summary is retained alongside the cells. + for _, scope := range e.history.Scopes { + out.EvidenceBytes += stringBytes(scope.Name) + int64(unsafe.Sizeof(RegionsEvidenceScope{})) + + sliceHeaderBytes + int64(len(scope.Buckets))*int64(unsafe.Sizeof(RegionsEvidenceBucket{})) + } + } + } + if h := snapshot.history; h != nil { + out.Truncated = out.Truncated || h.truncated || h.scopesTruncated || h.bucketsTruncated + for key, scopes := range h.nodes { + out.HistoryNodes++ + out.HistoryBytes += stringBytes(key) + mapEntryBytes + sliceHeaderBytes + for _, scope := range scopes { + out.HistoryBytes += stringBytes(scope.Name) + int64(unsafe.Sizeof(RegionsHistoryScope{})) + + regionsHistoryBucketsBytes(scope.Buckets) + out.HistoryCells += len(scope.Buckets) + } + } + } + if r := snapshot.routes; r != nil { + out.Truncated = out.Truncated || r.stepsTruncated || r.bucketsTruncated || r.evidence.SourceTruncated + out.EvidenceIncomplete = r.evidence.Incomplete + out.Capture = r.capture + out.StepsBytes += sliceHeaderBytes + for _, step := range r.steps { + out.Steps++ + out.StepCells += len(step.Buckets) + out.StepsBytes += stringBytes(step.From) + stringBytes(step.To) + stringBytes(step.Scope) + + int64(unsafe.Sizeof(RegionsRouteStep{})) + regionsHistoryBucketsBytes(step.Buckets) + } + out.EstimatedStepPayloadBytes = int64(out.Steps)*regionsRouteStepRowBytes + int64(out.StepCells)*regionsRouteStepCellBytes + } + out.TotalBytes = out.MembershipsBytes + out.HistoryBytes + out.StepsBytes + out.EvidenceBytes + out.TotalMB = math.Round(float64(out.TotalBytes)/1048576*100) / 100 + return out +} + +// GetRegionsSnapshotMemory sizes the retained Regions snapshot, or returns nil +// before the first background relay snapshot is published. It takes only +// repeaterEnrichMu, and releases it before walking the immutable snapshot, so +// it never blocks ingest or eviction. +func (s *PacketStore) GetRegionsSnapshotMemory() *RegionsSnapshotMemory { + s.repeaterEnrichMu.Lock() + snapshot := s.repeaterRelaySnapshot + s.repeaterEnrichMu.Unlock() + if snapshot == nil || snapshot.routes == nil { + return nil + } + return measureRegionsSnapshot(snapshot) +} diff --git a/cmd/server/regions_routes.go b/cmd/server/regions_routes.go index 2bbf63837..98ae84139 100644 --- a/cmd/server/regions_routes.go +++ b/cmd/server/regions_routes.go @@ -12,12 +12,72 @@ import ( ) const ( - maxRegionsRouteSteps = 5000 - maxRegionsRouteCells = 100000 - maxRegionsRoutePaths = 250000 - maxRegionsRoutePathsPerTx = 128 + // Published directed-step payload budget (#508). The retired fixed caps + // (5,000 rows and 100,000 hourly cells) were both reached on real data at + // 48 h retention, which left directed steps permanently incomplete, and a + // larger fixed row count gives no payload guarantee. Rows and hourly cells + // now share one estimated-byte budget, so a snapshot whose rows have few + // populated hours publishes many more rows while the response-size + // guarantee stays explicit and measurable. + regionsRouteStepRowBytes = 180 // from/to/scope plus the row's JSON scaffolding + regionsRouteStepCellBytes = 75 // one {index, count, lastRelayed} bucket + maxRegionsRouteStepBytes = 16 << 20 + // Publication ceilings implied by the byte budget, for the OpenAPI + // description and the browser's defensive bounds. + maxRegionsRouteSteps = maxRegionsRouteStepBytes / regionsRouteStepRowBytes + maxRegionsRouteCells = maxRegionsRouteStepBytes / regionsRouteStepCellBytes + // Visible step endpoints published with a snapshot. Steps whose endpoints + // are omitted are reported as missing-endpoint steps, never dropped + // silently. + maxRegionsRouteEndpoints = 10000 + + // Distinct observation paths retained for one transmission (#508/#510). + // The retired value was 128, which a flood packet reaches on an ordinary + // mesh: staging carries 65 observers, each of which can hear one flood + // over several relay chains and log every copy, so a single transmission + // offers hundreds of distinct chains and the surplus was reported capped. + // It is not a memory bound -- every retained path already costs one unit + // of the per-hour reference budget, which bounds the snapshot -- it is the + // share of an hour one transmission may spend. 4,096 is a sixteenth of + // that hour, which at 65 observers still allows 63 chains each. + maxRegionsRoutePathsPerTx = 4096 maxRegionsRoutePairsPerTx = 8192 maxRegionsRoutePathBytes = 8192 + // Interning table for observation path strings: a bounded dedup cache, not + // a budget. Identical paths share one entry, so one relay chain observed by + // thousands of transmissions is parsed and resolved once. Past the table a + // path is retained un-interned instead of being refused — it costs a path + // slot and a reference rather than a map entry. Refusing would reintroduce + // exactly the starvation #508 reported, at a lower ceiling: real path + // diversity is high (e2e-fixture.db, real captured traffic, holds 376 + // distinct path strings in 399 path-carrying observations), so a global + // table fills on ordinary traffic and would then cap every later hour. + maxRegionsRoutePathVariants = 1 << 17 + // Retained path references per hourly bucket (#508). The retired global + // 250,000-reference budget was exhausted by 48 h of real traffic (~805k + // observations): every hour after it was reported capped, which withheld + // the Health and Timeline rates for #dk and for all scopes. A per-hour + // budget makes the ceiling scale with retention and stops one busy hour + // from starving the rest of the horizon. 65,536 is about four times the + // ~16,800 observations/hour that 48 h of real traffic averaged across all + // scopes, so even a peak hour has headroom if every observation were + // transport-scoped and carried a distinct path. It is a fairness and + // pathology net, not the memory bound: capture retains at most one + // reference per eligible observation and one header per transmission + // already in the packet store, at roughly 3-5 % of the store state for the + // same rows, so packetStore.maxMemoryMB binds first (the full disclosure, + // including the horizon product, is in docs/regions-observed.md). + maxRegionsRoutePathRefsPerHour = 1 << 16 + // Estimated retained cost of the transient capture state, for the capture + // diagnostic and for the disclosed horizon ceiling. Path content is shared + // with the packet store, so only headers and map overhead are counted. + regionsRoutePathRefBytes = 8 // one regionsRoutePathRef + regionsRoutePathSlotBytes = 24 // one retained path: header plus resolver slot + regionsRoutePathInternedBytes = 48 // its interning-table entry on top + regionsRouteTxBytes = 72 // one retained transmission header + // Hop tokens cached with resolved paths. Past it, paths are resolved + // without caching: slower, never truncated. + maxRegionsRouteResolvedTokens = 1 << 21 ) // Cache only the observation's own header/transport evidence. Full raw frames @@ -98,32 +158,191 @@ type regionsRouteSnapshot struct { steps []RegionsRouteStep stepsTruncated, bucketsTruncated bool evidence RegionsStepEvidence + capture RegionsRouteCapture memberships *regionsMembershipSnapshot history *regionsRelayHistory } -type regionsRoutePath struct { - path string - observations int +// RegionsRouteCapture is the transient capture state's fill against its own +// budgets (#510). Retained with the published snapshot and reported by the +// memory diagnostics, so an operator can see how close the budgets that +// decide `capped` are to their ceilings before one is reached. EstimatedBytes +// is the same deliberate estimate the store breakdown uses: struct headers, +// map overhead and resolver slots, with path content excluded because it is +// shared with the packet store. +type RegionsRouteCapture struct { + PathVariants int `json:"pathVariants"` + PathVariantsInterned int `json:"pathVariantsInterned"` + PathVariantInternBudget int `json:"pathVariantInternBudget"` + PathReferences int `json:"pathReferences"` + MaxHourPathReferences int `json:"maxHourPathReferences"` + HourPathReferenceBudget int `json:"hourPathReferenceBudget"` + Transmissions int `json:"transmissions"` + MembershipCells int `json:"membershipCells"` + MembershipCellBudget int `json:"membershipCellBudget"` + MembershipNodes int `json:"membershipNodes"` + MembershipNodeBudget int `json:"membershipNodeBudget"` + EstimatedBytes int64 `json:"estimatedBytes"` + // Per-transmission and per-path fill. A flood packet heard by many + // observers is the shape that reaches the per-transmission ceiling, and + // neither it nor the path-byte ceiling was observable before (#510). + // Both maxima are over admitted observations, so they are headroom + // against their budgets; what the budgets refused is counted below. + MaxTransmissionPaths int `json:"maxTransmissionPaths"` + TransmissionPathBudget int `json:"transmissionPathBudget"` + MaxPathBytes int `json:"maxPathBytes"` + PathByteBudget int `json:"pathByteBudget"` + // Capped* attribute every capped observation to the ceiling that refused + // it, so the cause of a withheld rate is read off the snapshot rather than + // bisected. The four observation counters sum to the snapshot-wide + // cappedObservations. CappedMembership* are dropped memberships, not + // observations: they qualify a bucket's coverage without being counted + // among them. + CappedPathBytes int `json:"cappedPathBytes"` + CappedTransmissionPaths int `json:"cappedTransmissionPaths"` + CappedHourReferences int `json:"cappedHourReferences"` + CappedOverlongPaths int `json:"cappedOverlongPaths"` + CappedMembershipCells int `json:"cappedMembershipCells"` + CappedMembershipNodes int `json:"cappedMembershipNodes"` + // SourceTruncated repeats the capture budget's own refusal flag, so the + // fill figures above are never read as a complete capture. + SourceTruncated bool `json:"sourceTruncated"` } + +// regionsRoutePathRef points one transmission at an interned observation path +// and counts the observations that reported it. Eight bytes per retained +// reference, so the deferred capture list stays a small fraction of what the +// packet store already holds for the same observations. +type regionsRoutePathRef struct{ path, observations int32 } + type regionsRouteTx struct { id int when time.Time scope string - paths []regionsRoutePath + paths []regionsRoutePathRef } type regionsRoutesCollector struct { start, end time.Time now time.Time txs []regionsRouteTx - pathCount int - evidence RegionsStepEvidence - cells *regionsEvidenceBuilder + // Interned observation paths, and this snapshot's per-hour reference + // budget. Indexed by hourly bucket, 0 through regionsHistoryBucketCount. + pathIDs map[string]int32 + pathList []string + hourRefs []int32 + // internBound is the interning table's size. A field rather than the + // constant so tests can exercise the un-interned fallback without building + // 131,072 distinct paths; capture never mutates it. + internBound int + // The remaining capture ceilings, as fields for the same reason: the + // per-reason capped counters below are exercised without a 4,096-chain + // transmission or a 10,000-relay mesh. Capture never mutates them. + txPathBound int + membershipCellBound int + membershipNodeBound int + // Per-reason capture refusals and their high-water marks, reported by + // captureStats. maxTxPaths and maxPathBytes are headroom, the rest are + // refusals; cappedHourRefs also carries the defensive out-of-horizon + // guard, which captureAt's window check already makes unreachable. + maxTxPaths, maxPathBytes int + cappedPathBytes, cappedTxPaths int + cappedHourRefs, cappedOverlongPaths int + // stepBytes is the directed-step payload budget for this snapshot. A field + // rather than the constant so tests can exercise truncation without + // building a 16 MiB snapshot; build() never mutates it. + stepBytes int + evidence RegionsStepEvidence + cells *regionsEvidenceBuilder } func newRegionsRoutesCollector(now time.Time) *regionsRoutesCollector { end := now.UTC().Truncate(time.Hour) - return ®ionsRoutesCollector{start: end.Add(-regionsHistoryBucketCount * time.Hour), end: end, now: now.UTC(), cells: newRegionsEvidenceBuilder()} + return ®ionsRoutesCollector{start: end.Add(-regionsHistoryBucketCount * time.Hour), end: end, now: now.UTC(), + pathIDs: make(map[string]int32), hourRefs: make([]int32, regionsHistoryBucketCount+1), + internBound: maxRegionsRoutePathVariants, txPathBound: maxRegionsRoutePathsPerTx, + membershipCellBound: maxRegionsMembershipCells, membershipNodeBound: maxRegionsMembershipNodes, + stepBytes: maxRegionsRouteStepBytes, cells: newRegionsEvidenceBuilder()} +} + +// pathRefs reports how many observation path references this snapshot +// retained, for the capture tests and benchmarks. +func (c *regionsRoutesCollector) pathRefs() int { + total := 0 + for _, used := range c.hourRefs { + total += int(used) + } + return total +} + +// captureStats reports how much of each capture budget this snapshot spent, so +// the budgets that decide `capped` are observable before they cap (#510). It +// is O(hourly buckets) and runs with the rest of build(), outside the packet +// lock. +func (c *regionsRoutesCollector) captureStats(memberships *regionsMembershipBuilder) RegionsRouteCapture { + out := RegionsRouteCapture{ + PathVariants: len(c.pathList), + PathVariantsInterned: len(c.pathIDs), + PathVariantInternBudget: c.internBound, + PathReferences: c.pathRefs(), + HourPathReferenceBudget: maxRegionsRoutePathRefsPerHour, + Transmissions: len(c.txs), + MembershipCells: len(memberships.cells), + MembershipCellBudget: c.membershipCellBound, + MembershipNodes: len(memberships.nodes), + MembershipNodeBudget: c.membershipNodeBound, + MaxTransmissionPaths: c.maxTxPaths, + TransmissionPathBudget: c.txPathBound, + MaxPathBytes: c.maxPathBytes, + PathByteBudget: maxRegionsRoutePathBytes, + CappedPathBytes: c.cappedPathBytes, + CappedTransmissionPaths: c.cappedTxPaths, + CappedHourReferences: c.cappedHourRefs, + CappedOverlongPaths: c.cappedOverlongPaths, + CappedMembershipCells: memberships.cellDrops, + CappedMembershipNodes: memberships.nodeDrops, + SourceTruncated: c.evidence.SourceTruncated, + } + for _, used := range c.hourRefs { + if int(used) > out.MaxHourPathReferences { + out.MaxHourPathReferences = int(used) + } + } + out.EstimatedBytes = int64(out.PathReferences)*regionsRoutePathRefBytes + + int64(out.PathVariants)*regionsRoutePathSlotBytes + + int64(out.PathVariantsInterned)*regionsRoutePathInternedBytes + + int64(out.Transmissions)*regionsRouteTxBytes + return out +} + +// retainPath spends this hour's reference budget and interns the path when it +// is new and the table has room. Identical paths share one interned entry, so +// a repeated relay chain is parsed and resolved once per snapshot; the only +// refusal is the per-hour reference budget, which is fair across the horizon. +// held is how many distinct paths this transmission already retained. The +// caller has already looked the path up in the table, so a path new to this +// transmission costs two string-map lookups and a repeat of one costs one. +func (c *regionsRoutesCollector) retainPath(path string, id int32, known bool, index, held int) (int32, bool) { + switch { + case held >= c.txPathBound: + c.cappedTxPaths++ + return 0, false + case index < 0 || index >= len(c.hourRefs) || c.hourRefs[index] >= maxRegionsRoutePathRefsPerHour: + c.cappedHourRefs++ + return 0, false + } + if !known { + id = int32(len(c.pathList)) + c.pathList = append(c.pathList, path) + // Past the interning table the path is still retained, just without a + // table entry: identical paths then cost one slot each instead of + // sharing one. The per-hour reference budget bounds that; nothing is + // refused for lack of table space. + if len(c.pathIDs) < c.internBound { + c.pathIDs[path] = id + } + } + c.hourRefs[index]++ + return id, true } // addEvidence records one transmission's evidence in its own FirstSeen hour @@ -158,7 +377,14 @@ func (c *regionsRoutesCollector) captureAt(tx *StoreTx, when time.Time, valid bo canonical, known := wireTransportCodes(tx.RawHex) known = known && canonical>>16 != 0 item := regionsRouteTx{id: tx.ID, when: when, scope: tx.ScopeName} + index := c.bucket(item) var counts RegionsEvidenceCounts + // Keyed by the path string, not by the interned id: past the interning + // table a path has no id to dedupe on, and the table is full for most of + // a 48 h capture on real traffic. Without this, every repeat of one chain + // inside a flood transmission took its own slot, its own reference and + // its own unit of the per-transmission budget (#510). The string header + // is shared with the packet store; the map lives only for this call. seen := make(map[string]int) for _, obs := range tx.Observations { switch classifyRegionsObservation(obs, canonical, known) { @@ -170,21 +396,29 @@ func (c *regionsRoutesCollector) captureAt(tx *StoreTx, when time.Time, valid bo continue } if len(obs.PathJSON) > maxRegionsRoutePathBytes { + c.cappedPathBytes++ counts.Capped++ continue } - if index, exists := seen[obs.PathJSON]; exists { - item.paths[index].observations++ + if len(obs.PathJSON) > c.maxPathBytes { + c.maxPathBytes = len(obs.PathJSON) + } + if at, exists := seen[obs.PathJSON]; exists { + item.paths[at].observations++ continue } - if len(item.paths) >= maxRegionsRoutePathsPerTx || c.pathCount >= maxRegionsRoutePaths { + id, interned := c.pathIDs[obs.PathJSON] + id, admitted := c.retainPath(obs.PathJSON, id, interned, index, len(item.paths)) + if !admitted { c.evidence.SourceTruncated = true counts.Capped++ continue } seen[obs.PathJSON] = len(item.paths) - item.paths = append(item.paths, regionsRoutePath{path: obs.PathJSON, observations: 1}) - c.pathCount++ + item.paths = append(item.paths, regionsRoutePathRef{path: id, observations: 1}) + } + if len(item.paths) > c.maxTxPaths { + c.maxTxPaths = len(item.paths) } if len(tx.Observations) == 0 { counts.Unprovable++ @@ -201,15 +435,71 @@ type regionsRouteCell struct { last time.Time } +// regionsResolvedPath is one interned observation path, parsed and identity- +// resolved once per snapshot instead of once per transmission that observed +// it. keys keeps path order; an empty key is an unresolved hop token, which +// breaks adjacency without qualifying resolved memberships. +type regionsResolvedPath struct { + keys []string + bad bool // malformed path JSON + over bool // more hop tokens than the adjacency scan accepts +} + +// regionsPathResolver runs outside the packet lock. The cache is bounded by +// retained hop tokens; past that budget paths are resolved uncached, which +// costs time but never truncates evidence. +type regionsPathResolver struct { + resolve *relayTokenResolver + paths []string + cached []*regionsResolvedPath + remaining int +} + +func newRegionsPathResolver(paths []string, pm *prefixMap) *regionsPathResolver { + return ®ionsPathResolver{resolve: newRelayTokenResolver(pm), paths: paths, + cached: make([]*regionsResolvedPath, len(paths)), remaining: maxRegionsRouteResolvedTokens} +} + +func (r *regionsPathResolver) at(id int32) *regionsResolvedPath { + if id < 0 || int(id) >= len(r.paths) { + return ®ionsResolvedPath{bad: true} + } + if hit := r.cached[id]; hit != nil { + return hit + } + out := ®ionsResolvedPath{} + var tokens []string + if err := json.Unmarshal([]byte(r.paths[id]), &tokens); err != nil { + out.bad = true + } else if len(tokens) > 64 { + out.over = true + } else { + out.keys = make([]string, len(tokens)) + for i, token := range tokens { + if len(r.resolve.cache) >= 4096 { + clear(r.resolve.cache) + } + out.keys[i] = r.resolve.key(token) + } + } + if r.remaining >= len(out.keys) { + r.remaining -= len(out.keys) + r.cached[id] = out + } + return out +} + func (c *regionsRoutesCollector) build(pm *prefixMap) *regionsRouteSnapshot { out := ®ionsRouteSnapshot{steps: []RegionsRouteStep{}} sort.Slice(c.txs, func(i, j int) bool { return c.txs[i].id < c.txs[j].id }) - resolve := newRelayTokenResolver(pm) - memberships := newRegionsMembershipBuilder() + paths := newRegionsPathResolver(c.pathList, pm) + memberships := newRegionsMembershipBuilder(c.membershipCellBound, c.membershipNodeBound) rows := make(map[regionsRouteKey]map[int]regionsRouteCell) seen := make(map[regionsRouteKey]bool) seenMembers := make(map[string]bool) - cellCount := 0 + // Published rows and hourly cells share one estimated-payload budget, so a + // row cap can no longer truncate a snapshot that would fit (#508). + payload := c.stepBytes // Directed-step caps are counted only as step appearances: they never // reach membership coverage evidence. capStep := func(truncated *bool) { @@ -220,22 +510,19 @@ func (c *regionsRoutesCollector) build(pm *prefixMap) *regionsRouteSnapshot { clear(seen) clear(seenMembers) index := c.bucket(tx) - for _, path := range tx.paths { - var tokens []string - if err := json.Unmarshal([]byte(path.path), &tokens); err != nil { - c.addEvidence(tx, RegionsEvidenceCounts{Unprovable: path.observations}) + for _, ref := range tx.paths { + resolved := paths.at(ref.path) + if resolved.bad { + c.addEvidence(tx, RegionsEvidenceCounts{Unprovable: int(ref.observations)}) continue } - if len(tokens) > 64 { - c.addEvidence(tx, RegionsEvidenceCounts{Capped: path.observations}) + if resolved.over { + c.cappedOverlongPaths += int(ref.observations) + c.addEvidence(tx, RegionsEvidenceCounts{Capped: int(ref.observations)}) continue } previous := "" - for _, token := range tokens { - if len(resolve.cache) >= 4096 { - clear(resolve.cache) - } - key := resolve.key(token) + for _, key := range resolved.keys { if key == "" { previous = "" c.evidence.UnresolvedHops++ @@ -243,9 +530,9 @@ func (c *regionsRoutesCollector) build(pm *prefixMap) *regionsRouteSnapshot { } if !seenMembers[key] { seenMembers[key] = true - if !memberships.add(key, tx) { + if !memberships.add(key, tx.scope, index, tx.when) { // A dropped membership, not an observation: no step count. - c.cells.add(tx.scope, c.bucket(tx), RegionsEvidenceCounts{Capped: 1}) + c.cells.add(tx.scope, index, RegionsEvidenceCounts{Capped: 1}) } } // Latest includes the current hour. Completed hourly history and @@ -263,22 +550,27 @@ func (c *regionsRoutesCollector) build(pm *prefixMap) *regionsRouteSnapshot { } seen[step] = true byIndex := rows[step] + cell, exists := byIndex[index] + // A new row pays for its scaffolding and its first + // bucket; an existing bucket costs nothing more. + cost := 0 if byIndex == nil { - if len(rows) >= maxRegionsRouteSteps { - capStep(&out.stepsTruncated) - previous = key - continue - } - byIndex = make(map[int]regionsRouteCell) + cost = regionsRouteStepRowBytes + regionsRouteStepCellBytes + } else if !exists { + cost = regionsRouteStepCellBytes } - cell, exists := byIndex[index] - if !exists { - if cellCount >= maxRegionsRouteCells { + if cost > payload { + if byIndex == nil { + capStep(&out.stepsTruncated) + } else { capStep(&out.bucketsTruncated) - previous = key - continue } - cellCount++ + previous = key + continue + } + payload -= cost + if byIndex == nil { + byIndex = make(map[int]regionsRouteCell) } cell.count++ if tx.when.After(cell.last) { @@ -320,6 +612,7 @@ func (c *regionsRoutesCollector) build(pm *prefixMap) *regionsRouteSnapshot { } out.steps = append(out.steps, step) } + out.capture = c.captureStats(memberships) out.memberships, out.history = memberships.build(c.now, c.cells.build(c.start)) out.evidence = c.evidence // Directed-step completeness only. Unresolved hops break adjacency, so they @@ -375,10 +668,11 @@ func regionsRoutesSchemas() map[string]*openAPISchema { for key, value := range regionsHistorySchemas()["RegionsHistoryResponse"].Properties { properties[key] = value } - properties["steps"] = &openAPISchema{Type: "array", Items: openAPIRef("RegionsRouteStep")} + properties["steps"] = &openAPISchema{Type: "array", Items: openAPIRef("RegionsRouteStep"), + Description: "Directed step/scope rows, sorted by from, to and scope. Rows and their hourly buckets share one 16 MiB estimated-payload budget (at most 93206 rows or 223696 buckets), not a fixed row count; stepsTruncated and stepBucketsTruncated disclose when it was reached."} properties["stepNodes"] = &openAPISchema{Type: "array", Items: openAPIRef("RegionsRouteNode"), Description: "At most 10000 eligible visible endpoints, independent of legacy relay history scope attribution."} - properties["stepsTruncated"] = &openAPISchema{Type: "boolean"} - properties["stepBucketsTruncated"] = &openAPISchema{Type: "boolean"} + properties["stepsTruncated"] = &openAPISchema{Type: "boolean", Description: "A directed step/scope row was omitted by the published payload budget or the per-transmission pair cap."} + properties["stepBucketsTruncated"] = &openAPISchema{Type: "boolean", Description: "An hourly bucket of a published row was omitted by the published payload budget."} properties["stepEvidence"] = openAPIRef("RegionsStepEvidence") return map[string]*openAPISchema{ "RegionsRoutesResponse": {Type: "object", Properties: properties}, @@ -388,10 +682,10 @@ func regionsRoutesSchemas() map[string]*openAPISchema { "RegionsStepEvidence": {Type: "object", Description: "Snapshot-wide directed-step diagnostics, not selected-window totals or scope-transition evidence. Per-bucket membership evidence is coverage.evidence.", Properties: map[string]*openAPISchema{ "excludedObservations": {Type: "integer", Description: "Known DIRECT/unscoped or mismatched-transport-code observations."}, "unprovableObservations": {Type: "integer", Description: "Observations with unknown or malformed own/canonical evidence."}, - "cappedObservations": {Type: "integer", Description: "Observations omitted by source path limits."}, + "cappedObservations": {Type: "integer", Description: "Observations omitted by the per-hour capture budget, the per-transmission path cap or the path size bound."}, "unresolvedHops": {Type: "integer", Description: "Hop tokens without a unique relay identity; they break adjacency but not resolved memberships."}, "sourceTruncated": {Type: "boolean"}, - "cappedStepAppearances": {Type: "integer", Description: "Directed step appearances omitted by the step, per-transmission pair or cell caps. Never part of coverage.evidence."}, + "cappedStepAppearances": {Type: "integer", Description: "Directed step appearances omitted by the published payload budget or the per-transmission pair cap. Never part of coverage.evidence."}, "incomplete": {Type: "boolean", Description: "Directed steps may be missing: unprovable/capped observations, unresolved hops, step caps or missing endpoints."}, "missingEndpointSteps": {Type: "integer"}}}, } diff --git a/cmd/server/regions_routes_test.go b/cmd/server/regions_routes_test.go index 591769734..d78fcfdaf 100644 --- a/cmd/server/regions_routes_test.go +++ b/cmd/server/regions_routes_test.go @@ -3,8 +3,11 @@ package main import ( "encoding/json" "fmt" + "math/rand" "net/http/httptest" "reflect" + "runtime" + "strconv" "strings" "testing" "time" @@ -300,50 +303,69 @@ func TestRegionRoutesSourceAndRowBounds(t *testing.T) { a, b := strings.Repeat("ab", 32), strings.Repeat("cd", 32) path := `["` + a + `","` + b + `"]` tx := routeTransmission(1, now.Add(-time.Hour)) + // Distinct path strings that parse to the same two hops. The padding + // encodes i as JSON whitespace rather than as i spaces, so its length is + // logarithmic: a per-transmission ceiling anywhere near + // maxRegionsRoutePathBytes would otherwise make these paths oversize and + // refuse them for their byte count instead of for the ceiling under test. + pad := func(i int) string { + return strings.NewReplacer("0", " ", "1", "\t").Replace(strconv.FormatInt(int64(i), 2)) + } for i := 0; i < maxRegionsRoutePathsPerTx+1; i++ { - tx.Observations = append(tx.Observations, routeObservation(strings.Repeat(" ", i)+path, "14aabbccdd00")) + tx.Observations = append(tx.Observations, routeObservation(pad(i)+path, "14aabbccdd00")) } collector := newRegionsRoutesCollector(now) collector.capture(tx) - if collector.pathCount != maxRegionsRoutePathsPerTx || !collector.evidence.SourceTruncated { + if collector.pathRefs() != maxRegionsRoutePathsPerTx || !collector.evidence.SourceTruncated { t.Fatal("per-transmission source cap missing") } if got := collector.build(nil); len(got.steps) != 1 || got.steps[0].Buckets[0].Count != 1 || !got.evidence.Incomplete { t.Fatal("source bound altered per-transmission dedup") } + // Identical paths are interned once, so repeated relay chains do not spend + // the variant table (#508). The per-hour reference budget still counts + // every retained reference. + collector = newRegionsRoutesCollector(now) + for i := 0; i < 50; i++ { + collector.capture(routeTransmission(i+2, now.Add(-time.Hour), path)) + } + if len(collector.pathList) != 1 || collector.pathRefs() != 50 || collector.evidence.SourceTruncated { + t.Fatalf("path interning: variants=%d refs=%d", len(collector.pathList), collector.pathRefs()) + } + // The per-hour capture budget, exhausted by one hour of its own. collector = newRegionsRoutesCollector(now) - collector.pathCount = maxRegionsRoutePaths + collector.hourRefs[regionsHistoryBucketCount-1] = maxRegionsRoutePathRefsPerHour collector.capture(routeTransmission(2, now.Add(-time.Hour), path)) if len(collector.txs) != 0 || !collector.evidence.SourceTruncated { - t.Fatal("global path budget missing") + t.Fatal("per-hour path reference budget missing") } collector = newRegionsRoutesCollector(now) - for i := 0; i < maxRegionsRouteSteps+1; i++ { + // One row with its first bucket fits; the rest are budget truncation. + collector.stepBytes = regionsRouteStepRowBytes + regionsRouteStepCellBytes + for i := 0; i < 3; i++ { to := fmt.Sprintf("%064x", i+1) - collector.txs = append(collector.txs, regionsRouteTx{id: i + 1, when: now.Add(-time.Hour), scope: "#sample", paths: []regionsRoutePath{{path: `["` + a + `","` + to + `"]`, observations: 1}}}) + collector.capture(routeTransmission(i+1, now.Add(-time.Hour), `["`+a+`","`+to+`"]`)) } got := collector.build(nil) - if len(got.steps) != maxRegionsRouteSteps || !got.stepsTruncated || !got.evidence.Incomplete { - t.Fatal("step row budget missing") - } - for i, j := 0, len(collector.txs)-1; i < j; i, j = i+1, j-1 { - collector.txs[i], collector.txs[j] = collector.txs[j], collector.txs[i] - } - if other := collector.build(nil); !reflect.DeepEqual(got.steps, other.steps) { - t.Fatal("row selection depends on map/source encounter order") + if len(got.steps) != 1 || !got.stepsTruncated || !got.evidence.Incomplete || got.evidence.CappedStepAppearances != 2 { + t.Fatalf("step payload budget missing: steps=%d %+v", len(got.steps), got.evidence) } } -func TestRegionRoutesActualGlobalCellBound(t *testing.T) { +// The payload budget is spent exactly, on rows and hourly cells together, and +// its selection does not depend on map or source iteration order (#508). +func TestRegionRoutesActualStepPayloadBound(t *testing.T) { now := time.Date(2026, 10, 16, 12, 35, 0, 0, time.UTC) collector := newRegionsRoutesCollector(now) a := strings.Repeat("ab", 32) + // Two rows and five hourly cells between them; the budget is spent exactly. + collector.stepBytes = 2*regionsRouteStepRowBytes + 5*regionsRouteStepCellBytes id := 0 - for i := 0; i < 200; i++ { + for i := 0; i < 5; i++ { path := `["` + a + `","` + fmt.Sprintf("%064x", i+1) + `"]` - for hour := 0; hour < 501; hour++ { + for hour := 0; hour < 3; hour++ { id++ - collector.txs = append(collector.txs, regionsRouteTx{id: id, when: collector.start.Add(time.Duration(hour) * time.Hour), scope: "#sample", paths: []regionsRoutePath{{path: path, observations: 1}}}) + collector.capture(routeTransmission(id, collector.start.Add(time.Duration(hour)*time.Hour+time.Minute), path)) } } got := collector.build(nil) @@ -351,8 +373,17 @@ func TestRegionRoutesActualGlobalCellBound(t *testing.T) { for _, step := range got.steps { cells += len(step.Buckets) } - if cells != maxRegionsRouteCells || !got.bucketsTruncated || !got.evidence.Incomplete { - t.Fatalf("cell limit=%d truncated=%v", cells, got.bucketsTruncated) + if len(got.steps) != 2 || cells != 5 || !got.stepsTruncated || !got.bucketsTruncated || !got.evidence.Incomplete { + t.Fatalf("rows=%d cells=%d steps-truncated=%v buckets-truncated=%v", len(got.steps), cells, got.stepsTruncated, got.bucketsTruncated) + } + if spent := len(got.steps)*regionsRouteStepRowBytes + cells*regionsRouteStepCellBytes; spent != collector.stepBytes { + t.Fatalf("published payload estimate %d, budget %d", spent, collector.stepBytes) + } + for i, j := 0, len(collector.txs)-1; i < j; i, j = i+1, j-1 { + collector.txs[i], collector.txs[j] = collector.txs[j], collector.txs[i] + } + if other := collector.build(nil); !reflect.DeepEqual(got.steps, other.steps) { + t.Fatal("payload selection depends on map/source encounter order") } } @@ -491,13 +522,28 @@ func BenchmarkRegionsRoutesCapture30K2K(b *testing.B) { } } +// Worst-case published step payload, in the two shapes the byte budget admits +// (#508): one bucket per row, and few rows with many buckets. The reported +// payload-bytes validate the 180/75 per-row and per-bucket estimates the +// budget spends against real encoded JSON. func BenchmarkRegionsRoutesCapPayload(b *testing.B) { - response := RegionsRoutesResponse{Steps: make([]RegionsRouteStep, 5000), StepNodes: []RegionsRouteNode{}} + benchmarkRegionsRoutesCapPayload(b, 1) +} + +func BenchmarkRegionsRoutesCapPayloadCellHeavy(b *testing.B) { + benchmarkRegionsRoutesCapPayload(b, 504) +} + +func benchmarkRegionsRoutesCapPayload(b *testing.B, cellsPerRow int) { + rows := maxRegionsRouteStepBytes / (regionsRouteStepRowBytes + cellsPerRow*regionsRouteStepCellBytes) + response := RegionsRoutesResponse{Steps: make([]RegionsRouteStep, rows), StepNodes: []RegionsRouteNode{}} + cells := 0 for i := range response.Steps { - buckets := make([]RegionsHistoryBucket, 20) + buckets := make([]RegionsHistoryBucket, cellsPerRow) for j := range buckets { - buckets[j] = RegionsHistoryBucket{Index: 480 + j, Count: 12345, LastRelayed: "2026-10-16T11:59:59.123456789Z"} + buckets[j] = RegionsHistoryBucket{Index: j, Count: 12345, LastRelayed: "2026-10-16T11:59:59.123456789Z"} } + cells += len(buckets) response.Steps[i] = RegionsRouteStep{From: fmt.Sprintf("%064x", i), To: fmt.Sprintf("%064x", i+1), Scope: "#sample", Buckets: buckets} } encoded, err := json.Marshal(response) @@ -512,6 +558,41 @@ func BenchmarkRegionsRoutesCapPayload(b *testing.B) { } } b.ReportMetric(float64(len(encoded)), "payload-bytes") + b.ReportMetric(float64(rows*regionsRouteStepRowBytes+cells*regionsRouteStepCellBytes), "budgeted-bytes") + b.ReportMetric(float64(rows), "steps") +} + +// BenchmarkRegionsSnapshotRetainedMemory reports what the published Regions +// snapshot retains for the 30k-transmission / 2k-relay mixed fixture, plus the +// allocations the whole paired background recomputation makes and the heap it +// leaves behind, so #508's memory claim is measured rather than estimated. The +// timed loop is the diagnostic walk itself, which must stay cheap. +func BenchmarkRegionsSnapshotRetainedMemory(b *testing.B) { + srv := regionsRoutesBenchmarkFixtureWith(b, true) + var before, peak, after runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&before) + snapshot := srv.store.computeRepeaterRelaySnapshot(24) + runtime.ReadMemStats(&peak) + runtime.GC() + runtime.ReadMemStats(&after) + measured := measureRegionsSnapshot(snapshot) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + measureRegionsSnapshot(snapshot) + } + b.StopTimer() + runtime.KeepAlive(snapshot) + // Reported after the timed loop: ResetTimer discards custom metrics. + b.ReportMetric(float64(measured.TotalBytes), "regions-retained-bytes") + b.ReportMetric(float64(measured.MembershipsBytes), "membership-bytes") + b.ReportMetric(float64(measured.HistoryBytes), "history-bytes") + b.ReportMetric(float64(measured.StepsBytes), "step-bytes") + b.ReportMetric(float64(measured.EvidenceBytes), "evidence-bytes") + b.ReportMetric(float64(measured.EstimatedStepPayloadBytes), "step-payload-bytes") + b.ReportMetric(float64(peak.TotalAlloc-before.TotalAlloc), "recompute-alloc-bytes") + b.ReportMetric(float64(after.HeapAlloc)-float64(before.HeapAlloc), "paired-snapshot-heap-bytes") } func TestRegionsRoutesUnavailable(t *testing.T) { @@ -528,3 +609,166 @@ func TestRegionsRoutesUnavailable(t *testing.T) { t.Fatal("missing typed error response") } } + +// BenchmarkRegions168HBudgets measures the shape #508 targets: 168 h of +// retention at the per-hour rate real data showed at 48 h (about 5,200 +// retained path references per hour). It reports the retained snapshot size, +// the aggregation's own allocations, the transient capture state and whether +// any budget truncated, so the before/after claim is measured end to end. +// +// The two sub-benchmarks are the two diversity shapes that matter (#510 F2). +// "shared-paths" repeats 400 relay chains, which is the cheap case interning +// was built for. "distinct-paths" gives every transmission its own chain, 94 % +// of which is what e2e-fixture.db's real traffic looks like (376 distinct path +// strings in 399 path-carrying observations), and is the shape the round-1 +// global interning table capped. Capture order is shuffled in both, because +// production walks s.byPathHop by map iteration. +func BenchmarkRegions168HBudgets(b *testing.B) { + const hours, perHour, relays = 168, 5200, 400 + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + start := now.UTC().Truncate(time.Hour).Add(-regionsHistoryBucketCount * time.Hour) + keys := make([]string, relays) + for i := range keys { + keys[i] = fmt.Sprintf("%064x", i+1) + } + type item struct { + tx *StoreTx + when time.Time + } + build := func(distinct bool) []item { + items := make([]item, 0, hours*perHour) + id, n := 0, 0 + for hour := regionsHistoryBucketCount - hours; hour < regionsHistoryBucketCount; hour++ { + at := start.Add(time.Duration(hour)*time.Hour + time.Minute) + for i := 0; i < perHour; i++ { + id, n = id+1, n+1 + var path string + if distinct { + // A distinct 5-hop chain per transmission, drawn from a + // 20-relay pool so the published step rows stay inside the + // payload budget while the chains stay unique. + hops := make([]string, 5) + for h, d := 0, n; h < len(hops); h, d = h+1, d/20 { + hops[h] = keys[d%20] + } + path = `["` + strings.Join(hops, `","`) + `"]` + } else { + j := i % relays + path = `["` + keys[j] + `","` + keys[(j+1)%relays] + `"]` + } + items = append(items, item{routeTransmission(id, at, path), at}) + } + } + rand.New(rand.NewSource(1)).Shuffle(len(items), func(i, j int) { items[i], items[j] = items[j], items[i] }) + return items + } + for _, tc := range []struct { + name string + distinct bool + }{{"shared-paths", false}, {"distinct-paths", true}} { + b.Run(tc.name, func(b *testing.B) { + items := build(tc.distinct) + var measured *RegionsSnapshotMemory + var truncated, steps, cells int + var allocated uint64 + b.ReportAllocs() + b.ResetTimer() + for n := 0; n < b.N; n++ { + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + collector := newRegionsRoutesCollector(now) + for _, it := range items { + collector.captureAt(it.tx, it.when, true) + } + snapshot := collector.build(nil) + runtime.ReadMemStats(&after) + allocated = after.TotalAlloc - before.TotalAlloc + measured = measureRegionsSnapshot(&repeaterRelaySnapshot{generatedAt: now, + regionMemberships: snapshot.memberships, history: snapshot.history, routes: snapshot}) + steps, cells = len(snapshot.steps), measured.StepCells + truncated = 0 + if snapshot.evidence.Incomplete { + truncated = 1 + } + } + b.StopTimer() + b.ReportMetric(float64(measured.TotalBytes), "regions-retained-bytes") + b.ReportMetric(float64(measured.HistoryCells), "history-cells") + b.ReportMetric(float64(steps), "steps") + b.ReportMetric(float64(cells), "step-cells") + b.ReportMetric(float64(truncated), "incomplete") + b.ReportMetric(float64(allocated), "aggregate-alloc-bytes") + b.ReportMetric(float64(measured.Capture.PathVariants), "path-variants") + b.ReportMetric(float64(measured.Capture.MaxHourPathReferences), "max-hour-refs") + b.ReportMetric(float64(measured.Capture.EstimatedBytes), "capture-bytes") + }) + } +} + +// BenchmarkRegions48HFloodFanout measures the shape #510 round 3 targets: 48 h +// of retention — staging's configuration — where a fraction of transmissions +// are floods heard by every observer on the mesh over several relay chains +// each, with one repeat per observer. Staging carries 65 observers and +// reported ~1,080 capped observations with directed steps complete, which is +// what a per-transmission path ceiling below the observer fan-out looks like. +// The interning table is deliberately left at its real size, so it fills part +// way through exactly as it does on a 48 h capture. +func BenchmarkRegions48HFloodFanout(b *testing.B) { + const hours, perHour, observers, chains, quiet, pool = 48, 800, 65, 4, 6, 512 + now := time.Date(2026, 10, 17, 12, 35, 0, 0, time.UTC) + start := now.UTC().Truncate(time.Hour).Add(-regionsHistoryBucketCount * time.Hour) + type item struct { + tx *StoreTx + when time.Time + } + items := make([]item, 0, hours*perHour) + id, n := 0, 0 + for hour := regionsHistoryBucketCount - hours; hour < regionsHistoryBucketCount; hour++ { + at := start.Add(time.Duration(hour)*time.Hour + time.Minute) + for i := 0; i < perHour; i++ { + id, n = id+1, n+1 + if i%20 == 0 { // 5 % of transmissions reach the whole mesh + tx, _ := floodTransmission(id, at, observers, chains, pool) + items = append(items, item{tx, at}) + continue + } + paths := make([]string, quiet) + for p := range paths { + paths[p] = fmt.Sprintf(`["%064x","%064x","%064x"]`, n%pool+1, (n*quiet+p)%pool+1, pool+n%observers+1) + } + items = append(items, item{routeTransmission(id, at, paths...), at}) + } + } + rand.New(rand.NewSource(1)).Shuffle(len(items), func(i, j int) { items[i], items[j] = items[j], items[i] }) + var measured *RegionsSnapshotMemory + var steps, capped int + var allocated uint64 + b.ReportAllocs() + b.ResetTimer() + for k := 0; k < b.N; k++ { + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + collector := newRegionsRoutesCollector(now) + for _, it := range items { + collector.captureAt(it.tx, it.when, true) + } + snapshot := collector.build(nil) + runtime.ReadMemStats(&after) + allocated = after.TotalAlloc - before.TotalAlloc + measured = measureRegionsSnapshot(&repeaterRelaySnapshot{generatedAt: now, + regionMemberships: snapshot.memberships, history: snapshot.history, routes: snapshot}) + steps, capped = len(snapshot.steps), snapshot.evidence.CappedObservations + } + b.StopTimer() + c := measured.Capture + b.ReportMetric(float64(capped), "capped-observations") + b.ReportMetric(float64(c.CappedTransmissionPaths), "capped-tx-paths") + b.ReportMetric(float64(c.MaxTransmissionPaths), "max-tx-paths") + b.ReportMetric(float64(c.PathVariants), "path-variants") + b.ReportMetric(float64(c.PathVariantsInterned), "path-variants-interned") + b.ReportMetric(float64(c.MaxHourPathReferences), "max-hour-refs") + b.ReportMetric(float64(c.EstimatedBytes), "capture-bytes") + b.ReportMetric(float64(measured.TotalBytes), "regions-retained-bytes") + b.ReportMetric(float64(steps), "steps") + b.ReportMetric(float64(allocated), "aggregate-alloc-bytes") +} diff --git a/cmd/server/routes.go b/cmd/server/routes.go index be8b14d24..fbfb44863 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -27,7 +27,7 @@ import ( // memBreakdownNote is the static accounting caveat attached to the opt-in // /api/perf?mem=1 store memory breakdown (PerfResponse.MemoryBreakdownNote). -const memBreakdownNote = "Per-component *MB count string content + one Go string header each and are a deliberate upper bound (the header is also in the *EstimatedMB base figures). struct/index/map overhead, the neighbor graph and analytics caches are excluded, so the totals sit below goHeapInuseMB. floodTxSharePct sizes the flood-forward multiplication: each flood hop is stored as its own transmission." +const memBreakdownNote = "Per-component *MB count string content + one Go string header each and are a deliberate upper bound (the header is also in the *EstimatedMB base figures). struct/index/map overhead, the neighbor graph and analytics caches are excluded, so the totals sit below goHeapInuseMB. floodTxSharePct sizes the flood-forward multiplication: each flood hop is stored as its own transmission. regionsSnapshot is accounted separately and is not part of these totals: it sizes the retained Regions route/membership/history/evidence snapshot (published struct sizes, owned string content and a flat per-map-entry allowance), not the transient aggregation that builds it." // Server holds shared state for route handlers. type Server struct { @@ -60,6 +60,11 @@ type Server struct { // Admits one /api/perf?mem=1 store scan at a time plus a cooldown (#371). perfMem perfMemLimiter + // Test-only override for the published step-endpoint ceiling (#510), so + // the ceiling and its missing-endpoint disclosure can be exercised without + // a 10,001-node fixture. Zero means maxRegionsRouteEndpoints. + stepEndpointLimit int + // Throttles the log of failed GET /api/nodes/{pubkey} 404 lookups (#208). missingNodeLog missingNodeLookupLog @@ -1254,11 +1259,15 @@ func (s *Server) handlePerf(w http.ResponseWriter, r *http.Request) { // Released as soon as the walk ends: the cooldown runs from scan // completion, and a client slow to read the response cannot hold the slot. var memBreakdown *StoreMemoryBreakdown + var regionsSnapshot *RegionsSnapshotMemory var breakdownNote string if releaseMem != nil { memBreakdown = s.store.GetStoreMemoryBreakdown() breakdownNote = memBreakdownNote releaseMem() + // Sizing the published Regions snapshot is O(published cells) and takes + // no packet lock, so it runs after the gated store walk is released. + regionsSnapshot = s.store.GetRegionsSnapshotMemory() } // SQLite stats @@ -1279,6 +1288,7 @@ func (s *Server) handlePerf(w http.ResponseWriter, r *http.Request) { Sqlite: sqliteStats, MemoryBreakdown: memBreakdown, MemoryBreakdownNote: breakdownNote, + RegionsSnapshot: regionsSnapshot, GoRuntime: func() *GoRuntimeStats { ms := s.getMemStats() return &GoRuntimeStats{ diff --git a/cmd/server/types.go b/cmd/server/types.go index 798a6939f..87626221e 100644 --- a/cmd/server/types.go +++ b/cmd/server/types.go @@ -741,6 +741,11 @@ type PerfResponse struct { // MemoryBreakdownNote documents the accounting scope of MemoryBreakdown. // It lives here (one occurrence) rather than repeating in every breakdown. MemoryBreakdownNote string `json:"memoryBreakdownNote,omitempty"` + // RegionsSnapshot is populated only for /api/perf?mem=1, alongside + // MemoryBreakdown: the retained size of the published Regions route, + // membership, hourly history and per-bucket evidence snapshot (#508). nil + // before the first background relay snapshot is published. + RegionsSnapshot *RegionsSnapshotMemory `json:"regionsSnapshot,omitempty"` } // StoreMemoryBreakdown is the opt-in /api/perf?mem=1 diagnostic: the diff --git a/docs/api-spec.md b/docs/api-spec.md index b1e24acf5..69a64d2f6 100644 --- a/docs/api-spec.md +++ b/docs/api-spec.md @@ -352,7 +352,36 @@ Detailed performance metrics per endpoint. | Param | Type | Description | |-------|------|-------------| -| `mem` | string | `1` adds `memoryBreakdown` and `memoryBreakdownNote`: a walk of every stored transmission and observation under the packet-store read lock. Ignored when the server has no packet store. | +| `mem` | string | `1` adds `memoryBreakdown` and `memoryBreakdownNote`: a walk of every stored transmission and observation under the packet-store read lock. It also adds `regionsSnapshot`. Ignored when the server has no packet store. | + +`regionsSnapshot` sizes the retained Regions route, membership, hourly history +and per-bucket evidence snapshot published with the background relay snapshot +(#508): published counts, per-part and total retained bytes, `totalMB`, and +`estimatedStepPayloadBytes` against `stepPayloadBudgetBytes` so the directed-step +payload budget can be read against what a snapshot actually used. `truncated` is +budget truncation only — the hourly-history and directed-step publication flags +plus the capture budget's own refusal — and `evidenceIncomplete` carries the +snapshot's evidence qualification separately, because unresolved hops and +unprovable observations raise that on ordinary data. `capture` reports the +transient capture state's fill against its own budgets (#510): retained paths +and how many are interned (`pathVariants`, `pathVariantsInterned`, +`pathVariantInternBudget`), retained references with the per-hour high-water +mark (`pathReferences`, `maxHourPathReferences`, `hourPathReferenceBudget`), +retained transmissions, aggregated membership cells and identities against +their budgets, `estimatedBytes` and `sourceTruncated`. It also reports the +per-transmission and per-path fill a flood packet reaches first +(`maxTransmissionPaths`, `transmissionPathBudget`, `maxPathBytes`, +`pathByteBudget`) and attributes every capped observation to the ceiling that +refused it: `cappedPathBytes`, `cappedTransmissionPaths`, +`cappedHourReferences` and `cappedOverlongPaths`, which together account for +the snapshot's `cappedObservations`, plus `cappedMembershipCells` and +`cappedMembershipNodes`, which are dropped memberships rather than +observations. The walk is O(published cells), +touches no packet state and takes no packet lock, so it is not part of the +admission-controlled store scan; it is absent before the first background relay +snapshot. Byte figures are struct sizes plus owned string content plus a flat +per-map-entry allowance — an estimate of retained heap, not an allocator +measurement, and not the transient aggregation that builds the snapshot. `mem=1` is admission-controlled per server process (#371): one scan at a time, and none for 5 s after a scan completes. A refused request gets diff --git a/docs/regions-observed.md b/docs/regions-observed.md index a7ce2c785..ebb1cfffb 100644 --- a/docs/regions-observed.md +++ b/docs/regions-observed.md @@ -244,19 +244,98 @@ scope, step and populated-hour limits must be disclosed as incomplete evidence. No database writes, schema changes, retention extension, dependencies or new operator configuration are required. -The route collector retains at most 250,000 path variants globally and 128 per -transmission, with an 8,192-byte input-path safety bound. The capture includes -the current hour for Latest, but completed-hour history and steps exclude it. -Membership aggregation has independent budgets of 1,000,000 transmission/node -entries and 10,000 identities; step limits cannot suppress a valid singleton -membership. Each identity retains its 32 newest scope names. Source and -membership limits are counted as `capped` in the affected transmission's hour -and scope; the directed-step, per-transmission pair and cell caps are counted -separately as `stepEvidence.cappedStepAppearances` and never affect -memberships or history coverage. Capped capture is a -retained subset, not a ranking of all possible evidence. Publication retains -at most 5,000 directed step/scope rows, 100,000 populated hourly cells and 10,000 -endpoint identities. Endpoint metadata is independent of relay-history +Capture retains at most 65,536 observation-path references **per hourly +bucket** and 128 per transmission, with an 8,192-byte input-path safety bound. +A per-hour budget replaces the single global 250,000-variant budget, which 48 h +of real traffic exhausted and which then reported `capped` evidence for every +later hour and withheld the Health and Timeline rates (#508). The per-hour +ceiling makes retained capture scale with retention and stops one busy hour +from starving the rest of the horizon. 65,536 is roughly four times the ~16,800 +observations per hour that 48 h of real traffic averaged across all scopes, so +an hour has headroom even if every observation were transport-scoped and +carried a distinct path. The capture includes the current hour for Latest, but +completed-hour history and steps exclude it. + +Identical paths are interned, so one relay chain observed by thousands of +transmissions is parsed and identity-resolved once per snapshot. The interning +table (131,072 entries) is a bounded dedup cache, not a budget: past it a path +is retained **un-interned** — it costs a path slot and a reference instead of a +table entry — and no observation is ever refused for lack of table space. A +table that refused new paths when full would reintroduce exactly the starvation +#508 reported, at a lower ceiling, because real path diversity is high: +`e2e-fixture.db`, real captured traffic, holds 376 distinct path strings in 399 +path-carrying observations. + +### Transient capture memory + +The capture state is transient — it lives only while a background snapshot is +being built — but it is the part that decides whether evidence is reported as +`capped`, so its cost is disclosed rather than implied. Per retained row it is +8 bytes per observation-path reference, 24 for a retained path (a string header +plus one resolver slot; path content is shared with the packet store, not +copied), 48 more for a path that got an interning-table entry, and 72 per +retained transmission header. The store's own memory estimate +(`estimateStoreTxBytesTypical`, the figure `packetStore.maxMemoryMB` divides by) +assigns 3,214 bytes to a transmission with one observation and 8,506 to one with +ten. Capture therefore retains roughly **3–5 % of the store state for the rows +it mirrors**, and it cannot mirror rows the store does not hold, so the store +budget binds first on every host. + +The per-hour budget's own nominal ceiling, if every hourly bucket in the 504 h +horizon were filled to it, is 33.1 M references: about 253 MiB of references, +up to 758 MiB of path slots if every reference carried a distinct path, 6 MiB +of interning table and up to 2.22 GiB of transmission headers if each reference +belonged to its own single-observation transmission — about 3.2 GiB in total. +That ceiling is unreachable in practice, because reaching it needs the packet +store to hold 33 M transport-scoped observations across 505 distinct hours, +which is on the order of a hundred gigabytes of store by the estimate above and +orders of magnitude past any `packetStore.maxMemoryMB` a CoreScope host runs +with. That is why the store is the operative bound and the per-hour budget is a +fairness and pathology net rather than a memory budget. At the rate real data +showed — 168 h at ~16,800 observations per hour, every one transport-scoped +with a distinct path — the same arithmetic gives about 300 MiB of capture state, +against several gigabytes of store for the same rows. A measured 168 h +synthetic snapshot at 5,200 references per hour with a distinct 5-hop chain per +transmission retains 97 MiB of capture state and caps nothing. +`/api/perf?mem=1` reports the fill of every capture budget as +`regionsSnapshot.capture`, so the headroom is observable per snapshot instead of +inferred from these figures. It also attributes every capped observation to the +ceiling that refused it — `cappedPathBytes`, `cappedTransmissionPaths`, +`cappedHourReferences` and `cappedOverlongPaths`, which together account for +the snapshot's `cappedObservations`, with `cappedMembershipCells` and +`cappedMembershipNodes` for dropped memberships — so a withheld rate names its +own cause. + +One transmission retains at most 4,096 distinct observation paths. That is a +share of its hour's reference budget (a sixteenth of it), not a memory bound: +every retained path already costs one of that hour's references. The retired +value was 128, which a flood packet reaches on an ordinary mesh — a 65-observer +staging instance offers hundreds of distinct relay chains for one flood — and +the surplus was reported `capped`, which withheld the scope's rates. Repeats of +one chain inside a transmission share a single path, reference and budget unit +whether or not the chain is in the interning table. +`regionsSnapshot.capture.maxTransmissionPaths` reports the high-water mark +against it. The per-transmission directed-step pair cap (8,192) is independent +of it rather than, as before, exactly 128 × 64 hops; it is a publication +budget and is counted as `cappedStepAppearances` only. + +Memberships are aggregated straight into (relay, scope, hour) cells, so the +budget — 500,000 cells and 10,000 identities — grows with populated hours +rather than with observations; nothing per transmission is retained. Step +limits cannot suppress a valid singleton membership. Each identity retains its +32 newest scope names. Source and membership limits are counted as `capped` in +the affected transmission's hour and scope; the published-payload budget and +the per-transmission pair cap are counted separately as +`stepEvidence.cappedStepAppearances` and never affect memberships or history +coverage. Capped capture is a retained subset, not a ranking of all possible +evidence. Published directed steps share one 16 MiB estimated-payload budget +across rows and hourly buckets (at most 93,206 rows or 223,696 buckets, from +180 estimated bytes per row and 75 per bucket) instead of the fixed 5,000-row +and 100,000-cell caps, which real data also reached; `/api/perf?mem=1` reports +what a snapshot actually spent as `regionsSnapshot.estimatedStepPayloadBytes` +against `stepPayloadBudgetBytes`. Publication also retains at most 10,000 +endpoint identities; steps whose endpoints are omitted are reported as +`stepEvidence.missingEndpointSteps`. Endpoint metadata is independent of relay-history membership, so a valid flood observation of a canonically direct transmission can still be shown. The UI retains at most 1,000 shared scope pairs, paginates both tables in 25-row pages and maps at most 200 positioned step rows. A capped @@ -266,6 +345,24 @@ hops, capped step appearances and missing endpoints) are not selected-interval traffic totals. Complete retained history does not imply complete radio reception or that all paths can be attributed. +The published snapshot's retained size is an operator diagnostic rather than a +guess: `/api/perf?mem=1` reports `regionsSnapshot` with published counts and +per-part retained bytes for memberships, hourly history, directed steps and +per-bucket evidence. Publishing this snapshot raised staging container RSS from +about 1.55 to about 1.72 GiB, and production hosts are smaller, so the figure +is read against the host rather than against the budgets alone. +`regionsSnapshot.capture` adds the transient capture state's fill against its +own budgets — retained paths and how many of them are interned against the +table size, retained references with the per-hour high-water mark against the +per-hour budget, retained transmissions, aggregated membership cells and +identities against their budgets, and an estimate of the capture state's bytes +— because those are the budgets that decide `capped`, and publication sizes +alone do not show how close they are. `regionsSnapshot.truncated` is budget +truncation only (the publication flags plus the capture budget's own refusal); +`regionsSnapshot.evidenceIncomplete` carries the snapshot's evidence +qualification separately, because unresolved hops and unprovable observations +raise that on ordinary real data. + Future work can add independently evidenced scope transitions and comparisons of observed routes. A change in timeline relay membership alone does not establish a changed route. diff --git a/public/regions-routes.js b/public/regions-routes.js index e1192ac68..879e2bade 100644 --- a/public/regions-routes.js +++ b/public/regions-routes.js @@ -3,7 +3,13 @@ 'use strict'; (function (root) { const historyTools = typeof module !== 'undefined' && module.exports ? require('./regions-history.js') : root.RegionsHistory; - const MAX_STEPS = 5000, MAX_CELLS = 100000, MAX_PAIRS = 1000, PAGE_SIZE = 25, MAP_LIMIT = 200; + // Defensive client bounds mirroring the server's directed-step payload + // budget (#508): rows and hourly buckets share one 16 MiB estimate instead + // of a fixed 5,000-row / 100,000-cell cap, which real data reached at 48 h + // retention. MAX_STEP_NODES mirrors the server's endpoint ceiling. + const STEP_ROW_BYTES = 180, STEP_CELL_BYTES = 75, MAX_STEP_BYTES = 16 * 1024 * 1024; + const MAX_STEPS = Math.floor(MAX_STEP_BYTES / STEP_ROW_BYTES), MAX_CELLS = Math.floor(MAX_STEP_BYTES / STEP_CELL_BYTES); + const MAX_STEP_NODES = 10000, MAX_PAIRS = 1000, PAGE_SIZE = 25, MAP_LIMIT = 200; const keyPattern = /^[0-9a-f]{64}$/i; function indexRoutes(data, history, validScope, validPosition) { @@ -16,8 +22,8 @@ if (count != null && (!Number.isSafeInteger(count) || count < 0 || count > 0)) limited = true; } if (Array.isArray(data.stepNodes)) { - if (data.stepNodes.length > MAX_STEPS * 2) limited = true; - for (const node of data.stepNodes.slice(0, MAX_STEPS * 2)) { + if (data.stepNodes.length > MAX_STEP_NODES) limited = true; + for (const node of data.stepNodes.slice(0, MAX_STEP_NODES)) { if (!node || !keyPattern.test(node.publicKey) || !['repeater', 'room'].includes(node.role)) { limited = true; continue; } const positioned = node.positionSource === 'reported' && typeof validPosition === 'function' && validPosition(node); nodes.set(node.publicKey.toLowerCase(), { publicKey: node.publicKey.toLowerCase(), name: String(node.name || node.publicKey.slice(0, 8)), role: node.role, @@ -295,7 +301,7 @@ return { panel, update, renderMap, fitMap, clear, clearMap }; } - const api = { indexRoutes, selectRoutes, createPanel, PAGE_SIZE, MAP_LIMIT }; + const api = { indexRoutes, selectRoutes, createPanel, PAGE_SIZE, MAP_LIMIT, MAX_STEPS, MAX_CELLS, MAX_STEP_NODES }; if (typeof module !== 'undefined' && module.exports) module.exports = api; root.RegionsRoutes = api; })(typeof window === 'undefined' ? globalThis : window); diff --git a/test-regions-map.js b/test-regions-map.js index 355798c1a..22638ee96 100644 --- a/test-regions-map.js +++ b/test-regions-map.js @@ -420,6 +420,16 @@ assert.equal(fullConnections.mapSteps.length, 200, 'the map has a separate bound assert.equal(fullRoutes.steps.reduce((sum, step) => sum + step.indices.length, 0), 100000); const cappedSteps = routes.indexRoutes({ steps: [...populatedSteps, populatedSteps[0]], stepEvidence: routeData.stepEvidence }, cliqueHistory, regions.validScope); assert.equal(cappedSteps.limited, true, 'row caps cannot silently imply complete evidence'); +// #508: rows and hourly buckets share one estimated-payload budget, so the +// client no longer discards a snapshot the server published beyond 5,000 rows. +assert.equal(routes.MAX_STEPS, Math.floor(16 * 1024 * 1024 / 180), 'client row bound mirrors the server payload budget'); +assert.equal(routes.MAX_CELLS, Math.floor(16 * 1024 * 1024 / 75), 'client cell bound mirrors the server payload budget'); +assert.equal(routes.MAX_STEP_NODES, 10000, 'client endpoint bound mirrors the server endpoint ceiling'); +const wideSteps = Array.from({ length: 6000 }, (_, index) => ({ from: (index % 5000).toString(16).padStart(64, '0'), + to: ((index + 1) % 5000).toString(16).padStart(64, '0'), scope: '#scope' + Math.floor(index / 5000), buckets: [bucket(503)] })); +const wideRoutes = routes.indexRoutes({ steps: wideSteps, stepEvidence: routeData.stepEvidence }, cliqueHistory, regions.validScope); +assert.equal(wideRoutes.steps.length, 6000, 'more than 5,000 published directed step rows are retained'); +assert.equal(wideRoutes.limited, false, 'a snapshot inside the payload budget is not reported as limited'); const diverseNodes = Array.from({ length: 80 }, (_, i) => historical((10000 + i).toString(16).padStart(64, '0'), [], { scopes: Array.from({ length: 32 }, (_, j) => ({ name: '#group' + i + 's' + j, buckets: [bucket(503)] })), }));