From 3d77986bd4789488500f29f570a7629b68616e7d Mon Sep 17 00:00:00 2001 From: Jonathan Stokes Date: Wed, 12 Aug 2026 17:59:10 -0500 Subject: [PATCH 1/3] feat: a persistent broadcast subscription at the protocol layer The transport's three inbound hooks are all one-shot and scoped to a trigger: request() correlates one reply, await_broadcast() waits for one push, collect() gathers for a fixed number of seconds. A message no waiter expects is dropped at debug level. A push-fed cache needs the opposite - every decoded message, for the life of the connection - so add_listener() provides it. * Transport.add_listener / remove_listener, with QuadCortex passing both through so the layer above never reaches into _t. * Listeners are notified first and consume nothing: collectors and waiters behave exactly as they do with no listener registered. * Listeners run on the RX thread, so the RX rules cover them. One that raises is logged and skipped; its peers and the message's waiter lose nothing. * request, await_broadcast and collect now refuse to run on the RX thread. That is what makes "a listener never reads from the device" enforced rather than requested, and it can refuse nothing that ever worked: the RX thread is the thread that delivers replies, so a wait from inside it only times out. * protocol.connect(before_handshake=...) calls back with the started transport before the handshake, the only moment early enough to hear its state burst. Measured on the unit (d14e / CorOS 4.0.1) and recorded in protocol.md: connect() returns at 2.0 s, the ModelRepo lands at 4.9 s and the seed preset at 10.1 s - 474 messages of 24 types by 15 s. A listener attached to the client connect() hands back has already missed the burst, which is why the hook exists. Offline tests cover the push nobody asked for, the reply a listener must not steal, the ordering a cache depends on, a raising listener, the refused read, and removal. tests/hardware/test_broadcast_listener.py proves it on the unit; it only listens, so it writes nothing and has nothing to restore (ADR-0005). ADR-0008 records why listeners run on the RX thread rather than behind a queue, and why the refusal is enforced instead of documented. Protocol-layer half of #11. The model-side cache is the other half and is not here. --- CLAUDE.md | 1 + changelog.md | 43 +++++ docs/ADR.md | 18 ++ docs/STEERING.md | 26 ++- docs/api.md | 1 + docs/architecture.md | 30 +++- docs/protocol.md | 8 + pyquadcortex/protocol/client.py | 27 +++ pyquadcortex/protocol/session.py | 17 +- pyquadcortex/protocol/transport.py | 160 ++++++++++++++++++ tests/hardware/conftest.py | 69 +++++++- tests/hardware/readme.md | 12 ++ tests/hardware/test_broadcast_listener.py | 196 ++++++++++++++++++++++ tests/test_client.py | 35 ++++ tests/test_session.py | 34 ++++ tests/test_transport.py | 180 ++++++++++++++++++-- 16 files changed, 839 insertions(+), 18 deletions(-) create mode 100644 tests/hardware/test_broadcast_listener.py diff --git a/CLAUDE.md b/CLAUDE.md index 0091bd9..76fd9c4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -18,6 +18,7 @@ Read `docs/STEERING.md` before non-trivial work (new operations, transport or fr - Grid mutations use the row/column-keyed pattern (`set_param` / `set_bypass`) - never extend the wholesale `write_preset` path. - Docstrings state their evidence: confirmed on hardware vs inferred from the schema. When you verify something on hardware, record it (docstring + coverage table) in the same change. - Code in the RX path preserves "the RX thread never dies": wrap every decode, skip unknown types at debug level, reset the reassembly buffer on anything malformed. +- A `Transport.add_listener` listener runs ON the RX thread (ADR-0008). It applies what the push carries, notes what needs re-reading, and returns. It never reads from the device - `request`, `await_broadcast` and `collect` refuse to run on that thread, and that refusal is not to be relaxed for convenience. Anything registering a listener that must see the connect handshake's burst registers it through `protocol.connect(before_handshake=...)`; by the time `connect()` returns, the burst is still seconds away. - Hardware sessions: quit Cortex Control first - it holds the HID interface exclusively. - Describe the protocol work as documenting the device's protocol as-is (recovered schema, observed traffic). Do not call it "reverse engineering" in docs, comments, commit messages, or issues. - Changed code under a path listed in `docs/STEERING.md` § Owned Paths? Diff and update STEERING/CLAUDE/ADR in the same PR. diff --git a/changelog.md b/changelog.md index 1d894d6..c6b5b53 100644 --- a/changelog.md +++ b/changelog.md @@ -85,6 +85,49 @@ To use both layers in one script, wrap a connection you already have with `Device.from_client(qc)`. It does not take ownership: closing the `Device` leaves your connection open. +### New: listen to everything the unit sends + +The unit talks without being asked. Turn a knob on its touchscreen, recall a +preset, let the metronome run, and it pushes messages about it. Until now those +messages were only reachable if you happened to be waiting for that exact one, +and anything else was dropped. `add_listener` hands you all of them: + +```python +from pyquadcortex import protocol + +def watch(message): + print(type(message).__name__) + +with protocol.connect() as qc: + stop = qc.add_listener(watch) + ... + stop() # or qc.remove_listener(watch) +``` + +Your function is called for every message, and it takes nothing away from the +rest of the library: a call that was waiting for a reply still gets it. + +Two rules, because your function runs on the thread that reads from the USB +device: + +- **Do not block in it.** Whatever it does delays the next message being read. +- **Do not read from the device in it.** That thread is the one that would have to + deliver the answer, so the call could only ever time out - with the connection + stalled behind it. Rather than let that happen, the library raises + `RuntimeError` if you try. Note what you need and read it from your own thread. + +To hear the burst of state the unit sends when a client connects - nearly +everything it knows, including the preset currently loaded - register before the +handshake, because it arrives seconds after `connect()` returns: + +```python +with protocol.connect(before_handshake=lambda t: t.add_listener(watch)) as qc: + ... +``` + +The decision behind the two rules is ADR-0008. This is the groundwork for the +model keeping itself current without asking twice. + ### Withdrawn: the Tempo menu's MODE is "not on the wire" The 0.23.0 entry below records, under **Settled**, that the Tempo menu's MODE diff --git a/docs/ADR.md b/docs/ADR.md index 79f66d6..60d3608 100644 --- a/docs/ADR.md +++ b/docs/ADR.md @@ -90,3 +90,21 @@ Records are append-only once `Decided` and built upon: a shipped decision is nev - Finding the MODE wire path becomes a prerequisite of M3's device-settings Epic, not of M1. **No tempo surface ships at M1**, so nothing in this record is user-visible yet. - Design principle 3 keeps its meaning and gains a boundary: omission is for behaviour we do not understand, refusal is for behaviour we understand and cannot yet drive. A record that says which one applies is now expected of anything the model leaves out. - This does not license modelling controls on a hunch. It applies where the unit's behaviour is confirmed and only the message is missing; a control we have not understood on the hardware is still omitted. + +## ADR-0008: Persistent listeners run on the RX thread, which may not read from the device + +- **Status:** Decided (2026-08-12) +- **Decision:** A persistent inbound subscription (`Transport.add_listener`) is called on the transport's RX thread, synchronously, before the message reaches any collector or waiter. The transport enforces the other half of that bargain: `request`, `await_broadcast` and `collect` raise `RuntimeError` when called from the RX thread, so a listener cannot read from the device. A listener that raises is logged and skipped, costing its peers and the message's waiter nothing. +- **Context:** The three existing inbound hooks are one-shot and scoped to a trigger, so a message nobody is expecting is dropped. A push-fed cache (`docs/domain-model.md` section 9) needs the opposite: every decoded message, for the life of the connection. Where that callback runs is the whole decision, because the RX thread is the one thread in this library that must never block and never die - a wedged read loop takes the connection with it, and every request outstanding. +- **Options:** + - **(a) Call listeners on the RX thread, and forbid reads from it - chosen.** Cheapest, and it keeps arrival order exact. It also gives the ordering the cache design assumes: a listener has already applied a push before the caller that provoked it wakes up, so no caller can observe a reply that its own cache has not seen. + - **(b) A delivery thread with a queue.** The RX thread would be immune to a slow listener, at the cost of a thread, an unbounded queue, and a cache that lags the reply that fed it. It buys immunity from a listener that blocks while making one that blocks harmless enough to survive unnoticed. + - **(c) Call listeners on the RX thread and document the rules without enforcing them.** The rule that matters ("do not read from the device here") is invisible when broken: the read appears to work, then times out, having stopped the read loop for the whole timeout. This project's oldest lesson is that a failure which looks like success is the expensive kind. +- **Open Questions:** Whether a listener should be notified when the device is lost. Today it simply stops receiving, which is enough for a cache whose owner learns about loss from the exception on its next call. Reconnect (issue #15) is where this gets answered. +- **Rationale:** The design already says the RX thread applies pushes and notes what needs re-reading while the caller's thread does the re-reading. Option (a) is that sentence in code. The enforcement costs one thread-identity comparison per call and converts a silent, connection-wide stall into an immediate error naming the rule - and it can refuse nothing that ever worked, because a correlated wait issued from the thread that delivers replies can only ever time out. +- **Consequences:** + - A listener must return promptly. Anything expensive belongs on the caller's thread, reached by noting what needs doing rather than doing it. + - Nothing may weaken the refusal to keep a convenience. A future listener that wants a value it did not receive marks it for re-reading; it does not fetch it. + - `send` is deliberately not refused: it is fire-and-forget and cannot deadlock. A listener that writes owns the delay it adds to the read loop. + - Registering in time for the connect handshake's burst needs `protocol.connect(before_handshake=...)`, because the burst arrives after `connect()` returns. + - Existing behaviour is untouched: listeners consume nothing, and `send`/`request`/`collect`/`await_broadcast` answer exactly as they did with no listener registered. diff --git a/docs/STEERING.md b/docs/STEERING.md index ca4d799..b1580c7 100644 --- a/docs/STEERING.md +++ b/docs/STEERING.md @@ -37,7 +37,7 @@ Inside the protocol layer, a strict one-concern-per-file layering: `cli` → `se ### Data and state -The protocol layer is stateless between calls: every read is a live exchange, and the unit is the source of truth. The model layer (design in [`domain-model.md`](domain-model.md)) introduces a broadcast-fed write-through cache above `protocol/client.py`; at the time of writing the model is a skeleton and that cache is not built, so callers still hold whatever state they need. +The protocol layer is stateless between calls: every read is a live exchange, and the unit is the source of truth. It does carry one hook for a caller who wants to be told rather than to ask - `Transport.add_listener`, a subscription that sees every message the unit pushes for the life of the connection (ADR-0008) - but the transport stores none of it. The model layer (design in [`domain-model.md`](domain-model.md)) introduces a broadcast-fed write-through cache above `protocol/client.py`; at the time of writing the model is a skeleton and that cache is not built, so callers still hold whatever state they need. ## 4. Owned Paths @@ -80,6 +80,7 @@ Decisions for this area are recorded in [`ADR.md`](ADR.md): | ADR-0005 | A hardware-in-the-loop integration suite, state-neutral on success | | ADR-0006 | The domain model takes the top-level namespace; the protocol layer moves to `pyquadcortex.protocol` | | ADR-0007 | The model may represent a control whose wire path is still open | +| ADR-0008 | Persistent listeners run on the RX thread, which may not read from the device | ## 8. Open Questions @@ -119,6 +120,29 @@ Single-device, single-connection USB HID at interactive rates (129-byte reports) ## Change Log +### 2026-08-12 - A persistent broadcast subscription at the protocol layer, and ADR-0008 + +**What changed:** +- `Transport.add_listener` / `remove_listener`: a subscription that sees every decoded inbound message for the life of the connection, including the unsolicited pushes `_dispatch` used to drop for want of a waiter. `QuadCortex` passes both through so the layer above never reaches into `_t` +- The transport now refuses `request`, `await_broadcast` and `collect` when they are called from the RX thread. That is what makes "a listener never reads from the device" enforced rather than requested +- `protocol.connect(before_handshake=...)` calls back with the started transport before the handshake runs, which is the only moment early enough to hear the handshake's own state burst +- ADR.md: ADR-0008 - listeners run on the RX thread, and the RX thread may not read; the queue-and-delivery-thread alternative and the document-but-do-not-enforce alternative are recorded with why each was rejected +- Section 3's "Data and state" names the one hook that is not a live exchange; section 7's table gained the ADR-0008 row +- `docs/protocol.md` "Connect burst, measured" gained the fact that decided the hook: `connect()` returns at 2.0 s, the ModelRepo lands at 4.9 s and the seed preset at 10.1 s, so a listener attached to the returned client has missed the burst it wanted +- `tests/hardware/` gained `test_broadcast_listener.py`, and the suite's connection fixture now records the burst - it cannot be attached on demand later, because the burst happens during `connect()` + +**Why:** +- M1 Epic (stokes-audio/pyquadcortex#8), Story #11. This is the protocol-layer half of that story, carved out because it is independent of the model work: `docs/domain-model.md` section 9 needs a push-fed cache, and a cache cannot be fed by three hooks that are all one-shot and scoped to a trigger + +**Scope of impact:** +- **Updated:** `pyquadcortex/protocol/transport.py`, `client.py`, `session.py`, `tests/test_transport.py`, `tests/test_client.py`, `tests/test_session.py`, `tests/hardware/conftest.py`, `tests/hardware/test_broadcast_listener.py` (new), `tests/hardware/readme.md`, ADR.md, CLAUDE.md, STEERING.md, architecture.md, api.md, protocol.md, changelog.md +- **Not updated (intentionally):** ADR-0002 - the offline suite still imports no `hid` and the new tests run against `FakeHid` like the rest; ADR-0005 - the new hardware tests only listen, so they write nothing and have nothing to restore, which meets the contract rather than changing it; `docs/domain-model.md` - section 9 designed this and needed no correction; the coverage table in `protocol.md` - no new message type is involved + +**Downstream to consider:** +- The model-side cache (the other half of #11) is the intended consumer and is being written separately. It registers through `before_handshake` so the burst warms it for free +- `tests/hardware/test_write_echo.py` still taps `Transport._dispatch` by monkeypatching it, which predates this and could now be an ordinary listener. Left alone deliberately: it is a working measurement harness, and `tests/test_scene_echo_predicates.py` imports it offline +- ADR-0008 leaves one question open on purpose - whether a listener hears about device loss. It stops receiving today, and the answer belongs with reconnect (#15) + ### 2026-08-11 - The namespace flip lands, and ADR-0007 **What changed:** diff --git a/docs/api.md b/docs/api.md index 1391e1a..5cd6408 100644 --- a/docs/api.md +++ b/docs/api.md @@ -60,6 +60,7 @@ already read and need no connection; calling them as methods raises | **Per-beat accents** | `set_beat(n, MetronomeBeat.ACCENT)`, `set_beats([...])`, `protocol.beats(preset)` | | **Inspect a preset** (module functions) | `protocol.blocks(preset)`, `protocol.splits(preset)`, `protocol.free_rows(preset)`, `protocol.row_status(preset)`, `protocol.bypass_state(preset, row, column)`, `protocol.param_state(preset, row, column, index)`, `protocol.param_options(preset, row, column, index)`, `protocol.input_chain_rows(preset, input)`, `protocol.params_equal(a, b, option_count=)`, `protocol.field_present(msg, field)` | | **Wait for the device** | `wait_for_listing(setlist, until=...)` | +| **Watch what the unit pushes** | `add_listener(fn)`, `remove_listener(fn)` - your `fn` is called with every message the unit sends, asked for or not. It runs on the transport's read thread, so it must not block and may not read from the device. To catch the connect handshake's own burst of state, register before it with `protocol.connect(before_handshake=...)` | | **Scenes** | `copy_scene(from_scene, to_scene, swap=False)`, `set_scene_label(scene, label)`, `set_scene_color(scene, argb)` | | **Global settings** | `settings()`, `update_settings(**fields)`, `set_scene_bypass_behavior()`, `set_global_bypass()`, `set_master_volume_assignment()`, `mode()`, `set_mode()`, `set_mode_cycle()`, `set_gig_view()` | | **Global EQ** | `global_eq()`, `set_global_eq(band, gain=, frequency=, q=, filter_type=, enabled=)`, `set_global_eq_output(level=, out12=, out34=)`, `set_global_eq_bypassed()` | diff --git a/docs/architecture.md b/docs/architecture.md index fe1bbc0..a70093f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -106,6 +106,10 @@ concurrent: gunzips frame-level compressed payloads, parses the protobuf, and dispatches; - correlation: `request()` waiters keyed by `request_id`, plus `await_broadcast()` waiters keyed by message class and an optional predicate; +- persistent subscriptions: `add_listener()` registers a callable that sees every + decoded message for as long as the connection lasts, including the unsolicited + pushes no waiter is expecting. It consumes nothing, so waiters and collectors + behave exactly as they do with no listener registered; - a keepalive thread; - tolerating the device's benign write STALL (see [protocol.md](protocol.md#the-benign-write-stall)): write errors are logged at @@ -117,6 +121,16 @@ types and non-protobuf pushes are skipped at debug level, and the reassembly buffer is reset on anything malformed so one bad frame cannot wedge the stream. If you add code to the RX path, preserve that property. +Listeners run on that thread, so the same rule covers them: one that raises is +logged and skipped, its peers still get the message, and the message still +reaches its waiter. A listener may also not read from the device - +`request`, `await_broadcast` and `collect` raise `RuntimeError` when called from +the RX thread, because the RX thread is the one that would have to deliver the +answer, so such a call could only ever time out with the read loop stopped behind +it. A listener applies what a push carries and notes what needs re-reading; the +caller's thread does the re-reading (see [domain-model.md](domain-model.md) +section 9, and ADR-0008). + ### registry.py The only place that knows the mapping between the schema's @@ -153,6 +167,14 @@ remembers what it opened (`_owned_resources`) so `close()` and the context manager tear down only what `connect()` created. A client built around a caller-supplied transport owns nothing and `close()` is a no-op. +`connect(before_handshake=...)` is the hook for anything that has to be watching +before the handshake runs. The subscription burst the handshake sends is what +makes the unit start pushing state, and that state arrives AFTER `connect()` has +returned (measured: the client comes back at 2 s, the ModelRepo lands at 4.9 s and +the current preset at 10.1 s - see [protocol.md](protocol.md), "Connect burst, +measured"). So a listener registered on the returned client has already missed it; +one registered through this hook has not. + `import hid` lives *inside* `open_device()`. That laziness is a contract, not an accident: see [Testing philosophy](#testing-philosophy). @@ -200,18 +222,22 @@ device.read() -> one 129-byte input report -> framing.decode_reports(buffer) -> (message_type, payload) -> gunzip payload if it starts 1f 8b -> registry.class_for(message_type) -> parse - -> _dispatch: a request_id waiter, else a broadcast waiter, else dropped + -> _dispatch: every listener, then collectors, then a request_id waiter, + else a broadcast waiter, else dropped ``` ## send vs request vs await_broadcast -Choosing correctly is most of the work of adding an operation. +Choosing correctly is most of the work of adding an operation. The first three +rows serve ONE exchange, which is what an operation needs. The last one is not an +operation at all: it is how a long-lived caller watches the link. | Transport method | Use when | Blocking | Correlation | |---|---|---|---| | `send(msg)` | The device acts on the message and you do not need its answer: scene switch, grid edits, recall, keepalive. | No | None | | `request(msg, timeout=)` | The device answers a message of the **same type**: `Version` READ, `ResetCommsBuffers`, the `File` mutations. | Yes | Fresh `request_id` is assigned and registered before the write. Reply is the first inbound message of the same type whose `request_id`, if present on both sides, matches. | | `await_broadcast(cls, trigger, timeout=, match=)` | The answer arrives as a **push of a different type**, or as an unsolicited broadcast the device emits in response to an action: the `RecallPreset` push that carries a full preset, the `File` folder listings. | Yes | By message class, plus your optional `match` predicate. A right-type message the predicate rejects is left undelivered so a later one can satisfy the waiter. | +| `add_listener(fn)` | You want EVERY message for the life of the connection, not the answer to one call: a cache fed by the unit's own pushes, or a log of the link. | No, but `fn` runs on the RX thread | None. Every message, every type, whether or not a waiter also gets it. Removed with the returned callable or `remove_listener(fn)`. | Two gotchas the current code already encodes, and that new operations must respect: diff --git a/docs/protocol.md b/docs/protocol.md index 15952ab..ae8f6f3 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -1766,6 +1766,14 @@ once - including the seed RecallPreset - at about **9 s after connect**, consist several sessions on d14e. (Earlier sessions recorded 10-25 s for lazily-serviced pushes; keep timeouts generous, but 9 s is the typical seed arrival.) +**`connect()` returns before the burst arrives.** Re-measured 2026-08-12 on d14e with a +transport listener registered before the handshake: `connect()` handed back its client 2.0 s +in, having seen only the `ResetCommsBuffers` echo and the unit's own `Version` READ. The +ModelRepo landed at 4.9 s, the 399 `File` listings and most settings at 5.1 s, and the seed +`RecallPreset` at 10.1 s - 474 messages of 24 distinct types by 15 s. So a listener attached +to the client `connect()` returns is about 3 s too late for the ModelRepo and 8 s too late +for the current preset, which is why `connect(before_handshake=...)` exists. + Two ambient-traffic facts for anyone instrumenting the link: `GlobalTempo` streams one pair of messages per BEAT - 1.5 s apart at 40 bpm - so its rate follows the tempo and it is a poor heartbeat but a decent liveness hint; and a single knob turn on the touchscreen diff --git a/pyquadcortex/protocol/client.py b/pyquadcortex/protocol/client.py index d5192dd..e4f42f9 100644 --- a/pyquadcortex/protocol/client.py +++ b/pyquadcortex/protocol/client.py @@ -239,6 +239,33 @@ def __exit__(self, exc_type, exc, tb): self.close() return False + # -- live state pushes ----------------------------------------------------- + + def add_listener(self, listener): + """Call ``listener(message)`` for every message the device sends. + + A pass-through to + :meth:`~pyquadcortex.protocol.transport.Transport.add_listener`, which is + where the contract lives and is worth reading before you use this: the + listener runs on the transport's RX thread, so it must not block, and it + may not read from the device (the transport refuses that outright). It is + removed with the returned callable or with :meth:`remove_listener`. + + This is how a long-lived caller sees the state the unit pushes without + being asked - a touchscreen edit, a preset recall, the metronome's tempo + stream - rather than the one-shot answer to a call it just made. To catch + the connect handshake's own burst of state, register before the handshake + with ``protocol.connect(before_handshake=...)``. + """ + return self._t.add_listener(listener) + + def remove_listener(self, listener): + """Stop calling ``listener``; return True if it had been registered. + + Never raises, so teardown can call it unconditionally. + """ + return self._t.remove_listener(listener) + # -- session ------------------------------------------------------------- # State types the device only PUSHES to a client that has subscribed by diff --git a/pyquadcortex/protocol/session.py b/pyquadcortex/protocol/session.py index edb347d..9103fbb 100644 --- a/pyquadcortex/protocol/session.py +++ b/pyquadcortex/protocol/session.py @@ -80,7 +80,8 @@ def open_device(): def connect(*, timeout: float = 5.0, settle: float = 2.0, - handshake_patience: float = 30.0) -> QuadCortex: + handshake_patience: float = 30.0, + before_handshake=None) -> QuadCortex: """Open a Quad Cortex and return a connected, ready-to-use client. Finds and opens the device, starts the transport, and performs the connect @@ -107,6 +108,16 @@ def connect(*, timeout: float = 5.0, settle: float = 2.0, failing, which is why the default is 30. Each attempt restarts the full handshake (safe: it begins with a fresh session id). Set to 0 for the old single-attempt behaviour. + before_handshake: optional ``callable(transport)``, called once with the + started :class:`~pyquadcortex.protocol.transport.Transport` after it + starts and before the handshake runs. This is the only way to + register a listener + (:meth:`~pyquadcortex.protocol.transport.Transport.add_listener`) + early enough to see the handshake's own burst of state, which + delivers one message of nearly every state type the unit has - the + cheapest way to learn what the unit is currently doing. Called once, + not once per handshake attempt. An exception from it aborts the + connect and releases the device, like any other bring-up failure. Returns: A connected :class:`~pyquadcortex.protocol.client.QuadCortex`. @@ -120,6 +131,10 @@ def connect(*, timeout: float = 5.0, settle: float = 2.0, owned = [device.close, transport.stop] try: transport.start() + # Before the handshake, so a listener registered here sees the state + # burst the handshake provokes rather than joining after it. + if before_handshake is not None: + before_handshake(transport) qc = QuadCortex(transport, _owned_resources=owned) deadline = time.monotonic() + handshake_patience attempt = 0 diff --git a/pyquadcortex/protocol/transport.py b/pyquadcortex/protocol/transport.py index 4c95f77..c1f9b98 100644 --- a/pyquadcortex/protocol/transport.py +++ b/pyquadcortex/protocol/transport.py @@ -8,6 +8,9 @@ * runs a background RX thread that reads input reports, reassembles multi-report messages, decodes them, and correlates responses to callers by ``request_id``; + * hands every decoded message to any persistent listener (``add_listener``), + which is how a long-lived caller sees the unsolicited pushes no waiter is + expecting; * runs a background keepalive thread that periodically pokes the device so it keeps the session alive. @@ -24,6 +27,10 @@ * The RX thread must never die: every per-message decode/parse is wrapped so a malformed frame or unknown message type is logged and skipped, and the reassembly buffer is reset so one bad frame cannot wedge the stream. + * Listeners run ON the RX thread, so the same rule covers them: one that raises + is logged and skipped, and one may not issue a correlated read (``request``, + ``await_broadcast``, ``collect`` refuse to run on that thread - see + ``_refuse_read_from_rx``). * The keepalive thread swallows send failures and keeps going. """ @@ -102,6 +109,10 @@ def __init__(self, device, keepalive_interval=5.0): # (expected_class, Event, [response|None]). self._type_waiters = [] self._collectors = [] + # Persistent listeners: called with EVERY decoded inbound message until + # removed. Unlike the three above, not scoped to one trigger or one + # reply. See add_listener. + self._listeners = [] self._lock = threading.Lock() # guards _pending / _ids (state only) # Serializes device writes so each logical message's reports are written # as an atomic group (a keepalive can't slip between a multi-report @@ -174,6 +185,30 @@ def _confirm_lost(self, error): for _cls, _match, ev, _slot in waiters: ev.set() + # -- the RX thread may not read -------------------------------------------- + + def _refuse_read_from_rx(self, what): + """Refuse a correlated wait attempted from the RX thread. + + The RX thread is the only thread that delivers a reply, so a wait issued + from inside it can never be satisfied: it would sit out its entire + timeout with the read loop stopped behind it, which is the "the RX thread + never blocks" rule broken in the worst way available. Listeners + (:meth:`add_listener`) are the only caller code that runs on that thread, + so this guard is what makes the listener contract enforced rather than + merely requested (ADR-0008). + + Cheap enough to leave in every entry point: one identity comparison. + """ + if threading.current_thread() is self._rx: + raise RuntimeError( + f"{what}() was called from the RX thread, which is the thread " + f"that would have to deliver the answer - so it can only ever " + f"time out. A listener applies what a push carries and notes " + f"what needs re-reading; the caller's thread does the " + f"re-reading (docs/domain-model.md section 9)." + ) + # -- outbound ------------------------------------------------------------ def send(self, message): @@ -238,7 +273,11 @@ def request(self, message, timeout=5.0): the same request_id before the SetlistPosition echo). So the reply is the first inbound message whose TYPE matches the request's, and whose request_id - if present on both sides - matches too. + + Refused when called from the RX thread, where it could only ever time out + (see ``_refuse_read_from_rx``). """ + self._refuse_read_from_rx("request") self._check_lost() ev = threading.Event() slot = [None] @@ -281,7 +320,12 @@ def collect(self, expected_class, trigger, seconds, match=None): Returns the messages in arrival order. Unlike a waiter, a collector does not consume messages - they still reach any waiter or other collector. + + Refused when called from the RX thread, which would stall the read loop + for the whole window and so collect nothing (see + ``_refuse_read_from_rx``). """ + self._refuse_read_from_rx("collect") self._check_lost() got = [] entry = (expected_class, match, got) @@ -317,7 +361,11 @@ def await_broadcast(self, expected_class, trigger, timeout=40.0, match=None): the id on the push). The device services large pushes lazily (10-25s observed), hence the generous default timeout. Raises ``TimeoutError`` on no matching broadcast. + + Refused when called from the RX thread, where it could only ever time out + (see ``_refuse_read_from_rx``). """ + self._refuse_read_from_rx("await_broadcast") self._check_lost() ev = threading.Event() slot = [None] @@ -345,6 +393,113 @@ def await_broadcast(self, expected_class, trigger, timeout=40.0, match=None): self._check_lost() # woken by _confirm_lost, not by a broadcast return slot[0] + # -- persistent listeners -------------------------------------------------- + + def add_listener(self, listener): + """Register ``listener`` to see EVERY decoded inbound message. + + The transport's other three inbound hooks are one-shot and scoped to a + trigger: :meth:`request` correlates one reply, :meth:`await_broadcast` + waits for one push, :meth:`collect` gathers for a fixed number of + seconds. A listener is none of those. It stays registered until it is + removed and sees every message the RX thread decodes - including the + unsolicited pushes no waiter is expecting, which :meth:`_dispatch` would + otherwise drop at debug level. That is what a push-fed cache needs (see + ``docs/domain-model.md`` section 9). + + ``listener`` is called as ``listener(message)`` with the parsed protobuf. + + Additive by construction: a listener does not CONSUME a message. It is + notified first, and the message then reaches every collector and waiter + exactly as it would have with no listener registered. + + Registration and removal are safe while the RX thread is running. + Returns a zero-argument callable that removes this registration; + :meth:`remove_listener` does the same job for a caller who kept the + listener rather than the callable. + + **Listeners run on the RX thread**, synchronously, in registration order, + before the message reaches its waiter (so a cache fed by a listener is + already current when the blocked caller wakes). Two consequences: + + * **A listener must not block.** It spends the RX thread's time: whatever + it does delays the next report being read. Apply the push and return. + * **A listener may not read from the device**, and that is enforced + rather than asked for: :meth:`request`, :meth:`await_broadcast` and + :meth:`collect` raise ``RuntimeError`` when called from the RX thread + (see ``_refuse_read_from_rx``). Such a call could never have worked - + the RX thread is the one that delivers replies, so a wait from inside + it only ever times out - and the rule it breaks is older than this + method: the RX thread applies pushes and notes what needs re-reading, + and the caller's thread does the re-reading + (``docs/domain-model.md`` section 9). :meth:`send` is NOT refused, + being fire-and-forget, but a listener that writes owns the delay it + adds to the read loop. + + A listener that raises is logged and skipped: the RX thread survives, the + other listeners still see that message, and the message still reaches its + waiter. Same contract as every other step in this module's RX path. + + A listener lives only as long as the connection. Device loss neither + removes nor notifies listeners - there is simply nothing further to + deliver - and a new connection means a new ``Transport`` and a new + registration. + + Evidence: the mechanism is proven offline against ``FakeHid`` + (``tests/test_transport.py``). That a listener registered before the + connect handshake sees the handshake's state burst is confirmed on + hardware (``tests/hardware/test_broadcast_listener.py``). Registering + that early needs ``protocol.connect(before_handshake=...)``, because + ``connect`` runs the handshake before it hands the client back. + + Raises: + DeviceLostError: if the device is already known to be gone. + """ + self._check_lost() + with self._lock: + self._listeners.append(listener) + return lambda: self.remove_listener(listener) + + def remove_listener(self, listener): + """Unregister ``listener``; return True if it had been registered. + + Never raises and is safe to call twice, so a teardown path can call it + unconditionally. Removal is by equality, which is what makes + ``remove_listener(self._apply)`` work for a bound method: each attribute + access builds a new object, and those compare equal. + + Safe to call from inside a listener, though the message being delivered + may still reach the listener being removed - notification runs over a + snapshot taken before the first listener was called. + """ + with self._lock: + try: + self._listeners.remove(listener) + except ValueError: + return False + return True + + def _notify_listeners(self, message): + """Hand ``message`` to every listener (RX thread). + + Snapshot under the lock, call outside it: a listener that registers or + removes a listener would otherwise deadlock on the non-reentrant state + lock, and holding that lock across arbitrary caller code is exactly what + the rest of this module avoids. + """ + with self._lock: + listeners = list(self._listeners) + for listener in listeners: + try: + listener(message) + except Exception: + # The RX thread never dies, and one broken listener never costs + # its peers or the waiter their copy of this message. + log.exception( + "inbound listener %r raised on %s; skipping it for this " + "message", listener, type(message).__name__ + ) + # -- inbound ------------------------------------------------------------- def _read_loop(self): @@ -471,7 +626,12 @@ def _dispatch(self, message): that id belongs to the waiter. Cascade messages of other types (which echo the id of the request that caused them) and unsolicited broadcasts simply find no waiter and are dropped at debug level. + + Persistent listeners (:meth:`add_listener`) are notified FIRST, before any + collector or waiter, and consume nothing: the routing below runs exactly + as it would with no listener registered. """ + self._notify_listeners(message) with self._lock: collectors = [c for c in self._collectors if c[0] is type(message) and (c[1] is None or c[1](message))] diff --git a/tests/hardware/conftest.py b/tests/hardware/conftest.py index daa422e..c63030b 100644 --- a/tests/hardware/conftest.py +++ b/tests/hardware/conftest.py @@ -12,6 +12,7 @@ Without the flag nothing here is collected, so the offline suite stays honest with no unit attached. """ +import threading import time import pytest @@ -22,17 +23,77 @@ def pytest_ignore_collect(collection_path, config): return not config.getoption("--hardware") +class HandshakeBurst: + """Records the type name of every message the unit pushes, from the start. + + Attached by the connection fixture through + ``protocol.connect(before_handshake=...)``, which is the only moment early + enough to catch the handshake's burst - by the time ``connect`` returns, the + burst is over. + + Records NAMES rather than messages, and stops at ``LIMIT``. The metronome's + tempo stream never stops, so an unbounded recorder on a connection that lives + for the whole run would keep growing all run; the burst is only the first few + hundred messages of it. + + Runs on the transport's RX thread, so it does the least it can: take the + lock, append, return. + """ + + LIMIT = 4000 + + def __init__(self): + self._lock = threading.Lock() + self._names = [] + self.dropped = 0 + + def __call__(self, message): + with self._lock: + if len(self._names) < self.LIMIT: + self._names.append(type(message).__name__) + else: + self.dropped += 1 + + def names(self): + """A snapshot of what has been recorded, in arrival order.""" + with self._lock: + return list(self._names) + + @pytest.fixture(scope="session") -def qc(): - """One connection for the whole run; the handshake is expensive. +def _connection(): + """The run's single connection, with the handshake burst recorded. + + One connection, because the handshake is expensive - and because the unit + only lets one process hold the HID interface, so a test that opened a second + one would fail on whatever order it ran in. This is a PROTOCOL-level suite, so it connects through :mod:`pyquadcortex.protocol` and gets a ``QuadCortex``. ``pyquadcortex.connect()`` returns the model's ``Device`` instead (ADR-0006). + + The burst recorder is attached for every run, not just the tests that read + it: registering it costs one list append per message, and it cannot be + attached later on demand, because the burst happens during ``connect``. """ from pyquadcortex import protocol - with protocol.connect() as client: - yield client + burst = HandshakeBurst() + with protocol.connect( + before_handshake=lambda transport: transport.add_listener(burst) + ) as client: + yield client, burst + + +@pytest.fixture(scope="session") +def qc(_connection): + """The connected ``QuadCortex`` every test in this suite drives.""" + return _connection[0] + + +@pytest.fixture(scope="session") +def handshake_burst(_connection): + """The :class:`HandshakeBurst` that listened through the connect handshake.""" + return _connection[1] @pytest.fixture diff --git a/tests/hardware/readme.md b/tests/hardware/readme.md index d4f24ee..cc6afc7 100644 --- a/tests/hardware/readme.md +++ b/tests/hardware/readme.md @@ -26,6 +26,18 @@ could not put back, rather than aborting on the first. That message is the list to fix by hand. Global settings are the ones worth checking first, since they survive a preset recall. +## One connection, and why it records the connect burst + +Every test shares one connection, because the unit lets only one process hold the +HID interface - a test that opened a second one would fail on whatever order it +happened to run in. + +That connection attaches a listener before the handshake and records the type of +every message the unit pushes. It is attached on every run, not only for the tests +that read it, because it cannot be attached later: the burst happens during +`connect()`. It records type names and stops at 4000 of them, so the metronome's +endless tempo stream cannot grow it for the length of the run. + ## Why the control test exists `test_parameter_echo_latency_is_the_control` measures a write whose latency was diff --git a/tests/hardware/test_broadcast_listener.py b/tests/hardware/test_broadcast_listener.py new file mode 100644 index 0000000..5cfa2f4 --- /dev/null +++ b/tests/hardware/test_broadcast_listener.py @@ -0,0 +1,196 @@ +"""What a persistent listener actually hears from a real unit. + +The offline tests prove the wiring: a listener is called, it consumes nothing, +and neither a raise nor an attempted device read can hurt the read loop +(``tests/test_transport.py``). Only the unit can prove there is anything to hear. +Two facts are checked here: + +* a listener registered BEFORE the connect handshake sees the handshake's state + burst - the one moment the unit volunteers nearly everything it knows, and the + reason ``protocol.connect(before_handshake=...)`` exists; +* a listener registered on a live connection keeps receiving, takes nothing away + from an ordinary request, and stops the moment it is removed. + +State-neutral by construction (ADR-0005): this file only listens. It sends no +write of any kind, so there is nothing to snapshot and nothing to restore. +""" +import threading +import time + +from pyquadcortex.protocol.proto import ProductionAutomation_pb2 as pa + +#: The metronome clock always runs, so the unit pushes GlobalTempo in pairs, one +#: pair per beat, on every connection - 1.5 s apart at the slowest tempo the unit +#: offers (40 bpm). Ten seconds is several beats even there. +UNSOLICITED_PATIENCE = 10.0 + +#: How long to wait for the handshake's state burst. Measured on this unit +#: (2026-08-12, CorOS 4.0.1 / d14e): ModelRepo at 4.9 s, the ~399 File listings +#: and most settings by 5.1 s, and the current preset at 10.1 s - 474 messages of +#: 24 distinct types by 15 s. Three times the measured figure, because the wait +#: ends as soon as the preset lands and a fixed sleep would only make the suite +#: slower. +BURST_PATIENCE = 30.0 + + +class Recorder: + """Collects messages from the RX thread for the calling thread to read.""" + + def __init__(self): + self._lock = threading.Lock() + self._messages = [] + + def __call__(self, message): + with self._lock: + self._messages.append(message) + + def messages(self): + with self._lock: + return list(self._messages) + + def names(self): + return [type(m).__name__ for m in self.messages()] + + +def _wait_until(predicate, timeout): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.05) + return predicate() + + +def test_a_listener_registered_before_connecting_sees_the_handshake_burst( + handshake_burst): + """The burst is what makes a push-fed cache warm for free. + + Note what the wait below says about the hook: ``connect()`` returns about 2 s + in, with only the ResetCommsBuffers echo and the unit's own Version READ + recorded, and the state burst starts arriving about 5 s in. So a listener + registered on the client ``connect()`` hands back is already too late, which + is the whole reason ``before_handshake`` exists. + + Confirmed on this unit (2026-08-12, CorOS 4.0.1 / d14e): 474 messages of 24 + distinct types by 15 s. The floors below sit well under that, because a test + pinned to the exact tally would fail on a unit with a different number of + presets rather than on a real regression. + """ + def tally(): + names = handshake_burst.names() + return {name: names.count(name) for name in sorted(set(names))} + + # The current preset lands last of the burst (10.1 s measured), so it is the + # signal that everything else has already arrived. + _wait_until(lambda: "RecallPresetMessage" in tally(), BURST_PATIENCE) + + counted = tally() + total = sum(counted.values()) + report = f"recorded {total} message(s) in {BURST_PATIENCE}s: {counted}" + + assert total >= 100, report + assert len(counted) >= 15, f"too few distinct state types in the burst - {report}" + # Nothing in the handshake REQUESTS these. The subscription is a burst of + # fire-and-forget READs, so almost every message here is one _dispatch would + # have dropped for want of a waiter. + assert "FileMessage" in counted, report # the folder enumeration + assert "RecallPresetMessage" in counted, report # the preset on the grid now + assert handshake_burst.dropped == 0, ( + f"the recorder hit its {handshake_burst.LIMIT}-message cap, so this tally " + f"is not the burst alone") + + +def test_a_listener_on_a_live_connection_hears_pushes_nobody_asked_for(qc): + """The tempo stream is the cheapest proof: the unit sends it unprompted.""" + listening = Recorder() + qc.add_listener(listening) + try: + assert _wait_until(lambda: listening.messages(), UNSOLICITED_PATIENCE), ( + f"nothing arrived in {UNSOLICITED_PATIENCE}s with no host request " + f"outstanding. The metronome's tempo stream should be enough on its " + f"own") + finally: + qc.remove_listener(listening) + + +def test_a_listener_does_not_take_the_reply_away_from_the_caller(qc): + listening = Recorder() + qc.add_listener(listening) + try: + reply = qc.version() + assert reply.app_fw_version, "version() lost its reply to the listener" + assert _wait_until( + lambda: "VersionMessage" in listening.names(), 2.0), ( + f"the listener never saw the reply it did not consume; it saw " + f"{listening.names()}") + finally: + qc.remove_listener(listening) + + +def test_removing_a_listener_stops_it_while_the_unit_keeps_pushing(qc): + """Removal has to stop delivery, and the proof has to survive a quiet unit. + + A second listener registered at the moment the first is removed is what makes + this honest: if it hears nothing either, the unit went quiet and the test says + so instead of crediting the removal. + """ + removed, still_listening = Recorder(), Recorder() + qc.add_listener(removed) + try: + assert _wait_until(lambda: removed.messages(), UNSOLICITED_PATIENCE), \ + "the unit pushed nothing at all, so this test cannot say anything" + qc.add_listener(still_listening) + assert qc.remove_listener(removed) is True + after_removal = len(removed.messages()) + + assert _wait_until(lambda: len(still_listening.messages()) >= 2, + UNSOLICITED_PATIENCE), ( + "the unit stopped pushing during the window, so a frozen count " + "proves nothing about removal") + assert len(removed.messages()) == after_removal, \ + "a removed listener was still being called" + finally: + qc.remove_listener(removed) + qc.remove_listener(still_listening) + + +def test_a_listener_may_not_read_from_the_device_on_a_real_link(qc): + """The rule the design doc states, on the real RX thread. + + Offline this is a thread-identity check against a fake. Here it is the actual + read loop of an actual connection, which is where a listener that tried to + re-read would take the link down with it. + """ + refusals = {} + done = threading.Event() + + def tries_to_read(message): + if refusals: + return + for name, attempt in ( + ("request", lambda: qc.version(timeout=0.5)), + ("await_broadcast", lambda: qc._t.await_broadcast( + pa.SceneMessage, lambda: None, timeout=0.5)), + ("collect", lambda: qc._t.collect(pa.SceneMessage, lambda: None, 0.5)), + ): + try: + attempt() + refusals[name] = None + except Exception as exc: # noqa: BLE001 - the type is the point + refusals[name] = exc + done.set() + + qc.add_listener(tries_to_read) + try: + assert done.wait(UNSOLICITED_PATIENCE), \ + "no push arrived, so the listener never ran" + finally: + qc.remove_listener(tries_to_read) + + for name, exc in refusals.items(): + assert isinstance(exc, RuntimeError), f"{name} was not refused: {exc!r}" + assert not isinstance(exc, TimeoutError), f"{name} waited instead of refusing" + assert "RX thread" in str(exc) + # The link is unharmed: a refusal costs one message's listener call, not the + # connection. + assert qc.version().app_fw_version diff --git a/tests/test_client.py b/tests/test_client.py index 35baaf3..a4ee567 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -25,6 +25,7 @@ def __init__(self, canned=None): self.canned = canned or {} self.broadcast = None self.last_match = None # the match predicate read_preset passed, if any + self.listeners = [] self._ids = itertools.count(1) def send(self, msg): @@ -42,6 +43,16 @@ def await_broadcast(self, expected_class, trigger, timeout=40.0, match=None): trigger() return self.broadcast + def add_listener(self, listener): + self.listeners.append(listener) + return lambda: self.remove_listener(listener) + + def remove_listener(self, listener): + if listener in self.listeners: + self.listeners.remove(listener) + return True + return False + # -- 5.1 read_current_preset ------------------------------------------------- @@ -3130,3 +3141,27 @@ def test_set_master_volume_accepts_both_ends(edge): qc = client.QuadCortex(FakeTransport()) qc.set_master_volume(edge) assert qc._t.sent[-1].volume == edge + + +# -- listening to what the device pushes --------------------------------------- + + +def test_add_and_remove_listener_pass_straight_through_to_the_transport(): + # The client's whole job here is to spare the caller reaching into qc._t. The + # contract - RX thread, no blocking, no reads - belongs to the transport and is + # tested there. + fake = FakeTransport() + qc = client.QuadCortex(fake) + seen = [] + + drop = qc.add_listener(seen.append) + assert fake.listeners == [seen.append] + + drop() + assert fake.listeners == [] + assert qc.remove_listener(seen.append) is False, \ + "removing what is not registered reports so rather than raising" + + qc.add_listener(seen.append) + assert qc.remove_listener(seen.append) is True + assert fake.listeners == [] diff --git a/tests/test_session.py b/tests/test_session.py index 8d51c43..d8a84dc 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -95,6 +95,40 @@ def boom(self, *a, **kw): assert fake_stack.closed, "device must be closed when bring-up fails" +def test_before_handshake_runs_after_start_and_before_the_handshake(fake_stack): + """The hook exists so a listener can catch the handshake's own state burst. + + Registered a moment later - after connect() returns - and the burst is over. + So what matters is the ORDER: the transport is started (its RX thread is + reading) and nothing of the handshake has been sent yet. + """ + calls = [] + + def before(t): + calls.append((t, t.started, list(t.sent))) + + qc = session.connect(settle=0, before_handshake=before) + t = FakeTransport.instances[0] + assert len(calls) == 1, "the hook runs once, not once per handshake attempt" + got, started, sent_by_then = calls[0] + assert got is t, "the hook gets the transport a listener registers on" + assert started, "the RX thread must already be reading" + assert sent_by_then == [], "the hook ran after the handshake had begun" + assert type(t.sent[0]).__name__ == "ResetCommsBuffersMessage" + qc.close() + + +def test_a_failing_before_handshake_hook_does_not_leak_the_device(fake_stack): + def boom(t): + raise RuntimeError("the listener could not be registered") + + with pytest.raises(RuntimeError, match="could not be registered"): + session.connect(settle=0, before_handshake=boom) + t = FakeTransport.instances[0] + assert t.stopped, "transport must be stopped when the hook fails" + assert fake_stack.closed, "device must be closed when the hook fails" + + def test_client_with_caller_supplied_transport_does_not_own_it(): """A hand-wired QuadCortex must not close a transport it did not open.""" device = FakeDevice() diff --git a/tests/test_transport.py b/tests/test_transport.py index c36befd..9c80ae9 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -411,16 +411,11 @@ def test_collect_gathers_every_matching_push_without_consuming_them(): # One request can provoke hundreds of pushes (a File READ enumerates the # device's whole folder tree), so collect() accumulates rather than taking # the first, and leaves messages available to waiters. - import threading - from pyquadcortex.protocol import transport as tmod - from pyquadcortex.protocol.proto import ProductionAutomation_pb2 as pa - - t = tmod.Transport.__new__(tmod.Transport) - t._lock = threading.RLock() - t._pending = {} - t._type_waiters = [] - t._collectors = [] - t._device_lost = None + # + # A real Transport, never started: the trigger dispatches on the calling + # thread, so no RX thread is needed and none of the transport's state has to + # be faked. + t = transport.Transport(FakeHid(), keepalive_interval=QUIET_KEEPALIVE) def trigger(): for i in range(3): @@ -436,6 +431,171 @@ def trigger(): assert t._collectors == [], "the collector is removed when done" +# -- persistent listeners ------------------------------------------------------ +# add_listener is the only inbound hook that is not scoped to one trigger or one +# reply, so what these tests protect is mostly what it must NOT do: consume a +# message, block the read loop, or take its peers down with it. + + +def test_a_listener_sees_an_unsolicited_push_no_waiter_wanted(): + # The case the other three hooks cannot serve: a push nobody asked for. With + # no listener this message reaches _dispatch, matches nothing, and is dropped + # at debug level. + fake = FakeHid() + t = transport.Transport(fake, keepalive_interval=QUIET_KEEPALIVE) + seen = [] + t.start() + try: + t.add_listener(seen.append) + fake.inject(*_recall_broadcast("unsolicited", rid=None)) + assert _wait_until(lambda: len(seen) == 1), "the push never reached the listener" + finally: + t.stop() + assert isinstance(seen[0], pa.RecallPresetMessage) + assert seen[0].preset.name == "unsolicited" + + +def test_a_listener_does_not_steal_a_message_from_its_waiter(): + # A listener is additive: the reply still lands in request()'s hands. + fake = FakeHid() + t = transport.Transport(fake, keepalive_interval=QUIET_KEEPALIVE) + seen = [] + t.start() + try: + t.add_listener(seen.append) + resp = t.request( + pa.VersionMessage(action=pa.MessageAction.READ), timeout=REQUEST_TIMEOUT + ) + finally: + t.stop() + assert resp.request_id == 1, "the waiter did not get its reply" + assert [type(m) for m in seen] == [pa.VersionMessage] + assert seen[0] is resp, "the listener and the waiter get the same message" + + +def test_a_listener_has_already_run_when_the_blocked_caller_wakes(): + # The ordering a push-fed cache depends on: listeners are notified before the + # waiter's event is set, so a cache fed by a listener is current by the time + # the caller that provoked the reply gets it back. No sleeps - if the order + # were the other way round, the list would still be empty here. + fake = FakeHid() + t = transport.Transport(fake, keepalive_interval=QUIET_KEEPALIVE) + seen = [] + t.start() + try: + t.add_listener(seen.append) + t.request( + pa.VersionMessage(action=pa.MessageAction.READ), timeout=REQUEST_TIMEOUT + ) + assert len(seen) == 1, "the caller woke before the listener had the message" + finally: + t.stop() + + +def test_a_raising_listener_costs_nobody_else_the_message(caplog): + # Wrap and log, like every other decode step in the RX path: the peers still + # see the message, the waiter still gets its reply, and the read loop is still + # running afterwards. + fake = FakeHid() + t = transport.Transport(fake, keepalive_interval=QUIET_KEEPALIVE) + before, after = [], [] + + def explodes(message): + raise RuntimeError("a listener with a bug in it") + + t.start() + try: + t.add_listener(before.append) + t.add_listener(explodes) + t.add_listener(after.append) + with caplog.at_level("ERROR", logger="pyquadcortex.protocol.transport"): + first = t.request( + pa.VersionMessage(action=pa.MessageAction.READ), + timeout=REQUEST_TIMEOUT, + ) + # A second round trip through the same RX thread: had the raise killed + # it, this would time out rather than answer. + second = t.request( + pa.VersionMessage(action=pa.MessageAction.READ), + timeout=REQUEST_TIMEOUT, + ) + finally: + t.stop() + assert first is not None and second is not None + assert len(before) == 2, "a listener registered before the raiser lost a message" + assert len(after) == 2, "a listener registered after the raiser lost a message" + assert "a listener with a bug in it" in caplog.text, \ + "a raising listener must be logged, not silently swallowed" + + +def test_a_listener_cannot_read_from_the_device_on_the_rx_thread(): + # The design rule this enforces: the RX thread applies pushes and notes what + # needs re-reading, and the caller's thread does the re-reading. All three + # correlated waits are refused, and refused with RuntimeError rather than left + # to time out - the RX thread is the thread that would have to deliver the + # answer, so waiting for one from inside it can only stall the whole link. + fake = FakeHid() + t = transport.Transport(fake, keepalive_interval=QUIET_KEEPALIVE) + refusals = {} + + def tries_to_read(message): + attempts = { + "request": lambda: t.request( + pa.VersionMessage(action=pa.MessageAction.READ), timeout=0.1), + "await_broadcast": lambda: t.await_broadcast( + pa.SceneMessage, lambda: None, timeout=0.1), + "collect": lambda: t.collect(pa.SceneMessage, lambda: None, 0.1), + } + for name, attempt in attempts.items(): + try: + attempt() + refusals[name] = None # allowed through: the guard is not working + except Exception as exc: # noqa: BLE001 - the point is what type it is + refusals[name] = exc + + t.start() + try: + t.add_listener(tries_to_read) + fake.inject(*_recall_broadcast("push", rid=None)) + assert _wait_until(lambda: len(refusals) == 3) + # The link still works, which is the whole point of refusing rather than + # letting a listener sit in a wait. + assert t.request(pa.VersionMessage(action=pa.MessageAction.READ), + timeout=REQUEST_TIMEOUT) is not None + finally: + t.stop() + for name, exc in refusals.items(): + assert isinstance(exc, RuntimeError), f"{name} was not refused: {exc!r}" + assert not isinstance(exc, TimeoutError), f"{name} waited instead of refusing" + assert "RX thread" in str(exc) + + +def test_removing_a_listener_actually_removes_it(): + fake = FakeHid() + t = transport.Transport(fake, keepalive_interval=QUIET_KEEPALIVE) + by_callable, by_identity = [], [] + t.start() + try: + drop = t.add_listener(by_callable.append) + t.add_listener(by_identity.append) + fake.inject(*_recall_broadcast("first", rid=None)) + assert _wait_until(lambda: len(by_callable) == 1 and len(by_identity) == 1) + + drop() # the handle add_listener returned + assert t.remove_listener(by_identity.append) is True # ... or by equality + assert t.remove_listener(by_identity.append) is False, \ + "removing twice must report that there was nothing to remove" + + fake.inject(*_recall_broadcast("second", rid=None)) + assert _wait_until(lambda: fake.pending_reads() == 0) + assert t.request(pa.VersionMessage(action=pa.MessageAction.READ), + timeout=REQUEST_TIMEOUT) is not None # the push was handled + finally: + t.stop() + assert [m.preset.name for m in by_callable] == ["first"] + assert [m.preset.name for m in by_identity] == ["first"] + + # -- device loss --------------------------------------------------------------- # The unplug sequence as measured on macOS: the FIRST read exception carries the # same text as the benign write stall; the SECOND says "Device is disconnected". From 68d358e26e8b7a3c2576c3cea269ec69f1fce34f Mon Sep 17 00:00:00 2001 From: Jonathan Stokes Date: Wed, 12 Aug 2026 18:14:42 -0500 Subject: [PATCH 2/3] review: close the findings from the PR #20 review Fixed: * add_listener now says the message is not a copy. It is the same object the next listener and the waiter receive, so a listener that tidies it in place changes what they see - and the first consumer is a cache that merges fields out of partial pushes, which is exactly the code most likely to want to. * The refusal's error text said the call "can only ever time out". True of request and await_broadcast, wrong for collect, which would return empty having stalled the link for its full duration. Corrected in the message, the docstring, ADR-0008 and the changelog. * remove_listener documents what removal-by-equality opens: the same callable registered twice is called twice and needs two removals, and a listener class with __eq__ can have an equal-but-different registration removed. * The _lock comment still read "guards _pending / _ids (state only)" and had been understating for a while. * test_before_handshake's "runs once, not once per handshake attempt" was vacuous - the fake handshake succeeded first try, so one attempt happened. Now a separate test drives three attempts, and it fails if the hook moves inside session.connect's retry loop (verified by moving it). The hardware suite's burst recorder was asserting on "everything since connect", not on the burst: it never stopped recording, so the tally held the tempo stream and every other test's traffic, and "the seed preset is in there" was true only because of alphabetical file order. Now the connection fixture waits for the burst to finish and stops the recorder before the first test runs, so the recording is the burst whatever order the suite runs in. That wait costs about 8 s once and buys more than it costs: connect() returns about 3 s before the unit starts streaming several hundred messages, so without it every latency measurement in the suite is taken on a link still answering the handshake. Suite wall time is unchanged, because the burst test used to do this waiting itself. The stop cannot be checked with a unit attached - the hardware test reads the recording afterwards and its assertions are floors, which contamination satisfies too - so tests/test_handshake_burst_recorder.py pins it offline, the way test_scene_echo_predicates.py pins the echo predicates. Declined, with reasons: * Skipping the listener snapshot's lock when no listeners are registered. Two uncontended lock operations at a peak of about 80 messages/s is not a cost worth paying for a GIL-dependent fast path and the comment explaining why the race is benign. * Forwarding before_handshake through the model's connect(). The model half of #11 is being written separately and will need it in the file it is already editing; adding it here only makes a conflict. --- changelog.md | 9 +- docs/ADR.md | 2 +- pyquadcortex/protocol/transport.py | 41 ++++-- tests/hardware/conftest.py | 86 +++++++++---- tests/hardware/readme.md | 19 ++- tests/hardware/test_broadcast_listener.py | 37 +++--- tests/test_handshake_burst_recorder.py | 147 ++++++++++++++++++++++ tests/test_session.py | 25 +++- 8 files changed, 302 insertions(+), 64 deletions(-) create mode 100644 tests/test_handshake_burst_recorder.py diff --git a/changelog.md b/changelog.md index c6b5b53..fd08833 100644 --- a/changelog.md +++ b/changelog.md @@ -112,9 +112,12 @@ device: - **Do not block in it.** Whatever it does delays the next message being read. - **Do not read from the device in it.** That thread is the one that would have to - deliver the answer, so the call could only ever time out - with the connection - stalled behind it. Rather than let that happen, the library raises - `RuntimeError` if you try. Note what you need and read it from your own thread. + deliver the answer, so the call could never be answered - and the connection + would stall behind it for as long as it waited. Rather than let that happen, the + library raises `RuntimeError` if you try. Note what you need and read it from + your own thread. +- **Treat the message as read-only.** It is the same object the rest of the + library sees, not a copy. To hear the burst of state the unit sends when a client connects - nearly everything it knows, including the preset currently loaded - register before the diff --git a/docs/ADR.md b/docs/ADR.md index 60d3608..0f2ffd3 100644 --- a/docs/ADR.md +++ b/docs/ADR.md @@ -101,7 +101,7 @@ Records are append-only once `Decided` and built upon: a shipped decision is nev - **(b) A delivery thread with a queue.** The RX thread would be immune to a slow listener, at the cost of a thread, an unbounded queue, and a cache that lags the reply that fed it. It buys immunity from a listener that blocks while making one that blocks harmless enough to survive unnoticed. - **(c) Call listeners on the RX thread and document the rules without enforcing them.** The rule that matters ("do not read from the device here") is invisible when broken: the read appears to work, then times out, having stopped the read loop for the whole timeout. This project's oldest lesson is that a failure which looks like success is the expensive kind. - **Open Questions:** Whether a listener should be notified when the device is lost. Today it simply stops receiving, which is enough for a cache whose owner learns about loss from the exception on its next call. Reconnect (issue #15) is where this gets answered. -- **Rationale:** The design already says the RX thread applies pushes and notes what needs re-reading while the caller's thread does the re-reading. Option (a) is that sentence in code. The enforcement costs one thread-identity comparison per call and converts a silent, connection-wide stall into an immediate error naming the rule - and it can refuse nothing that ever worked, because a correlated wait issued from the thread that delivers replies can only ever time out. +- **Rationale:** The design already says the RX thread applies pushes and notes what needs re-reading while the caller's thread does the re-reading. Option (a) is that sentence in code. The enforcement costs one thread-identity comparison per call and converts a silent, connection-wide stall into an immediate error naming the rule - and it can refuse nothing that ever worked, because a wait issued from the thread that delivers replies can never be satisfied. `request` and `await_broadcast` would time out; `collect` would return empty, having stalled the link for its full duration. - **Consequences:** - A listener must return promptly. Anything expensive belongs on the caller's thread, reached by noting what needs doing rather than doing it. - Nothing may weaken the refusal to keep a convenience. A future listener that wants a value it did not receive marks it for re-reading; it does not fetch it. diff --git a/pyquadcortex/protocol/transport.py b/pyquadcortex/protocol/transport.py index c1f9b98..b3b62c8 100644 --- a/pyquadcortex/protocol/transport.py +++ b/pyquadcortex/protocol/transport.py @@ -113,7 +113,9 @@ def __init__(self, device, keepalive_interval=5.0): # removed. Unlike the three above, not scoped to one trigger or one # reply. See add_listener. self._listeners = [] - self._lock = threading.Lock() # guards _pending / _ids (state only) + # Guards every registry above plus _ids. State only: never held across + # blocking device I/O, and never held while calling a listener. + self._lock = threading.Lock() # Serializes device writes so each logical message's reports are written # as an atomic group (a keepalive can't slip between a multi-report # message's header and its continuation reports). SEPARATE from _lock: @@ -190,10 +192,12 @@ def _confirm_lost(self, error): def _refuse_read_from_rx(self, what): """Refuse a correlated wait attempted from the RX thread. - The RX thread is the only thread that delivers a reply, so a wait issued - from inside it can never be satisfied: it would sit out its entire - timeout with the read loop stopped behind it, which is the "the RX thread - never blocks" rule broken in the worst way available. Listeners + The RX thread is the only thread that delivers a message to a waiter, so a + wait issued from inside it can never be satisfied: it sits out its whole + window with the read loop stopped behind it, which is the "the RX thread + never blocks" rule broken in the worst way available. ``request`` and + ``await_broadcast`` would time out; ``collect`` would return empty, having + stalled the link for its full duration. Listeners (:meth:`add_listener`) are the only caller code that runs on that thread, so this guard is what makes the listener contract enforced rather than merely requested (ADR-0008). @@ -203,10 +207,11 @@ def _refuse_read_from_rx(self, what): if threading.current_thread() is self._rx: raise RuntimeError( f"{what}() was called from the RX thread, which is the thread " - f"that would have to deliver the answer - so it can only ever " - f"time out. A listener applies what a push carries and notes " - f"what needs re-reading; the caller's thread does the " - f"re-reading (docs/domain-model.md section 9)." + f"that would have to deliver the answer - so the wait can never " + f"be satisfied, and the read loop stops for its whole duration. " + f"A listener applies what a push carries and notes what needs " + f"re-reading; the caller's thread does the re-reading " + f"(docs/domain-model.md section 9)." ) # -- outbound ------------------------------------------------------------ @@ -413,6 +418,11 @@ def add_listener(self, listener): notified first, and the message then reaches every collector and waiter exactly as it would have with no listener registered. + **Treat the message as read-only.** It is not a copy: the object handed to + a listener is the same one the next listener and the waiter receive, so a + listener that normalizes or tidies it in place changes what they see. + Read what you need out of it and merge that into your own state. + Registration and removal are safe while the RX thread is running. Returns a zero-argument callable that removes this registration; :meth:`remove_listener` does the same job for a caller who kept the @@ -429,7 +439,7 @@ def add_listener(self, listener): :meth:`collect` raise ``RuntimeError`` when called from the RX thread (see ``_refuse_read_from_rx``). Such a call could never have worked - the RX thread is the one that delivers replies, so a wait from inside - it only ever times out - and the rule it breaks is older than this + it can never be satisfied - and the rule it breaks is older than this method: the RX thread applies pushes and notes what needs re-reading, and the caller's thread does the re-reading (``docs/domain-model.md`` section 9). :meth:`send` is NOT refused, @@ -464,9 +474,14 @@ def remove_listener(self, listener): """Unregister ``listener``; return True if it had been registered. Never raises and is safe to call twice, so a teardown path can call it - unconditionally. Removal is by equality, which is what makes - ``remove_listener(self._apply)`` work for a bound method: each attribute - access builds a new object, and those compare equal. + unconditionally. Removal takes the FIRST registration equal to + ``listener``, which is what makes ``remove_listener(self._apply)`` work + for a bound method: each attribute access builds a new object, and those + compare equal. Two consequences of that, neither of them a problem unless + it is a surprise: registering the same callable twice registers it twice + and it is then called twice per message, needing one removal each; and a + listener whose class defines ``__eq__`` can have an equal-but-different + registration removed instead of the one passed. Safe to call from inside a listener, though the message being delivered may still reach the listener being removed - notification runs over a diff --git a/tests/hardware/conftest.py b/tests/hardware/conftest.py index c63030b..aae48af 100644 --- a/tests/hardware/conftest.py +++ b/tests/hardware/conftest.py @@ -24,35 +24,73 @@ def pytest_ignore_collect(collection_path, config): class HandshakeBurst: - """Records the type name of every message the unit pushes, from the start. + """Records the type of every message the unit pushes DURING the connect burst. Attached by the connection fixture through ``protocol.connect(before_handshake=...)``, which is the only moment early - enough to catch the handshake's burst - by the time ``connect`` returns, the - burst is over. + enough to catch the burst - by the time ``connect`` returns, the burst has not + even started. - Records NAMES rather than messages, and stops at ``LIMIT``. The metronome's - tempo stream never stops, so an unbounded recorder on a connection that lives - for the whole run would keep growing all run; the burst is only the first few - hundred messages of it. + It stops recording and takes itself off the transport as soon as the burst is + over, which is what makes the recording mean "the burst" rather than "the + traffic so far". The metronome's tempo stream never stops, so a recorder left + running would hold the whole run, and a test asserting on it would really be + asserting on whatever other tests had provoked first. Stopping also keeps it + out of the read path of the latency measurements in ``test_write_echo.py``, + which are calibrated numbers. - Runs on the transport's RX thread, so it does the least it can: take the - lock, append, return. - """ + Removing a listener from inside a listener is safe by contract - see + ``Transport.add_listener`` and ADR-0008. - LIMIT = 4000 + Runs on the RX thread, so it does the least it can: append and return. + """ def __init__(self): self._lock = threading.Lock() self._names = [] - self.dropped = 0 + self._detach = None + self.closed = False + self.settled_in = None # seconds the burst took, or None if it timed out + + def attach(self, transport): + """Register on ``transport``. Called before the handshake runs.""" + self._detach = transport.add_listener(self) def __call__(self, message): with self._lock: - if len(self._names) < self.LIMIT: - self._names.append(type(message).__name__) - else: - self.dropped += 1 + if self.closed: + # The RX thread notifies from a snapshot, so a message can still + # arrive after removal. It must not reopen the recording. + return + self._names.append(type(message).__name__) + + def record_until(self, sentinel, patience): + """Record until a ``sentinel``-typed message arrives, then stop. + + The seed ``RecallPresetMessage`` is the tail of the burst - measured + 2026-08-12 on d14e: ModelRepo at 4.9 s, the folder listings and settings + at 5.1 s, the current preset at 10.1 s - so waiting for it means the whole + burst has been recorded, however long the unit takes about it. + + Stops on ``patience`` seconds regardless, so a unit that never sends it + cannot hang the run. ``settled_in`` says which of the two happened. + """ + started = time.monotonic() + deadline = started + patience + while time.monotonic() < deadline: + if sentinel in self.names(): + self.settled_in = time.monotonic() - started + break + time.sleep(0.1) + self.close() + + def close(self): + """Stop recording and come off the transport. Idempotent.""" + with self._lock: + already = self.closed + self.closed = True + if not already and self._detach is not None: + self._detach() def names(self): """A snapshot of what has been recorded, in arrival order.""" @@ -73,14 +111,20 @@ def _connection(): ``pyquadcortex.connect()`` returns the model's ``Device`` instead (ADR-0006). The burst recorder is attached for every run, not just the tests that read - it: registering it costs one list append per message, and it cannot be - attached later on demand, because the burst happens during ``connect``. + it, because it cannot be attached later on demand: the burst happens during + ``connect``. + + The fixture then waits for the burst to finish before handing the connection + over, so the recording is exactly the burst whatever order the tests run in. + It costs about 8 s once per run and buys more than it costs: `connect()` + returns roughly 3 s before the unit starts streaming several hundred messages, + so without the wait every latency measurement in this suite would be taken on + a link that is still busy answering the handshake. """ from pyquadcortex import protocol burst = HandshakeBurst() - with protocol.connect( - before_handshake=lambda transport: transport.add_listener(burst) - ) as client: + with protocol.connect(before_handshake=burst.attach) as client: + burst.record_until("RecallPresetMessage", patience=30.0) yield client, burst diff --git a/tests/hardware/readme.md b/tests/hardware/readme.md index cc6afc7..f297e92 100644 --- a/tests/hardware/readme.md +++ b/tests/hardware/readme.md @@ -35,8 +35,23 @@ happened to run in. That connection attaches a listener before the handshake and records the type of every message the unit pushes. It is attached on every run, not only for the tests that read it, because it cannot be attached later: the burst happens during -`connect()`. It records type names and stops at 4000 of them, so the metronome's -endless tempo stream cannot grow it for the length of the run. +`connect()`. + +The fixture then waits for the burst to finish before handing the connection to +the first test, and stops the recorder there. The recording is therefore exactly +the burst, whatever order the tests run in. The metronome's tempo stream never +stops, so a recorder left running would hold the whole run's traffic and a test +asserting on it would really be asserting on whatever other tests provoked first. + +The wait costs about 8 seconds once per run and buys more than it costs. +`connect()` returns roughly 3 seconds before the unit starts streaming several +hundred messages, so without it every latency measurement below would be taken on +a link still busy answering the handshake. + +The hardware suite cannot check that the recorder stopped cleanly - it reads the +recording afterwards and its assertions are floors, which contamination satisfies +too. So the stop is pinned offline, in +`tests/test_handshake_burst_recorder.py`. ## Why the control test exists diff --git a/tests/hardware/test_broadcast_listener.py b/tests/hardware/test_broadcast_listener.py index 5cfa2f4..9721611 100644 --- a/tests/hardware/test_broadcast_listener.py +++ b/tests/hardware/test_broadcast_listener.py @@ -24,13 +24,6 @@ #: offers (40 bpm). Ten seconds is several beats even there. UNSOLICITED_PATIENCE = 10.0 -#: How long to wait for the handshake's state burst. Measured on this unit -#: (2026-08-12, CorOS 4.0.1 / d14e): ModelRepo at 4.9 s, the ~399 File listings -#: and most settings by 5.1 s, and the current preset at 10.1 s - 474 messages of -#: 24 distinct types by 15 s. Three times the measured figure, because the wait -#: ends as soon as the preset lands and a fixed sleep would only make the suite -#: slower. -BURST_PATIENCE = 30.0 class Recorder: @@ -75,29 +68,27 @@ def test_a_listener_registered_before_connecting_sees_the_handshake_burst( distinct types by 15 s. The floors below sit well under that, because a test pinned to the exact tally would fail on a unit with a different number of presets rather than on a real regression. - """ - def tally(): - names = handshake_burst.names() - return {name: names.count(name) for name in sorted(set(names))} - - # The current preset lands last of the burst (10.1 s measured), so it is the - # signal that everything else has already arrived. - _wait_until(lambda: "RecallPresetMessage" in tally(), BURST_PATIENCE) - counted = tally() - total = sum(counted.values()) - report = f"recorded {total} message(s) in {BURST_PATIENCE}s: {counted}" - - assert total >= 100, report + The recorder is already closed by the time any test runs - the connection + fixture waits for the burst and then stops it - so this reads the burst itself + and not the traffic other tests have provoked since. + """ + names = handshake_burst.names() + counted = {name: names.count(name) for name in sorted(set(names))} + report = (f"recorded {len(names)} message(s), settled in " + f"{handshake_burst.settled_in}s: {counted}") + + assert handshake_burst.closed, "the recorder was still running - see conftest" + assert handshake_burst.settled_in is not None, ( + f"the seed preset never arrived, so the burst was cut off by the " + f"fixture's patience rather than by finishing - {report}") + assert len(names) >= 100, report assert len(counted) >= 15, f"too few distinct state types in the burst - {report}" # Nothing in the handshake REQUESTS these. The subscription is a burst of # fire-and-forget READs, so almost every message here is one _dispatch would # have dropped for want of a waiter. assert "FileMessage" in counted, report # the folder enumeration assert "RecallPresetMessage" in counted, report # the preset on the grid now - assert handshake_burst.dropped == 0, ( - f"the recorder hit its {handshake_burst.LIMIT}-message cap, so this tally " - f"is not the burst alone") def test_a_listener_on_a_live_connection_hears_pushes_nobody_asked_for(qc): diff --git a/tests/test_handshake_burst_recorder.py b/tests/test_handshake_burst_recorder.py new file mode 100644 index 0000000..f106b1c --- /dev/null +++ b/tests/test_handshake_burst_recorder.py @@ -0,0 +1,147 @@ +"""The hardware suite's connect-burst recorder, checked offline. + +``tests/hardware/conftest.py`` attaches a listener before the handshake, records +what the unit pushes, and stops once the burst is over. Stopping is what makes the +recording mean "the burst" rather than "the traffic so far", and it has three +parts: recording stops, the listener comes off the transport, and a message that +arrives after both - the RX thread notifies from a snapshot - does not reopen it. + +The hardware suite cannot check any of that. It reads the recording after the +fixture has closed it and has no way to tell "closed correctly" from "closed and +then quietly kept recording"; the assertions are floors, which contamination +satisfies too. So if the close broke, the burst test would go back to asserting on +whatever the rest of the suite provoked and would still pass. That is the trap +``tests/test_scene_echo_predicates.py`` was written for. + +Timing is real here rather than faked, so the code under test is the code that +runs on the unit. +""" +import importlib.util +import threading +import time +from pathlib import Path + +import pytest + +from pyquadcortex.protocol.proto import ProductionAutomation_pb2 as pa + +_HARDWARE_CONFTEST = ( + Path(__file__).resolve().parent / "hardware" / "conftest.py") + + +@pytest.fixture(scope="module") +def recorder_class(): + """The real ``HandshakeBurst``, loaded from the hardware suite's conftest. + + Loaded by path under its own module name: the hardware conftest is not + collected at all without ``--hardware``, so there is no other way to reach it + from the offline suite, and pytest's own copy is untouched by this. + """ + spec = importlib.util.spec_from_file_location( + "hardware_conftest", _HARDWARE_CONFTEST) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.HandshakeBurst + + +class FakeTransport: + """The two methods ``HandshakeBurst`` uses, with the same removal contract.""" + + def __init__(self): + self.listeners = [] + + def add_listener(self, listener): + self.listeners.append(listener) + return lambda: self.remove_listener(listener) + + def remove_listener(self, listener): + if listener in self.listeners: + self.listeners.remove(listener) + return True + return False + + +def _file_push(): + return pa.FileMessage(action=pa.MessageAction.UPDATE) + + +def test_closing_stops_the_recording_and_takes_the_listener_off(recorder_class): + transport = FakeTransport() + burst = recorder_class() + burst.attach(transport) + assert transport.listeners == [burst], "attach did not register the recorder" + + burst(_file_push()) + assert burst.names() == ["FileMessage"] + + burst.close() + assert transport.listeners == [], "the recorder stayed on the transport" + + # A message can still arrive after the removal, because the RX thread notifies + # from a snapshot taken before the first listener ran. + burst(pa.SceneMessage(action=pa.MessageAction.UPDATE, selected_scene=1)) + assert burst.names() == ["FileMessage"], "recorded after being closed" + + burst.close() # idempotent: teardown must not care how it got here + assert transport.listeners == [] + + +def test_record_until_stops_when_the_sentinel_arrives(recorder_class): + transport = FakeTransport() + burst = recorder_class() + burst.attach(transport) + + def push_the_burst(): + for _ in range(3): + burst(_file_push()) + burst(pa.RecallPresetMessage(action=pa.MessageAction.UPDATE)) + + feeder = threading.Timer(0.1, push_the_burst) + feeder.start() + started = time.monotonic() + burst.record_until("RecallPresetMessage", patience=5.0) + took = time.monotonic() - started + feeder.join() + + assert took < 5.0, "it waited out its patience instead of noticing the sentinel" + assert burst.settled_in is not None + assert burst.closed + assert burst.names() == ["FileMessage"] * 3 + ["RecallPresetMessage"] + + +def test_record_until_gives_up_rather_than_hanging_on_a_silent_unit(recorder_class): + # A unit that never sends the sentinel must not hold the whole run. The + # give-up is reported rather than swallowed, so the hardware test can say the + # burst was cut off instead of asserting on half of it. + transport = FakeTransport() + burst = recorder_class() + burst.attach(transport) + burst(_file_push()) + + burst.record_until("RecallPresetMessage", patience=0.2) + + assert burst.settled_in is None, "it reported settling on a sentinel it never saw" + assert burst.closed + assert transport.listeners == [] + + +def test_the_recorder_is_safe_to_call_from_more_than_one_thread(recorder_class): + # It runs on the RX thread while the test thread reads names(). Nothing here + # is subtle; the point is that the lock covers both sides. + transport = FakeTransport() + burst = recorder_class() + burst.attach(transport) + + def push(): + for _ in range(200): + burst(pa.FileMessage(action=pa.MessageAction.UPDATE)) + + writers = [threading.Thread(target=push) for _ in range(4)] + for writer in writers: + writer.start() + while any(writer.is_alive() for writer in writers): + assert all(name == "FileMessage" for name in burst.names()) + for writer in writers: + writer.join() + + assert len(burst.names()) == 800 diff --git a/tests/test_session.py b/tests/test_session.py index d8a84dc..18d871c 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -109,7 +109,6 @@ def before(t): qc = session.connect(settle=0, before_handshake=before) t = FakeTransport.instances[0] - assert len(calls) == 1, "the hook runs once, not once per handshake attempt" got, started, sent_by_then = calls[0] assert got is t, "the hook gets the transport a listener registers on" assert started, "the RX thread must already be reading" @@ -118,6 +117,30 @@ def before(t): qc.close() +def test_before_handshake_runs_once_however_many_handshake_attempts_it_takes( + monkeypatch, fake_stack): + """Registering twice would register the listener twice, and it would then be + called twice per message. The device can be openable but silent for ~9-17s + after a boot, so a connect taking three handshake attempts is ordinary rather + than exotic - and a hook called per attempt would still pass a test where the + first attempt succeeds. + """ + attempts = {"n": 0} + + def flaky_hello(self, timeout=5.0, settle=2.0): + attempts["n"] += 1 + if attempts["n"] < 3: + raise TimeoutError("no response for request_id=1") + + monkeypatch.setattr(client.QuadCortex, "_hello", flaky_hello) + calls = [] + qc = session.connect(settle=0, handshake_patience=30.0, + before_handshake=calls.append) + assert attempts["n"] == 3, "the handshake was not actually retried" + assert len(calls) == 1, "the hook runs once, not once per handshake attempt" + qc.close() + + def test_a_failing_before_handshake_hook_does_not_leak_the_device(fake_stack): def boom(t): raise RuntimeError("the listener could not be registered") From b22a9ee02d8f6c32b5ecfd64a049305449955713 Mon Sep 17 00:00:00 2001 From: Jonathan Stokes Date: Wed, 12 Aug 2026 18:33:33 -0500 Subject: [PATCH 3/3] review: close the findings from the PR #20 re-review The re-review mutation-tested the previous round's fix rather than reading it, which found the hole below and two claims that were the wrong way round. * A listener raising outside Exception killed the RX thread. _notify_listeners caught Exception, so pytest.fail() and sys.exit() - both ordinary things for caller code to do, and BaseException subclasses - went through it, out of _handle_message, and out of the read loop. What a caller saw next was a TimeoutError with device_lost unset: the connection dead and nothing saying why. Listeners are the first arbitrary caller code to run on that thread, and "the RX thread never dies" is absolute, so this one site catches BaseException. The new test fails against the narrow catch (verified). * The coverage split for the burst recorder was documented backwards in two places. The hardware burst test covers the WIRING: its `closed` and `settled_in` assertions are the only ones that are not floors, so they are what fails if the fixture stops waiting for the burst. Both files now say that, and say not to delete those lines as redundant. The offline pin covers the stopping itself, which nothing on hardware can see. * test_the_recorder_is_safe_to_call_from_more_than_one_thread claimed to prove the lock covers both sides. list.append and list() are atomic under the GIL, so it passes with no lock at all - the same vacuous shape the first review caught, in the commit that fixed it. Renamed to what it does prove, with the limit written down. * HandshakeBurst's docstring still said removing a listener from inside a listener is safe by contract, which is true but no longer describes this class - and it invited the natural next edit (close as soon as the sentinel lands, from inside __call__) which deadlocks permanently on the non-reentrant lock. Replaced with a warning on close() saying where the stop must not go. * record_until copied the whole recording every 100 ms to test one membership. Scans in place now, so the poll stops contending with the RX thread at its busiest. * STEERING's change-log entry was missing the new offline test file. Declined: pinning the fixture WIRING offline as well, via the fixture's __wrapped__ and a fake connect. The wiring is already covered where it fails loudly - see the second point above - and reaching into pytest fixture internals to cover it twice buys less than it costs to read. --- docs/STEERING.md | 2 +- pyquadcortex/protocol/transport.py | 20 ++++++++++++++---- tests/hardware/conftest.py | 24 ++++++++++++++++----- tests/hardware/readme.md | 12 ++++++++--- tests/test_handshake_burst_recorder.py | 29 ++++++++++++++++---------- tests/test_transport.py | 27 ++++++++++++++++++++++++ 6 files changed, 90 insertions(+), 24 deletions(-) diff --git a/docs/STEERING.md b/docs/STEERING.md index b1580c7..61fa72d 100644 --- a/docs/STEERING.md +++ b/docs/STEERING.md @@ -135,7 +135,7 @@ Single-device, single-connection USB HID at interactive rates (129-byte reports) - M1 Epic (stokes-audio/pyquadcortex#8), Story #11. This is the protocol-layer half of that story, carved out because it is independent of the model work: `docs/domain-model.md` section 9 needs a push-fed cache, and a cache cannot be fed by three hooks that are all one-shot and scoped to a trigger **Scope of impact:** -- **Updated:** `pyquadcortex/protocol/transport.py`, `client.py`, `session.py`, `tests/test_transport.py`, `tests/test_client.py`, `tests/test_session.py`, `tests/hardware/conftest.py`, `tests/hardware/test_broadcast_listener.py` (new), `tests/hardware/readme.md`, ADR.md, CLAUDE.md, STEERING.md, architecture.md, api.md, protocol.md, changelog.md +- **Updated:** `pyquadcortex/protocol/transport.py`, `client.py`, `session.py`, `tests/test_transport.py`, `tests/test_client.py`, `tests/test_session.py`, `tests/test_handshake_burst_recorder.py` (new), `tests/hardware/conftest.py`, `tests/hardware/test_broadcast_listener.py` (new), `tests/hardware/readme.md`, ADR.md, CLAUDE.md, STEERING.md, architecture.md, api.md, protocol.md, changelog.md - **Not updated (intentionally):** ADR-0002 - the offline suite still imports no `hid` and the new tests run against `FakeHid` like the rest; ADR-0005 - the new hardware tests only listen, so they write nothing and have nothing to restore, which meets the contract rather than changing it; `docs/domain-model.md` - section 9 designed this and needed no correction; the coverage table in `protocol.md` - no new message type is involved **Downstream to consider:** diff --git a/pyquadcortex/protocol/transport.py b/pyquadcortex/protocol/transport.py index b3b62c8..781fd55 100644 --- a/pyquadcortex/protocol/transport.py +++ b/pyquadcortex/protocol/transport.py @@ -448,7 +448,10 @@ def add_listener(self, listener): A listener that raises is logged and skipped: the RX thread survives, the other listeners still see that message, and the message still reaches its - waiter. Same contract as every other step in this module's RX path. + waiter. Same contract as every other step in this module's RX path, and + wider than the rest of it - ``BaseException``, not just ``Exception``, + because ``pytest.fail()`` and ``sys.exit()`` are ordinary things for + caller code to do and neither may cost the connection its read loop. A listener lives only as long as the connection. Device loss neither removes nor notifies listeners - there is simply nothing further to @@ -507,9 +510,18 @@ def _notify_listeners(self, message): for listener in listeners: try: listener(message) - except Exception: - # The RX thread never dies, and one broken listener never costs - # its peers or the waiter their copy of this message. + except BaseException: + # BaseException, not Exception, and this is the only place in the + # module that goes that wide. Everything else on the RX path is + # our own code, where a BaseException means something genuinely + # fatal; a listener is arbitrary caller code, and the ways it can + # raise outside Exception are ordinary rather than exotic - + # pytest.fail() and sys.exit() both do. Letting one of those + # through kills the RX thread, and the failure the caller sees is + # a TimeoutError on the next request with device_lost unset: the + # connection is dead and nothing says why. "The RX thread never + # dies" is absolute, so it outranks the usual rule about not + # swallowing BaseException. log.exception( "inbound listener %r raised on %s; skipping it for this " "message", listener, type(message).__name__ diff --git a/tests/hardware/conftest.py b/tests/hardware/conftest.py index aae48af..40a38eb 100644 --- a/tests/hardware/conftest.py +++ b/tests/hardware/conftest.py @@ -39,9 +39,6 @@ class HandshakeBurst: out of the read path of the latency measurements in ``test_write_echo.py``, which are calibrated numbers. - Removing a listener from inside a listener is safe by contract - see - ``Transport.add_listener`` and ADR-0008. - Runs on the RX thread, so it does the least it can: append and return. """ @@ -78,20 +75,37 @@ def record_until(self, sentinel, patience): started = time.monotonic() deadline = started + patience while time.monotonic() < deadline: - if sentinel in self.names(): + if self._recorded(sentinel): self.settled_in = time.monotonic() - started break time.sleep(0.1) self.close() def close(self): - """Stop recording and come off the transport. Idempotent.""" + """Stop recording and come off the transport. Idempotent. + + Runs on the caller's thread, from :meth:`record_until`. If you ever move + the stop into :meth:`__call__` - closing the moment the sentinel lands, + which is tempting - it has to happen OUTSIDE that method's ``with + self._lock`` block: ``_lock`` is not reentrant, so closing from inside it + deadlocks the RX thread permanently. + """ with self._lock: already = self.closed self.closed = True if not already and self._detach is not None: self._detach() + def _recorded(self, name): + """Whether a message of type ``name`` has been recorded. + + Scans in place rather than going through :meth:`names`, which would copy + the whole recording on every poll, briefly contending with the RX thread + at the busiest moment it has. + """ + with self._lock: + return name in self._names + def names(self): """A snapshot of what has been recorded, in arrival order.""" with self._lock: diff --git a/tests/hardware/readme.md b/tests/hardware/readme.md index f297e92..46048f0 100644 --- a/tests/hardware/readme.md +++ b/tests/hardware/readme.md @@ -48,9 +48,15 @@ The wait costs about 8 seconds once per run and buys more than it costs. hundred messages, so without it every latency measurement below would be taken on a link still busy answering the handshake. -The hardware suite cannot check that the recorder stopped cleanly - it reads the -recording afterwards and its assertions are floors, which contamination satisfies -too. So the stop is pinned offline, in +The burst test's `assert handshake_burst.closed` and `settled_in is not None` are +what hold that up. They are not belt-and-braces: they are the only things that +fail if the fixture stops waiting for the burst, since every other assertion in +that test is a floor and contamination satisfies a floor. Do not delete them as +redundant. + +What they cannot see is a recorder that sets its flag and keeps recording anyway, +or one that stops recording but stays attached to the transport. Both read like a +working recorder from the outside, so both are pinned offline in `tests/test_handshake_burst_recorder.py`. ## Why the control test exists diff --git a/tests/test_handshake_burst_recorder.py b/tests/test_handshake_burst_recorder.py index f106b1c..8d6c95b 100644 --- a/tests/test_handshake_burst_recorder.py +++ b/tests/test_handshake_burst_recorder.py @@ -3,15 +3,19 @@ ``tests/hardware/conftest.py`` attaches a listener before the handshake, records what the unit pushes, and stops once the burst is over. Stopping is what makes the recording mean "the burst" rather than "the traffic so far", and it has three -parts: recording stops, the listener comes off the transport, and a message that -arrives after both - the RX thread notifies from a snapshot - does not reopen it. +parts: recording stops, the listener comes off the transport, and a message +arriving after both - the RX thread notifies from a snapshot - does not reopen it. -The hardware suite cannot check any of that. It reads the recording after the -fixture has closed it and has no way to tell "closed correctly" from "closed and -then quietly kept recording"; the assertions are floors, which contamination -satisfies too. So if the close broke, the burst test would go back to asserting on -whatever the rest of the suite provoked and would still pass. That is the trap -``tests/test_scene_echo_predicates.py`` was written for. +**Who covers what**, because it is not obvious and getting it backwards leads to +deleting the wrong assertion: + +* The hardware burst test covers the WIRING. If the fixture stopped calling + ``record_until``, its ``assert handshake_burst.closed`` and ``settled_in is not + None`` fail loudly - neither is a floor, so contamination cannot satisfy them. + Those two lines are load-bearing, not belt-and-braces. +* This file covers the STOPPING ITSELF, which no hardware test can see: a recorder + that sets its flag and keeps recording anyway, or one that stops recording but + stays on the transport, reads exactly like a working one from the outside. Timing is real here rather than faked, so the code under test is the code that runs on the unit. @@ -125,9 +129,12 @@ def test_record_until_gives_up_rather_than_hanging_on_a_silent_unit(recorder_cla assert transport.listeners == [] -def test_the_recorder_is_safe_to_call_from_more_than_one_thread(recorder_class): - # It runs on the RX thread while the test thread reads names(). Nothing here - # is subtle; the point is that the lock covers both sides. +def test_recording_and_reading_at_the_same_time_loses_nothing(recorder_class): + # A smoke test, and labelled as one deliberately. The recorder is written to + # from the RX thread while the test thread reads names(), so the paths do + # overlap - but list.append and list() are atomic under CPython's GIL, so + # removing the lock entirely leaves this green. It would NOT catch that, and + # an earlier version of this comment claimed it would. transport = FakeTransport() burst = recorder_class() burst.attach(transport) diff --git a/tests/test_transport.py b/tests/test_transport.py index 9c80ae9..55e9e3a 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -528,6 +528,33 @@ def explodes(message): "a raising listener must be logged, not silently swallowed" +def test_a_listener_raising_outside_exception_still_cannot_kill_the_rx_thread(): + # The reason this is not covered by the test above: a listener is arbitrary + # caller code, and two ordinary things it might do - pytest.fail() and + # sys.exit() - raise BaseException subclasses, which a plain `except + # Exception` lets through. That kills the read loop, and what the caller then + # sees is a TimeoutError with device_lost unset: the connection is dead and + # nothing says why. SystemExit stands in for both here. + fake = FakeHid() + t = transport.Transport(fake, keepalive_interval=QUIET_KEEPALIVE) + after = [] + + def bails_out(message): + raise SystemExit("a listener that called sys.exit()") + + t.start() + try: + t.add_listener(bails_out) + t.add_listener(after.append) + fake.inject(*_recall_broadcast("push", rid=None)) + assert _wait_until(lambda: after), "the peer listener lost the message" + assert t._rx.is_alive(), "the RX thread died" + assert t.request(pa.VersionMessage(action=pa.MessageAction.READ), + timeout=REQUEST_TIMEOUT) is not None + finally: + t.stop() + + def test_a_listener_cannot_read_from_the_device_on_the_rx_thread(): # The design rule this enforces: the RX thread applies pushes and notes what # needs re-reading, and the caller's thread does the re-reading. All three