feat(qwp): QWP ingest over WebSocket with store-and-forward durability + benchmarks

README.md

@@ -25,6 +25,55 @@ The Undici HTTP agent was introduced in 4.0.0, and it is the default HTTP transp
 The standard HTTP/HTTPS modules of Node.js are still supported for backwards compatibility.
 Use the <i>stdlib_http</i> option to switch to the standard HTTP/HTTPS modules.
 
+## Transport support matrix
+
+| Protocol | Transport | Notes |
+|---|---|---|
+| `http` / `https` | HTTP | ILP, request/response |
+| `tcp` / `tcps` | TCP | ILP, persistent |
+| `ws` / `wss` | WebSocket | **QWP** — columnar binary, store-and-forward |
+
+### QWP caveats
+
+These properties of the WebSocket (QWP) sender differ from what the HTTP/ILP
+senders do, and each one can surprise a user who assumes otherwise:
+
+- `flush()` resolves once rows are **published** to the send log, not when the
+  server acknowledges them. This differs from `http::`.
+- Delivery is **at-least-once**. A retried batch can duplicate rows; use a
+  `DEDUP` table if you need idempotence.
+- Symbol columns are serialized in **delta mode**: a connection dictionary is
+  shared with the buffer and persisted to `<sf_dir>/.symbol-dict` in disk mode,
+  so only newly-introduced symbols ride each frame. Recovery re-seeds it
+  positionally, keeping ids dense from 0 — so **don't** assume symbol ids are
+  stable across senders or restarts.
+- An acknowledgement means server-side **commit**, not object-store durability.
+  Set `request_durable_ack=on` if durability gates downstream work — the client
+  then requests durable ACKs and **fails fast at connect** if the server cannot
+  confirm them (`X-QWP-Durable-Ack: enabled`).
+- `sf_dir` alone is **not** power-loss durability. The default
+  `sf_durability=memory` never fsyncs, so bytes survive a *process* crash but
+  not a power loss. `sf_durability=periodic` (requires `sf_dir`) runs a
+  background barrier every `sf_sync_interval_millis` that fsyncs the active
+  segment and the ack watermark, making bytes survive an OS/power crash up to
+  that cadence.
+- `drain_orphans` (off by default) replays slots left by a crashed process: on
+  startup, the client scans `<sf_dir>/` for slot directories not held by a live
+  lock and hands each to a background drainer (bounded by
+  `max_background_drainers`) that opens its own WebSocket, re-registers the
+  recovered symbol dictionary, and replays unacked frames to the server before
+  releasing the slot. A terminal failure drops a `.failed` sentinel so the slot
+  is retried only after an operator clears it; a transient outage is retried
+  indefinitely. Requires `sf_dir`.
+- In disk mode, recovery cross-checks the scanned segment chain against
+  `sf-manifest.bin`, a crash-safe boundary record (magic `SFM1`) written on
+  each segment rotation using the same alternating-generation scheme as the ack
+  watermark, and quarantines the slot when the manifest's recorded chain head is
+  ahead of what is actually on disk (a vanished tail segment = real data loss).
+- `tls_roots` accepts PEM or PKCS#12.
+- `tls_roots` accepts PEM or PKCS#12. **JKS keystores are not supported** by
+  Node; convert them first.
+
 ## Configuration options
 
 Detailed description of the client's configuration options can be found in

benchmarks/README.md

@@ -0,0 +1,107 @@
+# QWP benchmarks
+
+Ad hoc. No CI job, no gate. Run them when you want a number.
+
+```bash
+# Point this at REAL storage, not tmpfs — see "Am I measuring disk?" below.
+export QWP_BENCH_DIR=/var/tmp/qwp-bench
+
+pnpm bench                       # encoder, buffer, sf  (no server needed)
+pnpm bench:e2e                   # needs QuestDB on :9000
+QDB_ADDR=host:9000 pnpm bench:e2e
+
+pnpm typecheck:bench             # tsc over src + benchmarks
+pnpm lint:bench                  # eslint over benchmarks/
+```
+
+**If `pnpm <script>` fails before running anything**, that is this repo's
+ignored-builds check, not your benchmark. Every earlier QWP handoff hit it. Call
+the binary directly instead — the scripts above are still the canonical
+definitions, and they work wherever the gate is satisfied:
+
+```bash
+./node_modules/.bin/vitest bench --run benchmarks/          # pnpm bench
+./node_modules/.bin/tsx benchmarks/e2e.ts                   # pnpm bench:e2e
+./node_modules/.bin/tsc -p tsconfig.bench.json --noEmit     # pnpm typecheck:bench
+```
+
+## Reading the output
+
+**`hz` is callbacks per second, not rows per second.** Each encoder callback
+covers 10,000 rows, so rows/s is `hz × 10_000`. This is the easiest way to
+write down a number that is wrong by four orders of magnitude.
+
+**The disk arm spikes every tenth iteration, and that is real.** `APPENDS × 4 KiB
+= 400 KiB` per iteration against `segmentBytes: 4 << 20`, so a segment roll fires
+on roughly every 10th body. A roll closes the active fd, creates the next `.sfa`,
+writes its header, and rewrites the whole 8 KiB `sf-manifest.bin` (`persistManifest`
+in `sf/engine.ts`, added by the C2 work in handoff 8 — *after* this plan was first
+drafted). At the default durability there is no `fsync` on that path, so the spike
+is syscalls and a file creation rather than a device round-trip. Read **p75 as
+steady-state append cost** and `max`/`p999` as roughly the cost of a roll: a
+bimodal distribution here is the benchmark working, not noise or GC.
+
+It also means **the memory arm is not the disk arm minus the disk.** Memory mode
+never reaches `persistFrame`, so it never rolls and never writes a manifest — the
+gap between the two arms is segment management *plus* writing, not writing alone.
+Don't quote the ratio as "the cost of durability".
+
+**`rme` is the relative margin of error.** Check it before believing a
+difference: a 5% gap between two arms each carrying ±3% rme is not a result.
+
+There is no p50 or p90 in `vitest bench` output. The e2e script computes its own
+percentiles, which is why it is a separate script.
+
+**sf-on is not always faster than sf-off.** `e2e.ts` runs the two flush
+contracts on one server, and the SF engagement is real (the slot dir gets a
+`.sfa` + `sf-manifest.bin`). On a same-host server the localhost round-trip can
+be *faster* than a disk-backed append, so sf-on can legitimately come out
+slower — check that SF is engaged (slot files appear) and the disk baseline is
+sane before reading anything into the sign of the gap. They answer different
+questions regardless: sf-on is "recoverable if I crash now"; sf-off is "the
+server has it". Do not quote one as the other.
+
+## Am I measuring disk?
+
+`os.tmpdir()` is `/tmp` on most Linux boxes, and `/tmp` is frequently **tmpfs —
+RAM**. If the SF benchmarks and the e2e sf-on arm run there, they measure memory
+while claiming to measure disk, and the `dd` baseline reports an *excellent*
+figure — so the guard confirms a false conclusion instead of catching it.
+
+```bash
+df -T "$QWP_BENCH_DIR"   # type must not be tmpfs
+```
+
+## Before you quote anything
+
+1. **Check the disk line.** `sf.bench.ts` prints a `dd` figure first. If it is
+   far below what the hardware should do, the SF numbers are meaningless —
+   fix the machine, do not average around it.
+2. **Check the spread.** `e2e.ts` runs every arm three times and prints a range.
+   If p99 varies by more than about 2× across repeats, the machine is too busy.
+3. **Do not mix the two e2e numbers.** With SF on, `flush()` returns once the
+   row is durable *locally*. With SF off it waits for the *server*. They answer
+   different questions and the SF-on number is not a faster version of the
+   SF-off one.
+
+## What the floors mean
+
+`encoder.bench.ts` compares against hand-written baselines that do the minimum
+byte movement — no null bitmap, no schema, no framing. The floor is not a
+target. The gap between floor and encoder is what the protocol costs.
+
+Both sides accumulate into a `sink` so neither can be optimised away. If the
+encoder still beats its floor, the comparison itself is wrong — not dead-code
+removal. (A flat LONG column lands *at* its floor by design: the encoder uses
+the same `writeBigInt64LE` primitive, so near-parity is the correct result.)
+
+## What this suite does not cover
+
+- **Gorilla's raw fallback.** Every workload has perfectly regular timestamps,
+  so delta-of-delta is always 0 and the compressed path always wins. The
+  fallback needs a delta-of-delta beyond signed int32 — about a 35-minute jump
+  at microsecond resolution. `validate.test.ts` asserts the behaviour; these
+  numbers say nothing about it.
+- **Reconnect, replay and drainers.** Dominated by network and disk timing
+  rather than client code, so a number there measures the environment.
+- **Concurrency.** Single sender, single connection throughout.

benchmarks/buffer.bench.ts

@@ -0,0 +1,87 @@
+import { bench, describe } from "vitest";
+import { QwpBuffer } from "../src/qwp/buffer";
+import { SymbolDict } from "../src/qwp/protocol/symbolDict";
+import { WORKLOADS } from "./workloads";
+
+const N = 10_000;
+const CAP = 1 << 30; // large enough that nothing splits
+
+/** See Task 5's note — consumes results so nothing is optimised away. */
+let sink = 0;
+
+/**
+ * Handles every column family the workloads generate — see Task 5's note.
+ *
+ * Note `Number(v)`: `intColumn` takes a `number` while the workloads carry
+ * `bigint`, so this layer pays a per-value conversion the encoder benchmarks in
+ * Task 5 do not (they push the bigint straight into `values`). The two tasks are
+ * therefore not directly comparable — some of the gap is representation, not
+ * buffer overhead. All generated values are well under 2^53, so no precision is
+ * lost; raise N or change a generator and re-check that.
+ */
+function fill(b: QwpBuffer, rows: ReturnType<typeof WORKLOADS.trades.rows>): void {
+  for (const row of rows) {
+    b.table(row.table);
+    for (const [n, v] of row.symbols) b.symbol(n, v);
+    for (const [n, v] of row.longs) b.intColumn(n, Number(v));
+    for (const [n, v] of row.doubles) b.floatColumn(n, v);
+    for (const [n, v] of row.strings) b.stringColumn(n, v);
+    // QwpBuffer.at() is synchronous (src/qwp/buffer.ts:173) — do NOT await it.
+    b.at(row.ts, "us");
+  }
+}
+
+describe("QwpBuffer build + seal", () => {
+  for (const name of ["trades", "wide", "sparse"] as const) {
+    const rows = WORKLOADS[name].rows(N);
+    bench(`build+seal / ${name}`, () => {
+      const b = new QwpBuffer();
+      fill(b, rows);
+      // sealFrames returns Buffer[]; sum BYTES, not the array length --
+      // `.length` here would be the frame count (always 1 under CAP) and
+      // would leave the encoded contents unobserved.
+      for (const f of b.sealFrames(CAP)) sink += f.length;
+    });
+  }
+});
+
+describe("dictionary mode", () => {
+  const rows = WORKLOADS.highCardSymbol.rows(N);
+
+  bench("full-dict (no dict attached)", () => {
+    const b = new QwpBuffer();
+    fill(b, rows);
+    for (const f of b.sealFrames(CAP)) sink += f.length;
+  });
+
+  // Steady state needs BOTH halves: a populated dictionary and a confirmed
+  // baseline. confirmedMaxId lives on the buffer, starts at -1, and only moves
+  // via confirmDeltaPublished() from the transport's publish path -- so a
+  // fresh buffer with a primed dict still ships every symbol, because
+  // entriesFrom(-1 + 1) returns everything. Priming the dict alone measures
+  // nothing new.
+  const primed = new SymbolDict();
+  {
+    const warm = new QwpBuffer();
+    warm.attachDict(primed);
+    fill(warm, rows);
+    warm.sealFrames(CAP);
+  }
+
+  bench("delta-dict (primed + baseline confirmed)", () => {
+    const b = new QwpBuffer();
+    b.attachDict(primed);
+    b.setConfirmedMaxId(primed.size() - 1);
+    fill(b, rows);
+    for (const f of b.sealFrames(CAP)) sink += f.length;
+  });
+
+  bench("delta-dict (cold, first batch)", () => {
+    const b = new QwpBuffer();
+    b.attachDict(new SymbolDict());
+    fill(b, rows);
+    for (const f of b.sealFrames(CAP)) sink += f.length;
+  });
+});
+
+export const _sink = () => sink;