M6-T2: window-protocol load harness (local-guarded, extensively tested)

docs/ops/load-test-plan.md

@@ -0,0 +1,84 @@
+# LumaLine load-test plan (M6-T2) — window-protocol write ceiling
+
+**Status: harness BUILT (`scripts/load/`), not yet run.** Running is a separate, gated step — it goes
+against a **local** stack only. Never run this against prod: it would create real windows/impressions
+and confound M5's first-charge reconcile. The harness enforces this with a hard guard (below).
+
+## What it measures
+
+The hot path is `window_open → window_beat×N → close_window` (SQL RPCs in
+`20260627025330_window_rpcs.sql`). Each honest window is ~5–6 DB writes: 1 insert (open) + ≥3 updates
+(beats) + 1 update + 1 insert (close). The harness drives many concurrent synthetic devices through
+that loop at honest cadence and reports:
+
+- **throughput** — total ops/s and successful **writes/s** (the number the ~15k-writes/s ceiling claim is about);
+- **latency** — p50 / p95 / p99 / max per op (`open`/`beat`/`close`);
+- **error taxonomy** — a histogram by `(op, code)` so a rate-limit 429, an anti-batch reject, or a
+  timeout are distinguished, not lumped.
+
+## The ceiling claim it validates
+
+The production-readiness handoff documents ~**15k writes/s @ ~50k concurrent devs** for the UNLOGGED
+`ad_windows` + direct-RPC beats design. That is unvalidated. Because each device completes one window
+per ~5s (5s dwell + ≥500ms×3 anti-batch spacing), a device sustains ~1–1.2 writes/s, so ~15k writes/s
+implies ~12–15k concurrent devices. The harness scales `--users` toward that and records where error
+rate / p99 latency start to climb — the real ceiling, and the trigger point for the migration below.
+
+## Honest cadence (why the harness can't just hammer)
+
+The server enforces, and the harness respects:
+- **anti-batch**: ≥500ms wall-clock between beats (harness uses 600ms);
+- **HMAC hash-chain**: each beat = `HMAC_sha256("seq|prev|activity", challenge)`, `prev` = previous
+  beat's hmac (or `window_id` for beat 1) — replicated in `scripts/load/lib/hmac-chain.mjs`, locked to
+  the server contract by `test/load-harness.test.mjs`;
+- **full dwell**: `close_window` credits only when elapsed ≥ `dwell_ms` (5000) with ≥3 beats +
+  activity progress. The harness produces genuinely creditable windows so it exercises the real
+  credit/insert path, not just rejects.
+
+## Safety guard (hard)
+
+`assertSafeTarget()` in `harness.mjs` refuses to run if the target URL contains the prod ref
+`prmsonskzrubqsazmpwd`, or if the host is not `127.0.0.1`/`localhost` (unless `LOAD_ALLOW_NONLOCAL=1`
+is set for a non-prod remote you own — the prod-ref check still fires regardless). Default target is
+the local stack `http://127.0.0.1:54321`.
+
+## Rate-limit tuning (`LUMALINE_RL_MAX_PER_MIN`, `rl_buckets`)
+
+The salted-IP fixed-window limiter (`20260627041000_rate_limit.sql`) guards the **anonymous self-promo
+feed**, not authenticated device windows. Tuning goal (task acceptance): the limit must **not 429 a
+legitimate single dev's cadence** while still capping a single-source flood. Method: run the harness's
+anon path (future extension) at one-dev cadence, confirm zero 429s; then at flood cadence, confirm the
+limiter clamps. `rl_hit` fails **open** on a missing hash, so a backend hiccup never blocks honest
+ticks — verify that holds under load.
+
+## Cloudflare Durable Objects migration trigger
+
+Document (and the harness helps set) the concrete trigger to move `ad_windows` off Postgres onto
+Cloudflare DO: when measured sustained writes/s approaches the ceiling at acceptable p99, OR when p99
+`beat` latency exceeds the heartbeat interval under target concurrency. Until then, UNLOGGED
+`ad_windows` + direct RPC is sufficient; the trigger is a number this harness produces, not a guess.
+
+## How to run (local, gated)
+
+```bash
+supabase start                                   # bring up the local stack
+# export local keys from `supabase status` (or accept the built-in local demo defaults):
+#   export LOAD_BASE=http://127.0.0.1:54321
+#   export LOAD_SERVICE_KEY=… LOAD_ANON_KEY=… LOAD_JWT_SECRET=…
+node scripts/load/harness.mjs --users 200 --duration 30 --beats 3
+node scripts/load/harness.mjs --users 2000 --duration 60      # step toward the ceiling
+# full JSON report is written to /tmp/lumaline-load-<ts>.json
+```
+
+The harness seeds synthetic `auth.users → publishers → devices` (publishers.auth_user_id is NOT NULL)
+via the GoTrue admin API + service-role inserts, mints per-device JWTs, then runs the loop. Seed data
+is local-only; drop it with `supabase db reset` after a run.
+
+## Acceptance (M6-T2)
+
+- harness sustains a target write rate without error spikes (error taxonomy clean at the target);
+- tuned limits don't 429 legitimate single-dev traffic;
+- the report records the **measured** ceiling + the DO-migration trigger number.
+
+**DEPENDS-ON: M5** for *running* against anything shared. Building + local runs are safe now; do not
+point it at prod until M5-T3/T4 are validated and prod is no longer single-owner.

scripts/load/harness.mjs

@@ -0,0 +1,159 @@
+#!/usr/bin/env node
+// LumaLine window-protocol load harness (M6-T2) — measures the write ceiling of the hot path
+// (window_open -> window_beat×N -> close_window) under concurrency, plus latency percentiles and an
+// error taxonomy. Drives PostgREST RPCs as authenticated synthetic devices, honest cadence.
+//
+// ⚠️ LOCAL-ONLY BY HARD GUARD. Running load against prod would pollute M5's live windows/impressions
+// and confound the first-charge reconcile — so this REFUSES any target that looks like prod
+// (ref prmsonskzrubqsazmpwd) or any non-localhost host unless LOAD_ALLOW_NONLOCAL=1 is set for a
+// non-prod remote you explicitly own. Default target is the local Supabase stack.
+//
+// PREREQUISITE (run step, not build step): `supabase start`, then export the local keys from
+// `supabase status` (LOAD_SERVICE_KEY / LOAD_ANON_KEY / LOAD_JWT_SECRET) or accept the well-known
+// local demo defaults below. This file is an ops tool; it is NOT shipped to npm.
+//
+//   node scripts/load/harness.mjs --users 200 --duration 30 --beats 3
+//   node scripts/load/harness.mjs --users 2000 --duration 60          # push toward the ceiling
+import { createHmac } from 'node:crypto';
+import { writeFileSync } from 'node:fs';
+import { buildChain } from './lib/hmac-chain.mjs';
+import { summarize, renderSummary } from './lib/metrics.mjs';
+
+const PROD_REF = 'prmsonskzrubqsazmpwd';
+const env = process.env;
+
+// ---- config (local defaults; override via env) ----------------------------------------------
+const BASE = env.LOAD_BASE || 'http://127.0.0.1:54321';
+// Well-known local Supabase demo credentials (safe: they only work against a local stack).
+const ANON_KEY = env.LOAD_ANON_KEY || 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyb2xlIjoiYW5vbiIsImlzcyI6InN1cGFiYXNlLWRlbW8iLCJpYXQiOjE2NDE3NjkyMDAsImV4cCI6MTc5OTUzNTYwMH0.dc_X5iR_VP_qT0zsiyj_I_OZ2T9FtRU2BBNWN8Bu4GE';
+const SERVICE_KEY = env.LOAD_SERVICE_KEY || 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyb2xlIjoic2VydmljZV9yb2xlIiwiaXNzIjoic3VwYWJhc2UtZGVtbyIsImlhdCI6MTY0MTc2OTIwMCwiZXhwIjoxNzk5NTM1NjAwfQ.DaYlNEoUrrEn2Ig7tqibS-PHK5vgusbcbo7X36XVt4Q';
+const JWT_SECRET = env.LOAD_JWT_SECRET || 'super-secret-jwt-token-with-at-least-32-characters-long';
+
+const args = process.argv.slice(2);
+const flag = (name, def) => { const i = args.indexOf(`--${name}`); return i >= 0 ? args[i + 1] : def; };
+const USERS = parseInt(flag('users', '50'), 10);       // concurrent virtual devices
+const DURATION_S = parseInt(flag('duration', '20'), 10);
+const BEATS = Math.max(3, parseInt(flag('beats', '3'), 10)); // >=3 to credit
+const BEAT_SPACING_MS = 600;   // > 500ms server anti-batch minimum
+const DWELL_MS = 5000;         // server requires elapsed >= dwell_ms to credit
+
+// ---- hard safety guard ------------------------------------------------------------------------
+function assertSafeTarget(base) {
+  if (base.includes(PROD_REF)) {
+    console.error(`FATAL: target ${base} is the PROD project (${PROD_REF}). Load against prod is forbidden.`);
+    process.exit(3);
+  }
+  let host;
+  try { host = new URL(base).hostname; } catch { console.error(`FATAL: bad LOAD_BASE ${base}`); process.exit(3); }
+  const isLocal = host === '127.0.0.1' || host === 'localhost' || host === '::1';
+  if (!isLocal && env.LOAD_ALLOW_NONLOCAL !== '1') {
+    console.error(`FATAL: ${host} is not local. Set LOAD_ALLOW_NONLOCAL=1 ONLY for a non-prod remote you own.`);
+    process.exit(3);
+  }
+}
+
+// ---- helpers ----------------------------------------------------------------------------------
+const b64url = (buf) => Buffer.from(buf).toString('base64').replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
+function mintDeviceJWT({ sub, publisher_id, device_id }) {
+  const header = b64url(JSON.stringify({ alg: 'HS256', typ: 'JWT' }));
+  const now = Math.floor(Date.now() / 1000);
+  const payload = b64url(JSON.stringify({ sub, role: 'authenticated', publisher_id, device_id, iat: now, exp: now + 3600 }));
+  const sig = b64url(createHmac('sha256', JWT_SECRET).update(`${header}.${payload}`).digest());
+  return `${header}.${payload}.${sig}`;
+}
+
+async function rpc(fn, body, jwt) {
+  const t0 = performance.now();
+  try {
+    const r = await fetch(`${BASE}/rest/v1/rpc/${fn}`, {
+      method: 'POST',
+      headers: { apikey: ANON_KEY, Authorization: `Bearer ${jwt}`, 'Content-Type': 'application/json' },
+      body: JSON.stringify(body || {}),
+    });
+    const ms = performance.now() - t0;
+    if (!r.ok) {
+      let code = `http_${r.status}`;
+      try { const j = await r.json(); if (j.code || j.message) code = j.code || j.message.slice(0, 40); } catch {}
+      return { ok: false, ms, code };
+    }
+    return { ok: true, ms, body: await r.json() };
+  } catch (e) {
+    return { ok: false, ms: performance.now() - t0, code: `net:${(e.message || 'err').slice(0, 30)}` };
+  }
+}
+
+async function svcInsert(table, row) {
+  const r = await fetch(`${BASE}/rest/v1/${table}`, {
+    method: 'POST',
+    headers: { apikey: SERVICE_KEY, Authorization: `Bearer ${SERVICE_KEY}`, 'Content-Type': 'application/json', Prefer: 'return=representation' },
+    body: JSON.stringify(row),
+  });
+  if (!r.ok) throw new Error(`insert ${table}: ${r.status} ${(await r.text()).slice(0, 160)}`);
+  return (await r.json())[0];
+}
+
+// Create auth user -> publisher -> device (publishers.auth_user_id is NOT NULL). Returns a device
+// descriptor with a ready-to-use device JWT. Uses the GoTrue admin API for the auth user.
+async function seedDevice(i) {
+  const email = `load-${i}-${Date.now()}@lumaline.invalid`;
+  const ur = await fetch(`${BASE}/auth/v1/admin/users`, {
+    method: 'POST',
+    headers: { apikey: SERVICE_KEY, Authorization: `Bearer ${SERVICE_KEY}`, 'Content-Type': 'application/json' },
+    body: JSON.stringify({ email, email_confirm: true, password: `Load!${i}!${Math.floor(performance.now())}` }),
+  });
+  if (!ur.ok) throw new Error(`create auth user: ${ur.status} ${(await ur.text()).slice(0, 160)}`);
+  const authUser = await ur.json();
+  const pub = await svcInsert('publishers', { auth_user_id: authUser.id, handle: `load_${i}_${authUser.id.slice(0, 8)}`, status: 'active' });
+  const dev = await svcInsert('devices', { publisher_id: pub.id, label: `load-${i}`, attested: true });
+  return { publisher_id: pub.id, device_id: dev.id, jwt: mintDeviceJWT({ sub: authUser.id, publisher_id: pub.id, device_id: dev.id }) };
+}
+
+const sleep = (ms) => new Promise((res) => setTimeout(res, ms));
+
+// One virtual user: open -> honest beats -> close, repeatedly until the deadline. Records samples.
+async function runVU(device, deadline, samples) {
+  while (performance.now() < deadline) {
+    const open = await rpc('window_open', { p_activity_snapshot: null }, device.jwt);
+    samples.push({ op: 'open', ok: open.ok, ms: open.ms, code: open.code });
+    if (!open.ok || !open.body?.window_id) { await sleep(50); continue; }
+    const { window_id, challenge } = open.body;
+    const chain = buildChain({ windowId: window_id, challenge, count: BEATS });
+    const openedAt = performance.now();
+    for (const beat of chain) {
+      await sleep(BEAT_SPACING_MS);
+      const res = await rpc('window_beat', { p_window_id: window_id, p_seq: beat.seq, p_hmac: beat.hmac, p_activity_delta: beat.activity }, device.jwt);
+      samples.push({ op: 'beat', ok: res.ok, ms: res.ms, code: res.code });
+    }
+    const remaining = DWELL_MS - (performance.now() - openedAt);
+    if (remaining > 0) await sleep(remaining + 20);
+    const close = await rpc('close_window', { p_window_id: window_id }, device.jwt);
+    samples.push({ op: 'close', ok: close.ok, ms: close.ms, code: close.code });
+  }
+}
+
+async function main() {
+  assertSafeTarget(BASE);
+  console.error(`Load harness → ${BASE}  users=${USERS} duration=${DURATION_S}s beats=${BEATS}`);
+  console.error('Seeding synthetic devices…');
+  const devices = [];
+  for (let i = 0; i < USERS; i++) {
+    try { devices.push(await seedDevice(i)); }
+    catch (e) { console.error(`  seed ${i} failed: ${e.message}`); }
+  }
+  if (devices.length === 0) { console.error('FATAL: seeded 0 devices — is the local stack up? (supabase start)'); process.exit(4); }
+  console.error(`Seeded ${devices.length}/${USERS} devices. Running…`);
+
+  const samples = [];
+  const startedAt = performance.now();
+  const deadline = startedAt + DURATION_S * 1000;
+  await Promise.all(devices.map((d) => runVU(d, deadline, samples)));
+  const durationMs = performance.now() - startedAt;
+
+  const report = summarize(samples, durationMs);
+  console.log(renderSummary(report));
+  const out = `/tmp/lumaline-load-${Math.floor(Date.now() / 1000)}.json`;
+  writeFileSync(out, JSON.stringify({ config: { BASE, USERS, DURATION_S, BEATS }, report }, null, 2));
+  console.error(`\nfull report → ${out}`);
+}
+
+main().catch((e) => { console.error('harness error:', e.message); process.exit(1); });

scripts/load/lib/hmac-chain.mjs

@@ -0,0 +1,43 @@
+// Synthetic-client heartbeat HMAC chain — must byte-match public.window_beat's server check
+// (supabase/migrations/20260627025330_window_rpcs.sql). The load harness (M6-T2) drives honest
+// windows, so it has to produce the exact hash-chain the server recomputes, or every beat 400s.
+//
+// Server contract (from the SQL):
+//   msg      = format('%s|%s|%s', seq, prev, activity_delta)
+//   hmac     = encode(hmac(msg, challenge, 'sha256'), 'hex')
+//   prev     = coalesce(prev_hash, window_id::text)   -- prev_hash is the PREVIOUS beat's hmac
+//   seq      = beats_count + 1                          -- 1-based, strictly increasing
+//   spacing  = >=500ms wall-clock between beats (anti-batch) — enforced by the harness scheduler
+//   activity ∈ {none, low, med, high}; at least one non-'none' beat sets activity_progress
+//
+// Pure + dependency-free (node:crypto only) so test/load-harness.test.mjs can pin it hermetically.
+import { createHmac } from 'node:crypto';
+
+export const ACTIVITY_BUCKETS = ['none', 'low', 'med', 'high'];
+
+// One beat's HMAC. prev = previous beat's hmac hex, or the window_id string for seq 1.
+export function beatHmac({ seq, prev, activity, challenge }) {
+  if (!ACTIVITY_BUCKETS.includes(activity)) throw new Error(`bad activity bucket: ${activity}`);
+  const msg = `${seq}|${prev}|${activity}`;
+  return createHmac('sha256', challenge).update(msg).digest('hex');
+}
+
+/**
+ * Build the full chain of `count` beats for a window.
+ * Returns [{ seq, prev, activity, hmac }] where each beat's `prev` is the prior beat's hmac
+ * (or window_id for the first), exactly as the server derives it.
+ * `activityFor(seq)` picks the bucket per beat (default: 'low' on beat 1 so activity_progress
+ * is set, 'none' thereafter — the minimum that credits).
+ */
+export function buildChain({ windowId, challenge, count, activityFor }) {
+  const pick = activityFor || ((seq) => (seq === 1 ? 'low' : 'none'));
+  const beats = [];
+  let prev = String(windowId);
+  for (let seq = 1; seq <= count; seq++) {
+    const activity = pick(seq);
+    const hmac = beatHmac({ seq, prev, activity, challenge });
+    beats.push({ seq, prev, activity, hmac });
+    prev = hmac; // chain head advances to this beat's hmac
+  }
+  return beats;
+}

scripts/load/lib/metrics.mjs

@@ -0,0 +1,76 @@
+// Pure load-test metrics: latency percentiles, throughput, error taxonomy. No I/O — the harness
+// feeds it recorded samples and it renders the summary. Hermetically tested (test/load-harness.test.mjs).
+
+// Nearest-rank percentile over a numeric array (p in [0,100]). Empty -> null.
+export function percentile(values, p) {
+  if (!values || values.length === 0) return null;
+  const sorted = [...values].sort((a, b) => a - b);
+  const rank = Math.ceil((p / 100) * sorted.length);
+  const idx = Math.min(sorted.length - 1, Math.max(0, rank - 1));
+  return sorted[idx];
+}
+
+/**
+ * summarize(samples, durationMs) -> report.
+ * samples: [{ op, ok, ms, code }]   op ∈ {open,beat,close}; code = error class on failure.
+ * Reports overall + per-op: count, ok/err, throughput (ops/s over durationMs), latency
+ * percentiles (over ok samples only), and an error histogram by (op, code).
+ */
+export function summarize(samples, durationMs) {
+  const secs = durationMs > 0 ? durationMs / 1000 : 0;
+  const ops = samples.length;
+  const okSamples = samples.filter((s) => s.ok);
+  const errSamples = samples.filter((s) => !s.ok);
+
+  const perOp = {};
+  for (const op of ['open', 'beat', 'close']) {
+    const forOp = samples.filter((s) => s.op === op);
+    const okOp = forOp.filter((s) => s.ok).map((s) => s.ms);
+    perOp[op] = {
+      count: forOp.length,
+      ok: forOp.filter((s) => s.ok).length,
+      err: forOp.filter((s) => !s.ok).length,
+      throughput_ops_s: secs ? +(forOp.length / secs).toFixed(1) : null,
+      p50_ms: percentile(okOp, 50),
+      p95_ms: percentile(okOp, 95),
+      p99_ms: percentile(okOp, 99),
+      max_ms: okOp.length ? Math.max(...okOp) : null,
+    };
+  }
+
+  const errorHistogram = {};
+  for (const s of errSamples) {
+    const key = `${s.op}:${s.code || 'unknown'}`;
+    errorHistogram[key] = (errorHistogram[key] || 0) + 1;
+  }
+
+  return {
+    duration_ms: durationMs,
+    total_ops: ops,
+    ok: okSamples.length,
+    errors: errSamples.length,
+    error_rate: ops ? +(errSamples.length / ops).toFixed(4) : 0,
+    throughput_ops_s: secs ? +(ops / secs).toFixed(1) : null,
+    writes_per_s: secs ? +(okSamples.length / secs).toFixed(1) : null, // each ok RPC = >=1 DB write
+    per_op: perOp,
+    error_histogram: errorHistogram,
+  };
+}
+
+// Render a summary as a compact text block.
+export function renderSummary(r) {
+  const ms = (x) => (x == null ? '—' : (Math.round(x * 10) / 10) + 'ms');
+  const L = [];
+  L.push(`total ops ${r.total_ops}  ok ${r.ok}  err ${r.errors} (${(r.error_rate * 100).toFixed(2)}%)  over ${(r.duration_ms / 1000).toFixed(1)}s`);
+  L.push(`throughput ${r.throughput_ops_s} ops/s  |  successful writes ${r.writes_per_s}/s`);
+  for (const op of ['open', 'beat', 'close']) {
+    const o = r.per_op[op];
+    L.push(`  ${op.padEnd(5)} n=${o.count} ok=${o.ok} err=${o.err}  p50=${ms(o.p50_ms)} p95=${ms(o.p95_ms)} p99=${ms(o.p99_ms)}`);
+  }
+  const eh = Object.entries(r.error_histogram);
+  if (eh.length) {
+    L.push('  errors:');
+    for (const [k, v] of eh.sort((a, b) => b[1] - a[1])) L.push(`    ${k}: ${v}`);
+  }
+  return L.join('\n');
+}