OM-M1.3 (protocol half): a persistent broadcast subscription

[jonathanstokes]

Part of #11 (M1 Epic #8). This is only the protocol-layer half of that story. The model-side cache is being written separately and is not here.

The gap

The unit talks without being asked. Turn a knob on its touchscreen, recall a preset, let the metronome run, and it pushes messages saying so.

The transport had three ways to hear inbound messages, and all three are one-shot and tied to something the caller just did:

• request() waits for the reply to one message it sent.
• await_broadcast() fires a trigger and waits for one push.
• collect() fires a trigger and gathers for a fixed number of seconds.

Anything else reaches _dispatch, matches no waiter, and is dropped with a debug log line. So nothing could watch the link for the life of a connection, which is exactly what a cache fed by the unit's own pushes needs.

What this adds

add_listener(fn) registers fn to be called with every decoded message until it is removed:

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)

Nothing existing changes behaviour. A listener consumes nothing: it is notified first, and the message then reaches every collector and waiter exactly as it would have with no listener registered. QuadCortex passes both methods through so the layer above never has to reach into a private attribute.

A listener runs on the thread that reads from the USB device, and that thread must never block and never die. Two consequences, both handled:

• A listener that raises is logged and skipped. Its peers still get that message, and so does the waiter it belongs to. Same as every other decode step on that path.
• A listener may not read from the device. request, await_broadcast and collect now raise RuntimeError when called from the read thread. This refuses nothing that ever worked: the read thread is the one that delivers replies, so a wait from inside it can never be satisfied, with the whole connection stalled behind it for as long as it waits. A listener applies what a push carries and notes what needs re-reading; the caller's thread does the re-reading. The rule was already in the design doc (section 9); this makes it enforced rather than requested, and the reason is in add_listener's docstring.

send is deliberately not refused. It is fire-and-forget and cannot deadlock, though a listener that writes owns the delay it adds.

The second addition, and why it is needed

protocol.connect(before_handshake=...) calls back with the started transport before the connect handshake runs.

The handshake is what makes the unit start pushing state, and it answers with a burst of nearly everything it knows. Measured on the unit for this change:

when	what arrived
2.0 s	connect() returned, with 2 messages seen
4.9 s	the model repository
5.1 s	399 folder listings and most settings
10.1 s	the preset currently on the grid
by 15 s	474 messages, 24 distinct message types

So a listener registered on the client connect() hands back is about 8 seconds too late for the preset. Without this hook, anything wanting the burst would have to assemble the layers by hand and reimplement the handshake retry logic. It is called once, before the first handshake attempt, and a failure in it releases the device like any other bring-up failure.

Tests

Offline, against the fake HID device and fake transport:

• a listener sees a push no waiter wanted
• a listener does not take the reply away from request()
• the listener has already run when the blocked caller wakes (the ordering a cache depends on)
• a listener that raises costs its peers and the waiter nothing, is logged, and the read loop survives two more round trips
• all three correlated waits are refused from the read thread, with RuntimeError rather than a timeout, and the link still works afterwards
• removal works by either handle, and reports honestly when there was nothing to remove
• the hook runs after the transport starts and before the handshake sends anything, once, and a failure in it does not leak the device

tests/test_transport.py::test_collect_gathers_every_matching_push_without_consuming_them now builds a real (unstarted) Transport instead of a half-initialised one made with __new__.

On the unit, tests/hardware/test_broadcast_listener.py (5 tests): the pre-handshake listener sees the burst, a live listener hears the tempo stream nobody asked for, a listener does not take version()'s reply, removal stops delivery while a second listener proves the unit was still pushing, and the refused read is refused on a real read loop.

That file only listens. It sends no write, so there is nothing to snapshot and nothing to restore - it meets ADR-0005's contract rather than changing it.

The suite's connection fixture records the burst on every run, because a listener cannot be attached on demand later. It then waits for the burst to finish before handing the connection to the first test and stops the recorder there, so the recording is the burst itself rather than the traffic other tests have provoked since. That wait costs about 8 seconds once and buys more than it costs: connect() returns about 3 seconds before the unit starts streaming several hundred messages, so without it every latency measurement in the suite would be taken on a link still answering the handshake. Suite wall time is unchanged, because the burst test used to do that waiting itself.

The stop cannot be checked with a unit attached, since the hardware test reads the recording afterwards and its assertions are floors that contamination satisfies too. So tests/test_handshake_burst_recorder.py pins it offline, the way test_scene_echo_predicates.py pins the echo predicates.

Results

Offline suite, no unit attached:

529 passed, 1 skipped in 6.92s

Whole hardware suite on the unit (d14e / CorOS 4.0.1), including the existing write-echo tests:

10 passed, 1 skipped in 15.14s

The skip is pre-existing: a block-bypass measurement that needs a preset whose first block already carries a stored bypass entry.

Documentation

• ADR-0009 records the decision: listeners on the read thread rather than behind a queue and a delivery thread, and the refusal enforced rather than documented. Both rejected options are written down with why.
• CLAUDE.md carries the rule for future work. STEERING.md gets the change-log entry and the ADR row.
• architecture.md: the transport section, the message-flow diagram, the hook-choosing table, and the session.py section.
• protocol.md: the re-measured connect burst, including the fact that connect() returns before it arrives.
• api.md and changelog.md for people using the library.

No version bump and no release.

Review

Reviewed twice, the second time over the full diff because the first round's fix commit had not been looked at. No merge blockers either time. Ten findings closed in 68d358e, seven more in b22a9ee.

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. 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 claimed 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, the ADR 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 defining __eq__ can have an equal-but-different registration removed.
• The _lock comment still read "guards _pending / _ids (state only)", which 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 either way. A separate test now drives three attempts, and it fails if the hook moves inside connect's retry loop. Verified by moving it.
• The 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" held only because of alphabetical file order. See the Tests section above for the fix and its offline pin.

Answered, not changed:

• Skipping the listener snapshot's lock when no listeners are registered. Two uncontended lock operations at a peak of about 80 messages a second is not worth a GIL-dependent fast path plus the comment explaining why the race is benign.
• Forwarding before_handshake through the model's connect(). Deliberate boundary. The model half of OM-M1.3: The model stays current with the unit without asking twice #11 is being written separately and will need it in a file it is already editing; adding it here only makes a conflict.

Second round

The second pass mutation-tested the first round's fix instead of reading it - it copied the recorder into a scratch directory, broke one thing at a time, and checked whether the new offline test noticed. That found one real hole 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, both BaseException subclasses - went straight through it 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 that one site now catches BaseException, with the reason written down. The new test fails against the narrow catch.
• The burst recorder's coverage split was documented backwards in two places. The hardware test covers the wiring: its closed and settled_in assertions are the only ones there that are not floors, so they are what fails if the fixture stops waiting for the burst. Both files now say so, 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 could not fail for the reason it claimed - list.append and list() are atomic under the GIL, so it passes with no lock at all. The same vacuous shape the first round caught, in the commit that fixed it. Renamed to what it does prove, with the limit written down.
• HandshakeBurst's docstring still said self-removal from inside a listener is safe by contract. True, but it 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. It scans in place now, so the poll stops contending with the RX thread at its busiest moment.
• STEERING's change-log entry was missing the new offline test file.

Declined: pinning the fixture wiring offline as well, through the fixture's __wrapped__ and a fake connect. The wiring is already covered where it fails loudly, and reaching into pytest's fixture internals to cover it twice buys less than it costs to read.

Both rounds independently confirmed the two findings declined in the first round still stand.

Merged main in

PR #19 landed while this was in review and took ADR-0008 for the generator floor, so the listener record here is renumbered ADR-0009 - in ADR.md and in every reference to it. The two review commits on this branch predate the renumber and still say 0008; STEERING's change-log entry says so rather than leaving a reader to work it out.

Both change-log entries and both ADR table rows are kept. Nothing else conflicted. Test numbers above are from after the merge, so they include main's new packaging tests, and the hardware suite was re-run on the merged tree.

Not in this change

• The model-side cache, which is the rest of OM-M1.3: The model stays current with the unit without asking twice #11.
• Device loss for listeners. A listener stops receiving today and is not told; that belongs with reconnect (OM-M1.7: Scripts survive the unit disconnecting, sleeping, and coming back #15) and is recorded as ADR-0009's open question.
• Forwarding before_handshake through pyquadcortex.connect(). The model half of OM-M1.3: The model stays current with the unit without asking twice #11 needs it and is being written separately.
• tests/hardware/test_write_echo.py still taps _dispatch by monkeypatching it, which predates this and could now be an ordinary listener. Left alone on purpose: it is a working measurement harness, and an offline test imports it.