perf(detection): charge the scan budget per pass and cut monitor overhead

guardcore/performance_monitor.go

@@ -527,10 +527,13 @@ func (m *PerformanceMonitor) sendAnomalyEvent(anomaly *Anomaly, agent AnomalyEve
 // exported BuildCallbackErrorEventData via sinks that observe callback
 // failures.
 func (m *PerformanceMonitor) notifyCallbacks(anomaly Anomaly, agent AnomalyEventSender, correlationID string) {
-	safe := SanitizeAnomalyData(anomaly)
 	m.mu.Lock()
 	callbacks := append([]func(map[string]any){}, m.anomalyCallbacks...)
 	m.mu.Unlock()
+	if len(callbacks) == 0 {
+		return
+	}
+	safe := SanitizeAnomalyData(anomaly)
 	for _, callback := range callbacks {
 		runAnomalyCallback(callback, safe, agent, correlationID)
 	}

guardcore/scanbudget.go

@@ -24,52 +24,75 @@ type detectionScanBudget struct {
 	depthWarned bool
 
 	// The time budget (detection_max_scan_seconds): spent is the summed
-	// scan-clock time of every finished pass, passStart the clock reading
-	// when the current pass began.
+	// scan-clock time charged so far, passMark the scan-clock reading at
+	// the last charge. The CPU clock is read when a pass opens, at each
+	// scanSliceSeconds wall-clock boundary (sliceStart, request scoped so it
+	// spans passes) and when the pass closes, never once per pattern; the
+	// per-pattern check reads only the cheap monotonic wall clock.
 	clock       func() float64
+	wall        func() float64
 	scanSeconds float64
 	spent       float64
-	passStart   float64
+	passMark    float64
+	sliceStart  float64
 	timeWarned  bool
 }
 
+// scanSliceSeconds is the wall-clock slice between CPU charges, the same
+// 2 ms the reference yields the event loop on (_EVENT_LOOP_SLICE_SECONDS).
+const scanSliceSeconds = 0.002
+
 func newDetectionScanBudget(cfg *SecurityConfig, clientIP string) *detectionScanBudget {
-	b := &detectionScanBudget{cfg: cfg, clientIP: clientIP, clock: scanCPUNow, scanSeconds: 10.0}
+	b := &detectionScanBudget{cfg: cfg, clientIP: clientIP, clock: scanCPUNow, wall: scanWallNow, scanSeconds: 10.0}
+	b.sliceStart = b.wall()
 	if cfg != nil && cfg.Detection.MaxScanSeconds > 0 {
 		b.scanSeconds = cfg.Detection.MaxScanSeconds
 	}
 	return b
 }
 
 // beginPass starts timing one regex pass (a view scan or the runtime
-// registry scan) and returns the function that ends it, adding the pass to
-// the spent total. Only regex work runs inside a pass, so decoding and
-// semantic analysis are not counted, like the reference. With a per-thread
-// CPU clock the goroutine is pinned to its OS thread for the pass so both
-// readings come from the same thread.
+// registry scan) and returns the function that ends it with a final
+// charge. Only regex work runs inside a pass, so decoding and semantic
+// analysis are not counted, like the reference. With a per-thread CPU
+// clock the goroutine is pinned to its OS thread for the pass so every
+// reading comes from the same thread.
 func (b *detectionScanBudget) beginPass() func() {
 	if b == nil {
 		return func() {}
 	}
 	if clockIsThreadCPU {
 		runtime.LockOSThread()
 	}
-	b.passStart = b.clock()
+	b.passMark = b.clock()
 	return func() {
-		b.spent += b.clock() - b.passStart
+		b.charge()
 		if clockIsThreadCPU {
 			runtime.UnlockOSThread()
 		}
 	}
 }
 
-// timeExhausted mirrors _scan_time_budget_exhausted: true once the time
-// spent in finished passes plus the running pass reaches the cap.
+// charge adds the scan-clock time since the last charge to spent.
+func (b *detectionScanBudget) charge() {
+	now := b.clock()
+	b.spent += now - b.passMark
+	b.passMark = now
+}
+
+// timeExhausted mirrors _scan_time_budget_exhausted: true once the charged
+// time reaches the cap. It reads only the monotonic wall clock, and charges
+// the CPU clock when a slice has elapsed, so the lag behind the true spend
+// is at most one slice plus one pattern.
 func (b *detectionScanBudget) timeExhausted() bool {
 	if b == nil {
 		return false
 	}
-	return b.spent+(b.clock()-b.passStart) >= b.scanSeconds
+	if now := b.wall(); now-b.sliceStart >= scanSliceSeconds {
+		b.charge()
+		b.sliceStart = now
+	}
+	return b.spent >= b.scanSeconds
 }
 
 // timeBudgetThreat mirrors _scan_time_budget_exhausted_threat plus the

guardcore/scanclock.go

@@ -9,6 +9,10 @@ import "time"
 // drive it.
 var scanCPUNow = platformScanClock
 
+// scanWallNow is the cheap monotonic clock the per-pattern check and the
+// slice boundary read; a variable so tests can drive it.
+var scanWallNow = wallClockSeconds
+
 var scanClockEpoch = time.Now()
 
 // wallClockSeconds is the fallback scan clock: seconds on the monotonic

guardcore/scantimebudget_test.go

@@ -8,16 +8,23 @@ import (
 )
 
 // jumpClock is a fake scan clock that advances by step on every read, so a
-// small cap is spent without burning real CPU.
+// small cap is spent without burning real CPU. The wall clock the slice
+// boundary reads advances by step too, so every pattern check crosses a
+// slice and charges the CPU clock (the per-pattern check itself reads only
+// the wall clock; see TestCPUClockIsReadPerPassNotPerPattern).
 func jumpClock(t *testing.T, step float64) {
 	t.Helper()
-	prev := scanCPUNow
-	now := 0.0
+	prevCPU, prevWall := scanCPUNow, scanWallNow
+	cpu, wall := 0.0, 0.0
 	scanCPUNow = func() float64 {
-		now += step
-		return now
+		cpu += step
+		return cpu
 	}
-	t.Cleanup(func() { scanCPUNow = prev })
+	scanWallNow = func() float64 {
+		wall += step
+		return wall
+	}
+	t.Cleanup(func() { scanCPUNow, scanWallNow = prevCPU, prevWall })
 }
 
 func TestScanSecondsDefaultAndRange(t *testing.T) {
@@ -241,3 +248,61 @@ func TestBudgetThreatIsFlaggedByWeightNotUnconditionally(t *testing.T) {
 		t.Fatalf("a sub-threshold weight (summed over the view passes) must not flag the request, got %v", r.Threats)
 	}
 }
+
+// The CPU clock is read when a pass opens, at slice boundaries and when it
+// closes, not once per pattern: a pass over many patterns inside one slice
+// reads it twice, and a stalled pattern adds one read per elapsed slice.
+func TestCPUClockIsReadPerPassNotPerPattern(t *testing.T) {
+	prevCPU, prevWall := scanCPUNow, scanWallNow
+	t.Cleanup(func() { scanCPUNow, scanWallNow = prevCPU, prevWall })
+	cpuReads, wall := 0, 0.0
+	scanCPUNow = func() float64 { cpuReads++; return 0 }
+	scanWallNow = func() float64 { return wall }
+	budget := newDetectionScanBudget(nil, "203.0.113.7")
+	end := budget.beginPass()
+	for i := 0; i < 1000; i++ {
+		if budget.timeExhausted() {
+			t.Fatal("a zero CPU clock never exhausts the budget")
+		}
+		wall += 0.00001 // 10 us per pattern: 1000 patterns span 5 slices
+	}
+	end()
+	if cpuReads < 2 || cpuReads > 10 {
+		t.Fatalf("1000 patterns over ~5 slices must read the CPU clock a few times, got %d reads", cpuReads)
+	}
+	cpuReads = 0
+	end = budget.beginPass()
+	for i := 0; i < 1000; i++ {
+		budget.timeExhausted()
+	}
+	end()
+	if cpuReads != 2 {
+		t.Fatalf("a pass inside one slice reads the CPU clock only at open and close, got %d reads", cpuReads)
+	}
+}
+
+// A long pass stops within about one slice of the cap: the budget is
+// checked against the charged time, which lags the true spend by at most one
+// slice plus one pattern.
+func TestLongPassStopsWithinOneSliceOfTheBudget(t *testing.T) {
+	prevCPU, prevWall := scanCPUNow, scanWallNow
+	t.Cleanup(func() { scanCPUNow, scanWallNow = prevCPU, prevWall })
+	now := 0.0 // one fake clock drives both: each pattern costs 0.5 ms
+	scanCPUNow = func() float64 { return now }
+	scanWallNow = func() float64 { return now }
+	budget := newDetectionScanBudget(nil, "203.0.113.7")
+	budget.scanSeconds = 0.0105
+	end := budget.beginPass()
+	defer end()
+	patterns := 0
+	for !budget.timeExhausted() {
+		patterns++
+		now += 0.0005
+		if patterns > 1000 {
+			t.Fatal("the budget never stopped the pass")
+		}
+	}
+	if overshoot := now - budget.scanSeconds; overshoot < 0 || overshoot > scanSliceSeconds+0.0005 {
+		t.Fatalf("stopped %.4fs past the cap, want within one slice plus one pattern", overshoot)
+	}
+}