[DONE] 정리 4단계 PR 11 — MQTT 벡터가 시험 브로커를 거쳐 토픽과 역할을 가상 패널에 돌려 보고 check:mqtt-vectors 가 생긴다

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.

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 } } }
+  ]
+}

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] } } } }
+  ]
+}

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": ["<integer>", "<integer>", 750, "<integer>"], "paramActive": [false, false, true, false] } } },
+    { "send": { "mqtt": { "topic": "patternflow/param/3", "payload": "1000" } } },
+    { "send": { "wait": 20 }, "expect": { "status": { "params": ["<integer>", "<integer>", 1000, "<integer>"], "paramActive": [false, false, true, false] } } },
+    { "send": { "mqtt": { "topic": "patternflow/param/3", "payload": "0" } } },
+    { "send": { "wait": 20 }, "expect": { "status": { "params": ["<integer>", "<integer>", 0, "<integer>"], "paramActive": [false, false, true, false] } } },
+    { "send": { "mqtt": { "topic": "patternflow/param/3", "payload": "" } } },
+    { "send": { "wait": 20 }, "expect": { "status": { "paramActive": [false, false, false, false] } } }
+  ]
+}

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" } } }
+  ]
+}

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": "<absent>" }],
+      "input": { "given": { "deltas": [0, 0, 0, 0] } },
+      "status": { "active": "Origin", "paramActive": [false, false, false, false] } } }
+  ]
+}

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": "<string>", "retain": false } }] } }
+  ]
+}

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": "<absent>" }] } },
+    { "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 } }] } }
+  ]
+}

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 } } }
+  ]
+}

docs/mqtt-vectors/topics-sleep.json

@@ -0,0 +1,11 @@
+{
+  "contract": "docs/mqtt-spec.md#topics",
+  "says": "Publishing `1` to `<prefix>/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 } } }
+  ]
+}

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<net.Socket, Set<string> | null>();
+  const retained = new Map<string, Buffer>();
+  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<void>((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<void>((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<void>((done) => server.close(() => done()));
+    },
+  };
+};

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);