Copy, don't retain, message data carried across calls

[cboulay]

Fixes #214.

BinnedAggregateTransformer and DiffTransformer kept message.data (or a view of it) in their state and read it back on the next call. Message data is not the transformer's to keep: over a cross-process link it is a view into a shared-memory slot the publisher recycles (PEP 574 out-of-band buffers, msg_id % num_buffers ring), so the retained samples silently become whatever overwrote them — valid arrays of the right shape and dtype, wrong numbers, no exception.

Only cross-process links are affected; in-process publishers pass the object by reference with no serialization, which is part of why this survived testing.

Fix

Copy at each point of retention, via the existing backend-portable util.array.xp_copy:

• binned_aggregate.py — the carry is None branch when no bin completes (the xp.concat branch already allocates), and the leftover tail after the last completed bin, which was a view of message.data whenever there was no prior carry. Copying the tail also stops a couple of carried samples from pinning a whole chunk alive.
• diff.py — the trailing sample kept for the next message's first diff, and the _reset_state slice, which was not exposed (overwritten within the same call) but left state aliasing the message.

This matches the convention already used elsewhere in the package: Sampler, Resample, ResampleConcat and Window all buffer through HybridBuffer with update_strategy="immediate" precisely so nothing is retained by reference.

Tests

Nothing in the suite could catch this, so the fix comes with a harness for the whole bug class.

tests/helpers/recycled_shm.py marshals each message through ezmsg's own Marshal into a single slot — the publisher's 32-slot ring compressed so reuse is deterministic on the very next message rather than 32 later. assert_survives_buffer_recycling(make_proc, messages) runs the same inputs through two fresh transformers, one on owned arrays and one on recycled ones, and requires bit-identical outputs: same arithmetic on the same numbers, so any difference at all means state was read back from recycled memory.

tests/unit/test_buffer_recycling.py applies it to:

• diff, both scale_by_fs settings, plus one test pinning the actual value of the cross-message boundary diff
• binned_aggregate: no-bin-completed, leftover-tail, fractional and sample-locked grids at an off-nominal 1013 Hz, and stacked MIN/MAX

plus two tests of the harness itself — a deliberately-retaining canary transformer that must trip the assertion, and a check that the slot really aliases and really gets overwritten — so it cannot go blind and let the others pass for the wrong reason.

All 8 target tests fail against the unfixed source and pass with it. Full unit suite: 4049 passed, 6 skipped.

One caveat worth knowing: whether corruption of a small carry is observable depends on the data (a two-sample corruption can average away under MEAN, or fall inside the bin's range under MIN/MAX). The multi-op test uses a seed where it is observable — 6 of 8 seeds tried detect it. The leftover-tail test covers the same code path without that sensitivity.

🤖 Generated with Claude Code