From fc7e12a113dbd862baea5e81b2110ac6d660bbe2 Mon Sep 17 00:00:00 2001 From: dborup Date: Mon, 28 Sep 2026 17:23:53 +0200 Subject: [PATCH 1/2] test(anomaly): guard the fixture rule's search work, both ways Follow-up to PR #108, which raised the fixture rule's alternating shape from 72.96 to 103.0 gaps visited per pulse (+41%) without any guard seeing it: TestPeriodicSearchWorkIsBounded ran only the max and worst rules, and only checked the upper side. - The guard now also runs the fixture rule that BenchmarkPeriodicSearch uses for its "fixture/..." cases; measuredSteps is keyed by rule name and shape (the fixture and max rules share MinCoverage 0.6, so the old shape/MinCoverage key could not tell them apart). - measuredSteps holds master's actual totals (e.g. worst/jitter 6,242,089 -> 6,434,483 after #108). - The 25% margin is two-sided. More work fails as before; markedly fewer gaps visited fails too, with a message that it means lost search coverage and that a deliberate optimization updates measuredSteps. Reintroducing the seed's tolerance in chainRefined now fails here (fixture/alternating 73,761 < 104,271 - 25%). - The comment states that the counters count visited gaps only: removing the periodRange memo leaves them unchanged, and TestRangeMemoIsNeverStale covers the memo. Co-Authored-By: Claude Opus 5.5 --- internal/anomaly/periodic_bench_test.go | 67 +++++++++++++++++-------- 1 file changed, 46 insertions(+), 21 deletions(-) diff --git a/internal/anomaly/periodic_bench_test.go b/internal/anomaly/periodic_bench_test.go index 0547d8bd0..803133fb2 100644 --- a/internal/anomaly/periodic_bench_test.go +++ b/internal/anomaly/periodic_bench_test.go @@ -116,28 +116,50 @@ func periodicWorkBound(r *PeriodicRule) (chains, steps uint64) { return chains, chains*uint64(r.HistoryLen-1) + seeds*recentGaps } +// workGuardRules are the rules TestPeriodicSearchWorkIsBounded runs: the +// fixture rule of BenchmarkPeriodicSearch's "fixture/..." cases, and the +// largest allowed ring and MaxMissing ("max"), also where no chain ever meets +// the criteria ("worst"). +var workGuardRules = []struct { + name string + rule PeriodicRule +}{ + {"fixture", experimentalPeriodic()}, + {"max", maxPeriodicRule()}, + {"worst", worstPeriodicRule()}, +} + // measuredSteps is the total number of gaps visited by the search over the // deterministic runs of TestPeriodicSearchWorkIsBounded (4*maxHistoryLen -// pulses per shape). The test allows 25% above it, so a change that makes -// the search visit markedly more chains or gaps fails here even though it -// stays within the loose analytic periodicWorkBound. Work these counters do -// not count (more CPU per visited gap, say) is not caught; only -// BenchmarkPeriodicSearch measures it. +// pulses per rule and shape). The test allows 25% either way. More means the +// search visits markedly more chains or gaps, even though it may stay within +// the loose analytic periodicWorkBound. Less means it visits markedly fewer: +// lost search coverage (a range or seed dropped too early, say), unless it +// is a deliberate optimization, which then updates these values. +// +// The counters count visited gaps only. Work per visited gap is not caught: +// removing the periodRange memo, for example, leaves every count unchanged +// (TestRangeMemoIsNeverStale covers the memo), and only +// BenchmarkPeriodicSearch measures the CPU. var measuredSteps = map[string]uint64{ - "periodic/0.6": 236631, "alternating/0.6": 1466800, "jitter/0.6": 247574, - "missing/0.6": 243414, "bursty/0.6": 199730, "noise/0.6": 148177, - "periodic/1": 236631, "alternating/1": 1466800, "jitter/1": 6242089, - "missing/1": 1174373, "bursty/1": 199730, "noise/1": 148177, + "fixture/periodic": 33289, "fixture/alternating": 104271, "fixture/jitter": 54979, + "fixture/missing": 34850, "fixture/bursty": 82025, "fixture/noise": 67588, + "max/periodic": 236631, "max/alternating": 1466835, "max/jitter": 247953, + "max/missing": 243414, "max/bursty": 199882, "max/noise": 147811, + "worst/periodic": 236631, "worst/alternating": 1466835, "worst/jitter": 6434483, + "worst/missing": 1174373, "worst/bursty": 199882, "worst/noise": 147811, } -// Every single pulse stays within periodicWorkBound, at the largest allowed -// ring and MaxMissing, for every traffic shape, including the configuration -// where no chain ever meets the criteria and the search runs on every pulse; -// and the total work stays within 25% of measuredSteps. +// Every single pulse stays within periodicWorkBound, for every guard rule and +// traffic shape, including the configuration where no chain ever meets the +// criteria and the search runs on every pulse; and the total work stays +// within 25% of measuredSteps, either way. func TestPeriodicSearchWorkIsBounded(t *testing.T) { - for _, r := range []PeriodicRule{maxPeriodicRule(), worstPeriodicRule()} { + for _, g := range workGuardRules { + r := g.rule maxChains, maxSteps := periodicWorkBound(&r) for _, shape := range benchShapes { + key := g.name + "/" + shape f := newPulseFeeder([]PeriodicRule{r}, shape) var peakC, peakS uint64 for i := 0; i < 4*maxHistoryLen; i++ { @@ -145,17 +167,20 @@ func TestPeriodicSearchWorkIsBounded(t *testing.T) { f.feed(1) dc, ds := f.sc.chains-c0, f.sc.steps-s0 if dc > maxChains || ds > maxSteps { - t.Fatalf("%s MinCoverage=%g: pulse %d did %d chains and %d steps, bound %d and %d", - shape, r.MinCoverage, i, dc, ds, maxChains, maxSteps) + t.Fatalf("%s: pulse %d did %d chains and %d steps, bound %d and %d", key, i, dc, ds, maxChains, maxSteps) } peakC, peakS = max(peakC, dc), max(peakS, ds) } - key := fmt.Sprintf("%s/%g", shape, r.MinCoverage) - if m, ok := measuredSteps[key]; !ok || f.sc.steps > m+m/4 { - t.Errorf("%s: %d steps in total, measured %d (+25%% allowed)", key, f.sc.steps, m) + switch m, ok := measuredSteps[key]; { + case !ok: + t.Errorf("%s: %d steps in total, no measuredSteps entry", key, f.sc.steps) + case f.sc.steps > m+m/4: + t.Errorf("%s: %d steps in total, more than measured %d +25%%: the search does markedly more work", key, f.sc.steps, m) + case f.sc.steps < m-m/4: + t.Errorf("%s: %d steps in total, less than measured %d -25%%: the search visits markedly fewer gaps, which means lost search coverage; if this is a deliberate optimization, update measuredSteps", key, f.sc.steps, m) } - t.Logf("%-11s MinCoverage=%g: peak %d chains (bound %d), %d steps (bound %d), total %d steps", - shape, r.MinCoverage, peakC, maxChains, peakS, maxSteps, f.sc.steps) + t.Logf("%-19s peak %d chains (bound %d), %d steps (bound %d), total %d steps", + key, peakC, maxChains, peakS, maxSteps, f.sc.steps) } } } From fee2964ca37f4972d85fc1371dcb86bdbcc01477 Mon Sep 17 00:00:00 2001 From: dborup Date: Mon, 28 Sep 2026 17:38:30 +0200 Subject: [PATCH 2/2] test(anomaly): measure BytesPerStream once per sub-benchmark so -bench . finishes BenchmarkBytesPerStream ran its whole measurement (a new Detector, 4*n events, two forced GCs) inside the b.N loop with the timer stopped. The timed work was ~0, so the framework kept raising b.N towards 1e9 and repeated the measurement b.N times: `go test -run '^$' -bench . ./...` never finished (on 18a13264, aborted after 13.5 min, still in the first sub-benchmark; -timeout does not cover benchmarks). This predates #108. Each sub-benchmark now takes one heap measurement (bytesPerStream) outside the b.N loop, reuses it when the framework calls it again with a larger b.N, runs an empty b.N loop and reports B/stream with b.ReportMetric as before (ns/op is meaningless here). b.Loop is not used: the module targets Go 1.22. B/stream is unchanged (760.7, 1345, 747.9, 1332, 739.0, 1323 for 1k, 10k and 100k streams without and with periodicity, the same as master with -benchtime=1x); the full `-bench . -count=1` now completes in 63.5s. Co-Authored-By: Claude Opus 5.5 --- internal/anomaly/bench_test.go | 57 ++++++++++++++++++++++------------ 1 file changed, 37 insertions(+), 20 deletions(-) diff --git a/internal/anomaly/bench_test.go b/internal/anomaly/bench_test.go index abb834ee3..14daf574b 100644 --- a/internal/anomaly/bench_test.go +++ b/internal/anomaly/bench_test.go @@ -89,36 +89,53 @@ func heapInUse() uint64 { // BenchmarkBytesPerStream reports steady-state heap per stream with full // 1/5/15/60-minute windows (4 entries each) and optionally periodicity. +// +// B/stream is one heap measurement per sub-benchmark, taken outside the b.N +// loop and reused when the framework calls the sub-benchmark again with a +// larger b.N; nothing is timed, so ns/op means nothing here. (Measuring inside +// the b.N loop with the timer stopped never finished: the timed work was ~0, +// so the framework kept raising b.N and repeated the whole measurement b.N +// times.) func BenchmarkBytesPerStream(b *testing.B) { for _, n := range []int{1_000, 10_000, 100_000} { for _, per := range []bool{false, true} { + var perStream float64 + measured := false b.Run(fmt.Sprintf("streams=%d/periodic=%v", n, per), func(b *testing.B) { - for it := 0; it < b.N; it++ { - b.StopTimer() - before := heapInUse() - d, _ := New(benchConfig(2*n, per, false)) - clock := t0 - id := 0 - for r := 0; r < 4; r++ { - for s := 0; s < n; s++ { - clock = clock.Add(time.Millisecond) - d.Observe(streamEvent(id, s, clock)) - id++ - } - } - // the dedup set is bounded separately (MaxDedupIDs); drop it so - // only per-stream state is measured - d.dedup, d.dedupLog, d.dedupHead = map[TxID]int64{}, nil, 0 - after := heapInUse() - b.ReportMetric(float64(after-before)/float64(n), "B/stream") - runtime.KeepAlive(d) - b.StartTimer() + b.StopTimer() + if !measured { + perStream, measured = bytesPerStream(n, per), true } + for range b.N { + } + b.ReportMetric(perStream, "B/stream") }) } } } +// bytesPerStream is the heap growth per stream of a Detector that has seen +// four events on each of n streams, without its dedup set. +func bytesPerStream(n int, per bool) float64 { + before := heapInUse() + d, _ := New(benchConfig(2*n, per, false)) + clock := t0 + id := 0 + for r := 0; r < 4; r++ { + for s := 0; s < n; s++ { + clock = clock.Add(time.Millisecond) + d.Observe(streamEvent(id, s, clock)) + id++ + } + } + // the dedup set is bounded separately (MaxDedupIDs); drop it so only + // per-stream state is measured + d.dedup, d.dedupLog, d.dedupHead = map[TxID]int64{}, nil, 0 + after := heapInUse() + runtime.KeepAlive(d) + return float64(after-before) / float64(n) +} + // zipf picks streams with a skewed distribution. func zipfStreams(n, count int, seed int64) []int { r := rand.New(rand.NewSource(seed))