diff --git a/docs/mqtt-vectors/README.md b/docs/mqtt-vectors/README.md new file mode 100644 index 00000000..a8d9d00c --- /dev/null +++ b/docs/mqtt-vectors/README.md @@ -0,0 +1,13 @@ +# MQTT vectors + +Each file here is one sentence of [mqtt-spec.md](../mqtt-spec.md) and the steps that show a panel doing it, written as [contract-vectors.md](../contract-vectors.md) sets out. `npm run check:mqtt-vectors` in `web/` replays every one against a fresh [virtual panel](../../tools/virtual-panel/README.md) built with the `mqtt` feature, and prints what the vector expects beside what the panel did. CI runs it on every pull request with the other drift checks. When the spec rewords a sentence a vector quotes, that vector fails until someone brings it along. + +A Patternflow is a broker's client, so the suite brings the broker. `web/scripts/drift/_broker.ts` starts one for each vector on a free port. It speaks MQTT 3.1.1 at QoS 0 and keeps retained messages, and a topic matches only itself. Before the first step the suite points the panel at it with `POST /api/mqtt`, host and port only, which leaves the role off. A vector picks the role in its first step, `POST /api/mqtt?role=subscriber` for one, and the panel dials on its next frame, so a `wait` comes next. A message the vector sends is `{ "topic": "patternflow/sleep", "payload": "1" }`, with `"retain": true` to keep it on the broker. It reaches the panel only on a topic the panel subscribed to, as on any broker. `out` is each message the panel published, with its `retain` flag beside the topic and payload. + +A file's name starts with the part of the spec it reads. `topics-` is the topic table and the two roles, and `channels-` is the channels and the snapshot. `/api/mqtt` is REST, so its vectors are in [rest-api-vectors](../rest-api-vectors/README.md) and start with `mqtt-`. + +The banner a `message` puts up is read with `GET /vp/mqtt`. That route is the virtual panel's own. The firmware has none, and `/frame.png` does not draw the banner. + +Two vectors keep the spec where the firmware does something else. `topics-param.json` empties `param/3` 20 ms after writing it, and the firmware keeps a channel written in the last 250 ms (`ABSOLUTE_RELEASE_GRACE_MS` in `core_bus.h`). `topics-knob.json` sends a count 27 past the one before, and the firmware cuts that to 16 (`KNOB_PENDING_FLUSH` in `core_mqtt.h`). + +Some of the spec has no vectors. The Director's topics (`query`, `select`, `select/ack`) are for one tool, and every other client ignores them. How soon a panel dials again after losing its broker is the firmware's to decide, not the spec's. The broker takes any user and password, so a refused login is not replayed. diff --git a/docs/mqtt-vectors/topics-either-role.json b/docs/mqtt-vectors/topics-either-role.json new file mode 100644 index 00000000..301c4bbd --- /dev/null +++ b/docs/mqtt-vectors/topics-either-role.json @@ -0,0 +1,17 @@ +{ + "contract": "docs/mqtt-spec.md#topics", + "says": "`sleep` and `message` are the two topics a panel obeys in **either** role", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=subscriber" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttConnected": true } } }, + { "send": { "mqtt": { "topic": "patternflow/message", "payload": "Doors at nine" } } }, + { "send": { "http": "GET /vp/mqtt" }, "expect": { "reply": { "body": { "banner": "Doors at nine" } } } }, + { "send": { "http": "POST /api/mqtt?role=publisher" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttRole": "publisher", "mqttConnected": true } } }, + { "send": { "mqtt": { "topic": "patternflow/message", "payload": "Last call" } } }, + { "send": { "http": "GET /vp/mqtt" }, "expect": { "reply": { "body": { "banner": "Last call" } } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "1" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "sleep": true } } } + ] +} diff --git a/docs/mqtt-vectors/topics-knob.json b/docs/mqtt-vectors/topics-knob.json new file mode 100644 index 00000000..687c1b07 --- /dev/null +++ b/docs/mqtt-vectors/topics-knob.json @@ -0,0 +1,19 @@ +{ + "contract": "docs/mqtt-spec.md#which-topic-to-write", + "says": "The panel diffs it against the last value it received and injects the difference as a detent delta.", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=subscriber" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttConnected": true } } }, + { "send": { "mqtt": { "topic": "patternflow/knob/2", "payload": "0" } } }, + { "send": { "wait": 20 }, "expect": { "input": { "given": { "deltas": [0, 0, 0, 0] } } } }, + { "send": { "mqtt": { "topic": "patternflow/knob/2", "payload": "5" } } }, + { "send": { "wait": 20 }, "expect": { "input": { "given": { "deltas": [0, 5, 0, 0] } } } }, + { "send": { "mqtt": { "topic": "patternflow/knob/2", "payload": "5" } } }, + { "send": { "wait": 20 }, "expect": { "input": { "given": { "deltas": [0, 5, 0, 0] } } } }, + { "send": { "mqtt": { "topic": "patternflow/knob/2", "payload": "-2" } } }, + { "send": { "wait": 20 }, "expect": { "input": { "given": { "deltas": [0, -2, 0, 0] } } } }, + { "send": { "mqtt": { "topic": "patternflow/knob/2", "payload": "25" } } }, + { "send": { "wait": 20 }, "expect": { "input": { "given": { "deltas": [0, 25, 0, 0] } } } } + ] +} diff --git a/docs/mqtt-vectors/topics-param.json b/docs/mqtt-vectors/topics-param.json new file mode 100644 index 00000000..6b232676 --- /dev/null +++ b/docs/mqtt-vectors/topics-param.json @@ -0,0 +1,17 @@ +{ + "contract": "docs/mqtt-spec.md#topics", + "says": "absolute `0`–`1000`; empty payload releases the hold", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=subscriber" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttConnected": true } } }, + { "send": { "mqtt": { "topic": "patternflow/param/3", "payload": "750" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "params": ["", "", 750, ""], "paramActive": [false, false, true, false] } } }, + { "send": { "mqtt": { "topic": "patternflow/param/3", "payload": "1000" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "params": ["", "", 1000, ""], "paramActive": [false, false, true, false] } } }, + { "send": { "mqtt": { "topic": "patternflow/param/3", "payload": "0" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "params": ["", "", 0, ""], "paramActive": [false, false, true, false] } } }, + { "send": { "mqtt": { "topic": "patternflow/param/3", "payload": "" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "paramActive": [false, false, false, false] } } } + ] +} diff --git a/docs/mqtt-vectors/topics-pattern.json b/docs/mqtt-vectors/topics-pattern.json new file mode 100644 index 00000000..30ec819c --- /dev/null +++ b/docs/mqtt-vectors/topics-pattern.json @@ -0,0 +1,15 @@ +{ + "contract": "docs/mqtt-spec.md#topics", + "says": "the pattern's display name or slug", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=subscriber" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttConnected": true } } }, + { "send": { "mqtt": { "topic": "patternflow/pattern", "payload": "Wave Saw" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "active": "Wave Saw" } } }, + { "send": { "mqtt": { "topic": "patternflow/pattern", "payload": "Origin" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "active": "Origin" } } }, + { "send": { "mqtt": { "topic": "patternflow/pattern", "payload": "wave_saw" } } }, + { "send": { "wait": 20 }, "expect": { "status": { "active": "Wave Saw" } } } + ] +} diff --git a/docs/mqtt-vectors/topics-publisher-ignores.json b/docs/mqtt-vectors/topics-publisher-ignores.json new file mode 100644 index 00000000..8c4522f1 --- /dev/null +++ b/docs/mqtt-vectors/topics-publisher-ignores.json @@ -0,0 +1,16 @@ +{ + "contract": "docs/mqtt-spec.md#which-topic-to-write", + "says": "`knob`, `param` and `pattern` are obeyed **only in Subscriber role**; a Publisher ignores them silently.", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=publisher" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttRole": "publisher", "mqttConnected": true } } }, + { "send": { "mqtt": { "topic": "patternflow/knob/1", "payload": "5" } } }, + { "send": { "mqtt": { "topic": "patternflow/param/2", "payload": "800" } } }, + { "send": { "mqtt": { "topic": "patternflow/pattern", "payload": "Wave Saw" } } }, + { "send": { "wait": 20 }, "expect": { + "out": [{ "mqtt": "" }], + "input": { "given": { "deltas": [0, 0, 0, 0] } }, + "status": { "active": "Origin", "paramActive": [false, false, false, false] } } } + ] +} diff --git a/docs/mqtt-vectors/topics-publisher.json b/docs/mqtt-vectors/topics-publisher.json new file mode 100644 index 00000000..670286e7 --- /dev/null +++ b/docs/mqtt-vectors/topics-publisher.json @@ -0,0 +1,13 @@ +{ + "contract": "docs/mqtt-spec.md#transport", + "says": "`publisher` (sends knobs, pattern and, on a show channel, a snapshot heartbeat)", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=publisher" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttRole": "publisher", "mqttConnected": true } } }, + { "send": { "hand": { "knob": 2, "turn": -3 } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/knob/2", "payload": "-3", "retain": false } }] } }, + { "send": { "hand": { "knob": 2, "turn": 1 } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/knob/2", "payload": "-2", "retain": false } }] } }, + { "send": { "http": "GET /api/patterns/select?index=1" } }, + { "send": { "wait": 20 }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/pattern", "payload": "", "retain": false } }] } } + ] +} diff --git a/docs/mqtt-vectors/topics-sleep-state.json b/docs/mqtt-vectors/topics-sleep-state.json new file mode 100644 index 00000000..d5de0b48 --- /dev/null +++ b/docs/mqtt-vectors/topics-sleep-state.json @@ -0,0 +1,13 @@ +{ + "contract": "docs/mqtt-spec.md#topics", + "says": "`0` or `1`, on every change and once per connection", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=subscriber" } }, + { "send": { "wait": 20 }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "0", "retain": false } }, { "mqtt": "" }] } }, + { "send": { "http": "POST /api/sleep?on=1" } }, + { "send": { "wait": 20 }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1", "retain": false } }] } }, + { "send": { "http": "POST /api/mqtt?role=publisher" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttRole": "publisher", "mqttConnected": true }, "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1", "retain": false } }] } } + ] +} diff --git a/docs/mqtt-vectors/topics-sleep-words.json b/docs/mqtt-vectors/topics-sleep-words.json new file mode 100644 index 00000000..6ceaa4be --- /dev/null +++ b/docs/mqtt-vectors/topics-sleep-words.json @@ -0,0 +1,19 @@ +{ + "contract": "docs/mqtt-spec.md#topics", + "says": "`1` / `on` / `true` / `sleep` · `0` / `off` / `false` / `wake` · `toggle`", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=subscriber" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttConnected": true } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "1" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1" }, "within": 20 }], "status": { "sleep": true } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "0" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "0" }, "within": 20 }], "status": { "sleep": false } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "on" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1" }, "within": 20 }], "status": { "sleep": true } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "off" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "0" }, "within": 20 }], "status": { "sleep": false } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "true" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1" }, "within": 20 }], "status": { "sleep": true } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "false" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "0" }, "within": 20 }], "status": { "sleep": false } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "sleep" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1" }, "within": 20 }], "status": { "sleep": true } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "wake" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "0" }, "within": 20 }], "status": { "sleep": false } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "toggle" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1" }, "within": 20 }], "status": { "sleep": true } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "toggle" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "0" }, "within": 20 }], "status": { "sleep": false } } } + ] +} diff --git a/docs/mqtt-vectors/topics-sleep.json b/docs/mqtt-vectors/topics-sleep.json new file mode 100644 index 00000000..09f29507 --- /dev/null +++ b/docs/mqtt-vectors/topics-sleep.json @@ -0,0 +1,11 @@ +{ + "contract": "docs/mqtt-spec.md#topics", + "says": "Publishing `1` to `/sleep` puts the panel to sleep, `0` wakes it, and `sleep/state` mirrors the result", + "features": ["mqtt"], + "steps": [ + { "send": { "http": "POST /api/mqtt?role=subscriber" } }, + { "send": { "wait": 20 }, "expect": { "status": { "mqttConnected": true, "sleep": false } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "1" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "1" }, "within": 20 }], "status": { "sleep": true } } }, + { "send": { "mqtt": { "topic": "patternflow/sleep", "payload": "0" } }, "expect": { "out": [{ "mqtt": { "topic": "patternflow/sleep/state", "payload": "0" }, "within": 20 }], "status": { "sleep": false } } } + ] +} diff --git a/web/scripts/drift/_broker.ts b/web/scripts/drift/_broker.ts new file mode 100644 index 00000000..48a29948 --- /dev/null +++ b/web/scripts/drift/_broker.ts @@ -0,0 +1,130 @@ +/** + * check:mqtt-vectors' broker, and the Sender it hands _vectors.ts (docs/contract-vectors.md). A Patternflow + * is a broker's client, so each vector gets a broker of its own on a free port: MQTT 3.1.1 at QoS 0 on + * node:net, enough to connect, subscribe, publish and retain. A topic matches only itself, since a panel + * subscribes with no wildcards (docs/mqtt-spec.md line 12). open() points the panel at the broker with + * POST /api/mqtt and leaves the role off for the vector to pick. A step's mqtt goes to each connection + * subscribed to its topic, and heard is each PUBLISH the panel sent. + */ +import net, { type AddressInfo } from "node:net"; + +import type { Panel, Sender } from "./_vectors"; + +/** A step's mqtt, and what heard holds. A send keeps it on the broker with retain: true. */ +type Message = { topic: string; payload: string; retain?: boolean }; + +// The framing: a type byte, the length in groups of seven bits, the body. A string is two length bytes and UTF-8. +const u16 = (n: number) => Buffer.from([n >> 8, n & 255]); +const str = (s: string) => Buffer.concat([u16(Buffer.byteLength(s)), Buffer.from(s)]); +function packet(type: number, ...parts: Buffer[]) { + const [body, head] = [Buffer.concat(parts), [type]]; + let n = body.length; + do head.push((n % 128) | (n >= 128 ? 128 : 0)); + while ((n = Math.floor(n / 128))); + return Buffer.concat([Buffer.from(head), body]); +} +const [CONNECT, PUBLISH, SUBSCRIBE, PINGREQ, DISCONNECT] = [1, 3, 8, 12, 14]; +const SETTLE_MS = 2000; // the longest a dial may take before settle() gives up on it + +const isMessage = (m: unknown): m is Message => + typeof m === "object" && m !== null && Object.keys(m).every((k) => ["topic", "payload", "retain"].includes(k)) && + typeof (m as Message).topic === "string" && typeof (m as Message).payload === "string" && [undefined, true, false].includes((m as Message).retain); + +export const mqtt: Sender = async () => { + // Every connection, and once it has sent CONNECT, the topics it subscribed to. + const clients = new Map | null>(); + const retained = new Map(); + const heard: Message[] = []; + let [moved, panel] = [0, undefined as Panel | undefined]; // bytes in and out, which settle() watches + + const write = (s: net.Socket, b: Buffer) => ((moved += b.length), s.write(b)); + // To each subscriber with the retain flag clear, and kept for later ones when it is set. An empty one clears what was kept. + function publish(topic: string, payload: Buffer, retain: boolean) { + if (retain && payload.length) retained.set(topic, payload); + else if (retain) retained.delete(topic); + for (const [s, topics] of clients) if (topics?.has(topic)) write(s, packet(0x30, str(topic), payload)); + } + + function received(s: net.Socket, type: number, flags: number, body: Buffer) { + const topics = clients.get(s); + if (type === CONNECT) return void (clients.set(s, new Set()), write(s, packet(0x20, Buffer.from([0, 0])))); + if (!topics) return void s.destroy(); // anything before CONNECT + if (type === PUBLISH) { + const at = 2 + body.readUInt16BE(0); + const [topic, payload, retain] = [body.toString("utf8", 2, at), body.subarray(at + (flags & 6 ? 2 : 0)), (flags & 1) === 1]; + heard.push({ topic, payload: payload.toString(), retain }); + publish(topic, payload, retain); + // QoS 0 has no answer, and a PINGRESP is one a client takes for nothing. Bytes going back carry the TCP ACK + // at once, where the kernel would hold it 40 ms and the panel's next write behind it, by Nagle's rule. + return void write(s, packet(0xd0)); + } + if (type === SUBSCRIBE) { + // Each filter is granted QoS 0, then gets what was kept on its topic, with the retain flag set. + const filters: string[] = []; + for (let at = 2; at < body.length; at += 3 + body.readUInt16BE(at)) filters.push(body.toString("utf8", at + 2, at + 2 + body.readUInt16BE(at))); + filters.forEach((f) => topics.add(f)); + write(s, packet(0x90, body.subarray(0, 2), Buffer.alloc(filters.length))); + for (const f of filters) if (retained.has(f)) write(s, packet(0x31, str(f), retained.get(f)!)); + } else if (type === PINGREQ) write(s, packet(0xd0)); + else if (type === DISCONNECT) s.end(); + } + + const server = net.createServer((s) => { + clients.set(s, null); + s.setNoDelay(true); // with Nagle on, a write behind one not yet acknowledged waits out a 40 ms delayed ACK + let inbox: Buffer = Buffer.alloc(0); + s.on("error", () => s.destroy()); + s.on("close", () => clients.delete(s)); + s.on("data", (chunk: Buffer) => { + moved += chunk.length; + inbox = Buffer.concat([inbox, chunk]); + for (;;) { + let [length, i, scale] = [0, 1, 1]; + for (; i < inbox.length && i < 5; i++, scale *= 128) { + length += (inbox[i] & 127) * scale; + if (inbox[i] < 128) break; + } + if (i === 5) return void s.destroy(); // a length past four bytes is not MQTT + if (i >= inbox.length || inbox.length < i + 1 + length) return; + const [type, flags, body] = [inbox[0] >> 4, inbox[0] & 15, inbox.subarray(i + 1, i + 1 + length)]; + inbox = inbox.subarray(i + 1 + length); + try { + received(s, type, flags, body); + } catch { + return void s.destroy(); // a body shorter than its own fields say + } + } + }); + }); + await new Promise((done, fail) => server.once("error", fail).listen(0, "127.0.0.1", () => done())); + const port = (server.address() as AddressInfo).port; + + return { + heard, + async open(p) { + panel = p; + const r = await p.http("POST /api/mqtt", `host=127.0.0.1&port=${port}`); + if (r.status !== 200) throw new Error(`POST /api/mqtt answered ${r.status}, so the panel has no mqtt feature to point at the broker`); + }, + async send(what) { + if (!isMessage(what)) throw new Error(`mqtt sends { "topic", "payload" }, and "retain": true to keep it, not ${JSON.stringify(what)}`); + publish(what.topic, Buffer.from(what.payload), what.retain === true); + return undefined; + }, + // A dial, its CONNECT and CONNACK and the SUBSCRIBEs after them take real time, the panel's clock or not. Settled + // once the panel is not mid-dial and a round trip over HTTP moved nothing on the broker. + async settle() { + for (const from = Date.now(); ; ) { + const before = moved; + for (let i = 0; i < 3; i++) await new Promise((done) => setImmediate(done)); + const { state } = (await panel!.http("GET /api/mqtt")).body as { state?: string }; + if (state !== "connecting" && moved === before) return; + if (Date.now() - from > SETTLE_MS) throw new Error(`the panel's MQTT is still ${state} after ${SETTLE_MS} ms`); + } + }, + async close() { + for (const s of clients.keys()) s.destroy(); + await new Promise((done) => server.close(() => done())); + }, + }; +}; diff --git a/web/scripts/drift/mqtt-vectors.ts b/web/scripts/drift/mqtt-vectors.ts new file mode 100644 index 00000000..25d30449 --- /dev/null +++ b/web/scripts/drift/mqtt-vectors.ts @@ -0,0 +1,10 @@ +/** + * check:mqtt-vectors: docs/mqtt-vectors/ against a fresh virtual panel each + * (docs/contract-vectors.md), and a broker of its own for each (_broker.ts). + * HTTP, a hand and the clock are _vectors.ts's own. + */ +import { mqtt } from "./_broker"; +import { done } from "./_lib"; +import { replay } from "./_vectors"; + +replay("docs/mqtt-vectors", { mqtt }).then(done);