From f7bdc3c1588e4e9840ea37067be8c88fba375408 Mon Sep 17 00:00:00 2001 From: casablanque-code Date: Mon, 2 Nov 2026 01:47:55 +0300 Subject: [PATCH] perf(edge): allocation-free window path, guarded by tests --- .github/workflows/ci.yml | 3 + .gitignore | 5 ++ README.md | 2 +- docs/ARCHITECTURE.md | 1 + docs/DOC.md | 4 +- docs/TESTING.md | 3 +- edge/internal/anomaly/band_detector.go | 27 +++++-- .../anomaly/band_update_alloc_test.go | 20 +++++ edge/internal/anomaly/band_update_test.go | 67 ++++++++++++++++ edge/internal/filter/sdt.go | 17 +++- edge/internal/filter/sdt_into_alloc_test.go | 18 +++++ edge/internal/filter/sdt_into_test.go | 47 +++++++++++ edge/internal/sensor/alloc_test.go | 78 +++++++++++++++++++ edge/internal/sensor/bands_snapshot_test.go | 52 +++++++++++++ edge/internal/sensor/bench_test.go | 31 ++++++++ edge/internal/sensor/worker.go | 28 ++++++- 16 files changed, 387 insertions(+), 16 deletions(-) create mode 100644 edge/internal/anomaly/band_update_alloc_test.go create mode 100644 edge/internal/anomaly/band_update_test.go create mode 100644 edge/internal/filter/sdt_into_alloc_test.go create mode 100644 edge/internal/filter/sdt_into_test.go create mode 100644 edge/internal/sensor/alloc_test.go create mode 100644 edge/internal/sensor/bands_snapshot_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e084afa..0ff1480 100644 --- a/.github/workflows/ci.yml +++ b/.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}') diff --git a/.gitignore b/.gitignore index 1a54a7e..41b23a7 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,8 @@ __pycache__/ # Node (docs diagram checker) node_modules/ + +# Go test artifacts +*.test +coverage.out +test.log diff --git a/README.md b/README.md index 128d137..dc6038f 100644 --- a/README.md +++ b/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. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 0e06f4c..1474407 100644 --- a/docs/ARCHITECTURE.md +++ b/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. diff --git a/docs/DOC.md b/docs/DOC.md index 4cc69f4..ccc435c 100644 --- a/docs/DOC.md +++ b/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 --- diff --git a/docs/TESTING.md b/docs/TESTING.md index d6d9d3f..9cd0bb6 100644 --- a/docs/TESTING.md +++ b/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 ``` diff --git a/edge/internal/anomaly/band_detector.go b/edge/internal/anomaly/band_detector.go index 48397aa..a253bf1 100644 --- a/edge/internal/anomaly/band_detector.go +++ b/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 diff --git a/edge/internal/anomaly/band_update_alloc_test.go b/edge/internal/anomaly/band_update_alloc_test.go new file mode 100644 index 0000000..0f65e81 --- /dev/null +++ b/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) + } +} diff --git a/edge/internal/anomaly/band_update_test.go b/edge/internal/anomaly/band_update_test.go new file mode 100644 index 0000000..b3382f0 --- /dev/null +++ b/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") + } +} diff --git a/edge/internal/filter/sdt.go b/edge/internal/filter/sdt.go index 301a187..dacc9ff 100644 --- a/edge/internal/filter/sdt.go +++ b/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. diff --git a/edge/internal/filter/sdt_into_alloc_test.go b/edge/internal/filter/sdt_into_alloc_test.go new file mode 100644 index 0000000..ccee9ca --- /dev/null +++ b/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) + } +} diff --git a/edge/internal/filter/sdt_into_test.go b/edge/internal/filter/sdt_into_test.go new file mode 100644 index 0000000..ddad49f --- /dev/null +++ b/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)) + } +} diff --git a/edge/internal/sensor/alloc_test.go b/edge/internal/sensor/alloc_test.go new file mode 100644 index 0000000..2b4eb09 --- /dev/null +++ b/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) + } +} diff --git a/edge/internal/sensor/bands_snapshot_test.go b/edge/internal/sensor/bands_snapshot_test.go new file mode 100644 index 0000000..4209593 --- /dev/null +++ b/edge/internal/sensor/bands_snapshot_test.go @@ -0,0 +1,52 @@ +package sensor + +import ( + "math" + "reflect" + "testing" +) + +// The band states inside a WorkerEvent cross goroutines through the events +// channel, so later windows must never modify them. +func TestEventBandStatesAreNotAliased(t *testing.T) { + cfg := DefaultSensorConfig("alias") + cfg.SDTFlushEvery = 1 + w := newSensorWorker(cfg) + dt := 1 / float64(cfg.SampleRate) + var clock float64 + + window := func(amp float64) []float64 { + vals := make([]float64, cfg.FFTWindow) + for i := range vals { + ts := float64(i) / float64(cfg.SampleRate) + vals[i] = 0.5*math.Sin(2*math.Pi*50*ts) + amp*math.Sin(2*math.Pi*200*ts) + } + return vals + } + run := func(vals []float64) *WorkerEvent { + var last *WorkerEvent + for _, v := range vals { + if ev := w.ingestSample(Sample{T: clock, Value: v}); ev != nil { + last = ev + } + clock += dt + } + return last + } + + var first *WorkerEvent + for i := 0; i < cfg.BaselineWindows+5; i++ { + first = run(window(0.01)) + } + if first == nil || len(first.BandResult.Bands) == 0 { + t.Fatal("expected a flush event carrying band states") + } + frozen := append(first.BandResult.Bands[:0:0], first.BandResult.Bands...) + + for i := 0; i < 20; i++ { // a strong fault changes every later window + run(window(2.0)) + } + if !reflect.DeepEqual(frozen, first.BandResult.Bands) { + t.Fatal("band states of an earlier event changed after later windows") + } +} diff --git a/edge/internal/sensor/bench_test.go b/edge/internal/sensor/bench_test.go index 57e3d4f..74eb9d4 100644 --- a/edge/internal/sensor/bench_test.go +++ b/edge/internal/sensor/bench_test.go @@ -30,3 +30,34 @@ func BenchmarkIngestWindow(b *testing.B) { } } } + +// BenchmarkIngestWindowNoFlush measures the steady-state per-window path with +// a telemetry flush only every 1000 windows, so the flush cost is amortised to +// well below one allocation per window. It must report 0 allocs/op. +func BenchmarkIngestWindowNoFlush(b *testing.B) { + cfg := DefaultSensorConfig("bench") + cfg.SDTFlushEvery = 1000 + w := newSensorWorker(cfg) + n := cfg.FFTWindow + vals := make([]float64, n) + 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) + } + var t float64 + dt := 1 / float64(cfg.SampleRate) + feed := func() { + for _, v := range vals { + w.ingestSample(Sample{T: t, Value: v}) + t += dt + } + } + for i := 0; i < cfg.SDTFlushEvery; i++ { // one full cycle: buffers reach steady size + feed() + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + feed() + } +} diff --git a/edge/internal/sensor/worker.go b/edge/internal/sensor/worker.go index bb40334..9c80a78 100644 --- a/edge/internal/sensor/worker.go +++ b/edge/internal/sensor/worker.go @@ -137,11 +137,25 @@ type SensorWorker struct { ringBuf []float64 ringFilled int + // Reusable scratch buffers: the per-window path must not allocate. + origBuf []float64 // DC-included copy of the window for the RMS detector + sdtOut []filter.Point // SDT result, only its length is used + totalWindows int totalRaw int totalCompressed int } +// maxPreallocWindows bounds how many windows of SDT history are allocated up +// front. A longer flush interval still works: the buffers grow once and are +// then reused, so steady state does not allocate. +const maxPreallocWindows = 64 + +// bufferWindows returns how many windows of samples to preallocate for. +func bufferWindows(cfg SensorConfig) int { + return min(max(cfg.SDTFlushEvery, 1), maxPreallocWindows) +} + // newSensorWorker creates an initialised SensorWorker for the given config. // All buffers are allocated here — the hot loop is allocation-free. func newSensorWorker(cfg SensorConfig) *SensorWorker { @@ -168,8 +182,10 @@ func newSensorWorker(cfg SensorConfig) *SensorWorker { bandCfg, ), scoreThr: scoring.DefaultThresholds(), - sdt: make([]filter.Point, 0, cfg.FFTWindow*20), + sdt: make([]filter.Point, 0, cfg.FFTWindow*bufferWindows(cfg)), + sdtOut: make([]filter.Point, 0, cfg.FFTWindow*bufferWindows(cfg)), ringBuf: make([]float64, cfg.FFTWindow), + origBuf: make([]float64, cfg.FFTWindow), } } @@ -194,10 +210,10 @@ func (w *SensorWorker) ingestSample(s Sample) *WorkerEvent { // FFT (zero-alloc via Processor, in-place DC removal). removeDCInPlace(w.ringBuf) freqs, power := w.fftProc.PowerSpectrum(w.ringBuf) - bandResult := w.bandDet.IngestSpectrum(freqs, power) + bandResult := w.bandDet.Update(freqs, power) // allocation-free summary // Anomaly detection on original values (DC-included). - origVals := make([]float64, w.cfg.FFTWindow) + origVals := w.origBuf tail := w.sdt[len(w.sdt)-w.cfg.FFTWindow:] for i, p := range tail { origVals[i] = p.Value @@ -220,12 +236,15 @@ func (w *SensorWorker) ingestSample(s Sample) *WorkerEvent { // periodic data in ClickHouse so Grafana has something to draw. var flushEvent *WorkerEvent if w.totalWindows%w.cfg.SDTFlushEvery == 0 { - compressed := filter.SDT(w.sdt, w.cfg.SDTEpsilon) + // Only the length is needed, so reuse one buffer instead of allocating. + w.sdtOut = filter.SDTInto(w.sdtOut, w.sdt, w.cfg.SDTEpsilon) + compressed := w.sdtOut ratio := float32(filter.CompressionRatio(len(w.sdt), len(compressed))) w.totalRaw += len(w.sdt) w.totalCompressed += len(compressed) w.sdt = w.sdt[:0] + bandResult.Bands = w.bandDet.Bands() // owned copy: the event leaves this goroutine flushEvent = &WorkerEvent{ SensorID: w.cfg.ID, Event: event, // may be nil — that's fine @@ -247,6 +266,7 @@ func (w *SensorWorker) ingestSample(s Sample) *WorkerEvent { // Flush event already carries this transition — no duplicate needed. return flushEvent } + bandResult.Bands = w.bandDet.Bands() // owned copy: the event leaves this goroutine return &WorkerEvent{ SensorID: w.cfg.ID, Event: event,