perf(edge): allocation-free window path, guarded by tests

.github/workflows/ci.yml

@@ -84,6 +84,9 @@ jobs:
           set -o pipefail
           go test -race -count=1 -coverprofile=coverage.out ./internal/... | tee test.log
 
+      - name: Allocation guards (without the race detector)
+        run: go test -count=1 -run 'AllocFree|AllocsBounded' ./internal/...
+
       - name: Coverage gate
         run: |
           total=$(go tool cover -func=coverage.out | awk '/^total:/ {gsub("%", "", $3); print $3}')

.gitignore

@@ -18,3 +18,8 @@ __pycache__/
 
 # Node (docs diagram checker)
 node_modules/
+
+# Go test artifacts
+*.test
+coverage.out
+test.log

README.md

@@ -4,7 +4,7 @@
 
 [![CI](https://github.com/casablanque-code/avenge/actions/workflows/ci.yml/badge.svg)](https://github.com/casablanque-code/avenge/actions/workflows/ci.yml)
 
-A Go-based edge computing engine for real-time vibration analytics, inline signal compression, and predictive anomaly detection. Zero external runtime dependencies, zero-heap-allocation hot path.
+A Go-based edge computing engine for real-time vibration analytics, inline signal compression, and predictive anomaly detection. Zero external runtime dependencies, allocation-free per-window processing (enforced by tests).
 
 > Concept, motivation, and the broader case for edge-first vibration analytics live in [`docs/DOC.md`](docs/DOC.md). This README covers what's running and how to run it.
 

docs/ARCHITECTURE.md

@@ -288,5 +288,6 @@ sequenceDiagram
 
 - **Anomaly events come from the RMS detector only.** A message on `sm/anomaly/...` is published when the RMS detector changes state, and its `event_type` is `onset` only while that detector reports an anomaly; every other transition, including a first warning, is labelled `clear`. The combined severity (RMS and frequency bands) travels in the telemetry messages, so a fault visible only in a frequency band shows up in telemetry but may produce no anomaly event.
 - **Outage buffer.** With the default cadence the queue of 512 messages holds roughly `512 × 20 × 512 / sample_rate` seconds of telemetry: about 87 minutes at 1 kHz and 55 minutes at 1.6 kHz. This is derived from the code, not measured.
+- **Allocations.** After initialization the per-window path (ring buffer, FFT, both detectors, scoring) does not allocate; tests enforce this. A window that emits an event allocates the event and a copy of the band states (a few small objects), because they are handed to another goroutine. This is a measured property of the code on amd64 with Go 1.22; the Raspberry Pi has not been measured.
 - **Timestamps from the serial port are synthesized.** The firmware sends no timestamps; the reader derives them from a counter and `SENSOR_SAMPLE_RATE`, so that value must match the firmware's real rate.
 - **No transport encryption.** MQTT is plaintext, authenticated and ACL-restricted; run it on a trusted network.

docs/DOC.md

@@ -192,9 +192,9 @@ This allows detection of:
 ## 8. Performance Characteristics
 
 * Constant memory usage proportional to FFT window size
-* No heap allocations in the hot path after initialization
+* No heap allocations on the per-window path after initialization (enforced by tests); the periodic telemetry flush and an anomaly transition allocate a few small objects that are handed to the publisher
 * Low-latency processing suitable for real-time systems
-* Scalable to thousands of sensor streams (with proper orchestration)
+* One worker goroutine per sensor; how many sensors a single device can serve has not been measured
 
 ---
 

docs/TESTING.md

@@ -6,7 +6,7 @@ Every pull request and every push to `main` runs the same checks. The single req
 
 | Job        | What it checks |
 | ---------- | -------------- |
-| `edge`     | `gofmt`, `go vet`, build, `go test -race` with a coverage floor on `edge/internal/...`, and cross-build plus `go vet` for `linux/arm64` (Raspberry Pi), `darwin/arm64` and `windows/amd64` (development machines; serial input is Linux-only and stubbed elsewhere). Runs on the Go version in `go.mod` and on the latest stable Go. |
+| `edge`     | `gofmt`, `go vet`, build, `go test -race` with a coverage floor on `edge/internal/...`, allocation guards (the per-window path must not allocate), and cross-build plus `go vet` for `linux/arm64` (Raspberry Pi), `darwin/arm64` and `windows/amd64` (development machines; serial input is Linux-only and stubbed elsewhere). Runs on the Go version in `go.mod` and on the latest stable Go. |
 | `vuln`     | `govulncheck` against the Go vulnerability database. |
 | `infra`    | Compose files resolve (`docker compose config`), `yamllint`, `shellcheck`, `hadolint`, `actionlint`, dashboard JSON validity, that every Mermaid diagram in the docs parses, and `tests/lint/repo-checks.sh` (every required `.env` variable is in `.env.example`, no tracked `.env`, `passwd` or key files, every path and link in README/docs exists). |
 | `image`    | Builds the edge image, verifies it runs as non-root, and replays a seeded signal through it (`--no-mqtt`) to check that no sample is dropped. |
@@ -30,6 +30,7 @@ Edge unit tests and benchmarks (Go only):
 ```
 cd edge
 go vet ./... && go test -race -count=1 ./...
+go test -count=1 -run 'AllocFree|AllocsBounded' ./internal/...   # allocation guards, without -race
 go test ./internal/... -run '^$' -bench . -benchmem
 ```
 

edge/internal/anomaly/band_detector.go

@@ -122,6 +122,16 @@ func NewBandDetector(bands []FreqBand, cfg BandConfig) *BandDetector {
 // freqs and power must come from fft.PowerSpectrum (same length, same order).
 // Returns a BandResult describing the anomaly state of each band.
 func (bd *BandDetector) IngestSpectrum(freqs, power []float64) BandResult {
+	r := bd.Update(freqs, power)
+	r.Bands = bd.Bands()
+	return r
+}
+
+// Update processes one spectrum like IngestSpectrum but does not allocate:
+// the returned BandResult carries the summary fields only and Bands is nil.
+// Call Bands when the per-band states are needed (for example when an event
+// leaves the worker goroutine).
+func (bd *BandDetector) Update(freqs, power []float64) BandResult {
 	// Compute band-limited RMS for each band.
 	for i := range bd.states {
 		bd.states[i].BandRMS = bandRMS(freqs, power, bd.states[i].Band)
@@ -244,13 +254,18 @@ func bandRMS(freqs, power []float64, band FreqBand) float64 {
 	return math.Sqrt(v)
 }
 
-func (bd *BandDetector) buildResult() BandResult {
-	r := BandResult{
-		Bands:           make([]BandState, len(bd.states)),
-		OverallSeverity: SeverityNormal,
-	}
-	copy(r.Bands, bd.states)
+// Bands returns a copy of the current per-band states. The copy is owned by
+// the caller, so it is safe to hand to another goroutine.
+func (bd *BandDetector) Bands() []BandState {
+	out := make([]BandState, len(bd.states))
+	copy(out, bd.states)
+	return out
+}
 
+// buildResult summarises the current states. It does not allocate and leaves
+// Bands nil.
+func (bd *BandDetector) buildResult() BandResult {
+	r := BandResult{OverallSeverity: SeverityNormal}
 	for _, st := range bd.states {
 		if st.IsAnomaly {
 			r.AnyAnomaly = true

edge/internal/anomaly/band_update_alloc_test.go

@@ -0,0 +1,20 @@
+//go:build !race
+
+package anomaly
+
+import (
+	"math/rand"
+	"testing"
+)
+
+func TestBandUpdateAllocFree(t *testing.T) {
+	bd := NewBandDetector(DefaultBands(1000), DefaultBandConfig())
+	f, p := randomSpectrum(rand.New(rand.NewSource(1)), false)
+	for i := 0; i < 50; i++ { // get past warm-up
+		bd.Update(f, p)
+	}
+	allocs := testing.AllocsPerRun(50, func() { bd.Update(f, p) })
+	if allocs != 0 {
+		t.Fatalf("BandDetector.Update allocated %v times per call, want 0", allocs)
+	}
+}

edge/internal/anomaly/band_update_test.go

@@ -0,0 +1,67 @@
+package anomaly
+
+import (
+	"math/rand"
+	"reflect"
+	"testing"
+)
+
+func randomSpectrum(r *rand.Rand, spike bool) (freqs, power []float64) {
+	freqs = make([]float64, 256)
+	power = make([]float64, 256)
+	for i := range freqs {
+		freqs[i] = float64(i) * 1000 / 512
+		power[i] = 1e-3 * (1 + 0.1*r.Float64())
+	}
+	power[26] = 1 // ~50 Hz
+	if spike {
+		power[100] = 5 // ~195 Hz, inside the fault band
+	}
+	return freqs, power
+}
+
+// Update followed by Bands must equal IngestSpectrum on every window,
+// through warm-up, steady state, a fault and recovery.
+func TestUpdateAndBandsMatchIngestSpectrum(t *testing.T) {
+	a := NewBandDetector(DefaultBands(1000), DefaultBandConfig())
+	b := NewBandDetector(DefaultBands(1000), DefaultBandConfig())
+	r1 := rand.New(rand.NewSource(7))
+	r2 := rand.New(rand.NewSource(7))
+
+	for w := 0; w < 120; w++ {
+		spike := w >= 60 && w < 80
+		f1, p1 := randomSpectrum(r1, spike)
+		f2, p2 := randomSpectrum(r2, spike)
+
+		want := a.IngestSpectrum(f1, p1)
+		got := b.Update(f2, p2)
+		if got.Bands != nil {
+			t.Fatalf("window %d: Update must not attach Bands", w)
+		}
+		got.Bands = b.Bands()
+		if !reflect.DeepEqual(want, got) {
+			t.Fatalf("window %d: Update+Bands differs from IngestSpectrum", w)
+		}
+	}
+}
+
+// A copy returned by Bands must not change when the detector keeps running:
+// it is handed to another goroutine inside an event.
+func TestBandsReturnsIndependentCopy(t *testing.T) {
+	bd := NewBandDetector(DefaultBands(1000), DefaultBandConfig())
+	r := rand.New(rand.NewSource(3))
+	for w := 0; w < 40; w++ {
+		f, p := randomSpectrum(r, false)
+		bd.Update(f, p)
+	}
+	snapshot := bd.Bands()
+	frozen := append([]BandState(nil), snapshot...)
+
+	for w := 0; w < 40; w++ {
+		f, p := randomSpectrum(r, true)
+		bd.Update(f, p)
+	}
+	if !reflect.DeepEqual(frozen, snapshot) {
+		t.Fatal("a Bands copy changed after later windows were processed")
+	}
+}

edge/internal/filter/sdt.go

@@ -27,11 +27,24 @@ func SDT(points []Point, epsilon float64) []Point {
 	if len(points) == 0 {
 		return nil
 	}
+	return SDTInto(nil, points, epsilon)
+}
+
+// SDTInto is SDT that appends its result to dst[:0] and returns it, so a
+// caller that only needs the result transiently (for example its length) can
+// reuse one buffer and avoid allocating on every call. dst may be nil; when
+// its capacity is at least the number of points the call does not allocate.
+// The result aliases dst, so it is only valid until dst is reused.
+func SDTInto(dst []Point, points []Point, epsilon float64) []Point {
+	result := dst[:0]
+	if len(points) == 0 {
+		return result
+	}
 	if len(points) == 1 {
-		return []Point{points[0]}
+		return append(result, points[0])
 	}
 
-	result := []Point{points[0]}
+	result = append(result, points[0])
 	anchor := points[0]
 
 	// Slope limits for the two doors.

edge/internal/filter/sdt_into_alloc_test.go

@@ -0,0 +1,18 @@
+//go:build !race
+
+package filter
+
+import "testing"
+
+// With enough capacity SDTInto must not allocate. Skipped under the race
+// detector, which changes allocation behaviour.
+func TestSDTIntoAllocFree(t *testing.T) {
+	pts := noisySine(10240, 3)
+	buf := make([]Point, 0, len(pts))
+	allocs := testing.AllocsPerRun(20, func() {
+		buf = SDTInto(buf, pts, 0.05)
+	})
+	if allocs != 0 {
+		t.Fatalf("SDTInto allocated %v times per call, want 0", allocs)
+	}
+}

edge/internal/filter/sdt_into_test.go

@@ -0,0 +1,47 @@
+package filter
+
+import (
+	"math"
+	"math/rand"
+	"reflect"
+	"testing"
+)
+
+func noisySine(n int, seed int64) []Point {
+	r := rand.New(rand.NewSource(seed))
+	pts := make([]Point, n)
+	for i := range pts {
+		t := float64(i) / 1000
+		pts[i] = Point{T: t, Value: math.Sin(2*math.Pi*5*t) + 0.05*r.NormFloat64()}
+	}
+	return pts
+}
+
+// SDTInto must produce exactly what SDT produces, for every input size.
+func TestSDTIntoMatchesSDT(t *testing.T) {
+	for _, n := range []int{0, 1, 2, 3, 10, 500, 10240} {
+		pts := noisySine(n, int64(n)+1)
+		want := SDT(pts, 0.05)
+		got := SDTInto(nil, pts, 0.05)
+		if len(want) == 0 && len(got) == 0 {
+			continue
+		}
+		if !reflect.DeepEqual(want, got) {
+			t.Fatalf("n=%d: SDTInto differs from SDT (%d vs %d points)", n, len(got), len(want))
+		}
+	}
+}
+
+// A reused destination buffer must not leak points from a previous call.
+func TestSDTIntoReusesBuffer(t *testing.T) {
+	first := noisySine(2000, 1)
+	second := noisySine(300, 2)
+	buf := make([]Point, 0, 2000)
+
+	buf = SDTInto(buf, first, 0.05)
+	buf = SDTInto(buf, second, 0.05)
+
+	if want := SDT(second, 0.05); !reflect.DeepEqual(want, buf) {
+		t.Fatalf("reused buffer holds %d points, want %d", len(buf), len(want))
+	}
+}

edge/internal/sensor/alloc_test.go

@@ -0,0 +1,78 @@
+//go:build !race
+
+package sensor
+
+import (
+	"math"
+	"testing"
+)
+
+// Allocation guards for the per-window path. They are excluded under the race
+// detector (which changes allocation behaviour) and run in their own CI step.
+
+func steadyWindow(cfg SensorConfig) []float64 {
+	vals := make([]float64, cfg.FFTWindow)
+	for i := range vals {
+		t := float64(i) / float64(cfg.SampleRate)
+		vals[i] = 0.5*math.Sin(2*math.Pi*50*t) + 0.02*math.Sin(2*math.Pi*333*t)
+	}
+	return vals
+}
+
+func feedWindow(w *SensorWorker, vals []float64, t *float64, dt float64) *WorkerEvent {
+	var last *WorkerEvent
+	for _, v := range vals {
+		if ev := w.ingestSample(Sample{T: *t, Value: v}); ev != nil {
+			last = ev
+		}
+		*t += dt
+	}
+	return last
+}
+
+// In steady state, with no telemetry flush and no state change, processing a
+// window must not allocate at all.
+func TestHotPathAllocFree(t *testing.T) {
+	const flushEvery = 1000
+	cfg := DefaultSensorConfig("alloc")
+	cfg.SDTFlushEvery = flushEvery
+	w := newSensorWorker(cfg)
+	vals := steadyWindow(cfg)
+	dt := 1 / float64(cfg.SampleRate)
+	var clock float64
+
+	// Run through one full flush cycle: baselines settle and the buffers grow
+	// to their steady-state size. Right after the flush the next one is far
+	// away, so the measured windows below contain no flush.
+	for i := 0; i < flushEvery; i++ {
+		feedWindow(w, vals, &clock, dt)
+	}
+	allocs := testing.AllocsPerRun(20, func() { feedWindow(w, vals, &clock, dt) })
+	if allocs != 0 {
+		t.Fatalf("per-window path allocated %v times per window, want 0", allocs)
+	}
+}
+
+// A window that emits an event may allocate the event and its band snapshot,
+// nothing else. The bound catches accidental per-window garbage.
+func TestFlushWindowAllocsBounded(t *testing.T) {
+	cfg := DefaultSensorConfig("alloc")
+	cfg.SDTFlushEvery = 1 // every window flushes
+	w := newSensorWorker(cfg)
+	vals := steadyWindow(cfg)
+	dt := 1 / float64(cfg.SampleRate)
+	var clock float64
+
+	for i := 0; i < cfg.BaselineWindows+30; i++ {
+		feedWindow(w, vals, &clock, dt)
+	}
+	allocs := testing.AllocsPerRun(20, func() {
+		if feedWindow(w, vals, &clock, dt) == nil {
+			t.Fatal("expected a flush event on every window")
+		}
+	})
+	const maxAllocs = 3
+	if allocs > maxAllocs {
+		t.Fatalf("flush window allocated %v times, want at most %d", allocs, maxAllocs)
+	}
+}