diff --git a/CLAUDE.md b/CLAUDE.md index 8437adf..9907dd3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -11,15 +11,18 @@ Read `docs/STEERING.md` before non-trivial work (new operations, transport or fr - Two namespaces, one package (ADR-0006): `pyquadcortex` is the model of the unit, `pyquadcortex.protocol` is the message-level API. The model's code lives in `pyquadcortex/device/` - not `model/`, because in this codebase the identifier `model` means an amp or pedal block (`protocol/models.py`, `catalog.Model`, `ModelCatalog`, `set_block(model=...)`). The model imports the protocol layer; nothing under `pyquadcortex/protocol/` may import from `pyquadcortex/device/`. - Every conversion between a screen value and a wire value lives in `pyquadcortex/device/translate.py` and nowhere else in the package outside `pyquadcortex/protocol/` - the whole package, not just `device/`, because a rule scoped to a directory is satisfied by moving the code one directory up. It covers rows 1-4, slots 1-8, scene and footswitch letters, preset addresses and display units. Outside the boundary that means no `+1`/`-1` on a coordinate AND none of the other spellings of the same conversion (`ord`/`chr`, a letter table in any container, `divmod` on a position, a one-based `enumerate`, `ROWS.index(...)`), and no module reaching past the boundary for a protocol-layer name that carries a coordinate or a raw scale - the converters and also the readers that hand back wire indexes, such as `protocol.stomp_assignments`. `tests/test_translation.py` reads the source and proves both, and pins where each check stops seeing rather than implying it sees everything. A model API takes `FootswitchLetter`, never a bare footswitch integer, because a footswitch index and a block's column are different numbers that usually agree. A new conversion goes in that module with its own test, however small it is, and its protocol-layer name joins that file's allowlist in the same commit. - The model represents what the unit shows, in the unit's own words, and never guesses. A control we understand but cannot yet drive is modelled and REFUSES the operation (ADR-0007); a control we do not understand is omitted, with the reason recorded in `docs/domain-model.md`'s appendix. Nothing ships with a "this might be stale or wrong" caveat. -- A model property that reads a device field checks the field is PRESENT (`protocol.field_present`) before reporting it. Most of this schema sits in synthetic `oneof`s, so protobuf returns `""` or `0` for a field the unit never sent, and reporting that as the answer is the guess the rule above forbids. Never cache a reply that came back incomplete - a retry has to be able to recover. -- Anything the model caches is valid only while its connection is. A closed `Device` refuses reads rather than answering from cache, because a model that reports the unit's state through an object with no unit behind it is the failure the whole layer exists to avoid. +- A model property that reads a device field checks the field is PRESENT (`protocol.field_present`) before reporting it. Most of this schema sits in synthetic `oneof`s, so protobuf returns `""` or `0` for a field the unit never sent, and reporting that as the answer is the guess the rule above forbids. `device/state.py` does this structurally - an absent field is simply not in the entry's copy - so a property reads through it rather than checking by hand. The exception is a field the schema gives no presence at all, where absent and default are the same bytes: those are declared in the entry's `FieldPlan.no_presence` WITH the recorded evidence for what the default means, and `tests/test_state.py` holds the declaration against the schema. Never assume; declare. +- Never cache an incomplete reply AS IF IT WERE COMPLETE. Per field is the rule: the cache keeps what the unit actually sent and re-reads for what it did not, so a retry recovers the missing half without re-asking for the half already answered. What must never happen is a field the unit never sent being handed back as a value. +- Model state lives in `device/state.py`, and what is tracked is a `StateEntry` in `device/entries.py` - not an attribute a property fills in itself (ADR-0011). Pushes MERGE (an absent field means "not mentioned", never "reset to default"); a read REPLACES, because it is the unit's whole answer. A message that sets any field the entry does not keep - a schema field or a field number the bindings have never heard of - marks that entry for one re-read; there is no "harmless field" category, and adding one is a guess with a table around it. A message type no entry tracks is ignored outright, which is what makes the metronome's tempo stream free. +- A new entry decides for itself what `action` means on the types that feed it. The shared `SCAFFOLDING` skip covers it today because the two tracked types give it no meaning, and that is NOT true of `Grid`, where `action: DELETE` is what removes a block and an `UPDATE` with the same payload does nothing. Never widen `SCAFFOLDING` to make a new entry quiet. +- Anything the model caches is valid only while its connection is. A closed `Device` refuses reads rather than answering from cache, because a model that reports the unit's state through an object with no unit behind it is the failure the whole layer exists to avoid. `Device.close()` closes the state layer first, so a `Device` built by `from_client` stops listening on a connection it never owned. - `import hid` appears exactly once, lazily, inside `session.open_device()`. Never import `hid` at module scope. A new module that needs it imports it inside the function that opens the device; `tests/test_import_cleanliness.py` walks the whole package and proves it. - Never gitignore or delete `pyquadcortex/protocol/proto/*_pb2.py` - the generated bindings are committed on purpose (ADR-0001, written before the proto directory was moved). Regenerate only via `scripts/compile_protos.sh`, and bump the `protobuf` pin in `pyproject.toml` in the same commit as regenerated bindings. The `grpcio-tools` floor in the dev extra is part of that same commit: `grpcio-tools` carries its own protoc, so the installed version decides the gencode, and an older one emits older gencode that still imports and quietly walks the pin backwards (ADR-0008). Both directions are now guarded - the script refuses to write a downgrade, and `tests/test_packaging.py` proves the committed gencode equals the pin floor - so trust the failure and fix the cause rather than working around either. Never read the floor off `grpcio-tools` metadata; 1.82.1 declares `protobuf>=7.35.1` and emits 7.35.0. Run the compiler and read the stamp. CI's `build` job runs `scripts/check_artifacts.py`, which proves the bindings are inside the wheel and the sdist. - New operations follow `docs/architecture.md` "How to add a new operation": register the type, add a thin client method (no HID, no bytes, no sleeps in `protocol/client.py`), add an offline test asserting the exact wire shape, then verify on hardware and update the coverage table in `docs/protocol.md`. - 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-0009). 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. +- A `Transport.add_listener` listener runs ON the RX thread (ADR-0009). 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. That binds `device/state.py` as much as the transport: `apply_push` and everything it calls merge and mark, and the caller's thread does the reading. 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 33c90e6..d715ae9 100644 --- a/changelog.md +++ b/changelog.md @@ -85,6 +85,41 @@ 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. +### The model keeps up with the unit on its own + +Anything a `Device` tells you is what the unit is doing now, including changes you +make on its touchscreen while your script is running. You do not have to re-read +anything, and nothing you read comes with a "this might be out of date" warning. + +It works because the unit says when things change, and the model listens from the +moment it connects. Connecting is also when the unit volunteers most of what it +knows, in one burst, so the model usually has your answer before you ask for it. +Where the unit says nothing - its firmware version, for one - the model asks, once, +the first time you want it. + +```python +import pyquadcortex + +with pyquadcortex.connect() as device: + print(device.firmware) # asks the unit + print(device.firmware) # free +``` + +Two things it will not do. It will not hand you a value the unit never sent: a +field the unit left out raises rather than coming back as an empty string, and +asking again can still succeed. And it will not answer at all once you close the +`Device` - what it remembers stopped being true of the unit the moment the +connection went away. + +If the unit mentions something the model does not yet understand, the model stops +trusting that part of what it remembers and asks the unit next time you read it. +Slower, and right. `device.state` shows you what it currently holds and what it is +about to re-read. + +Presets, the grid and the Directory are not in the cache yet - they arrive with the +surfaces that read them. Nor is reconnecting after the unit sleeps or the cable +comes out; that is still your code's job for now. + ### New: listen to everything the unit sends The unit talks without being asked. Turn a knob on its touchscreen, recall a diff --git a/docs/ADR.md b/docs/ADR.md index fe0b840..144b5a4 100644 --- a/docs/ADR.md +++ b/docs/ADR.md @@ -145,3 +145,24 @@ Records are append-only once `Decided` and built upon: a shipped decision is nev - ADR-0007 keeps its status and its rule. It currently has no instance, which is the healthy state for it. - Epic #8's dependency on the TEMPO MODE wire path is resolved. It never gated M1; it no longer gates M3. - A negative result about device traffic now states which question the instrument answered. "The unit does not announce X" and "X is not on the wire" are separate claims and the second needs a READ. + +## ADR-0011: A push merges, a read replaces, and anything the cache cannot place forces one read + +- **Status:** Decided (2026-08-14) +- **Decision:** The model's cache treats an inbound message and a read answer differently, on purpose. A **push merges**: the fields the model keeps are applied, and everything else it holds is left alone. A **read replaces**: the answer is the unit's whole account of that entry, so a field it does not carry is dropped rather than kept from before. The per-field check that decides whether our copy is still trustworthy is **conservative and has no exemptions**: a message that sets any field the entry does not keep - including a field number the recovered schema has never heard of - marks that entry, and the next read goes to the unit. The only fields skipped are `action` and `request_id`, which belong to the transport and describe no state. A mark is consumed by **exactly one** read, and what makes that safe is a count rather than a hope: a read clears the mark only if nothing arrived for that entry beyond the read's own answer. +- **Context:** `docs/domain-model.md` section 9 gives the rules and the reason for them - "applying the half of a message we understand and silently dropping the rest is the one failure mode that leaves the cache confidently wrong" - and issue #11 restates the check as per FIELD, not per message type. What section 9 does not settle is what happens when the read that resolves a mark carries the same unkept fields that caused it. Answered naively, the entry re-arms its own mark from its own answer and never caches anything again, so every access becomes a round trip and the cache is decorative. The other half of the problem is the opposite failure: clearing the mark unconditionally throws away any push that landed while the read was in flight, and that push is the only record of a change the replacement has just overwritten. +- **Options:** + - **(a) Push merges, read replaces, one read consumes the mark, guarded by a count of what arrived - chosen.** Every message for an entry is counted as the listener handles it. The read path notes the count before it asks and, when the answer comes back, clears the mark only if the count moved by no more than the answer itself. Because the transport notifies listeners before waking the thread that asked (ADR-0009), the answer is already counted by then, and a push that arrives in the microseconds afterwards is counted under the same lock - so there is no window, rather than a small one. + - **(b) Declare, field by field, which changes cannot affect what we hold.** A `Version` reply carries a bootloader version and a MAC address; neither can make the firmware string wrong, so an entry could say so and skip the read. It is also a judgement call per field with nothing to check it against, made by whoever adds the field, and wrong quietly. Section 9 exists because the model does not guess, and this is a guess with a table around it. + - **(c) Tell a read answer from a push by correlating it.** The transport already refuses to promise this: READ replies carry no `request_id` echo, which is recorded in `Transport.request`. Correlating by type and arrival order would be the same count as (a) with more machinery and a worse failure mode. + - **(d) Suppress the listener for an entry while it is being read.** Simple, and it silently drops any genuine push that lands in the read window - which is the exact change the cache most needs to hear about, because a replacement is about to overwrite it. +- **Open Questions:** Whether the conservative rule costs enough reads to be worth refining once an entry is fed by whole preset dumps rather than small keyed pushes. It is measurable rather than arguable: the log names the entry and the field on every forced re-read. Nothing should be refined before that measurement exists. +- **Rationale:** The distinction between merging and replacing is not an implementation detail, it is the difference between the two things the unit says. A push is a delta about what changed; a read is an answer about what is. Merging a delta needs us to understand every field it names, which is why an unrecognised one costs trust; replacing needs no such understanding, which is why the same fields in an answer cost nothing. That is also why one read is enough, and why the alternative that looked simplest - suppressing or ignoring the listener during a read - is the one that loses information. On the conservative rule: the cost of being wrong in the safe direction is one read, logged with the field that caused it. The cost of being wrong in the other direction is a value the caller believes and the unit disagrees with, and the device gives no error for either. +- **Consequences:** + - Unknown field numbers count. The schema here is recovered rather than published (ADR-0010), so a field the unit really sends and the bindings have never heard of is ordinary rather than hypothetical. It decodes into nothing at all, which makes it the quietest way to drop half a message, and the cache notices it by weighing the message rather than by reading it. That costs one message copy per push per entry - cheap for the small keyed pushes an edit produces, and worth measuring before an entry is fed by full preset dumps. + - A field with no wire presence needs recorded evidence before it is kept, because absent and default are the same bytes and there is nothing to check. `PresetDirty.is_dirty` is the first and only one; the evidence is the protocol layer's, watched flipping across a save on hardware. `tests/test_state.py` holds every such declaration against the schema, so a field that does have presence cannot be declared this way. + - An entry with no reader is not an entry. Section 9's table is longer than the registry, and each remaining row arrives with the surface that reads it - otherwise every push mentioning a field it did not keep would mark it for a read nobody had asked for. + - The mark is per entry, not per field. "Exactly that part of the cache" is the entry, which is what a single read replaces. + - A write the unit CONTRADICTS marks the entry too, which section 10 does not ask for. Section 10 asks for a log line, on the reasoning that a disagreement is a bug in our code rather than a stale cache. That is true of the field the unit named, whose value the echo has just put right. It is not true of the other fields in the same write: those went into the cache on our say-so, the echo did not carry them, and the write they belonged to is one the unit has just demonstrated it disagreed with. Leaving them there makes the one path that means "we have a bug" the one that cleans up after itself least, so it marks - which costs one read on a path that should never run. + - `action` and `request_id` are skipped on every entry, and only one of them is really the transport's. `request_id` always is. `action` is not: on `Grid` it is load-bearing state, because an `UPDATE` carrying `hash: 0` is transmitted and ignored while the same payload with `action: DELETE` removes the block. It is skipped today because the two tracked message types give it no meaning, and a `Grid` entry (issue #12) therefore cannot inherit the skip - two pushes with identical payloads and opposite meanings would apply identically and mark nothing. That entry gives `action` its own decision rather than widening the shared set. + - A field the wire gives no presence and the entry does not KEEP is undetectable rather than merely unkept: proto3 writes such a field only when it differs from its default, so a message leaving one at its default carries no bytes for it and no implementation of this check could see it. `tests/test_state.py` asserts no feeding type has one, which turns a limit of the wire into a question about our own code, checkable and checked. diff --git a/docs/STEERING.md b/docs/STEERING.md index 93460b8..c233840 100644 --- a/docs/STEERING.md +++ b/docs/STEERING.md @@ -2,7 +2,7 @@ > **What this is:** durable technical context for the pyquadcortex library - what the system is and why it is shaped this way. > **What this is not:** coding rules (see the repo-root `CLAUDE.md`) or decision rationale (see [`ADR.md`](ADR.md)). -> **Last reviewed:** 2026-08-13 by Stokes +> **Last reviewed:** 2026-08-14 by Stokes > **Owners:** Stokes ## 1. Purpose @@ -37,7 +37,9 @@ 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. 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-0009) - 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. +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-0009) - but the transport stores none of it. + +The model layer holds the state (design in [`domain-model.md`](domain-model.md) sections 9 and 10, decided in ADR-0011). `pyquadcortex/device/state.py` is a write-through cache above `protocol/client.py`, fed by one persistent listener registered before the connect handshake so it hears the handshake's burst. It applies what the unit pushes as data rather than as an invalidation signal, asks the unit directly for what the unit never announces, and stops trusting a part of its copy when a message names a field the model does not keep. Reads happen on the caller's thread; the RX thread only ever merges and marks. What is tracked is a registry in `device/entries.py` rather than code, and it currently holds two of section 9's rows - the unit's identity and the unsaved-changes flag. The rest arrive with the surfaces that read them, so callers still hold whatever state the model does not yet cover. ## 4. Owned Paths @@ -58,6 +60,7 @@ The protocol layer is stateless between calls: every read is a live exchange, an | Evidence-bearing docstrings | Each operation's docstring states what is confirmed on hardware vs inferred from the schema | The device gives no errors for wrong writes, so recorded evidence is the only trail | `QuadCortex.read_preset` in `pyquadcortex/protocol/client.py` | Non-protocol helpers (pure functions) carry ordinary docstrings | | Keyed grid edits | Mutations are row/column-keyed `Grid` UPDATEs | The device applies grid updates by key; wholesale preset writes are silently ignored (see [`architecture.md`](architecture.md), "write_preset is a trap") | `QuadCortex.set_bypass` in `pyquadcortex/protocol/client.py` | Read paths, and non-grid operations | | One translation boundary | Screen values become wire values in exactly one module, and a source-reading test proves no other module in the package does it - the whole package outside `protocol/`, not just `device/` | An off-by-one row is silent - the write lands on a real row and reads back perfectly - so a convention cannot be trusted to hold (design principle 5 in [`domain-model.md`](domain-model.md)) | `pyquadcortex/device/translate.py` | The protocol layer, which keeps its zero-based indexes and raw scales | +| Model state goes through the cache | A model property reads `Device.state.value(entry, field)`; what it tracks is a `StateEntry` in `device/entries.py`, not an attribute the property fills in itself | One account of what the model believes and how it learned it. A property with its own cached attribute answers from a copy nothing invalidates, and a closed connection cannot take it away (see ADR-0011) | `Device.firmware` in `pyquadcortex/device/device.py` | Values derived from an entry rather than read from the unit, which compute from `value()` rather than caching alongside it | ## 6. Constraints @@ -84,6 +87,7 @@ Decisions for this area are recorded in [`ADR.md`](ADR.md): | ADR-0008 | The generator floor joins the bindings/pin unit, with a gate at regeneration and a CI check on the pin | | ADR-0009 | Persistent listeners run on the RX thread, which may not read from the device | | ADR-0010 | A control with no known wire path gets a bounded search before it is modelled as refused | +| ADR-0011 | A push merges, a read replaces, and anything the cache cannot place forces one read | ## 8. Open Questions @@ -123,6 +127,45 @@ Single-device, single-connection USB HID at interactive rates (129-byte reports) ## Change Log +### 2026-08-14 - The model keeps its own copy of what the unit is doing (ADR-0011) + +**What changed:** +- `pyquadcortex/device/state.py`: the write-through cache every model read now goes + through. One persistent listener (ADR-0009), registered before the connect handshake + so it hears the burst, applies what the unit pushes into a per-entry copy. Reads + happen on the caller's thread; the RX thread only merges and marks. +- `pyquadcortex/device/entries.py`: what is tracked, as data rather than code - the + message types that carry each entry, the fields the model keeps from each, and the + read that fetches it. Two of `domain-model.md` section 9's rows so far. +- `pyquadcortex/device/watch.py`: the write side - a watcher per write with section + 10's three outcomes, and one watchdog thread per connection that does not start + until something is written. +- `pyquadcortex/device/device.py`: `Device.firmware` and `.serial` read through the + cache instead of holding their own reply; `Device.state` exposes the layer; + `connect()` subscribes before the handshake and `close()` unsubscribes. +- `tests/hardware/conftest.py`: the run's connection also carries a `DeviceState` + subscribed before the handshake, plus a snapshot of what the burst warmed, taken + before any test can read through it. + +**Why:** the model has to be right about a change somebody made on the touchscreen +while a script was connected, and no property may ship with a "might be stale" +caveat. Story OM-M1.3 (#11), Epic #8. + +**What this constrains going forward:** +- A model property reads through `Device.state`, and what it reads is a `StateEntry`. + A property that caches its own answer is a second account of the same fact, with + nothing to invalidate it and nothing to take it away when the connection closes. +- An entry with no read is not an entry. Section 9's table is longer than the + registry on purpose; each row lands with the surface that reads it. +- The RX thread's rule is now load-bearing in the model as well as the transport: + push-handling code merges and marks and returns, and never reads. +- A field with no wire presence needs recorded evidence before the model keeps it, + and `tests/test_state.py` holds every such declaration against the schema. + +**Not covered here:** reconnect and device loss (#15), the Directory, presets, the +grid and parameters (#12 and after), and the counters and event taxonomy (#16). The +log events this code emits are named for #16 to pick up, but nothing reads them yet. + ### 2026-08-13 - One translation boundary, and the model package is `device/` **What changed:** diff --git a/docs/architecture.md b/docs/architecture.md index 7ca8e9e..8a32f7c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -38,7 +38,19 @@ only about the layer directly below it. pyquadcortex/ THE MODEL - what import pyquadcortex hands back device/device.py connect(): opens the unit, returns a Device. | Speaks the unit's vocabulary, never the wire. - | (Directory, cache and grid land in later stories.) + | (Directory and grid land in later stories.) + | + device/state.py The write-through cache every model read goes + | through: applies what the unit pushes, asks for + | what it does not push, and knows when to stop + | trusting its copy + | + device/entries.py What the cache tracks, as data: which messages + | carry each part, which fields the model keeps, + | and how to read each one + | + device/watch.py Whether a write landed: the echo watcher and its + | three outcomes, plus the one thread that gives up | device/translate.py Screen values <-> wire values, and the ONLY place | either becomes the other: rows, slots, scene and @@ -200,9 +212,41 @@ a `Device`, which carries the unit's identity and owns the connection. does NOT take ownership of it. `Device.client` is the way back down to the message level for anything the model does not cover yet. -The rest of the model - the Directory, the write-through cache, the loaded preset -and the grid - is designed in [domain-model.md](domain-model.md) and is being -built story by story. Nothing is stubbed out to look finished. +Every value a `Device` reports comes out of the state layer below, reached as +`Device.state`. `connect()` builds that cache first and hands its subscription to +`protocol.connect(before_handshake=...)`, because the handshake's burst of state +starts seconds after `connect()` returns - a model that subscribed to the client +it is handed would miss all of it. `Device.from_client` cannot do that and does +not pretend to: it subscribes to the live connection and starts cold. + +The rest of the model - the Directory, the loaded preset and the grid - is +designed in [domain-model.md](domain-model.md) and is being built story by story. +Nothing is stubbed out to look finished. + +### device/state.py, device/entries.py, device/watch.py + +The state layer, designed in [domain-model.md](domain-model.md) sections 9 and 10 +and decided in ADR-0011. Someone turns a knob on the touchscreen while a script +is connected, and the library should not be wrong about it. + +`state.py` holds the cache. It registers one persistent listener (ADR-0009) and, +for each message, merges the fields the model keeps into its copy. A message that +sets a field the model does NOT keep marks that part of the cache, and the next +read of it goes to the unit - on the CALLER's thread. The RX thread never reads, +which the transport enforces rather than asks for. A message type no entry tracks +returns immediately, which is what makes the metronome's tempo stream - a pair +per beat, on every connection, forever - cost nothing. + +`entries.py` is the table of what is tracked: per entry, the message types that +carry it, the fields kept from each, and the read that fetches it. Two entries +today, `identity` and `dirty`; the rest of section 9's table arrives with the +surfaces that read it. + +`watch.py` is the write side. A write updates the cache immediately and the +unit's echo confirms it in the background, against one sentence: every field we +sent must come back with the value we sent. Not "the echo equals what we sent" - +the unit legitimately changes things nobody asked about. One watchdog thread for +the whole connection, started on the first write and never before it. ### device/translate.py @@ -473,6 +517,8 @@ How each layer is faked: | `session` | `open_device` and `Transport` monkeypatched | `tests/test_session.py` | | `cli` | `build_parser()` exercised directly | `tests/test_cli.py` | | `model` | `FakeClient`: answers the calls the model makes on a `QuadCortex`, plus the same monkeypatched device+transport as `session` | `tests/test_device.py` | +| the state layer | `LoopbackTransport`: canned replies under the REAL `QuadCortex`, notifying listeners before the caller wakes as the real transport does | `tests/test_state.py` | +| the state layer's threading | none - a real `Transport` over a fake HID link, because "the RX thread never reads" is a claim about a thread and a double cannot test it | `tests/test_state_rx.py` | | schema | asserts the enum integers the code relies on and that core messages instantiate | `tests/test_schema_compiles.py` | | namespaces | the pre-flip `__all__`, read verbatim from git, must all resolve under `pyquadcortex.protocol` | `tests/test_namespace.py` | | the translation boundary | none: it is pure functions, so it is called directly. Two of its tests take the package's SOURCE as their input instead, and read it with `ast` | `tests/test_translation.py` | diff --git a/docs/protocol.md b/docs/protocol.md index 2de3e61..44f5e11 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -1836,6 +1836,32 @@ and the asymmetry is load-bearing for anyone maintaining a cached preset: * Block bypass is still unmeasured: its test skips unless the target block already carries a stored bypass entry. See `tests/hardware/test_write_echo.py`. +## `PresetDirty` announces a CHANGE of flag, not an edit + +The `Grid` echo above arrives for every edit. `PresetDirty` does not, and the +difference matters to anything watching one to learn about the other. + +Measured 2026-08-14 on d14e, as a controlled pair inside one connection. The same +`set_param` write was made twice on the same block, with the RX path tapped for +everything the unit sent in the two seconds after each: + +| The flag before the write | What the unit sent | +|---|---| +| `false` | a `Grid` echo **and** a `PresetDirty` carrying `is_dirty: true` | +| `true` | a `Grid` echo, and nothing else at all | + +So `PresetDirty` is an announcement about the FLAG, and the flag only changes once +until something clears it. A save clears it (confirmed earlier, watched flipping +across a save). Writing an edited parameter back to the value it had did NOT clear +it within the same connection, which is worth knowing before treating a restore as +an undo. + +This corrects "also pushed unsolicited ... on edits", which was true of the edit +that first dirties a preset and read as though it were true of every edit. Anything +waiting for a `PresetDirty` to confirm an edit landed will wait out its timeout on +an already-dirty preset - correctly, since the unit really did not say anything. +The `Grid` echo is the per-edit signal. + ## Connect burst, measured About 3 s of quiet, then the ModelRepo payload as one huge message, then ~400 File @@ -2617,7 +2643,7 @@ visually on the device's own screen. | `io_settings` / `set_input_level` / `set_output_level` | `IOSettings{READ}` / `{UPDATE, settings{in_port` or `out_port{port_id, level}}}` | read-back | sparse and port-keyed; also reports impedance, type, ground lift and `plugged` | | `global_eq` / `set_global_eq_bypassed` | `GlobalEQ{READ}` / `{UPDATE, bypassed}` | read-back | five bands reported as 28 parameters | | `mode` / `set_mode` | `Mode{READ}` / `{UPDATE, mode}` | read-back | a slot index; `available_modes` lists the configured slots | -| `preset_dirty` | `PresetDirty{READ}` | request_id echo | answers as UPDATE in 2-11 ms (two hardware sessions); `is_dirty` has no presence, absent IS false; flips false across a save; also pushed unsolicited | +| `preset_dirty` | `PresetDirty{READ}` | request_id echo | answers as UPDATE in 2-11 ms (two hardware sessions); `is_dirty` has no presence, absent IS false; flips false across a save; also pushed unsolicited, but only when the flag CHANGES - see below | | `set_gig_view` | `ShowGigView{UPDATE, show}` | read-back + on-unit | `show` has no presence | | `set_input_gate` | `Grid{UPDATE, preset{chains{row, input_control{hash: 28000, params{index, param_values}}}}}` | read-back | the per-row noise gate; NOISE REDUCTION, BYPASS and INPUT GAIN all confirmed in both directions, per-scene included. GAIN REDUCTION is a meter (`grMeter`), not a control | | `free_rows` | reads `models[]` + `Chain.split_control_points` | read-back | rows available for an independent chain: excludes the lane row of a branch, which is spoken for even when empty | diff --git a/pyquadcortex/device/device.py b/pyquadcortex/device/device.py index f427175..eb122ce 100644 --- a/pyquadcortex/device/device.py +++ b/pyquadcortex/device/device.py @@ -8,13 +8,19 @@ with pyquadcortex.connect() as device: print(device.firmware, device.serial) +Everything a `Device` reports comes through the state layer +(:mod:`pyquadcortex.device.state`), which listens to what the unit announces and +asks it directly for the rest. So a value read here is what the unit is doing +now, not what it was doing when somebody last asked. + The `Device` is deliberately small right now. It carries the unit's identity and -owns the connection; the Directory, the live cache, the loaded preset and the -grid arrive in the stories that follow, per ``docs/domain-model.md``. What is -here is what has been built - nothing is stubbed out to look finished. +owns the connection; the Directory, the loaded preset and the grid arrive in the +stories that follow, per ``docs/domain-model.md``. What is here is what has been +built - nothing is stubbed out to look finished. """ from pyquadcortex import protocol +from pyquadcortex.device.state import DeviceState class Device: @@ -24,12 +30,19 @@ class Device: connection you already hold with :meth:`from_client`. """ - def __init__(self, client, *, _owns_client: bool = False): + def __init__(self, client, *, _owns_client: bool = False, _state=None): """Internal. Use :func:`connect` or :meth:`from_client`.""" self._client = client self._owns_client = _owns_client self._closed = False - self._version = None + # `connect` builds the cache itself so it can subscribe before the + # handshake, which is the only moment early enough to hear the burst. + # Anything else is joining a connection already up, so it subscribes to + # the client and starts cold. + self._state = _state if _state is not None else DeviceState() + if _state is None: + self._state.listen_on(client) + self._state.bind(client) def _check_open(self) -> None: """Refuse to answer through a `Device` the caller has finished with. @@ -76,46 +89,50 @@ def client(self): self._check_open() return self._client + @property + def state(self): + """The state layer this `Device` reads through. + + The place to look when you want to know what the model knows - what it + has cached, what it is about to re-read - rather than what the unit is + doing. Reading a property is the way to ask the unit. + + Raises ``RuntimeError`` once this `Device` is closed. + """ + self._check_open() + return self._state + @property def firmware(self) -> str: - """The firmware version the unit reports, e.g. ``"d14e"``.""" - return self._identity().app_fw_version + """The firmware version the unit reports, e.g. ``"d14e"``. + + The unit never announces this, so the first read asks it and every read + after that is free for as long as this connection lasts. Firmware and + serial cannot change while a connection is up: the only thing that + changes either is a firmware update, and the firmware `Updater` surface + is permanently out of scope for this library (repo-root ``CLAUDE.md``). + That is an inference from scope rather than a measurement, which is why + the state layer still treats it as ordinary cached state rather than as + something read once and settled. + """ + return self._identity("app_fw_version") @property def serial(self) -> str: """The unit's serial number.""" - return self._identity().device_serial_number - - def _identity(self): - """The unit's Version reply, read once per connection. - - Firmware and serial cannot change while a connection is up, so one read - answers both properties for as long as this `Device` is connected. That - rests on an inference, not a measurement: the only thing that changes - either value is a firmware update, and the firmware `Updater` surface is - permanently out of scope for this library (repo-root ``CLAUDE.md``), so - the claim cannot be tested here. It is why the reply is only cached once - it is known to be complete. - - Both fields sit in a synthetic ``oneof`` in the schema, so protobuf hands - back ``""`` for a field the unit never sent rather than complaining. An - empty string behind a signature promising a version is a guess, so a - reply missing either field raises and is not cached, leaving a retry able - to recover. + return self._identity("device_serial_number") + + def _identity(self, field: str) -> str: + """One field of the unit's identity, through the cache. + + Both fields sit in a synthetic ``oneof`` in the schema, so protobuf + hands back ``""`` for a field the unit never sent rather than + complaining. An empty string behind a signature promising a version is a + guess, so the state layer refuses a field the unit did not send and + caches nothing for it, leaving a retry able to recover. """ self._check_open() - if self._version is None: - reply = self._client.version() - missing = [f for f in ("app_fw_version", "device_serial_number") - if not protocol.field_present(reply, f)] - if missing: - raise RuntimeError( - f"the unit's Version reply did not carry {', '.join(missing)}, " - f"so its firmware and serial cannot be reported. Nothing was " - f"cached, so asking again can still succeed." - ) - self._version = reply - return self._version + return self._state.value("identity", field) def close(self) -> None: """Finish with this `Device`, releasing the unit if it opened it. @@ -129,8 +146,13 @@ def close(self) -> None: `close()` defines. A connection that goes away on its own - the cable pulled, the unit rebooted - is a different event with its own handling, and belongs to the reconnect story (#15). + + The state layer is closed first, so a `Device` built by + :meth:`from_client` stops listening on a connection it never owned + rather than quietly staying subscribed to somebody else's. """ self._closed = True + self._state.close() if self._owns_client: self._client.close() @@ -173,6 +195,13 @@ def connect(*, timeout: float = 5.0, settle: float = 2.0, while the unit is openable but silent. See :func:`pyquadcortex.protocol.connect`, which this passes through to. + The model subscribes to the unit's pushes BEFORE the handshake runs, which + is the only moment early enough to hear the handshake's own burst of state - + one message of nearly every state type the unit has, the current preset + included, arriving over about nine seconds. That is what makes the cache warm + for free: by the time a caller asks for something, the unit has usually + already said it. + Returns: A connected :class:`Device`. @@ -185,6 +214,8 @@ def connect(*, timeout: float = 5.0, settle: float = 2.0, rather than after it. Raising ``handshake_patience`` is the fix; the 30 second default already covers the measured window. """ + state = DeviceState() client = protocol.connect(timeout=timeout, settle=settle, - handshake_patience=handshake_patience) - return Device(client, _owns_client=True) + handshake_patience=handshake_patience, + before_handshake=state.listen_on) + return Device(client, _owns_client=True, _state=state) diff --git a/pyquadcortex/device/entries.py b/pyquadcortex/device/entries.py new file mode 100644 index 0000000..2163de6 --- /dev/null +++ b/pyquadcortex/device/entries.py @@ -0,0 +1,265 @@ +"""What the model tracks, and how each part of it stays current. + +This is ``docs/domain-model.md`` section 9's table written as data. Each entry +names one part of the cache and says three things about it: which message types +carry it, which of their fields the model keeps, and how to ask the unit for it +when we have nothing or have stopped trusting what we have. + +**Why the fields are listed one by one.** A push is often partial - the standby +announcement carries only ``power_option`` - so applying it means merging the +fields it names and leaving the rest alone. The failure that makes a cache worse +than no cache is applying the half of a message we understand and dropping the +rest, because the result is confidently wrong rather than obviously stale. So the +rule here is per FIELD: a field we keep is applied, a field we do not keep makes +the whole entry untrusted and the next read goes to the unit. There is no third +option and nothing is silently ignored, which is why nothing here is a category +called "harmless". + +That is deliberately conservative. A push that mentions one field we do not keep +costs one read, whether or not that field could really have made our copy wrong. +The alternative - declaring, field by field, which changes cannot affect what we +hold - is a judgement call per field with no way to check it, and the whole point +of section 9 is that the model does not guess. + +**Adding an entry.** Write it here, give it a read, list the fields, and add its +tests to ``tests/test_state.py``. The structural tests in that file hold every +entry to the same standard: a field is only read without a presence check if the +schema really gives it no presence, and every field named has to exist. +""" + +import dataclasses +import typing + +from pyquadcortex import protocol +from pyquadcortex.protocol.proto import ProductionAutomation_pb2 as pa + +#: Fields the per-field check skips, on every entry. They are on nearly every +#: message in this schema, and treating them as unkept state would mark every +#: entry for re-reading on every push - the thrash section 9 exists to avoid. +#: ``tests/test_state.py`` checks both are really on every feeding type, so the +#: skip cannot quietly forgive a field that never arrives. +#: +#: ``request_id`` is the transport's, always. **``action`` is not always.** It +#: says nothing on the two message types tracked today, which is why it is +#: skipped globally - but on ``Grid`` it is load-bearing state: an +#: ``UPDATE`` carrying ``hash: 0`` is transmitted and ignored, while the same +#: payload with ``action: DELETE`` removes the block +#: (``QuadCortex.remove_block``). So a ``Grid`` entry - issue #12 - cannot +#: inherit this skip: two pushes with identical payloads and opposite meanings +#: would apply identically and mark nothing. Give that entry its own decision +#: about ``action`` rather than widening this set, and see ADR-0011. +SCAFFOLDING = frozenset({"action", "request_id"}) + + +@dataclasses.dataclass(frozen=True) +class FieldPlan: + """What one message type carries for one entry. + + Args: + kept: fields the model holds, applied when the message says so. Each has + field presence in the schema, so "absent" is a fact rather than a + guess and the model can tell "not mentioned" from "set to zero". + no_presence: fields the wire cannot report as absent, applied on every + message of this type. Proto3 gives a plain scalar no presence, so its + default value and "unset" are the same bytes and there is nothing to + check - which means each of these needs recorded evidence for what + the default MEANS, on the entry that declares it. Empty for most. + """ + + kept: frozenset = frozenset() + no_presence: frozenset = frozenset() + + +@dataclasses.dataclass(frozen=True, eq=False) +class StateEntry: + """One part of the cache: what feeds it, and how to ask for it. + + Args: + name: how the rest of the model refers to this entry. + read: ``callable(client)`` returning a mapping of field name to value - + the unit's whole answer for this entry. Runs on the CALLER's thread, + never the RX thread. + feeds: message class -> :class:`FieldPlan`. + + Every entry's :attr:`read` is one request and one reply, which the read path + relies on to tell its own answer apart from a push that arrived while it was + waiting. An entry whose read provokes a STREAM instead - a ``File`` + enumeration, a preset dump - has to say how many messages that is, and this + class does not carry that yet because nothing needs it. It lands with the + first such entry, along with the test that a number other than one works. + """ + + name: str + read: typing.Callable + feeds: typing.Mapping + + def fields(self) -> frozenset: + """Every field name this entry holds, across all the types that feed it.""" + found = set() + for plan in self.feeds.values(): + found |= plan.kept | plan.no_presence + return frozenset(found) + + +def fields_applied(message, plan: FieldPlan) -> dict: + """The fields of ``message`` this plan keeps, as a mapping. + + A kept field appears only if the message actually carries it, so merging the + result leaves everything the push did not mention alone. A presence-free + field always appears, because there is no such thing as a message of its + type that does not carry it. + """ + found = {} + for name in plan.kept: + if protocol.field_present(message, name): + found[name] = getattr(message, name) + for name in plan.no_presence: + found[name] = getattr(message, name) + return found + + +def unkept_fields(message, plan: FieldPlan) -> list: + """Everything ``message`` sets that this plan does NOT keep, named. + + Two kinds, and the second is the reason this is not a set difference over + the descriptor: + + * a field in the schema that the entry does not keep. Tomorrow's field, or a + part of the unit this story does not model yet; + * a field number the schema does not have at all. The schema here is + recovered rather than published (ADR-0010), so a field the unit really + sends and our bindings have never heard of is ordinary. It decodes into + nothing, which makes it the quietest possible way to drop half a message, + and the only way to see it is to notice the bytes. + + Detecting the second costs a copy of the message per push, which is why this + runs once per entry rather than once per field. It is cheap for the small + keyed pushes an edit produces; an entry fed by whole preset dumps should + measure it before assuming the same. + """ + named = {field.name for field, _ in message.ListFields()} + found = sorted(named - plan.kept - plan.no_presence - SCAFFOLDING) + if _carries_unknown_fields(message): + found.append("a field number this schema does not have") + return found + + +def _carries_unknown_fields(message) -> bool: + """Whether ``message`` decoded with bytes our schema could not place. + + Asked by subtraction: discard the unknown fields from a copy and see whether + the message got smaller. The copy is why the caller does this once per + entry, not once per field. + + **The reason to keep it that way is recursion**, and this paragraph exists + because the obvious simplification loses it silently. + ``DiscardUnknownFields`` descends into submessages; + ``google.protobuf.unknown_fields.UnknownFieldSet`` reports only the top + level. Measured on protobuf 7.35.1: for an unknown field nested inside a + known submessage, subtraction says yes and ``UnknownFieldSet`` counts zero. + Nested is the case that will matter most, because the entries fed by whole + preset dumps are the ones with submessages in them. + + (``message.UnknownFields()``, the third way to ask, raises + ``NotImplementedError`` outright on the C implementation this project runs + on. That is why it is not used, but it is not the reason for the choice + between the other two.) + """ + probe = type(message)() + probe.CopyFrom(message) + before = probe.ByteSize() + probe.DiscardUnknownFields() + return probe.ByteSize() != before + + +# -- the entries ------------------------------------------------------------- + + +#: The unit's identity. Firmware and serial cannot change while a connection is +#: up: the only thing that changes either is a firmware update, and the firmware +#: `Updater` surface is permanently out of scope for this library (repo-root +#: ``CLAUDE.md``). That is an inference from scope rather than a measurement, +#: which is why the read is still the fallback rather than a one-time fill. +#: +#: The unit does not announce this. It sends a ``Version`` READ of its own during +#: the connect handshake - asking US for Cortex Control's version - and that +#: message carries none of the unit's own fields, so the burst does not warm this +#: entry and first access reads. That is the case section 9's third column exists +#: for: where the unit does not tell us, we ask. +_VERSION_FOR_IDENTITY = FieldPlan( + kept=frozenset({"app_fw_version", "device_serial_number"}), +) + + +def _read_identity(client) -> dict: + """``Version{READ}``: the unit's own firmware and serial.""" + return fields_applied(client.version(), _VERSION_FOR_IDENTITY) + + +IDENTITY = StateEntry( + name="identity", + read=_read_identity, + feeds={pa.VersionMessage: _VERSION_FOR_IDENTITY}, +) + + +#: Whether the live grid has edits nobody has saved. Section 9's table puts this +#: behind ``preset.has_unsaved_changes``, which arrives with the preset surface +#: (issue #12); the cache holds it now because it is the entry the unit pushes +#: most plainly - the connect burst delivers one, and every edit produces one. +#: +#: ``is_dirty`` is the model's one presence-free field. Proto3 gives a plain bool +#: no presence, so a clean grid and an unmentioned grid are the same bytes and +#: there is nothing to check. The evidence for reading it anyway is the protocol +#: layer's: ``QuadCortex.preset_dirty`` records that it reads true after an edit +#: and false after a clean save, watched flipping across a save on hardware, and +#: says in as many words that absent simply IS false. Treating an unset message +#: as "not mentioned" would leave the model stuck dirty for the life of the +#: connection. +_PRESET_DIRTY = FieldPlan(no_presence=frozenset({"is_dirty"})) + + +def _read_dirty(client) -> dict: + """``PresetDirty{READ}``: 2-11 ms on every measured poll. + + Goes through ``QuadCortex.preset_dirty``, which unwraps the reply to a bool, + so this builds the mapping by hand rather than through + :func:`fields_applied`. There is one field and it has no presence, so the two + routes agree by construction - and using the published reader keeps the + model off the transport. + """ + return {"is_dirty": client.preset_dirty()} + + +DIRTY = StateEntry( + name="dirty", + read=_read_dirty, + feeds={pa.PresetDirtyMessage: _PRESET_DIRTY}, +) + + +#: Everything the cache tracks. Section 9's table has more rows than this - the +#: preset on the grid, the active scene, the setlists, recents and favourites, +#: the device-level settings - and each arrives with the surface that reads it +#: (#12 and after). An entry with no reader would be a plan, not a fact, and +#: every push mentioning a field it did not keep would mark it for a read nobody +#: had asked for. +ENTRIES = (IDENTITY, DIRTY) + +ENTRY_BY_NAME = {entry.name: entry for entry in ENTRIES} + + +def _by_message_class(): + """message class -> the entries it feeds, with each one's plan.""" + found = {} + for entry in ENTRIES: + for message_class, plan in entry.feeds.items(): + found.setdefault(message_class, []).append((entry, plan)) + return {cls: tuple(pairs) for cls, pairs in found.items()} + + +#: Which entries a decoded message could touch. A type absent from here feeds +#: nothing and is ignored outright, which is what makes the metronome's tempo +#: stream free: it arrives in pairs on every beat of every connection and the +#: model has no tempo surface yet, so there is nothing for it to churn. +FEEDS = _by_message_class() diff --git a/pyquadcortex/device/state.py b/pyquadcortex/device/state.py new file mode 100644 index 0000000..0bccb47 --- /dev/null +++ b/pyquadcortex/device/state.py @@ -0,0 +1,447 @@ +"""The model's copy of what the unit is doing: ``docs/domain-model.md`` section 9. + +Someone turns a knob on the touchscreen while a script is connected. The library +should not be wrong about it. That is the whole job of this module. + +Three rules do it, and they are section 9's: + +1. **The unit tells us when things change**, so we listen and store what it says. + Reading a value later costs nothing. +2. **A message that mentions something we do not keep makes us stop trusting our + copy** of that part, and the next read goes to the unit. The check is per + FIELD, because applying the half of a message we understand and dropping the + rest is the one failure that leaves the cache confidently wrong rather than + obviously stale. +3. **A write updates our copy immediately** and the unit's echo confirms it in + the background (:mod:`pyquadcortex.device.watch`). + +Two things follow from where the code runs. + +**Pushes are applied as data, not as invalidation triggers.** The metronome +clock always runs, so the unit pushes ``GlobalTempo`` in pairs on every beat of +every connection. A cache that re-read on every inbound message would spend its +life re-reading. Applying pushes as data does not care, and a message type no +entry tracks is ignored outright. + +**The RX thread never asks the unit for anything.** :meth:`DeviceState.apply_push` +runs on it (ADR-0009) and only ever merges and marks; the caller's thread does +any reading. The transport enforces the other half - a read from the RX thread +raises rather than stalling the read loop for a timeout it could never survive. + +What is here is what the model reads today. Section 9's table is longer, and each +remaining row arrives with the surface that reads it - see +:mod:`pyquadcortex.device.entries`. +""" + +import itertools +import logging +import threading +import time + +from pyquadcortex.device import entries +from pyquadcortex.device.entries import fields_applied, unkept_fields +from pyquadcortex.device.watch import (WATCH_PATIENCE, WatchOutcome, Watchdog, + WriteWatch) + +log = logging.getLogger(__name__) + +#: What the write watchdog's thread is called, so a caller reading a stack dump +#: knows whose it is. +WATCHDOG_THREAD_NAME = "pyquadcortex-watchdog" + + +class _Slot: + """One entry's copy of the unit's state, and how much we trust it.""" + + __slots__ = ("fields", "needs_read", "witnessed", "_arrivals") + + def __init__(self): + #: field name -> value, holding only what the unit has actually said. + #: A field that is missing was never mentioned, which is not the same as + #: the unit reporting it empty - so it is read rather than answered. + self.fields = {} + #: Set when a message named something this entry does not keep. Cleared + #: by a read, not by another push. + self.needs_read = False + #: How many messages for this entry the listener has handled. The read + #: path uses the difference across a read to tell its own answer apart + #: from a push that arrived while it was waiting. + #: + #: Drawn from a counter rather than written `+= 1`, which is the + #: transport's idiom for the same job (`Transport._ids`). It also keeps + #: the package's one-translation-boundary check honest: that check + #: refuses every spelling of index arithmetic outside `translate.py` + #: precisely because it cannot tell a counter from an off-by-one on a + #: row, and a rule with an exemption for "but mine is fine" is not a + #: rule. Do not simplify this back. + self._arrivals = itertools.count(1) + self.witnessed = 0 + + def arrived(self) -> None: + """Note that one more message for this entry has been handled.""" + self.witnessed = next(self._arrivals) + + +class DeviceState: + """The cache one connection's worth of model reads through. + + Built by :func:`pyquadcortex.connect`, which registers it before the connect + handshake so it hears the handshake's burst of state - about 400 messages + covering nearly every state type, which is what makes the cache warm for + free. A `Device` built on a connection somebody else opened starts cold and + reads on first access, because the burst is already over by then. + + Valid only while its connection is. :meth:`close` stops it listening and + makes it refuse reads it could still have served from its copy, because a + model that reports the unit's state through an object with no unit behind it + is the failure this whole layer exists to avoid. + """ + + def __init__(self): + self._lock = threading.Lock() + self._slots = {entry.name: _Slot() for entry in entries.ENTRIES} + #: One per entry, so two threads asking for the same cold value make one + #: round trip rather than two. Always taken BEFORE :attr:`_lock`, and + #: never held by anything that holds :attr:`_lock`. + self._reading = {entry.name: threading.Lock() for entry in entries.ENTRIES} + self._client = None + self._detach = None + self._closed = False + self._watches = {entry.name: [] for entry in entries.ENTRIES} + self._watchdog = Watchdog(self._gave_up_on, WATCHDOG_THREAD_NAME) + + # -- wiring --------------------------------------------------------------- + + def listen_on(self, hub) -> None: + """Subscribe to every message ``hub`` decodes, for the connection's life. + + ``hub`` is anything with the transport's ``add_listener`` - the + `Transport` itself, which is what ``protocol.connect(before_handshake=)`` + hands over, or a `QuadCortex` for a connection already up. + + Registering before the handshake is the only way to hear its burst, so + this is called with the transport rather than the client on the path + that owns the connection. + """ + if self._detach is not None: + raise RuntimeError( + "this DeviceState is already listening - a second registration " + "would apply every message twice") + self._detach = hub.add_listener(self.apply_push) + + def bind(self, client) -> None: + """Use ``client`` for the reads this cache issues on a caller's thread.""" + with self._lock: + self._client = client + + def close(self) -> None: + """Stop listening, stop watching writes, and answer nothing further. + + Safe to call more than once. It does not close the connection: whoever + opened it closes it. + """ + with self._lock: + self._closed = True + self._slots = {name: _Slot() for name in self._slots} + self._client = None + in_flight = [watch for watches in self._watches.values() + for watch in watches] + self._watches = {name: [] for name in self._watches} + detach, self._detach = self._detach, None + if detach is not None: + detach() + # Outside the lock: the watchdog takes it to mark an entry, and stopping + # it joins its thread. + self._watchdog.stop() + for watch in in_flight: + # Released with no outcome, not timed out. Nothing can settle these + # now, so anybody waiting has to be let go - and calling them timed + # out would be a claim about the unit rather than a fact. + log.debug("watch.abandoned %s %s - the connection closed first", + watch.entry, sorted(watch.sent)) + watch.publish() + + # -- what the unit tells us (RX thread) ----------------------------------- + + def apply_push(self, message) -> None: + """Merge one decoded message into the cache. Runs on the RX THREAD. + + Section 9's rules 1 and 2, in the order they matter: apply the fields + this entry keeps, then - if the message named anything it does not - + mark the entry so the next read goes to the unit. Both, not either: the + value between the push and that read is otherwise the old one, which is + confidently wrong for a shorter while rather than not at all. + + A message type no entry tracks returns immediately. Each entry is + applied inside its own guard, so a bug in one costs that entry this + message and no more - the RX thread has to survive whatever happens + here, and the transport's own guard would skip every remaining entry. + + Never reads from the unit, and never can: the transport refuses a read + from this thread outright (ADR-0009). + """ + for entry, plan in entries.FEEDS.get(type(message), ()): + try: + self._apply_one(entry, plan, message) + except Exception: + # Logged, not raised: "the RX thread never dies" outranks + # surfacing this here, and there is no caller to surface it to. + log.exception("cache.push_failed %s from %s", entry.name, + type(message).__name__) + + def _apply_one(self, entry, plan, message) -> None: + applied = fields_applied(message, plan) + unkept = unkept_fields(message, plan) + with self._lock: + if self._closed: + return + slot = self._slots[entry.name] + slot.arrived() + was_empty = not slot.fields + slot.fields.update(applied) + if was_empty and slot.fields: + log.debug("cache.filled %s from a %s", entry.name, + type(message).__name__) + if applied: + log.debug("push.applied %s %s", entry.name, sorted(applied)) + if unkept: + slot.needs_read = True + log.info("push.forced_reread %s - a %s named %s, which the " + "model does not keep", entry.name, + type(message).__name__, ", ".join(unkept)) + settled = [(watch, watch.absorb(applied)) + for watch in self._watches[entry.name]] + self._watches[entry.name] = [watch for watch, outcome in settled + if outcome is None] + for watch, outcome in settled: + if outcome is None: + continue + self._report(watch) + if outcome is WatchOutcome.DIFFERENT: + # The echo we just applied put the unit's own answer in the + # cache for the field it disagreed about - but a write the unit + # contradicted is a write we did not understand, and any OTHER + # field it carried is still sitting there on our say-so, never + # confirmed by anything. Section 10 only asks for a log line + # here; leaving it at that is how the one path that means "we + # have a bug" ends up the one that cleans up least. + self.mark_for_reread( + watch.entry, + f"the unit disagreed with a write of {sorted(watch.sent)}") + watch.publish() + + # -- what we hand back (the caller's thread) ------------------------------ + + def value(self, entry_name: str, field: str): + """This entry's ``field``, reading from the unit if we cannot answer. + + Answers from the cache when the unit has told us and nothing has said + our copy is wrong. Otherwise it reads - on THIS thread, which is the + rule: the RX thread notes what needs re-reading and the caller's thread + does the reading. + + Raises: + KeyError: if no entry keeps ``field``. A programming error, not a + question about the unit, so it never becomes a round trip. + RuntimeError: if this cache is closed, is not bound to a connection, + or if the unit's answer did not carry ``field``. That last one + is deliberate: an absent string decodes as ``""`` and reporting + that would be a guess. Nothing is cached for it, so asking again + can still succeed. + """ + entry = self._entry(entry_name) + if field not in entry.fields(): + raise KeyError( + f"the model does not keep {field!r} on the {entry_name} entry - " + f"it keeps {sorted(entry.fields())}") + with self._reading[entry_name]: + with self._lock: + self._check_open() + slot = self._slots[entry_name] + if not slot.needs_read and field in slot.fields: + return slot.fields[field] + witnessed_before = slot.witnessed + client = self._client + if client is None: + raise RuntimeError( + f"this model is not connected to a unit, so {entry_name} " + f"cannot be read") + log.debug("read.proactive %s", entry_name) + answer = entry.read(client) + with self._lock: + self._check_open() + slot = self._slots[entry_name] + # A read is the unit's whole answer, so it REPLACES rather than + # merging: a field it did not carry is one the unit did not + # confirm, and leaving an older value there would report + # something no read has ever returned. + slot.fields = dict(answer) + # Our own answer came back through the listener too - listeners + # see a reply before the thread that asked for it wakes (ADR- + # 0009) - so exactly one message is expected here, every entry's + # read being one request and one reply. Anything beyond that is + # a push that landed while we waited, and clearing the mark + # regardless would throw it away with nothing left to recover it + # from. An entry whose read provokes a stream will have to say + # how many messages that is; see `StateEntry`. + extra = slot.witnessed - witnessed_before + slot.needs_read = extra > 1 + if slot.needs_read: + log.info("push.forced_reread %s - %d message(s) arrived " + "while it was being read", entry_name, extra) + if field in slot.fields: + return slot.fields[field] + raise RuntimeError( + f"the unit's answer for {entry_name} did not carry {field}, so it " + f"cannot be reported. Nothing was cached for it, so asking again " + f"can still succeed.") + + def cached(self, entry_name: str) -> dict: + """What this entry holds right now, without asking the unit. + + A snapshot for a caller that wants to know what the model knows - + logging, a health check, a test - rather than what the unit is doing. + Use :meth:`value` for that. + + Refuses once this cache is closed, like every other read here. An empty + mapping is an answer, and "the unit has told us nothing" is not what a + connection that has gone away means. + """ + with self._lock: + self._check_open() + return dict(self._slots[self._entry(entry_name).name].fields) + + def needs_read(self, entry_name: str) -> bool: + """Whether the next :meth:`value` on this entry will go to the unit. + + Refuses once closed, for the same reason: on a closed cache every slot + is empty, so this would answer "no" - "the next read is free" about a + read that raises. + """ + with self._lock: + self._check_open() + return self._slots[self._entry(entry_name).name].needs_read + + def mark_for_reread(self, entry_name: str, why: str) -> None: + """Stop trusting this entry's copy; the next read goes to the unit.""" + with self._lock: + self._check_open() + self._slots[self._entry(entry_name).name].needs_read = True + log.info("cache.forced_reread %s - %s", entry_name, why) + + # -- what we tell the unit ------------------------------------------------ + + def write_through(self, entry_name: str, fields: dict, send, + patience: float = WATCH_PATIENCE) -> WriteWatch: + """Apply ``fields`` to the cache, send the write, and watch for the echo. + + Section 9's third rule: our copy is updated immediately, because waiting + for the echo would make every write pay for information we almost always + already have. The echo confirms it in the background and the caller need + not look - a matching echo changes nothing, which is one code path + rather than two. + + Args: + entry_name: the part of the cache this write changes. + fields: what is being written, field name to value. This is the set + the watcher holds the unit to: every one of them must come back + with the value here. + send: ``callable()`` performing the protocol write. Called after the + cache is updated. If it raises, the write never reached the unit, + so the entry is marked for re-reading and the exception is passed + on - our copy would otherwise be the only place that value exists. + patience: seconds to wait for the echo before giving up. + + Returns: + A :class:`~pyquadcortex.device.watch.WriteWatch`. Ignoring it is + fine and normal; the outcomes are logged either way. + + Raises: + ValueError: if a field is not one this entry keeps. A write the + cache cannot hold would be applied nowhere and confirmed against + nothing. + RuntimeError: if this cache is closed. + + Nothing in M1 writes through here yet - ``momentary`` (#14) is the first + - but the cache is write-through by design, so the path exists and is + tested rather than being added under the pressure of a feature. + """ + entry = self._entry(entry_name) + unknown = sorted(set(fields) - entry.fields()) + if unknown: + raise ValueError( + f"the {entry_name} entry does not keep {unknown}, so writing " + f"them through the cache would apply them nowhere") + watch = WriteWatch(entry_name, fields, time.monotonic() + patience) + with self._lock: + self._check_open() + self._slots[entry_name].fields.update(fields) + self._watches[entry_name].append(watch) + self._watchdog.add(watch) + try: + send() + except BaseException: + with self._lock: + self._watches[entry_name] = [ + w for w in self._watches[entry_name] if w is not watch] + # Settled here rather than left for the watchdog: nothing is coming + # back, and a watcher that fired at its deadline would mark the + # entry a second time, long after a caller had put it right. + watch.time_out() + self.mark_for_reread( + entry_name, f"the write of {sorted(fields)} never reached the unit") + watch.publish() + raise + return watch + + def _gave_up_on(self, watch: WriteWatch) -> None: + """The watchdog's callback: no echo arrived. Runs on ITS thread. + + Returns quietly if the connection went away while it was deciding. + There is no copy left to mark, and raising here would land on a thread + with nobody to catch it. + """ + with self._lock: + if self._closed: + return + self._watches[watch.entry] = [ + w for w in self._watches[watch.entry] if w is not watch] + log.warning("watch.timeout %s - the unit never echoed %s", watch.entry, + sorted(watch.sent)) + self.mark_for_reread( + watch.entry, f"the unit never echoed a write of {sorted(watch.sent)}") + + def _report(self, watch: WriteWatch) -> None: + """Log a settled write. Runs on whichever thread settled it.""" + if watch.disagreement is not None: + field, sent, returned = watch.disagreement + log.warning("watch.different %s - we sent %s=%r and the unit " + "returned %r", watch.entry, field, sent, returned) + else: + log.debug("watch.confirmed %s %s", watch.entry, sorted(watch.sent)) + + # -- internals ------------------------------------------------------------ + + def _entry(self, entry_name: str): + try: + return entries.ENTRY_BY_NAME[entry_name] + except KeyError: + raise KeyError( + f"the model tracks no state called {entry_name!r} - it tracks " + f"{sorted(entries.ENTRY_BY_NAME)}") from None + + def _check_open(self) -> None: + if self._closed: + raise RuntimeError( + "this model's connection is closed, so nothing it remembers is " + "still true of the unit - open a new one with " + "pyquadcortex.connect()") + + def __repr__(self) -> str: + # Says nothing about the unit, only about this object: repr() is called + # by debuggers and logging and must never trigger a device read. + with self._lock: + state = "closed" if self._closed else "open" + warm = sorted(name for name, slot in self._slots.items() + if slot.fields and not slot.needs_read) + return f"" diff --git a/pyquadcortex/device/watch.py b/pyquadcortex/device/watch.py new file mode 100644 index 0000000..c822c0c --- /dev/null +++ b/pyquadcortex/device/watch.py @@ -0,0 +1,266 @@ +"""Knowing a write landed: ``docs/domain-model.md`` section 10. + +The unit accepts writes it does not understand and silently does nothing, so "no +error" proves nothing. What we have instead is the echo, and each write gets a +watcher that compares the echo against what we sent. + +**The bar is exactly one sentence.** + + Every field we sent must come back with the value we sent. + +Not "the echo equals what we sent". The unit legitimately changes things nobody +asked about - a gain-reduction meter, a mirrored parameter, NaN in unused slots, +dropdown values recomputed on rows we never touched - and all four are things we +did not send, so comparing only what we sent needs no exception for any of them. + +The watcher never blocks the write. It reports one of three outcomes, and the +caller can ignore all three: a confirmation changes nothing, because section 9's +third rule already applied the write to our copy. +""" + +import enum +import threading +import time + +#: What one watcher waits before giving up. Measured echo latency is 113-116 ms +#: for a parameter write and 290-420 ms for a block placement (section 10), and +#: ``PresetDirty`` answers in 2-11 ms, so this is several times the slowest thing +#: measured. It is a ceiling on how long a silently ignored write stays in our +#: copy, not a latency anybody waits on, so generous is the safe direction. +WATCH_PATIENCE = 2.0 + + +class WatchOutcome(enum.Enum): + """How a write ended. Section 10's three, and there is no fourth.""" + + #: Every field we sent came back with the value we sent. Our copy was + #: already right; nothing to do. + CONFIRMED = "confirmed" + + #: A field we sent came back with another value. That is a bug in our code, + #: now with a name and a location. The echo has been applied, so our copy + #: holds the unit's answer rather than our losing write. + DIFFERENT = "different" + + #: Nothing came back. The part of our copy the write touched is marked for + #: re-reading, so a silently ignored write self-corrects instead of + #: poisoning the cache. + TIMED_OUT = "timed out" + + +class WriteWatch: + """One write, and what the unit said about it. + + Built by ``DeviceState.write_through``; a caller holds one only to find out + how the write ended. Safe to read from any thread: :meth:`absorb` runs on the + RX thread and :meth:`time_out` on the watchdog, while the caller reads + :attr:`outcome` on its own. + """ + + def __init__(self, entry: str, sent: dict, deadline: float): + if not sent: + raise ValueError( + "a write with no fields has nothing to confirm - name the " + "fields being written, or do not go through the cache") + self.entry = entry + self.sent = dict(sent) + self.deadline = deadline + self._lock = threading.Lock() + self._settled = threading.Event() + self._outcome = None + self._disagreement = None + self._confirmed = set() + + @property + def outcome(self): + """The :class:`WatchOutcome`, or ``None`` while the write is in flight.""" + with self._lock: + return self._outcome + + @property + def disagreement(self): + """``(field, what we sent, what came back)``, or ``None``. + + Set only for :attr:`WatchOutcome.DIFFERENT`, and it is the whole point of + that outcome: it turns "a write did not stick" into a field name, a value + and a place to look. + """ + with self._lock: + return self._disagreement + + def settled(self, timeout=None) -> bool: + """Wait for an outcome; return whether there is one. + + Nothing in the model waits on this - the write already returned and the + cache is already right. It is here so a test, or a caller that wants to + be sure, can ask. + + When it returns TRUE, everything the outcome causes has already + happened: the log line is written and a timed-out write has already + marked its entry for re-reading. That is why :meth:`publish` is a + separate step from the two methods that decide the outcome. + + It returns FALSE either because the wait ran out or because the + connection closed with this write still in flight - see + :meth:`publish`. Nothing can settle a write once the connection that + would have echoed it is gone, so the wait ends rather than running to + the caller's timeout, which by default is forever. + """ + self._settled.wait(timeout) + return self.outcome is not None + + def absorb(self, applied: dict): + """Take one echo into account; return the outcome if this settles it. + + ``applied`` is what the cache took out of the echo, which is the same + thing the cache now holds - so a field the echo did not carry is simply + not yet confirmed, and a later echo can still confirm it. A write is + confirmed only when every field it sent has come back matching. + + Does NOT publish: the caller acts on the outcome, then calls + :meth:`publish`. + """ + with self._lock: + if self._outcome is not None: + return None + for field, value in self.sent.items(): + if field not in applied: + continue + if applied[field] != value: + self._disagreement = (field, value, applied[field]) + self._outcome = WatchOutcome.DIFFERENT + break + self._confirmed.add(field) + else: + if self._confirmed == set(self.sent): + self._outcome = WatchOutcome.CONFIRMED + return self._outcome + + def time_out(self) -> bool: + """Give up on this write; return whether this call is what ended it. + + False when an echo got there first, which is the race the watchdog runs + into every time a write is confirmed near its deadline. Does NOT + publish - see :meth:`absorb`. + """ + with self._lock: + if self._outcome is not None: + return False + self._outcome = WatchOutcome.TIMED_OUT + return True + + def publish(self) -> None: + """Release anyone waiting in :meth:`settled`. Idempotent. + + Called after an outcome's consequences have been applied, so a waiter + never wakes ahead of them. Also called with NO outcome when the + connection closes on a write still in flight: there is nothing left that + could answer it, so the honest thing is to stop the waiting rather than + to invent a third party's verdict. :meth:`settled` reports that as + false, which is what it says it reports. + """ + self._settled.set() + + def __repr__(self) -> str: + outcome = self.outcome + return (f"") + + +class Watchdog: + """One thread that gives up on writes the unit never echoed. + + One thread for the whole connection, not one per write, and it does not + start until something is written - a connection that only reads never has it. + It sleeps until the earliest deadline and wakes when a write is added. + + It marks and logs; it never reads from the unit. Section 9's rule is about + the RX thread, but the reason behind it - the thread that notices is not the + thread that asks - is why a read here would be just as wrong: it would put a + device round trip on a thread no caller knows exists. + """ + + def __init__(self, on_timeout, name: str): + self._on_timeout = on_timeout + self._name = name + self._wake = threading.Condition() + self._watches = [] + self._running = False + self._stopped = False + self._thread = None + + def add(self, watch: WriteWatch) -> None: + """Start watching ``watch``, starting the thread if this is the first. + + Once :meth:`stop` has run this watches nothing and starts nothing. That + is the race a write can lose: ``write_through`` finds the cache open, + the connection closes, and only then does the write reach here. Starting + a thread for it would leave one waiting out a deadline on a connection + nobody can reach, so the write is released with no outcome instead - + which is what the connection closing under a write means anyway. + """ + with self._wake: + if self._stopped: + late = watch + else: + late = None + self._watches.append(watch) + if self._thread is None: + self._running = True + self._thread = threading.Thread(target=self._loop, + name=self._name, daemon=True) + self._thread.start() + self._wake.notify_all() + if late is not None: + late.publish() # outside the lock: it wakes other threads + + def stop(self, join_timeout: float = 2.0) -> None: + """Stop the thread for good and forget every outstanding write. + + Idempotent, and permanent - a :class:`Watchdog` belongs to one + connection and a stopped one never watches again. + + The watches are dropped rather than timed out: the connection is going + away, so "the unit never answered" would be a claim about the unit + rather than a fact. Releasing whoever is waiting on them is the caller's + job, because the caller is the one that knows the connection has gone - + see ``DeviceState.close``. + """ + with self._wake: + self._running = False + self._stopped = True + self._watches = [] + thread, self._thread = self._thread, None + self._wake.notify_all() + if thread is not None and thread is not threading.current_thread(): + thread.join(join_timeout) + + def _loop(self) -> None: + while True: + with self._wake: + while True: + if not self._running: + return + now = time.monotonic() + self._watches = [w for w in self._watches + if w.outcome is None] + due = [w for w in self._watches if w.deadline <= now] + if due: + self._watches = [w for w in self._watches + if w.deadline > now] + break + # Computed and waited on inside one hold of the lock, so an + # add() cannot slip in between and have its notify missed - + # which with no other watch outstanding would be a wait with + # no timeout, and a write that never times out. + delay = (min(w.deadline for w in self._watches) - now + if self._watches else None) + self._wake.wait(delay) + for watch in due: + if watch.time_out(): + try: + self._on_timeout(watch) + finally: + # Published last, so a caller woken by settled() finds + # the entry already marked rather than racing us to it. + watch.publish() diff --git a/pyquadcortex/protocol/client.py b/pyquadcortex/protocol/client.py index b4622e3..dabbab8 100644 --- a/pyquadcortex/protocol/client.py +++ b/pyquadcortex/protocol/client.py @@ -926,7 +926,16 @@ def preset_dirty(self, timeout: float = 5.0) -> bool: 2-11 ms across every measured poll (two independent hardware sessions). Reads true after an edit and false after a clean save - confirmed by watching it flip across a save. Also pushed unsolicited in the connect - burst and on edits, so state trackers can subscribe rather than poll. + burst, so state trackers can subscribe rather than poll. + + **The push announces a CHANGE of the flag, not an edit.** Measured + 2026-08-14 as a controlled pair in one connection: the same ``set_param`` + write produced a ``Grid`` echo AND a ``PresetDirty`` when the preset was + clean, and a ``Grid`` echo and nothing else when it was already dirty. + So do not wait on this to confirm an edit landed - it will time out on an + already-dirty preset, correctly, because the unit said nothing. The + ``Grid`` echo is the per-edit signal. See ``protocol.md``, + "``PresetDirty`` announces a CHANGE of flag, not an edit". ``is_dirty`` has no field presence, so absent simply IS false - do not try to distinguish them. And like most reads here, the FIRST request diff --git a/tests/hardware/conftest.py b/tests/hardware/conftest.py index 40a38eb..e12c226 100644 --- a/tests/hardware/conftest.py +++ b/tests/hardware/conftest.py @@ -124,9 +124,16 @@ def _connection(): :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, because it cannot be attached later on demand: the burst happens during - ``connect``. + Two things are attached before the handshake, and neither can be attached + later on demand, because the burst happens during ``connect``: + + * the burst recorder, for every run rather than only the tests that read it; + * the model's state layer, which is what ``pyquadcortex.connect()`` does at + exactly this point. It stays attached for the whole run, which costs the + RX thread one small message copy per ``Version`` or ``PresetDirty`` push + and nothing at all for anything else - orders of magnitude under the + hundred-millisecond latencies ``test_write_echo.py`` measures. Its own + tests are in ``test_model_state.py``. 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. @@ -136,10 +143,26 @@ def _connection(): a link that is still busy answering the handshake. """ from pyquadcortex import protocol + from pyquadcortex.device import entries + from pyquadcortex.device.state import DeviceState + burst = HandshakeBurst() - with protocol.connect(before_handshake=burst.attach) as client: + cache = DeviceState() + + def subscribe(transport): + burst.attach(transport) + cache.listen_on(transport) + + with protocol.connect(before_handshake=subscribe) as client: + cache.bind(client) burst.record_until("RecallPresetMessage", patience=30.0) - yield client, burst + # Taken here, before any test can read through the cache, so "the burst + # warmed this" cannot later be confused with "some test read it". + warmed = {entry.name: cache.cached(entry.name) for entry in entries.ENTRIES} + try: + yield client, burst, cache, warmed + finally: + cache.close() @pytest.fixture(scope="session") @@ -154,6 +177,18 @@ def handshake_burst(_connection): return _connection[1] +@pytest.fixture(scope="session") +def model_cache(_connection): + """The model's ``DeviceState``, subscribed since before the handshake.""" + return _connection[2] + + +@pytest.fixture(scope="session") +def burst_warmed(_connection): + """What each cache entry held once the burst finished, before any test ran.""" + return _connection[3] + + @pytest.fixture def restores(): """Register undo callables; they run in reverse, failure or not. diff --git a/tests/hardware/readme.md b/tests/hardware/readme.md index 46048f0..8c48032 100644 --- a/tests/hardware/readme.md +++ b/tests/hardware/readme.md @@ -59,6 +59,22 @@ or one that stops recording but stays attached to the transport. Both read like working recorder from the outside, so both are pinned offline in `tests/test_handshake_burst_recorder.py`. +## The model's cache rides the same connection + +`test_model_state.py` covers the model's state layer, and the connection fixture +attaches a `DeviceState` before the handshake for the same reason it attaches the +burst recorder: that is the only moment early enough. It stays attached for the +whole run and costs the RX thread one small message copy per `Version` or +`PresetDirty` push - nothing at all for anything else, and orders of magnitude +under the latencies measured below. + +It needs one thing of the unit that nothing else here does: **a loaded preset with +no unsaved changes**. `PresetDirty` announces a CHANGE of the flag rather than an +edit, so only the first edit of a run produces an announcement, and the test that +proves an outside edit reaches the model needs that announcement. It skips with a +message saying so if the preset arrives already dirty. If you see that skip, save +or reload the preset on the unit and run again. + ## Why the control test exists `test_parameter_echo_latency_is_the_control` measures a write whose latency was diff --git a/tests/hardware/test_model_state.py b/tests/hardware/test_model_state.py new file mode 100644 index 0000000..3e5c170 --- /dev/null +++ b/tests/hardware/test_model_state.py @@ -0,0 +1,338 @@ +"""The model's cache against a real unit: does it hear, and does it stay quiet? + +The offline suite proves the rules: what merges, what forces a re-read, what a +write watcher makes of an echo. It cannot prove the two facts the whole design +rests on, because both are claims about a real unit: + +* **the connect burst warms the cache**, so a value read straight after + connecting costs no round trip; +* **a change the model did not make reaches it**, unasked, with no read. + +Both are checked here. So is the third claim the design leans on hardest - that +the metronome's tempo stream, which never stops, costs nothing at all. + +**State neutrality (ADR-0005).** Two tests edit the unit: each sets one parameter +on the first occupied block of the loaded preset and restores the value it read +first. That is the same edit ``test_write_echo.py`` makes, for the same reason - +it is the smallest change the unit announces. It leaves the loaded preset marked +as having unsaved changes, which is what an edit does and what discarding it +would have to un-do by throwing the owner's work away. Nothing here saves, +recalls, deletes or renames anything. + +**One flag transition per run.** ``PresetDirty`` announces a CHANGE of the flag, +not an edit (``docs/protocol.md``), so the first edit of a run gets an +announcement and the rest do not. The two editing tests are written around that +rather than against it: the first takes the transition and skips if the preset +arrived already dirty, and the second expects whichever outcome its own +before-reading predicts. + +One consequence worth knowing before reading a green run: in a whole-file run the +write-through test is the second edit, so it exercises the TIMED-OUT branch and +its self-correction. Its CONFIRMED branch needs to be the first edit of a run:: + + pytest tests/hardware --hardware -k write_through + +Both were run while this file was written (2026-08-14, d14e): CONFIRMED settled in +612 ms with the unit sending a ``PresetDirty``, and the timed-out branch saw the +unit send a ``Grid`` echo and nothing else. +""" +import collections +import threading +import time + +import pytest + +from pyquadcortex.device import entries +from pyquadcortex.device.watch import WatchOutcome + +#: The metronome clock always runs, so the unit pushes GlobalTempo in pairs, one +#: pair per beat - 1.5 s apart at the slowest tempo the unit offers (40 bpm). +#: Ten seconds is several beats even there. +PATIENCE = 10.0 + +#: Long enough to see the tempo stream several times over at any tempo. +QUIET_WATCH = 6.0 + + +class CountingClient: + """The real client, with every read the cache issues counted. + + Wraps rather than replaces, so the reads are real round trips to the unit - + this only records that they happened. "No round trip" is otherwise a claim + nothing on hardware can check. + """ + + def __init__(self, real): + self._real = real + self.reads = collections.Counter() + + def version(self, *args, **kwargs): + self.reads["version"] += 1 + return self._real.version(*args, **kwargs) + + def preset_dirty(self, *args, **kwargs): + self.reads["preset_dirty"] += 1 + return self._real.preset_dirty(*args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + +class Pushes: + """Records messages from the RX thread, for the test to read. + + ``type_name=None`` records everything, which is what a failing echo test + needs: "nothing came back" is a claim about the unit, and this file's + neighbour learned the hard way that such a claim can be wrong while the unit + is talking perfectly well (see ``test_write_echo.py``'s ``_landed``). + """ + + def __init__(self, type_name=None): + self._type_name = type_name + self._lock = threading.Lock() + self._seen = [] + + def __call__(self, message): + if self._type_name in (None, type(message).__name__): + with self._lock: + self._seen.append(message) + + def seen(self): + with self._lock: + return list(self._seen) + + def tally(self): + return dict(collections.Counter(type(m).__name__ for m in self.seen())) + + +@pytest.fixture +def counted(qc, model_cache): + """Count the cache's reads for one test, then give it the client back.""" + counting = CountingClient(qc) + model_cache.bind(counting) + try: + yield counting + finally: + model_cache.bind(qc) + + +def _wait_until(predicate, timeout=PATIENCE): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.05) + return predicate() + + +def _first_occupied_block(qc): + """Row 1's first block, as the wire indexes ``set_param`` wants. + + Wire coordinates on purpose: this drives the PROTOCOL client directly, which + keeps its zero-based indexes, so there is nothing here to convert. + """ + preset = qc.read_current_preset() + for column, model in enumerate(preset.chains[0].models): + if model.hash: + was = next(p.param_values[0].float_value + for p in model.params if p.index == 0) + return column, was + pytest.skip("row 1 of the loaded preset is empty, so there is nothing to edit") + + +# -- the burst ----------------------------------------------------------------- + + +def test_the_connect_burst_warms_the_cache(burst_warmed, handshake_burst, + record_property): + """The cheapest state the model ever gets: the unit volunteers it. + + Read from a snapshot the connection fixture took the moment the burst + finished, so this says the BURST filled it rather than some earlier test. + """ + record_property("burst_warmed", {name: sorted(fields) + for name, fields in burst_warmed.items()}) + counted = collections.Counter(handshake_burst.names()) + + assert counted.get("PresetDirtyMessage"), ( + f"the burst carried no PresetDirty, so there was nothing to warm the " + f"cache with - it recorded {dict(counted)}") + assert "is_dirty" in burst_warmed["dirty"], ( + f"the unit announced its unsaved-changes state during the burst and the " + f"model did not keep it - the cache held {burst_warmed}") + + +def test_nothing_the_burst_delivered_is_read_again_on_first_access( + model_cache, counted): + """The non-functional requirement, stated as a measurement.""" + assert model_cache.value("dirty", "is_dirty") in (True, False) + assert counted.reads["preset_dirty"] == 0, ( + "the unit had already said so during the handshake and the model asked " + "again, which is the round trip the cache exists to avoid") + + +def test_the_burst_does_not_warm_what_the_unit_never_announces(burst_warmed): + """The other half, and the reason the read path exists. + + The unit sends a ``Version`` of its own during the handshake, but it is a + READ asking US for Cortex Control's version and it carries none of the + unit's own fields. So identity is exactly the case section 9's third column + is for: where the unit does not tell us, we ask. + """ + assert burst_warmed["identity"] == {}, ( + f"the burst carried the unit's own identity after all, which is worth " + f"knowing - it held {burst_warmed['identity']}. If that is now true, " + f"the entry's docstring in device/entries.py is wrong.") + + +def test_state_the_unit_never_announces_costs_one_read_and_then_none( + model_cache, counted): + """No model property ships with a staleness caveat, so it asks - once.""" + model_cache.mark_for_reread("identity", "this test wants a cold read") + + firmware = model_cache.value("identity", "app_fw_version") + serial = model_cache.value("identity", "device_serial_number") + + assert firmware and serial + assert counted.reads["version"] == 1, ( + f"reading two fields of one entry took {counted.reads['version']} reads") + + +# -- a change the model did not make ------------------------------------------- + + +def test_an_edit_the_model_did_not_make_reaches_its_cache(qc, model_cache, + counted, restores, + record_property): + """The story's whole point, on the unit. + + The edit goes through the PROTOCOL client, so as far as the model is + concerned somebody else changed the unit - which is what a hand on the + touchscreen is. Nothing here asks the unit anything; the model finds out + because the unit says so. + + Needs a preset with no unsaved changes, because `PresetDirty` announces a + CHANGE of the flag rather than an edit (``protocol.md``). One transition is + available per run, and this test is the one that gets it - which is why it + comes before the write-through test in this file. + """ + if qc.preset_dirty(): + pytest.skip( + "the loaded preset already has unsaved changes, and the unit only " + "announces the flag when it CHANGES - so an edit now would tell the " + "model nothing and this test could not say anything. Save or reload " + "the preset on the unit and run this again.") + column, was = _first_occupied_block(qc) + restores("row 1 first block, parameter 0", lambda: qc.set_param(0, column, 0, was)) + + announcements = Pushes("PresetDirtyMessage") + everything = Pushes() + qc.add_listener(announcements) + qc.add_listener(everything) + try: + qc.set_param(0, column, 0, 0.75 if abs(was - 0.75) > 0.05 else 0.25) + assert _wait_until(lambda: announcements.seen()), ( + f"the unit said nothing about unsaved changes for an edit it " + f"accepted. It sent {everything.tally()} during the window, so read " + f"that before blaming the unit for saying nothing.") + finally: + qc.remove_listener(announcements) + qc.remove_listener(everything) + + announced = bool(announcements.seen()[-1].is_dirty) + record_property("unit_announced_is_dirty", announced) + assert announced is True, "the unit announced an edit as leaving no unsaved changes" + assert model_cache.cached("dirty")["is_dirty"] is True, ( + f"the unit said so and the model holds {model_cache.cached('dirty')}") + assert model_cache.value("dirty", "is_dirty") is True + assert counted.reads["preset_dirty"] == 0, "the model asked instead of listening" + + +def test_a_write_through_the_cache_is_settled_by_what_the_unit_says( + qc, model_cache, restores, record_property): + """Section 10's outcomes, against the unit that produces them. + + Which one is expected is decided by a state read BEFORE the write, not by + what came back - both branches assert something specific, and neither can + stand in for the other: + + * on a clean preset the edit changes the flag, the unit announces it, and the + write is CONFIRMED; + * on an already-dirty preset the unit announces nothing (``protocol.md``, + "``PresetDirty`` announces a CHANGE of flag"), so the write TIMES OUT and + the cache marks that entry - and the re-read that follows gets the truth + from the unit. That is the self-correction section 10 is for, on the one + occasion this suite can produce it honestly. + + ``different`` stays an offline test: it is by definition a bug in our code. + """ + column, was = _first_occupied_block(qc) + restores("row 1 first block, parameter 0", lambda: qc.set_param(0, column, 0, was)) + target = 0.75 if abs(was - 0.75) > 0.05 else 0.25 + already_dirty = qc.preset_dirty() + record_property("preset_was_already_dirty", already_dirty) + + everything = Pushes() + qc.add_listener(everything) + try: + started = time.monotonic() + watch = model_cache.write_through( + "dirty", {"is_dirty": True}, + send=lambda: qc.set_param(0, column, 0, target)) + + assert model_cache.cached("dirty")["is_dirty"] is True, ( + "the cache was not updated until the echo arrived, which is the " + "round trip section 9's third rule exists to avoid") + assert watch.settled(timeout=PATIENCE), "the watcher never settled" + finally: + qc.remove_listener(everything) + record_property("watch_settled_in_ms", (time.monotonic() - started) * 1000.0) + record_property("unit_sent", everything.tally()) + + assert everything.tally().get("GridMessage"), ( + f"the unit did not echo the parameter write at all, so this test is " + f"measuring a write that never landed - it sent {everything.tally()}") + + if already_dirty: + assert watch.outcome is WatchOutcome.TIMED_OUT, ( + f"the unit had nothing to announce and the watcher claimed " + f"{watch.outcome} anyway ({watch.disagreement})") + assert model_cache.needs_read("dirty") is True + assert model_cache.value("dirty", "is_dirty") is True, ( + "the re-read did not recover the unit's own answer") + else: + assert watch.outcome is WatchOutcome.CONFIRMED, ( + f"the unit answered {watch.outcome} ({watch.disagreement}). It sent " + f"{everything.tally()} during the window, so read that before " + f"blaming the unit for saying nothing.") + assert model_cache.needs_read("dirty") is False + + +# -- the stream that never stops ---------------------------------------------- + + +def test_the_metronome_stream_costs_the_cache_nothing(qc, model_cache, counted): + """The design's noisiest neighbour, measured rather than assumed. + + The metronome clock runs on every connection whether anybody asked for it or + not, so a cache that treated inbound messages as "something changed, go + re-read" would re-read for the length of the session. + """ + for entry in entries.ENTRIES: + model_cache.value(entry.name, sorted(entry.fields())[0]) + counted.reads.clear() + + stream = Pushes("GlobalTempoMessage") + qc.add_listener(stream) + try: + time.sleep(QUIET_WATCH) + finally: + qc.remove_listener(stream) + + assert len(stream.seen()) >= 2, ( + f"the unit pushed {len(stream.seen())} GlobalTempo message(s) in " + f"{QUIET_WATCH}s, so this test saw no stream to be quiet about") + marked = [entry.name for entry in entries.ENTRIES + if model_cache.needs_read(entry.name)] + assert not marked, f"the tempo stream marked {marked} for re-reading" + assert not counted.reads, f"the cache read {dict(counted.reads)} while idle" diff --git a/tests/test_device.py b/tests/test_device.py index eeb0fe5..1f8e784 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -13,13 +13,14 @@ class FakeClient: - """The two things the model asks of a protocol client, and no more. + """The four things the model asks of a protocol client, and no more. ``version()`` here decides the shape the model reads. The producer side of that agreement is pinned by ``test_the_real_client_returns_the_reply_shape_the_model_reads`` below, so this fake cannot drift away from what ``QuadCortex.version()`` actually - hands back. + hands back - and ``test_the_real_client_offers_what_the_model_subscribes_with`` + does the same job for the subscription. """ def __init__(self, firmware="d14e", serial="QCS0000001", omit=()): @@ -28,9 +29,13 @@ def __init__(self, firmware="d14e", serial="QCS0000001", omit=()): self._omit = set(omit) self.version_reads = 0 self.closed = False + self.listeners = [] def version(self, timeout=10.0): self.version_reads += 1 + return self.version_message() + + def version_message(self): reply = pa.VersionMessage(action=pa.MessageAction.UPDATE) if "app_fw_version" not in self._omit: reply.app_fw_version = self._firmware @@ -38,6 +43,22 @@ def version(self, timeout=10.0): reply.device_serial_number = self._serial return reply + def add_listener(self, listener): + self.listeners.append(listener) + return lambda: self.remove_listener(listener) + + def remove_listener(self, listener): + try: + self.listeners.remove(listener) + except ValueError: + return False + return True + + def push(self, message): + """The unit volunteering something, as the RX thread would deliver it.""" + for listener in list(self.listeners): + listener(message) + def close(self): self.closed = True @@ -51,13 +72,21 @@ class ReplyingTransport: def __init__(self, canned): self.canned = canned self.sent = [] + self.listeners = [] def send(self, msg): self.sent.append(msg) def request(self, msg, timeout=5.0): self.sent.append(msg) - return self.canned[type(msg).__name__] + reply = self.canned[type(msg).__name__] + for listener in list(self.listeners): + listener(reply) # listeners see a reply first (ADR-0009) + return reply + + def add_listener(self, listener): + self.listeners.append(listener) + return lambda: self.listeners.remove(listener) class FakeDevice: @@ -71,7 +100,11 @@ def close(self): class FakeTransport: - """Stands in for Transport: records lifecycle, swallows the handshake.""" + """Stands in for Transport: records lifecycle, swallows the handshake. + + ``happened`` is the running order of everything the connect path does to it, + which is what makes "before the handshake" checkable rather than assumed. + """ instances = [] @@ -79,20 +112,40 @@ def __init__(self, device, keepalive_interval=5.0): self.device = device self.started = False self.stopped = False + self.listeners = [] + self.happened = [] FakeTransport.instances.append(self) def start(self): self.started = True + self.happened.append("start") def stop(self, join_timeout=1.0): self.stopped = True def send(self, message): - pass + self.happened.append(f"send {type(message).__name__}") def request(self, message, timeout=None): + self.happened.append(f"request {type(message).__name__}") return message # the handshake only needs *a* reply + def add_listener(self, listener): + self.listeners.append(listener) + self.happened.append("add_listener") + return lambda: self.remove_listener(listener) + + def remove_listener(self, listener): + try: + self.listeners.remove(listener) + except ValueError: + return False + return True + + def push(self, message): + for listener in list(self.listeners): + listener(message) + @pytest.fixture def fake_stack(monkeypatch): @@ -235,8 +288,11 @@ def test_an_incomplete_version_reply_is_not_cached(): def test_a_closed_device_refuses_reads_it_could_have_served_from_cache(fake_stack): """The cache must not outlive the connection it was read over.""" device = pyquadcortex.connect(settle=0) - device._version = _fake_version_reply() # as if identity had been read + FakeTransport.instances[-1].push(_fake_version_reply()) # as the burst does + assert device.firmware == "d14e" + device.close() + for attribute in ("firmware", "serial", "client"): with pytest.raises(RuntimeError, match="closed"): getattr(device, attribute) @@ -270,6 +326,20 @@ def test_connect_hands_every_argument_to_the_protocol_layer(monkeypatch): seconds of patience instead of 30, inside a documented 9 to 17 second openable-but-silent window. """ + seen = _spy_on_protocol_connect(monkeypatch) + pyquadcortex.connect(timeout=1.5, settle=0.25, handshake_patience=45.0) + assert _without_the_hook(seen) == { + "timeout": 1.5, "settle": 0.25, "handshake_patience": 45.0} + + +def test_connect_passes_its_defaults_through_unchanged(monkeypatch): + seen = _spy_on_protocol_connect(monkeypatch) + pyquadcortex.connect() + assert _without_the_hook(seen) == { + "timeout": 5.0, "settle": 2.0, "handshake_patience": 30.0} + + +def _spy_on_protocol_connect(monkeypatch): seen = {} def spy(**kwargs): @@ -277,13 +347,90 @@ def spy(**kwargs): return FakeClient() monkeypatch.setattr(protocol, "connect", spy) - pyquadcortex.connect(timeout=1.5, settle=0.25, handshake_patience=45.0) - assert seen == {"timeout": 1.5, "settle": 0.25, "handshake_patience": 45.0} + return seen -def test_connect_passes_its_defaults_through_unchanged(monkeypatch): - seen = {} - monkeypatch.setattr(protocol, "connect", - lambda **kw: (seen.update(kw), FakeClient())[1]) - pyquadcortex.connect() - assert seen == {"timeout": 5.0, "settle": 2.0, "handshake_patience": 30.0} +def _without_the_hook(seen): + """The numeric arguments, with the subscription hook checked and removed. + + ``before_handshake`` gets its own tests below, where what matters is when it + runs rather than that it was passed. + """ + assert callable(seen.get("before_handshake")), ( + "the model did not subscribe before the handshake, so the connect " + "burst - nearly every state type the unit has - reaches nobody") + return {name: value for name, value in seen.items() + if name != "before_handshake"} + + +# -- the model is listening before the unit starts talking ------------------- + + +def test_the_model_subscribes_before_the_handshake_says_a_word(fake_stack): + """The burst is the only moment the unit volunteers nearly everything it + knows, and it does not start until seconds after `connect()` returns. A + model that subscribed on the client it is handed would miss all of it and + read every value back one at a time.""" + with pyquadcortex.connect(settle=0): + happened = FakeTransport.instances[-1].happened + + assert "add_listener" in happened, "the model never subscribed" + talking = [step for step in happened + if step.startswith("send") or step.startswith("request")] + assert talking, "the handshake sent nothing, so ordering proves nothing here" + assert happened.index("add_listener") < happened.index(talking[0]) + + +def test_a_push_the_burst_delivered_answers_the_first_read_for_free(fake_stack): + """The whole point of subscribing that early.""" + with pyquadcortex.connect(settle=0) as device: + FakeTransport.instances[-1].push(_fake_version_reply()) + before = len(FakeTransport.instances[-1].happened) + + assert device.firmware == "d14e" + assert device.serial == "QCS0000001" + + assert len(FakeTransport.instances[-1].happened) == before, ( + "the unit had already said so, and the model asked again") + + +def test_closing_the_device_stops_the_model_listening(fake_stack): + with pyquadcortex.connect(settle=0) as device: + transport = FakeTransport.instances[-1] + assert transport.listeners + assert transport.listeners == [] + + +def test_from_client_listens_on_the_connection_it_was_handed(fake_stack): + """The burst is long over by then, so this cache starts cold - but an edit + made from here on still reaches it without anybody asking.""" + qc = FakeClient() + device = Device.from_client(qc) + assert qc.listeners, "the model never subscribed to the connection" + + qc.push(qc.version_message()) + + assert device.firmware == "d14e" + assert qc.version_reads == 0 + + +def test_closing_a_borrowed_device_stops_it_listening(fake_stack): + """It does not own the connection, so it has to leave it as it found it.""" + qc = FakeClient() + Device.from_client(qc).close() + assert qc.listeners == [] + + +def test_the_real_client_offers_what_the_model_subscribes_with(fake_stack): + """Pins the producer, not just the fake. + + Every listening test here runs against ``FakeClient.add_listener``. Without + this, the real client could rename or lose the method and they would all + still pass. + """ + with protocol.connect(settle=0) as qc: + device = Device.from_client(qc) + transport = FakeTransport.instances[-1] + assert transport.listeners, "the model did not reach the transport" + device.close() + assert transport.listeners == [] diff --git a/tests/test_state.py b/tests/test_state.py new file mode 100644 index 0000000..80adf77 --- /dev/null +++ b/tests/test_state.py @@ -0,0 +1,874 @@ +"""The model's write-through cache: `docs/domain-model.md` sections 9 and 10. + +Everything here runs offline. The unit's side of the link is a +:class:`LoopbackTransport` that mirrors the two ``Transport`` guarantees the +cache is built on (ADR-0009): a listener sees every decoded message, and it sees +a reply BEFORE the thread that asked for it wakes up. Above it sits the REAL +``QuadCortex``, so ``client.version()`` and ``client.preset_dirty()`` are the +methods that run on hardware rather than stubs of them. + +``tests/test_state_rx.py`` covers the same cache on a real ``Transport`` and a +real RX thread, which is the only place the "never reads from the RX thread" +rule can actually be exercised. +""" +import collections +import logging +import threading +import time + +import pytest + +from pyquadcortex.device import entries, state +from pyquadcortex.device.watch import WatchOutcome +from pyquadcortex.protocol import client as protocol_client +from pyquadcortex.protocol.proto import ProductionAutomation_pb2 as pa + + +class LoopbackTransport: + """Canned replies, listeners notified first, every read counted. + + ``request`` answers from :attr:`replies`, keyed by the request's message + class name, and hands the reply to every listener BEFORE returning it - + which is the ordering the real transport guarantees and the ordering the + cache's read path depends on. + """ + + def __init__(self): + self.replies = {} + self.sent = [] + self.reads = collections.Counter() + self.listeners = [] + self._ids = iter(range(1, 1_000_000)) + + # -- the Transport surface the client and the model use ------------------- + + def add_listener(self, listener): + self.listeners.append(listener) + return lambda: self.remove_listener(listener) + + def remove_listener(self, listener): + try: + self.listeners.remove(listener) + except ValueError: + return False + return True + + def send(self, message): + self.sent.append(message) + + def next_request_id(self): + return next(self._ids) + + def request(self, message, timeout=5.0): + name = type(message).__name__ + self.sent.append(message) + self.reads[name] += 1 + try: + reply = self.replies[name] + except KeyError: # pragma: no cover - a test bug + raise AssertionError( + f"the test asked the unit for a {name} and set no reply for it") + if callable(reply): + reply = reply() + self.push(reply) # every listener sees it first... + return reply # ...and only then does the caller wake + + # -- the unit's side ------------------------------------------------------ + + def push(self, message): + """Deliver ``message`` to every listener, as the RX thread would.""" + for listener in list(self.listeners): + listener(message) + + +def version_reply(**fields): + """A ``VersionMessage`` the unit could have sent, carrying only ``fields``.""" + return pa.VersionMessage(action=pa.MessageAction.UPDATE, **fields) + + +def full_version_reply(): + return version_reply(app_fw_version="d14e", device_serial_number="QCS0000001") + + +def dirty_push(is_dirty): + return pa.PresetDirtyMessage(action=pa.MessageAction.UPDATE, is_dirty=is_dirty) + + +def tempo_pair(): + """One beat of the metronome stream: `GlobalTempo` arrives in pairs. + + The metronome clock always runs, so the unit pushes a pair per beat on + every connection whether or not anybody is listening - measured 1.5 s apart + at 40 bpm (``docs/domain-model.md`` section 9, smaller decision 7). + """ + beat = pa.GlobalTempoMessage(action=pa.MessageAction.UPDATE) + param = beat.params.add() + param.index = 0 + param.param_values.add().float_value = 0.4 # 120 bpm on the 40-240 scale + status = pa.GlobalTempoMessage(action=pa.MessageAction.UPDATE) + status.metronome_status.is_enabled = 1 + status.metronome_status.current_beat = 1 + return beat, status + + +def with_an_unknown_field(message, number=999, value=7): + """``message`` re-parsed with a field number the recovered schema lacks. + + Not hypothetical: `protocol/ProductionAutomation.proto` is recovered rather + than published (ADR-0010 says so in as many words), so a field the unit + really sends and our bindings have never heard of is the ordinary case, not + a future-firmware worry. + """ + tag = number << 3 # wire type 0, a varint + encoded = bytearray() + while tag > 0x7F: + encoded.append((tag & 0x7F) | 0x80) + tag >>= 7 + encoded.append(tag) + grown = type(message)() + grown.ParseFromString(message.SerializeToString() + bytes(encoded) + bytes([value])) + return grown + + +@pytest.fixture +def link(): + """A cache listening on a loopback link, over the real protocol client.""" + transport = LoopbackTransport() + transport.replies["VersionMessage"] = full_version_reply + transport.replies["PresetDirtyMessage"] = lambda: dirty_push(False) + qc = protocol_client.QuadCortex(transport) + cache = state.DeviceState() + cache.listen_on(transport) + cache.bind(qc) + try: + yield transport, cache + finally: + cache.close() + + +# -- pushes are data, not invalidation triggers ------------------------------- + + +def test_a_push_the_handshake_delivered_answers_the_first_read_for_free(link): + """The connect burst's whole value - section 9, smaller decision 1.""" + transport, cache = link + transport.push(dirty_push(True)) + + assert cache.value("dirty", "is_dirty") is True + assert transport.reads["PresetDirtyMessage"] == 0, ( + "the unit had already said so; asking again is the round trip the " + "cache exists to avoid") + + +def test_a_partial_push_merges_into_the_cache_rather_than_replacing_it(link): + """Section 9, smaller decision 2: an absent field means "not mentioned".""" + transport, cache = link + assert cache.value("identity", "device_serial_number") == "QCS0000001" + + transport.push(version_reply(app_fw_version="d15a")) + + assert cache.value("identity", "app_fw_version") == "d15a" + assert cache.value("identity", "device_serial_number") == "QCS0000001", ( + "the push did not mention the serial, which is not the same as the " + "unit reporting it empty") + assert transport.reads["VersionMessage"] == 1 + + +def test_a_field_the_wire_gives_no_presence_is_carried_by_every_push(link): + """`is_dirty` cannot be absent: proto3 gives it no presence, so False and + unset are the same bytes. The protocol layer's recorded evidence is that + absent IS false (``QuadCortex.preset_dirty``), so this is the one field the + cache reads without a presence check - declared, not assumed.""" + transport, cache = link + transport.push(dirty_push(True)) + assert cache.value("dirty", "is_dirty") is True + + transport.push(dirty_push(False)) + + assert cache.value("dirty", "is_dirty") is False, ( + "a clean save announces itself with a message that sets no field at " + "all; reading that as 'not mentioned' leaves the model stuck dirty") + assert transport.reads["PresetDirtyMessage"] == 0 + + +# -- a field we do not keep, checked per field -------------------------------- + + +def test_a_push_naming_a_field_the_model_does_not_keep_forces_a_reread(link): + """The failure this rule exists to catch: half a message applied.""" + transport, cache = link + assert cache.value("identity", "app_fw_version") == "d14e" + assert transport.reads["VersionMessage"] == 1 + + transport.push(version_reply(app_fw_version="d15a", + linux_kernel_version="5.10.0")) + + assert cache.needs_read("identity") is True + cache.value("identity", "app_fw_version") + assert transport.reads["VersionMessage"] == 2, ( + "the cache kept answering from a copy it had already been told was " + "incomplete") + + +def test_a_push_naming_a_field_the_schema_does_not_know_forces_a_reread(link): + """A recovered schema's own failure mode. The field is real on the unit and + absent from our bindings, so it decodes into nothing at all - the quietest + possible way to drop half a message.""" + transport, cache = link + assert cache.value("identity", "app_fw_version") == "d14e" + + transport.push(with_an_unknown_field(version_reply(app_fw_version="d15a"))) + + assert cache.needs_read("identity") is True + + +def test_the_half_of_the_push_we_do_understand_is_still_applied(link): + """Marking for re-read and applying what we read are not alternatives. + + If the kept half were dropped, the answer between the push and the next + read would be the OLD value - confidently wrong, just for a shorter while. + """ + transport, cache = link + assert cache.value("identity", "app_fw_version") == "d14e" + + transport.push(version_reply(app_fw_version="d15a", + linux_kernel_version="5.10.0")) + + assert cache.cached("identity")["app_fw_version"] == "d15a" + + +def test_it_marks_only_the_part_of_the_cache_the_push_named(link): + """"Exactly that part" - a Version surprise says nothing about the preset.""" + transport, cache = link + transport.push(dirty_push(True)) + cache.value("identity", "app_fw_version") + + transport.push(version_reply(linux_kernel_version="5.10.0")) + + assert cache.needs_read("identity") is True + assert cache.needs_read("dirty") is False + assert cache.value("dirty", "is_dirty") is True + assert transport.reads["PresetDirtyMessage"] == 0 + + +def test_the_forced_reread_names_the_field_that_forced_it(caplog, link): + """Section 10's standard for a log line: a bug with a name and a location. + The event name and the field are what issue #16's counters read.""" + transport, cache = link + cache.value("identity", "app_fw_version") + + with caplog.at_level(logging.INFO, logger="pyquadcortex.device.state"): + transport.push(version_reply(uboot_version="2019.04")) + + assert any("push.forced_reread" in r.message and "uboot_version" in r.message + for r in caplog.records), caplog.text + + +def test_one_reread_is_enough_and_the_cache_is_trusted_again(link): + """Section 9: "we discard our copy and read a fresh one. Slower, but right." + + Once, not on every access. The read's own answer carries the same fields we + do not keep, so an entry that re-armed the mark from its own reply would + never cache anything again. + """ + transport, cache = link + transport.replies["VersionMessage"] = lambda: version_reply( + app_fw_version="d14e", device_serial_number="QCS0000001", + uboot_version="2019.04") + transport.push(version_reply(uboot_version="2019.04")) + + cache.value("identity", "app_fw_version") + cache.value("identity", "app_fw_version") + cache.value("identity", "device_serial_number") + + assert transport.reads["VersionMessage"] == 1 + assert cache.needs_read("identity") is False + + +def test_a_push_that_lands_during_a_proactive_read_is_not_lost(link): + """The window the read path has to close. + + A read replaces our copy with an answer the unit composed before the push + arrived. Clearing the mark unconditionally would drop that push with + nothing left to recover it from. + """ + transport, cache = link + + def reply_but_a_push_first(): + transport.push(version_reply(app_fw_version="d15a")) + return full_version_reply() + + transport.replies["VersionMessage"] = reply_but_a_push_first + cache.value("identity", "app_fw_version") + + assert cache.needs_read("identity") is True + + +# -- the tempo stream --------------------------------------------------------- + + +def test_the_metronome_stream_causes_no_reads_and_no_churn(link): + """Section 9, smaller decision 7. At 40 bpm the unit pushes a pair every + 1.5 s for the life of every connection. An invalidation-based cache would + spend its life re-reading.""" + transport, cache = link + transport.push(dirty_push(True)) + assert cache.value("identity", "app_fw_version") == "d14e" + reads_before = dict(transport.reads) + + for _ in range(40): # a minute of beats at 40 bpm + for message in tempo_pair(): + transport.push(message) + + assert dict(transport.reads) == reads_before + assert cache.needs_read("identity") is False + assert cache.needs_read("dirty") is False + assert cache.value("dirty", "is_dirty") is True + assert cache.value("identity", "app_fw_version") == "d14e" + assert dict(transport.reads) == reads_before + + +def test_the_stream_above_really_reaches_the_cache(link): + """Guards the test above, which eighty pushes into a void would also pass. + + Same transport, same listeners, one message the cache does track: if the + delivery path were broken this fails and the churn test stops meaning + anything. + """ + transport, cache = link + for message in tempo_pair(): + transport.push(message) + transport.push(dirty_push(True)) + + assert cache.value("dirty", "is_dirty") is True + assert transport.reads["PresetDirtyMessage"] == 0 + + +def test_a_message_type_no_entry_tracks_is_ignored_outright(link): + """Section 9: "A message of a type we know nothing about is ignored + outright, which is what the RX thread already does." """ + transport, cache = link + cache.value("identity", "app_fw_version") + transport.push(dirty_push(True)) + + transport.push(pa.IOMeterMessage(action=pa.MessageAction.UPDATE)) + transport.push(pa.CPULoadMessage(action=pa.MessageAction.UPDATE)) + + assert cache.needs_read("identity") is False + assert cache.needs_read("dirty") is False + + +# -- state the unit does not volunteer --------------------------------------- + + +def test_state_the_unit_never_broadcasts_is_read_on_first_access(link): + """No model property ships with a staleness caveat, so the fallback is a + read rather than a shrug. Version is that case: the unit answers a READ and + never announces its own firmware.""" + transport, cache = link + assert transport.reads["VersionMessage"] == 0 + + assert cache.value("identity", "app_fw_version") == "d14e" + + assert transport.reads["VersionMessage"] == 1 + + +def test_a_second_access_of_the_same_entry_costs_no_round_trip(link): + transport, cache = link + cache.value("identity", "app_fw_version") + cache.value("identity", "device_serial_number") + cache.value("identity", "app_fw_version") + assert transport.reads["VersionMessage"] == 1 + + +def test_a_read_replaces_the_entry_rather_than_merging_into_it(link): + """A read is the unit's whole answer, so a field it does not carry is a + field the unit did not confirm. Leaving the old value in place would report + something no read has returned.""" + transport, cache = link + assert cache.value("identity", "device_serial_number") == "QCS0000001" + cache.mark_for_reread("identity", "this test") + transport.replies["VersionMessage"] = lambda: version_reply( + app_fw_version="d15a") + + assert cache.value("identity", "app_fw_version") == "d15a" + with pytest.raises(RuntimeError, match="device_serial_number"): + cache.value("identity", "device_serial_number") + + +def test_a_field_the_unit_did_not_send_is_refused_not_reported_empty(link): + """An absent string decodes as "", and reporting that is the guess this + whole layer exists to avoid.""" + transport, cache = link + transport.replies["VersionMessage"] = lambda: version_reply( + app_fw_version="d14e") + with pytest.raises(RuntimeError, match="device_serial_number"): + cache.value("identity", "device_serial_number") + + +def test_an_incomplete_answer_leaves_a_retry_able_to_recover(link): + transport, cache = link + transport.replies["VersionMessage"] = lambda: version_reply( + app_fw_version="d14e") + with pytest.raises(RuntimeError): + cache.value("identity", "device_serial_number") + + transport.replies["VersionMessage"] = full_version_reply + assert cache.value("identity", "device_serial_number") == "QCS0000001" + assert transport.reads["VersionMessage"] == 2 + + +def test_the_field_the_unit_did_send_is_still_answered_from_the_cache(link): + """Per field, here too: a reply that carried the firmware and not the serial + told us the firmware, and a retry is only owed for the half that is missing. + """ + transport, cache = link + transport.replies["VersionMessage"] = lambda: version_reply( + app_fw_version="d14e") + with pytest.raises(RuntimeError): + cache.value("identity", "device_serial_number") + + assert cache.value("identity", "app_fw_version") == "d14e" + assert transport.reads["VersionMessage"] == 1 + + +def test_a_read_before_the_cache_is_bound_to_a_connection_is_refused(link): + transport, _ = link + unbound = state.DeviceState() + with pytest.raises(RuntimeError, match="not connected"): + unbound.value("identity", "app_fw_version") + + +def test_a_field_no_entry_keeps_is_a_programming_error_not_a_read(link): + transport, cache = link + with pytest.raises(KeyError, match="power_option"): + cache.value("identity", "power_option") + assert transport.reads["VersionMessage"] == 0 + + +# -- what the entries declare ------------------------------------------------ + + +@pytest.mark.parametrize("entry", entries.ENTRIES, ids=lambda e: e.name) +def test_a_field_declared_presence_free_really_has_none(entry): + """The one exception to the presence rule, checked against the schema. + + A field listed here is read with no presence check, so if the schema gives + it presence the declaration downgrades a checkable answer to an unchecked + one - which is the guess the rule forbids. + """ + for message_class, plan in entry.feeds.items(): + for name in plan.no_presence: + field = message_class.DESCRIPTOR.fields_by_name[name] + assert not field.has_presence, ( + f"{message_class.__name__}.{name} does have presence - keep it " + f"in `kept` and let the presence check do its job") + + +@pytest.mark.parametrize("entry", entries.ENTRIES, ids=lambda e: e.name) +def test_a_kept_field_is_one_the_wire_can_report_absent(entry): + for message_class, plan in entry.feeds.items(): + for name in plan.kept: + field = message_class.DESCRIPTOR.fields_by_name[name] + assert field.has_presence, ( + f"{message_class.__name__}.{name} has no presence, so an unset " + f"message reports its default as an answer - declare it in " + f"`no_presence` with the evidence for what absent means") + + +@pytest.mark.parametrize("entry", entries.ENTRIES, ids=lambda e: e.name) +def test_no_field_an_entry_leaves_unkept_is_invisible_on_the_wire(entry): + """The one blind spot the per-field check genuinely has, held shut. + + A proto3 scalar with no presence is written only when it differs from its + default, so a message that leaves one at its default carries no bytes for it + at all - ``PresetDirty{is_dirty: False}`` serialises to two bytes, both of + them ``action``. Nothing can see a change to such a field, because there is + nothing to see: not ``ListFields``, not the unknown-field weigh-in, not a + hand-written parser. + + So a presence-free field has to be KEPT or it can never be noticed, and + "does the model keep it?" is a question about our code rather than about the + wire - which makes it checkable, which is this test. It fires the day an + entry is fed by a type carrying one it does not keep, and the answer then is + to keep it with the evidence for what its default means, exactly as + ``is_dirty`` is kept. + """ + for message_class, plan in entry.feeds.items(): + declared = plan.kept | plan.no_presence | entries.SCAFFOLDING + invisible = sorted(field.name for field in message_class.DESCRIPTOR.fields + if not field.has_presence and field.name not in declared) + assert not invisible, ( + f"{entry.name} does not keep {message_class.__name__}.{invisible}, " + f"which the wire cannot report as absent - so a change to it is " + f"undetectable rather than merely unkept") + + +@pytest.mark.parametrize("entry", entries.ENTRIES, ids=lambda e: e.name) +def test_every_field_an_entry_keeps_is_a_plain_value(entry): + """The cache stores what ``getattr`` hands back, and for a message or a + repeated field that is a live container INSIDE the message the RX thread + just decoded - shared with every other listener and read afterwards from + other threads. + + Scalars are copied by value, so today there is nothing to share. This fires + the first time an entry keeps something composite, which is when that has to + be answered rather than assumed. + """ + for message_class, plan in entry.feeds.items(): + for name in sorted(plan.kept | plan.no_presence): + field = message_class.DESCRIPTOR.fields_by_name[name] + assert not field.is_repeated, ( + f"{message_class.__name__}.{name} is repeated, so the cache " + f"would hold a live container from a message the RX thread owns") + assert field.type not in (field.TYPE_MESSAGE, field.TYPE_GROUP), ( + f"{message_class.__name__}.{name} is a submessage, so the cache " + f"would hold a live container from a message the RX thread owns") + + +@pytest.mark.parametrize("entry", entries.ENTRIES, ids=lambda e: e.name) +def test_every_field_an_entry_names_exists_on_the_message_that_feeds_it(entry): + """A misspelled field name is a field silently never applied.""" + for message_class, plan in entry.feeds.items(): + known = {f.name for f in message_class.DESCRIPTOR.fields} + missing = sorted((plan.kept | plan.no_presence) - known) + assert not missing, ( + f"{entry.name} keeps {missing}, which {message_class.__name__} " + f"does not have") + + +@pytest.mark.parametrize("entry", entries.ENTRIES, ids=lambda e: e.name) +def test_the_scaffolding_the_check_ignores_is_on_every_feeding_type(entry): + """`action` and `request_id` are the transport's, not the unit's state, so + the check skips them. If a feeding type lacked one, the skip would be + forgiving a field that never arrives - and it would hide the day a message + type starts carrying state in a field with one of those names.""" + for message_class in entry.feeds: + known = {f.name for f in message_class.DESCRIPTOR.fields} + assert entries.SCAFFOLDING <= known, ( + f"{message_class.__name__} lacks " + f"{sorted(entries.SCAFFOLDING - known)}") + + +def test_every_entry_answers_a_read_for_every_field_it_keeps(link): + """An entry the model cannot read is one it can only guess about.""" + transport, cache = link + for entry in entries.ENTRIES: + cache.mark_for_reread(entry.name, "this test") + for field in entry.fields(): + cache.value(entry.name, field) # raises if the read cannot serve it + + +# -- a closed connection answers nothing ------------------------------------- + + +def test_a_closed_cache_refuses_a_read_it_could_have_served(link): + """Anything the model caches is valid only while its connection is.""" + transport, cache = link + cache.value("identity", "app_fw_version") + cache.close() + with pytest.raises(RuntimeError, match="closed"): + cache.value("identity", "app_fw_version") + + +def test_a_closed_cache_stops_listening(link): + transport, cache = link + cache.close() + transport.push(dirty_push(True)) + assert transport.listeners == [] + + +def test_a_closed_cache_will_not_say_what_it_remembers(link): + """`cached()` is a read too, and a closed one answering `{}` is not a + refusal - it reads as "the unit told us nothing".""" + transport, cache = link + transport.push(dirty_push(True)) + assert cache.cached("dirty") == {"is_dirty": True} + cache.close() + with pytest.raises(RuntimeError, match="closed"): + cache.cached("dirty") + + +def test_a_closed_cache_will_not_say_what_it_would_do_next(link): + """`needs_read` flipped from True to False across `close()`, which is the + answer "the next read is free" about a cache whose next read raises.""" + transport, cache = link + cache.mark_for_reread("dirty", "this test") + assert cache.needs_read("dirty") is True + cache.close() + with pytest.raises(RuntimeError, match="closed"): + cache.needs_read("dirty") + + +def test_closing_twice_is_harmless(link): + transport, cache = link + cache.close() + cache.close() + + +# -- writes ------------------------------------------------------------------ + + +def test_a_write_updates_the_cache_before_any_echo_arrives(link): + """Section 9, rule 3. Waiting for the echo would make every write pay for + information we already have.""" + transport, cache = link + transport.push(dirty_push(False)) + + cache.write_through("dirty", {"is_dirty": True}, send=lambda: None) + + assert cache.value("dirty", "is_dirty") is True + assert transport.reads["PresetDirtyMessage"] == 0 + + +def test_the_write_reaches_the_unit(link): + transport, cache = link + cache.write_through("dirty", {"is_dirty": True}, + send=lambda: transport.send(dirty_push(True))) + assert [type(m).__name__ for m in transport.sent] == ["PresetDirtyMessage"] + + +def test_an_echo_carrying_every_field_we_sent_confirms_the_write(link): + transport, cache = link + watch = cache.write_through("dirty", {"is_dirty": True}, send=lambda: None) + + transport.push(dirty_push(True)) + + assert watch.outcome is WatchOutcome.CONFIRMED + + +def test_an_echo_carrying_the_units_own_request_id_still_confirms(link): + """Narrow on purpose, and worth saying what it does NOT pin. + + The scaffolding fields are stripped before the watcher sees the echo, so + this proves the stripping and nothing about the "every field we sent" rule - + an implementation demanding the whole echo equal what we sent passes it. + That rule needs an echo carrying a KEPT field we did not send, which no + entry here is wide enough to produce; ``tests/test_watch.py`` is where it + lives, and that is the file's stated reason for existing. + """ + transport, cache = link + watch = cache.write_through("dirty", {"is_dirty": True}, send=lambda: None) + + echo = dirty_push(True) + echo.request_id = 41 # the unit's own, not ours + transport.push(echo) + + assert watch.outcome is WatchOutcome.CONFIRMED + + +def test_an_echo_returning_another_value_for_a_field_we_sent_is_reported(link): + """A bug in our code, now with a name and a location.""" + transport, cache = link + watch = cache.write_through("dirty", {"is_dirty": True}, send=lambda: None) + + transport.push(dirty_push(False)) + + assert watch.outcome is WatchOutcome.DIFFERENT + assert watch.disagreement == ("is_dirty", True, False) + + +def test_the_unit_winning_a_disagreement_leaves_the_units_value_cached(link): + """Applying the whole echo is what handles section 10's four legitimate + cases for free, so a write the unit overrode must not be left behind.""" + transport, cache = link + cache.write_through("dirty", {"is_dirty": True}, send=lambda: None) + + transport.push(dirty_push(False)) + + assert cache.cached("dirty")["is_dirty"] is False + + +def test_a_disagreement_forces_a_reread_of_the_entry(link): + """A write the unit contradicted is a write we do not understand, so the + rest of what it claimed is not to be believed either.""" + transport, cache = link + cache.write_through("dirty", {"is_dirty": True}, send=lambda: None) + + transport.push(dirty_push(False)) + + assert cache.needs_read("dirty") is True + + +def test_a_disagreement_does_not_leave_an_unconfirmed_field_behind(link): + """The case the mark is really for. + + A write of two fields, an echo carrying only one of them, and that one + disagreeing. The other is a value we put in the cache ourselves, that the + unit never confirmed, in a write the unit has just demonstrated it disagreed + with - the "confidently wrong" state, reached down the one path that used to + clear up after itself least. + + ``identity`` is used because it is the only entry today wide enough to have + a second field; the rule is about the mechanism, not about that entry, and + nothing in the model writes firmware. + """ + transport, cache = link + cache.write_through("identity", + {"app_fw_version": "MINE", "device_serial_number": "MINE"}, + send=lambda: None) + + transport.push(version_reply(app_fw_version="THEIRS")) + + assert cache.cached("identity")["device_serial_number"] == "MINE" + assert cache.needs_read("identity") is True + assert cache.value("identity", "device_serial_number") == "QCS0000001", ( + "the unconfirmed field was never re-read, so the model kept answering " + "with a value it made up") + + +def test_an_echo_that_never_comes_times_out_and_forces_a_reread(link): + """A silently ignored write self-corrects instead of poisoning the cache.""" + transport, cache = link + watch = cache.write_through("dirty", {"is_dirty": True}, send=lambda: None, + patience=0.05) + + assert watch.settled(timeout=5.0) + assert watch.outcome is WatchOutcome.TIMED_OUT + assert cache.needs_read("dirty") is True + + +def test_the_watcher_does_not_block_the_write(link): + transport, cache = link + started = time.monotonic() + cache.write_through("dirty", {"is_dirty": True}, send=lambda: None, + patience=30.0) + assert time.monotonic() - started < 1.0 + + +def test_a_confirmed_write_does_not_force_a_reread(link): + transport, cache = link + transport.push(dirty_push(False)) + cache.write_through("dirty", {"is_dirty": True}, send=lambda: None, + patience=0.05) + transport.push(dirty_push(True)) + + time.sleep(0.2) # past the patience it was given + assert cache.needs_read("dirty") is False + + +def test_a_write_whose_send_fails_is_taken_back_out_of_the_cache(link): + """The unit never heard it, so our copy would be the only place it exists.""" + transport, cache = link + transport.push(dirty_push(False)) + + def send(): + raise TimeoutError("the unit did not take it") + + with pytest.raises(TimeoutError): + cache.write_through("dirty", {"is_dirty": True}, send=send) + + assert cache.needs_read("dirty") is True + + +def test_a_failed_send_leaves_no_watcher_to_time_out_later(link): + """The entry recovers on the next read and stays recovered. + + A watcher left behind for a write the unit never received would fire at its + deadline and mark the entry again, so a caller who had already put it right + would find it wrong once more for no reason. + """ + transport, cache = link + + def send(): + raise TimeoutError("the unit did not take it") + + with pytest.raises(TimeoutError): + cache.write_through("dirty", {"is_dirty": True}, send=send, patience=0.05) + assert cache.value("dirty", "is_dirty") is False # the read clears the mark + + time.sleep(0.25) # well past that patience + assert cache.needs_read("dirty") is False + + +def test_a_write_to_a_field_the_entry_does_not_keep_is_refused(link): + """A write the cache cannot hold would be applied nowhere and confirmed + against nothing.""" + transport, cache = link + with pytest.raises(ValueError, match="power_option"): + cache.write_through("dirty", {"power_option": 1}, send=lambda: None) + + +def test_a_write_through_a_closed_cache_is_refused(link): + transport, cache = link + cache.close() + with pytest.raises(RuntimeError, match="closed"): + cache.write_through("dirty", {"is_dirty": True}, send=lambda: None) + + +def test_the_watchdog_does_not_outlive_the_connection(link): + transport, cache = link + cache.write_through("dirty", {"is_dirty": True}, send=lambda: None, + patience=30.0) + assert _watchdog_threads(), "no watchdog was started, so this proves nothing" + + cache.close() + + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + if not _watchdog_threads(): + return + time.sleep(0.02) + pytest.fail("the write watchdog is still running after close()") + + +def test_a_write_still_in_flight_when_the_connection_closes_does_not_hang(link): + """Nothing can settle it once the connection is gone, so nobody may wait. + + ``settled()`` with its documented default waits forever, and the watchdog + deliberately does NOT call a closed connection's outstanding writes timed + out - that would be a claim about the unit rather than a fact. So the + waiting has to end without an outcome. + """ + transport, cache = link + watch = cache.write_through("dirty", {"is_dirty": True}, send=lambda: None, + patience=30.0) + + cache.close() + + started = time.monotonic() + assert watch.settled(timeout=5.0) is False + assert time.monotonic() - started < 1.0, ( + "it waited out the timeout it was given, which with settled()'s " + "documented default of None is forever") + assert watch.outcome is None + + +def test_a_write_that_races_the_close_leaves_no_thread_behind(link): + """The window between `write_through` checking the cache is open and the + watchdog starting its thread. A thread started there waits forever on a + connection nobody can reach.""" + transport, cache = link + cache.close() + + cache._watchdog.add(_a_watch_for_the_race()) + + assert not _watchdog_threads() + + +def _a_watch_for_the_race(): + from pyquadcortex.device.watch import WriteWatch + return WriteWatch("dirty", {"is_dirty": True}, time.monotonic() + 30.0) + + +def test_the_watchdog_firing_as_the_connection_closes_does_not_raise(link): + """The watchdog is inside its callback when `close()` lands. Marking a slot + that is gone would raise on a thread with nobody to catch it.""" + transport, cache = link + watch = cache.write_through("dirty", {"is_dirty": True}, send=lambda: None, + patience=30.0) + cache.close() + + cache._gave_up_on(watch) # no exception is the assertion + + +def test_no_watchdog_runs_until_something_is_written(link): + transport, cache = link + cache.value("identity", "app_fw_version") + transport.push(dirty_push(True)) + assert not _watchdog_threads() + + +def _watchdog_threads(): + return [t for t in threading.enumerate() + if t.name.startswith(state.WATCHDOG_THREAD_NAME)] diff --git a/tests/test_state_rx.py b/tests/test_state_rx.py new file mode 100644 index 0000000..4a4dca2 --- /dev/null +++ b/tests/test_state_rx.py @@ -0,0 +1,210 @@ +"""The cache on a real RX thread, over a real ``Transport`` and a fake HID link. + +``tests/test_state.py`` drives the cache through a loopback double, which is +where the merge rules belong: they are about message content, and a double keeps +them readable. This file exists for the rules a double cannot test at all, +because they are about a THREAD: + + The RX thread applies pushes and notes what needs re-reading; the caller's + thread does the re-reading. The RX thread still cannot block or die. + +A cache that re-read from its listener would pass every content test and would +stall the whole connection on hardware for the length of one timeout - a failure +that looks like success, which is the expensive kind (ADR-0009). + +No hardware and no ``hid``: the fake below is the unit's side of the HID link, +built with the real ``framing`` so the transport's reassembly path runs for real. +""" +import collections +import threading +import time + +import pytest + +from pyquadcortex.device import state +from pyquadcortex.protocol import client as protocol_client +from pyquadcortex.protocol import framing, registry, transport +from pyquadcortex.protocol.proto import ProductionAutomation_pb2 as pa + +#: High enough that the transport's keepalive never lands mid-test. +QUIET_KEEPALIVE = 3600.0 +PATIENCE = 2.0 + + +class FakeUnit: + """The unit's side of the link: answers a Version READ, pushes on demand. + + Only Version is answered, because it is the only thing the cache reads over + this link. Anything else the host writes is counted and dropped, which is + what makes ``asked_for`` a usable assertion about reads the model issued. + """ + + def __init__(self): + self._inbox = collections.deque() + self._lock = threading.Lock() + self.asked_for = collections.Counter() + + def write(self, report): + report = bytes(report) + message_type, payload = framing.decode_reports([report]) + message_class = registry.class_for(message_type) + message = message_class() + message.ParseFromString(payload) + with self._lock: + self.asked_for[message_class.__name__] += 1 + if message_class is pa.VersionMessage: + reply = pa.VersionMessage(action=pa.MessageAction.UPDATE, + app_fw_version="d14e", + device_serial_number="QCS0000001") + if message.HasField("request_id"): + reply.request_id = message.request_id + self.push(reply) + return len(report) + + def push(self, message): + """Queue ``message`` for the transport's read loop to pick up.""" + reports = framing.encode_message(registry.type_for(type(message)), + message.SerializeToString()) + with self._lock: + for report in reports: + # encode_message stamps the host->device report id; real input + # reports carry the device->host one. Same restamp as + # tests/test_transport.py's FakeHid. + self._inbox.append(bytes([framing.IN_REPORT_ID]) + + bytes(report)[1:]) + + def read(self, size, timeout=0): + with self._lock: + if self._inbox: + return list(self._inbox.popleft()) + time.sleep(0.005) # a blocking read that times out, unspun + return [] + + def close(self): + pass + + +def _wait_until(predicate, timeout=PATIENCE): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.005) + return predicate() + + +@pytest.fixture +def live(): + """A cache listening on a started transport, over the real client.""" + unit = FakeUnit() + link = transport.Transport(unit, keepalive_interval=QUIET_KEEPALIVE) + link.start() + cache = state.DeviceState() + cache.listen_on(link) + cache.bind(protocol_client.QuadCortex(link)) + try: + yield unit, link, cache + finally: + cache.close() + link.stop() + + +def test_the_rx_thread_applies_a_push_and_asks_the_unit_for_nothing(live): + unit, link, cache = live + + unit.push(pa.PresetDirtyMessage(action=pa.MessageAction.UPDATE, + is_dirty=True)) + + assert _wait_until(lambda: cache.cached("dirty").get("is_dirty") is True) + assert unit.asked_for["PresetDirtyMessage"] == 0 + assert cache.value("dirty", "is_dirty") is True + assert unit.asked_for["PresetDirtyMessage"] == 0 + + +def test_a_push_that_forces_a_reread_does_not_read_on_the_rx_thread(live): + """The rule, stated as a measurement: the mark appears, the read does not.""" + unit, link, cache = live + assert cache.value("identity", "app_fw_version") == "d14e" + assert unit.asked_for["VersionMessage"] == 1 + + unit.push(pa.VersionMessage(action=pa.MessageAction.UPDATE, + uboot_version="2019.04")) + + assert _wait_until(lambda: cache.needs_read("identity")) + time.sleep(0.2) # long enough for a listener to have read + assert unit.asked_for["VersionMessage"] == 1, ( + "the re-read was issued from the RX thread - on hardware that stalls " + "the read loop for its whole timeout and can never be satisfied") + + assert cache.value("identity", "app_fw_version") == "d14e" + assert unit.asked_for["VersionMessage"] == 2, "the caller's thread re-reads" + + +def test_the_read_loop_still_runs_after_a_push_that_forced_a_reread(live): + """"The RX thread never dies" is absolute, marks included.""" + unit, link, cache = live + unit.push(pa.VersionMessage(action=pa.MessageAction.UPDATE, + uboot_version="2019.04")) + assert _wait_until(lambda: cache.needs_read("identity")) + + assert cache.value("identity", "device_serial_number") == "QCS0000001" + assert link.device_lost is None + + +def test_the_read_loop_survives_a_push_the_cache_chokes_on(live, monkeypatch): + """A bug in the cache costs one message, not the connection. + + Forced rather than waited for: an exception from the listener is the case + the RX path's guarantee is written for, and there is no way to provoke it + from message content without a bug to provoke it with. + """ + unit, link, cache = live + + def explode(*args, **kwargs): + raise ValueError("a bug in the cache") + + monkeypatch.setattr(state, "fields_applied", explode) + unit.push(pa.PresetDirtyMessage(action=pa.MessageAction.UPDATE, + is_dirty=True)) + time.sleep(0.1) + + monkeypatch.undo() + assert cache.value("identity", "app_fw_version") == "d14e" + assert link.device_lost is None + + +def test_reading_the_cache_from_a_listener_is_refused_not_hung(live): + """The other half of ADR-0009's bargain, on the cache's own read path. + + A future listener that wants a value it did not receive marks it for + re-reading; it does not fetch it. This is what happens if it tries. + """ + unit, link, cache = live + tried = threading.Event() + refusal = {} + + def reads_from_the_rx_thread(message): + if tried.is_set(): + return + tried.set() + try: + cache.value("identity", "app_fw_version") + refusal["error"] = None + except BaseException as exc: # noqa: BLE001 - the type is the point + refusal["error"] = exc + + detach = link.add_listener(reads_from_the_rx_thread) + try: + unit.push(pa.PresetDirtyMessage(action=pa.MessageAction.UPDATE, + is_dirty=True)) + assert tried.wait(PATIENCE), "no push arrived, so the listener never ran" + assert _wait_until(lambda: "error" in refusal) + finally: + detach() + + error = refusal["error"] + assert isinstance(error, RuntimeError), f"not refused: {error!r}" + assert not isinstance(error, TimeoutError), "it waited instead of refusing" + assert "RX thread" in str(error) + # And the link is unharmed: the refusal cost one listener call. + assert cache.value("identity", "app_fw_version") == "d14e" diff --git a/tests/test_watch.py b/tests/test_watch.py new file mode 100644 index 0000000..538c1ed --- /dev/null +++ b/tests/test_watch.py @@ -0,0 +1,253 @@ +"""The write watcher's one sentence: `docs/domain-model.md` section 10. + + Every field we sent must come back with the value we sent. + +``tests/test_state.py`` drives the watcher through the cache, which is how it is +really used - but the cache's entries hold one or two fields, and the rule that +matters is about the fields we did NOT send. So the interesting cases are here, +against :class:`WriteWatch` itself. + +Why they matter: "the echo equals what we sent" would pass every test the cache +can currently reach, and would cry wolf constantly on hardware, because the unit +legitimately changes things nobody asked about. Section 10 lists four - a +gain-reduction meter, a mirrored parameter, NaN in unused slots, dropdown values +recomputed on rows we never touched - and every one of them is a field we did +not send. +""" +import threading +import time + +import pytest + +from pyquadcortex.device.watch import WatchOutcome, Watchdog, WriteWatch + + +def a_watch(sent, patience=30.0): + return WriteWatch("test entry", sent, time.monotonic() + patience) + + +# -- what the unit also changed ---------------------------------------------- + + +def test_an_echo_that_also_carries_something_we_did_not_send_confirms(): + """The mirrored parameter, the recomputed dropdown, the live meter.""" + watch = a_watch({"level": 0.5}) + + assert watch.absorb({"level": 0.5, "gain_reduction": -3.2}) \ + is WatchOutcome.CONFIRMED + + +def test_a_field_we_did_not_send_cannot_make_a_write_look_wrong(): + watch = a_watch({"level": 0.5}) + + watch.absorb({"level": 0.5, "mirrored": 999.0}) + + assert watch.disagreement is None + + +# -- every field we sent ------------------------------------------------------ + + +def test_an_echo_carrying_only_some_of_what_we_sent_is_not_a_confirmation(): + """Not yet confirmed, and not yet wrong. The echo is a sparse delta, so a + second one can still carry the rest.""" + watch = a_watch({"level": 0.5, "mix": 0.25}) + + assert watch.absorb({"level": 0.5}) is None + assert watch.outcome is None + + +def test_the_confirmation_completes_across_two_echoes(): + watch = a_watch({"level": 0.5, "mix": 0.25}) + watch.absorb({"level": 0.5}) + + assert watch.absorb({"mix": 0.25}) is WatchOutcome.CONFIRMED + + +def test_one_wrong_field_is_a_disagreement_even_when_the_others_match(): + watch = a_watch({"level": 0.5, "mix": 0.25}) + + assert watch.absorb({"level": 0.5, "mix": 0.9}) is WatchOutcome.DIFFERENT + assert watch.disagreement == ("mix", 0.25, 0.9) + + +def test_a_disagreement_names_the_field_the_value_sent_and_the_value_returned(): + """Section 10: a bug in our code, now with a name and a location. A watcher + that only said "it did not stick" would leave the reader no better off.""" + watch = a_watch({"level": 0.5}) + watch.absorb({"level": 0.0}) + + field, sent, returned = watch.disagreement + assert (field, sent, returned) == ("level", 0.5, 0.0) + + +def test_an_echo_that_mentions_nothing_we_sent_settles_nothing(): + watch = a_watch({"level": 0.5}) + + assert watch.absorb({"something_else": 1}) is None + assert watch.outcome is None + + +# -- a settled write stays settled ------------------------------------------- + + +def test_a_later_echo_cannot_overturn_a_confirmation(): + """The unit keeps talking after a write lands. A watcher that kept reading + would report a disagreement about a value somebody changed afterwards.""" + watch = a_watch({"level": 0.5}) + watch.absorb({"level": 0.5}) + + assert watch.absorb({"level": 0.9}) is None + assert watch.outcome is WatchOutcome.CONFIRMED + + +def test_a_write_the_unit_confirmed_cannot_then_time_out(): + watch = a_watch({"level": 0.5}) + watch.absorb({"level": 0.5}) + + assert watch.time_out() is False + assert watch.outcome is WatchOutcome.CONFIRMED + + +def test_a_write_that_timed_out_cannot_then_be_confirmed(): + """The race the watchdog runs into on every write confirmed near its + deadline; whichever gets there first is the answer.""" + watch = a_watch({"level": 0.5}) + assert watch.time_out() is True + + assert watch.absorb({"level": 0.5}) is None + assert watch.outcome is WatchOutcome.TIMED_OUT + + +def test_a_write_with_no_fields_is_refused(): + """It would confirm on the first echo of anything at all, having checked + nothing.""" + with pytest.raises(ValueError): + a_watch({}) + + +def test_a_waiter_is_released_by_publish_and_not_by_the_outcome(): + """`settled()` returning has to mean the outcome's consequences have already + happened - the entry marked, the line logged - so what wakes a waiter is the + separate publish, not the moment the outcome is decided. Otherwise a caller + woken by a timeout races the thread that is still marking the entry.""" + watch = a_watch({"level": 0.5}) + released = threading.Event() + threading.Thread(target=lambda: (watch.settled(timeout=5.0), released.set()), + daemon=True).start() + + watch.absorb({"level": 0.5}) + + assert not released.wait(0.2), "the waiter woke on the outcome alone" + watch.publish() + assert released.wait(2.0), "publish did not release the waiter" + + +def test_a_watch_released_without_an_outcome_reports_that_it_has_none(): + """What `close()` does to a write still in flight: release whoever is + waiting, because nothing can ever answer them now, and say plainly that + there is no outcome. `settled()` says it returns whether there is one.""" + watch = a_watch({"level": 0.5}) + + watch.publish() + + assert watch.settled(timeout=0) is False + assert watch.outcome is None + + +# -- the watchdog ------------------------------------------------------------- + + +def test_the_watchdog_gives_up_on_a_write_at_its_deadline(): + given_up_on = [] + dog = Watchdog(given_up_on.append, "pyquadcortex-test-watchdog") + watch = a_watch({"level": 0.5}, patience=0.05) + dog.add(watch) + try: + assert watch.settled(timeout=5.0) + finally: + dog.stop() + assert watch.outcome is WatchOutcome.TIMED_OUT + assert given_up_on == [watch] + + +def test_the_watchdog_leaves_a_confirmed_write_alone(): + given_up_on = [] + dog = Watchdog(given_up_on.append, "pyquadcortex-test-watchdog") + watch = a_watch({"level": 0.5}, patience=0.05) + dog.add(watch) + watch.absorb({"level": 0.5}) + try: + time.sleep(0.3) + finally: + dog.stop() + assert watch.outcome is WatchOutcome.CONFIRMED + assert given_up_on == [] + + +def test_a_write_added_while_the_watchdog_is_idle_still_times_out(): + """The lost-notify bug this is written to catch: with nothing outstanding + the thread waits with no timeout at all, so a write added at that moment + would wait for a wake-up that had already happened and never time out.""" + dog = Watchdog(lambda watch: None, "pyquadcortex-test-watchdog") + warm_up = a_watch({"level": 0.5}, patience=0.02) + dog.add(warm_up) + assert warm_up.settled(timeout=5.0) + time.sleep(0.05) # the thread is now waiting on nothing + + late = a_watch({"level": 0.5}, patience=0.05) + dog.add(late) + try: + assert late.settled(timeout=5.0), "the watchdog slept through it" + finally: + dog.stop() + + +def test_stopping_the_watchdog_ends_its_thread(): + dog = Watchdog(lambda watch: None, "pyquadcortex-test-watchdog") + dog.add(a_watch({"level": 0.5}, patience=30.0)) + assert _named_threads("pyquadcortex-test-watchdog") + + dog.stop() + + assert not _named_threads("pyquadcortex-test-watchdog") + + +def test_a_watchdog_that_has_stopped_does_not_start_again_for_a_late_write(): + """A write that races a `close()` reaches `add` after `stop`. Starting a + thread for it leaves one waiting forever on a connection that is gone.""" + dog = Watchdog(lambda watch: None, "pyquadcortex-test-watchdog") + dog.stop() + late = a_watch({"level": 0.5}, patience=30.0) + + dog.add(late) + + assert not _named_threads("pyquadcortex-test-watchdog") + assert late.settled(timeout=0) is False, "it should still report no outcome" + assert late.outcome is None + + +def test_a_write_added_after_the_watchdog_stopped_does_not_hang_its_caller(): + dog = Watchdog(lambda watch: None, "pyquadcortex-test-watchdog") + dog.stop() + late = a_watch({"level": 0.5}, patience=30.0) + + dog.add(late) + + assert late.settled(timeout=1.0) is False # returns rather than blocking + + +def test_a_watchdog_stopped_before_a_deadline_does_not_report_a_timeout(): + """The connection is going away, so "the unit never answered" would be a + claim about the unit rather than a fact.""" + given_up_on = [] + dog = Watchdog(given_up_on.append, "pyquadcortex-test-watchdog") + watch = a_watch({"level": 0.5}, patience=0.05) + dog.add(watch) + dog.stop() + time.sleep(0.2) + assert given_up_on == [] + + +def _named_threads(name): + return [t for t in threading.enumerate() if t.name == name]