1stvamp · June 15, 2026 00:17
diff --git a/issue-1-benchmark-plan.md b/issue-1-benchmark-plan.md
diff --git a/issue-1-phase0-results.md b/issue-1-phase0-results.md
diff --git a/issue-1-phase1-results.md b/issue-1-phase1-results.md
diff --git a/phase0.mjs b/phase0.mjs
 // Phase 0 POC harness for the run_head design.
 //  A0: simplest schema (single table, delete-on-terminal, no partitioning).
 //      Measures HOT-update ratio, bloat trend, size-vs-active, op latencies.
 //  A2: CAS correctness under raw concurrent contention (no lock at all),
 //      proving the head-row compare-and-swap serialises transitions by itself.
 //
 // Local-container run: throughput numbers are RELATIVE/shape only, not absolute.
 // HOT ratio, bloat trend, size-tracks-active, and CAS correctness are valid here.

 import { createRequire } from "module";
 const require = createRequire(import.meta.url);
 const { Pool } = require("/home/wes/projects/trigger/tech-test-wes-mason/node_modules/.pnpm/pg@8.16.3/node_modules/pg");

 const CONN = process.env.PG_URL || "postgres://postgres:poc@localhost:55432/poc";
 const MODE = process.env.MODE || "a0";
 const ACTIVE = parseInt(process.env.ACTIVE || "100000", 10);
 const DURATION_S = parseInt(process.env.DURATION_S || "60", 10);
 const CONCURRENCY = parseInt(process.env.CONCURRENCY || "8", 10);
 const FILLFACTOR = parseInt(process.env.FILLFACTOR || "90", 10);
 const SLEEPER_PCT = parseFloat(process.env.SLEEPER_PCT || "0.02");
 const T_TARGET = parseInt(process.env.T_TARGET || "5000", 10); // a2 rounds
 const TMIN = parseInt(process.env.TMIN || "5", 10);   // transitions per run before complete
 const TMAX = parseInt(process.env.TMAX || "20", 10);

 const randInt = (lo, hi) => lo + Math.floor(Math.random() * (hi - lo + 1));
 const pct = (arr, p) => {
  if (arr.length === 0) return 0;
  const s = arr.slice().sort((a, b) => a - b);
  return s[Math.min(s.length - 1, Math.floor((p / 100) * s.length))];
 };
 const fmt = (n) => n.toLocaleString("en-US");

 async function setup(pool) {
  await pool.query("DROP TABLE IF EXISTS run_head");
  await pool.query(`CREATE TABLE run_head (
      run_id text PRIMARY KEY,
      seq bigint NOT NULL DEFAULT 0,
      execution_status text NOT NULL,
      run_status text NOT NULL,
      attempt_number int,
      created_at timestamptz NOT NULL DEFAULT now(),
      updated_at timestamptz NOT NULL DEFAULT now()
    ) WITH (fillfactor = ${FILLFACTOR})`);
  await pool.query("CREATE EXTENSION IF NOT EXISTS pgstattuple");
 }

 async function statTable(pool) {
  const s = await pool.query(
    `SELECT n_tup_ins, n_tup_upd, n_tup_hot_upd, n_tup_del, n_live_tup, n_dead_tup,
            autovacuum_count, vacuum_count
       FROM pg_stat_user_tables WHERE relname = 'run_head'`
  );
  return s.rows[0] || {};
 }

 async function sizeSample(pool) {
  const sz = await pool.query(
    `SELECT pg_table_size('run_head') AS heap,
            pg_indexes_size('run_head') AS idx,
            pg_total_relation_size('run_head') AS total`
  );
  const st = await pool.query(
    `SELECT tuple_count, dead_tuple_count, dead_tuple_percent, free_percent
       FROM pgstattuple('run_head')`
  );
  return { ...sz.rows[0], ...st.rows[0] };
 }

 async function preload(pool, n) {
  const t0 = Date.now();
  const BATCH = 50000;
  for (let off = 0; off < n; off += BATCH) {
    const lim = Math.min(BATCH, n - off);
    await pool.query(
      `INSERT INTO run_head (run_id, seq, execution_status, run_status)
       SELECT 'r' || g, 0, 'RUN_CREATED', 'PENDING'
       FROM generate_series($1::bigint, $2::bigint) g`,
      [off + 1, off + lim]
    );
  }
  return Date.now() - t0;
 }

 async function runA0() {
  const pool = new Pool({ connectionString: CONN, max: CONCURRENCY + 2 });
  await setup(pool);
  console.log(`[A0] active=${fmt(ACTIVE)} concurrency=${CONCURRENCY} fillfactor=${FILLFACTOR} duration=${DURATION_S}s sleepers=${(SLEEPER_PCT * 100).toFixed(0)}%`);
  const loadMs = await preload(pool, ACTIVE);
  console.log(`[A0] preloaded ${fmt(ACTIVE)} rows in ${loadMs}ms`);

  // in-memory active set, sharded across workers; a fraction parked (sleepers)
  const entries = new Array(ACTIVE);
  for (let i = 0; i < ACTIVE; i++) {
    const parked = Math.random() < SLEEPER_PCT;
    entries[i] = { id: "r" + (i + 1), seq: 0, target: parked ? Infinity : randInt(TMIN, TMAX), parked };
  }

  const statBefore = await statTable(pool);
  const samples = [];
  const lat = { upd: [], ins: [], del: [] };
  let ops = { upd: 0, ins: 0, del: 0 };
  const deadline = Date.now() + DURATION_S * 1000;

  const sampler = setInterval(async () => {
    try { samples.push({ t: Math.round((Date.now() - (deadline - DURATION_S * 1000)) / 1000), ...(await sizeSample(pool)) }); } catch {}
  }, 5000);

  async function worker(wid) {
    const client = await pool.connect();
    let ptr = wid;
    try {
      while (Date.now() < deadline) {
        // round-robin across this worker's shard (stride = CONCURRENCY)
        if (ptr >= ACTIVE) ptr = wid;
        const e = entries[ptr];
        ptr += CONCURRENCY;
        if (!e || e.parked) continue;

        let s = process.hrtime.bigint();
        const r = await client.query(
          `UPDATE run_head SET seq = seq + 1, execution_status = 'EXECUTING',
                 run_status = 'EXECUTING', attempt_number = 1, updated_at = now()
            WHERE run_id = $1 AND seq = $2 RETURNING seq`,
          [e.id, e.seq]
        );
        lat.upd.push(Number(process.hrtime.bigint() - s) / 1e6);
        ops.upd++;
        if (r.rowCount === 1) e.seq = Number(r.rows[0].seq);

        if (e.seq >= e.target) {
          s = process.hrtime.bigint();
          await client.query(`DELETE FROM run_head WHERE run_id = $1`, [e.id]);
          lat.del.push(Number(process.hrtime.bigint() - s) / 1e6);
          ops.del++;
          const nid = `w${wid}-${ops.ins}`;
          s = process.hrtime.bigint();
          await client.query(
            `INSERT INTO run_head (run_id, seq, execution_status, run_status)
             VALUES ($1, 0, 'RUN_CREATED', 'PENDING')`,
            [nid]
          );
          lat.ins.push(Number(process.hrtime.bigint() - s) / 1e6);
          ops.ins++;
          entries[ptr - CONCURRENCY] = { id: nid, seq: 0, target: randInt(TMIN, TMAX), parked: false };
        }
      }
    } finally {
      client.release();
    }
  }

  const t0 = Date.now();
  await Promise.all(Array.from({ length: CONCURRENCY }, (_, w) => worker(w)));
  const elapsed = (Date.now() - t0) / 1000;
  clearInterval(sampler);
  const statAfter = await statTable(pool);
  const finalSize = await sizeSample(pool);

  const d = (k) => Number(statAfter[k] || 0) - Number(statBefore[k] || 0);
  const upd = d("n_tup_upd"), hot = d("n_tup_hot_upd");
  const totalOps = ops.upd + ops.ins + ops.del;

  console.log(`\n===== A0 RESULTS (active=${fmt(ACTIVE)}) =====`);
  console.log(`elapsed:            ${elapsed.toFixed(1)}s`);
  console.log(`ops total:          ${fmt(totalOps)}  (${fmt(Math.round(totalOps / elapsed))}/s)`);
  console.log(`  updates:          ${fmt(ops.upd)}  (${fmt(Math.round(ops.upd / elapsed))}/s)`);
  console.log(`  inserts:          ${fmt(ops.ins)}`);
  console.log(`  deletes:          ${fmt(ops.del)}`);
  console.log(`HOT-update ratio:   ${upd > 0 ? ((hot / upd) * 100).toFixed(2) : "n/a"}%   (hot ${fmt(hot)} / upd ${fmt(upd)})  <-- want ~100%`);
  console.log(`update latency ms:  p50 ${pct(lat.upd, 50).toFixed(2)}  p99 ${pct(lat.upd, 99).toFixed(2)}  p999 ${pct(lat.upd, 99.9).toFixed(2)}`);
  console.log(`insert latency ms:  p50 ${pct(lat.ins, 50).toFixed(2)}  p99 ${pct(lat.ins, 99).toFixed(2)}`);
  console.log(`delete latency ms:  p50 ${pct(lat.del, 50).toFixed(2)}  p99 ${pct(lat.del, 99).toFixed(2)}`);
  console.log(`final live tuples:  ${fmt(Number(statAfter.n_live_tup || 0))}   dead tuples: ${fmt(Number(statAfter.n_dead_tup || 0))}`);
  console.log(`autovacuum runs:    ${d("autovacuum_count")} (during test)`);
  console.log(`heap size:          ${(finalSize.heap / 1e6).toFixed(1)} MB   index: ${(finalSize.idx / 1e6).toFixed(1)} MB   total: ${(finalSize.total / 1e6).toFixed(1)} MB`);
  console.log(`pgstattuple:        live tuples ${fmt(Number(finalSize.tuple_count))}  dead% ${Number(finalSize.dead_tuple_percent).toFixed(2)}  free% ${Number(finalSize.free_percent).toFixed(2)}`);
  console.log(`\nsize / bloat over time (heapMB, idxMB, dead%, free%):`);
  for (const s of samples) {
    console.log(`  t=${String(s.t).padStart(3)}s  heap ${(s.heap / 1e6).toFixed(1)}  idx ${(s.idx / 1e6).toFixed(1)}  dead% ${Number(s.dead_tuple_percent).toFixed(2)}  free% ${Number(s.free_percent).toFixed(2)}`);
  }
  await pool.end();
 }

 async function runA2() {
  const pool = new Pool({ connectionString: CONN, max: CONCURRENCY + 2 });
  await setup(pool);
  await pool.query(`INSERT INTO run_head (run_id, seq, execution_status, run_status) VALUES ('cas', 0, 'RUN_CREATED', 'PENDING')`);
  console.log(`[A2] CAS correctness: ${CONCURRENCY} concurrent writers racing one run, target ${fmt(T_TARGET)} transitions, NO lock`);

  const winners = new Map(); // seq -> count, to detect dupes
  let conflicts = 0;
  let done = false;

  async function writer() {
    const client = await pool.connect();
    try {
      while (!done) {
        const cur = await client.query(`SELECT seq FROM run_head WHERE run_id = 'cas'`);
        const seq = Number(cur.rows[0].seq);
        if (seq >= T_TARGET) { done = true; break; }
        const r = await client.query(
          `UPDATE run_head SET seq = seq + 1, updated_at = now()
            WHERE run_id = 'cas' AND seq = $1 RETURNING seq`,
          [seq]
        );
        if (r.rowCount === 1) {
          const ns = Number(r.rows[0].seq);
          winners.set(ns, (winners.get(ns) || 0) + 1);
          if (ns >= T_TARGET) done = true;
        } else {
          conflicts++; // lost the race, retry
        }
      }
    } finally {
      client.release();
    }
  }

  const t0 = Date.now();
  await Promise.all(Array.from({ length: CONCURRENCY }, () => writer()));
  const elapsed = (Date.now() - t0) / 1000;

  const finalSeq = Number((await pool.query(`SELECT seq FROM run_head WHERE run_id = 'cas'`)).rows[0].seq);
  let dupes = 0, gaps = 0;
  for (const [, c] of winners) if (c > 1) dupes++;
  for (let i = 1; i <= T_TARGET; i++) if (!winners.has(i)) gaps++;

  console.log(`\n===== A2 RESULTS =====`);
  console.log(`final seq:           ${fmt(finalSeq)}  (target ${fmt(T_TARGET)})`);
  console.log(`distinct winners:    ${fmt(winners.size)}`);
  console.log(`duplicate seqs:      ${dupes}   <-- MUST be 0 (no forked transition)`);
  console.log(`gaps in 1..target:   ${gaps}   <-- MUST be 0 (no lost transition)`);
  console.log(`conflicts (retries): ${fmt(conflicts)}  (the CAS doing real work under contention)`);
  console.log(`result:              ${finalSeq === T_TARGET && dupes === 0 && gaps === 0 ? "PASS - CAS serialises transitions with no lock" : "FAIL"}`);
  console.log(`elapsed:             ${elapsed.toFixed(1)}s`);
  await pool.end();
 }

 (MODE === "a2" ? runA2() : runA0()).catch((e) => { console.error(e); process.exit(1); });
diff --git a/phase1.mjs b/phase1.mjs
 // Phase 1 + Workstream C: object-storage harness for the snapshot body log.
 // Works against any S3-compatible endpoint (MinIO, SeaweedFS, real AWS S3).
 //   B1  write latency/throughput for small objects across many run prefixes
 //   B2  batching tradeoff (objects-per-PUT) -> request count + wall time
 //   B3  cold reads: prefix LIST ("all states") and StartAfter ("since N") + GET fan-out
 //   B4  read-after-write and list-after-write consistency
 //
 // Local MinIO/SeaweedFS numbers are RELATIVE/shape + correctness only.
 // Absolute latency/throughput needs real S3 (run this same harness with S3 env).

 import { createRequire } from "module";
 const require = createRequire(import.meta.url);
 const SDK = process.env.SDK_PATH || "@aws-sdk/client-s3";
 const { S3Client, PutObjectCommand, GetObjectCommand, ListObjectsV2Command, CreateBucketCommand, HeadBucketCommand } = require(SDK);

 const BACKEND = process.env.BACKEND || "minio";
 const presets = {
  minio:   { endpoint: "http://localhost:9900", ak: "minioadmin", sk: "minioadmin" },
  seaweed: { endpoint: "http://localhost:8333", ak: "any",        sk: "anysecret12345" },
  s3:      { endpoint: process.env.ENDPOINT, ak: process.env.ACCESS_KEY, sk: process.env.SECRET_KEY },
 };
 const cfg = presets[BACKEND];
 const REGION = process.env.REGION || "us-east-1";
 const BUCKET = process.env.BUCKET || "snap-poc";
 const OBJ_SIZE = parseInt(process.env.OBJ_SIZE || "4096", 10);
 const N_OBJECTS = parseInt(process.env.N_OBJECTS || "3000", 10);
 const CONCURRENCY = parseInt(process.env.CONCURRENCY || "32", 10);
 const PREFIXES = parseInt(process.env.PREFIXES || "300", 10);
 const READ_PER_RUN = parseInt(process.env.READ_PER_RUN || "1000", 10);
 const BATCH = parseInt(process.env.BATCH || "20", 10);

 const fmt = (n) => Math.round(n).toLocaleString("en-US");
 const pctl = (arr, p) => { if (!arr.length) return 0; const s = arr.slice().sort((a, b) => a - b); return s[Math.min(s.length - 1, Math.floor((p / 100) * s.length))]; };
 const pad = (n) => String(n).padStart(8, "0");
 const body = Buffer.alloc(OBJ_SIZE, 7);

 const s3 = new S3Client({
  endpoint: cfg.endpoint, region: REGION, forcePathStyle: true,
  credentials: { accessKeyId: cfg.ak, secretAccessKey: cfg.sk },
 });

 async function runPool(items, concurrency, fn) {
  let i = 0; const lat = [];
  async function worker() {
    while (i < items.length) {
      const it = items[i++];
      const s = process.hrtime.bigint();
      await fn(it);
      lat.push(Number(process.hrtime.bigint() - s) / 1e6);
    }
  }
  await Promise.all(Array.from({ length: concurrency }, () => worker()));
  return lat;
 }

 async function ensureBucket() {
  try { await s3.send(new HeadBucketCommand({ Bucket: BUCKET })); }
  catch { try { await s3.send(new CreateBucketCommand({ Bucket: BUCKET })); } catch (e) { if (!/BucketAlreadyOwnedByYou|BucketAlreadyExists/.test(String(e))) throw e; } }
 }

 const put = (Key, Body) => s3.send(new PutObjectCommand({ Bucket: BUCKET, Key, Body }));
 async function get(Key) { const r = await s3.send(new GetObjectCommand({ Bucket: BUCKET, Key })); return (await r.Body.transformToByteArray()).length; }

 async function listAll(prefix, startAfter) {
  let token, pages = 0, keys = 0; const out = [];
  do {
    // EncodingType url avoids list-XML deserialization breaking on special chars
    const r = await s3.send(new ListObjectsV2Command({ Bucket: BUCKET, Prefix: prefix, EncodingType: "url", ContinuationToken: token, StartAfter: token ? undefined : startAfter }));
    pages++; keys += r.KeyCount || 0;
    for (const o of r.Contents || []) out.push(decodeURIComponent(o.Key));
    token = r.IsTruncated ? r.NextContinuationToken : undefined;
  } while (token);
  return { pages, keys, out };
 }

 async function main() {
  console.log(`\n######## BACKEND=${BACKEND} (${cfg.endpoint}) bucket=${BUCKET} objSize=${OBJ_SIZE}B concurrency=${CONCURRENCY} ########`);
  await ensureBucket();

  // B4 consistency
  const probeKey = `cc/${Date.now()}/probe`;
  await put(probeKey, body);
  let raw = "n/a", law = "n/a";
  try { raw = (await get(probeKey)) === OBJ_SIZE ? "OK" : "MISMATCH"; } catch { raw = "FAIL"; }
  try { const l = await listAll(`cc/`); law = l.out.includes(probeKey) ? "OK" : "MISSING"; } catch { law = "FAIL"; }
  console.log(`B4 read-after-write: ${raw}   list-after-write: ${law}`);

  // B1 write
  const keys = Array.from({ length: N_OBJECTS }, (_, i) => `runs/run${i % PREFIXES}/${pad(Math.floor(i / PREFIXES))}`);
  const t0 = Date.now();
  const wlat = await runPool(keys, CONCURRENCY, (k) => put(k, body));
  const welapsed = (Date.now() - t0) / 1000;
  console.log(`B1 write: ${fmt(N_OBJECTS)} objs in ${welapsed.toFixed(1)}s = ${fmt(N_OBJECTS / welapsed)} PUT/s`);
  console.log(`   PUT latency ms: p50 ${pctl(wlat, 50).toFixed(1)}  p99 ${pctl(wlat, 99).toFixed(1)}  p999 ${pctl(wlat, 99.9).toFixed(1)}`);

  // B2 batching: same logical snapshots, one-per-object vs BATCH-per-object
  const G = Math.min(2000, N_OBJECTS);
  const unbatchedKeys = Array.from({ length: G }, (_, i) => `b2/unbatched/${pad(i)}`);
  let s = Date.now();
  await runPool(unbatchedKeys, CONCURRENCY, (k) => put(k, body));
  const ub = (Date.now() - s) / 1000;
  const batchedKeys = Array.from({ length: Math.ceil(G / BATCH) }, (_, i) => `b2/batched/${pad(i)}`);
  const bigBody = Buffer.alloc(OBJ_SIZE * BATCH, 7);
  s = Date.now();
  await runPool(batchedKeys, CONCURRENCY, (k) => put(k, bigBody));
  const bt = (Date.now() - s) / 1000;
  console.log(`B2 batching (${fmt(G)} snapshots): unbatched ${fmt(G)} PUTs in ${ub.toFixed(1)}s  vs  batched ${fmt(batchedKeys.length)} PUTs in ${bt.toFixed(1)}s  (${BATCH}:1 -> ${(G / batchedKeys.length).toFixed(0)}x fewer requests, ${(ub / bt).toFixed(1)}x faster wall)`);

  // B3 cold reads: preload one run prefix with READ_PER_RUN objects
  const rprefix = `read-run/`;
  await runPool(Array.from({ length: READ_PER_RUN }, (_, i) => `${rprefix}${pad(i)}`), CONCURRENCY, (k) => put(k, body));
  s = Date.now(); const all = await listAll(rprefix); const listAllMs = Date.now() - s;
  const sinceFrom = `${rprefix}${pad(Math.floor(READ_PER_RUN / 2))}`;
  s = Date.now(); const since = await listAll(rprefix, sinceFrom); const sinceMs = Date.now() - s;
  s = Date.now(); await runPool(all.out, CONCURRENCY, (k) => get(k)); const getMs = Date.now() - s;
  console.log(`B3 cold reads (run with ${fmt(READ_PER_RUN)} states):`);
  console.log(`   all states: LIST ${all.keys} keys in ${all.pages} page(s), ${listAllMs}ms   |   since N/2: ${since.keys} keys in ${sinceMs}ms   |   GET fan-out of all: ${getMs}ms (${fmt(all.out.length / (getMs / 1000))} GET/s)`);
 }
 main().catch((e) => { console.error("ERROR", e?.message || e); process.exit(1); });
diff --git a/soakD.mjs b/soakD.mjs
 // Workstream D: Postgres long-soak harness for the run_head design.
 // Decides the partitioning-vs-simple question by watching long-horizon dynamics
 // the short A0 runs can't: autovacuum keep-up, PK-index bloat trajectory,
 // xid-freeze age, and long-open-transaction (xmin) holdback.
 //
 // Rate, not hardware, is what makes bloat/vacuum dynamics extrapolate, so this
 // decouples from the laptop fsync ceiling via synchronous_commit=off + group commit.
 // Emits a CSV time series to stdout (redirect to a file and plot).
 //
 // Env:
 //   ACTIVE=1000000  DURATION_S=3600  CONCURRENCY=16  FILLFACTOR=90
 //   SYNC_COMMIT=off  COMMIT_BATCH=10  TMIN=5 TMAX=20  SLEEPER_PCT=0.02
 //   SAMPLE_S=30  REINDEX_EVERY_S=0  HOLD_TXN=0  (1 = open a long idle txn to test xmin holdback)
 //   AV_RELOPTS="autovacuum_vacuum_scale_factor=0.02,autovacuum_vacuum_cost_limit=2000"

 import { createRequire } from "module";
 const require = createRequire(import.meta.url);
 const { Pool, Client } = require("/home/wes/projects/trigger/tech-test-wes-mason/node_modules/.pnpm/pg@8.16.3/node_modules/pg");

 const CONN = process.env.PG_URL || "postgres://postgres:poc@localhost:55432/poc";
 const ACTIVE = parseInt(process.env.ACTIVE || "1000000", 10);
 const DURATION_S = parseInt(process.env.DURATION_S || "3600", 10);
 const CONCURRENCY = parseInt(process.env.CONCURRENCY || "16", 10);
 const FILLFACTOR = parseInt(process.env.FILLFACTOR || "90", 10);
 const SYNC_COMMIT = (process.env.SYNC_COMMIT || "off").toLowerCase();
 const COMMIT_BATCH = parseInt(process.env.COMMIT_BATCH || "10", 10);
 const TMIN = parseInt(process.env.TMIN || "5", 10);
 const TMAX = parseInt(process.env.TMAX || "20", 10);
 const SLEEPER_PCT = parseFloat(process.env.SLEEPER_PCT || "0.02");
 const SAMPLE_S = parseInt(process.env.SAMPLE_S || "30", 10);
 const REINDEX_EVERY_S = parseInt(process.env.REINDEX_EVERY_S || "0", 10);
 const HOLD_TXN = process.env.HOLD_TXN === "1";
 const AV_RELOPTS = process.env.AV_RELOPTS || "";

 const randInt = (lo, hi) => lo + Math.floor(Math.random() * (hi - lo + 1));
 const fmt = (n) => Math.round(n).toLocaleString("en-US");
 const num = (v) => Number(v || 0);

 async function setup(pool) {
  await pool.query("DROP TABLE IF EXISTS run_head");
  const reloptions = [`fillfactor = ${FILLFACTOR}`, ...(AV_RELOPTS ? AV_RELOPTS.split(",").map((s) => s.trim()) : [])].join(", ");
  await pool.query(`CREATE TABLE run_head (
      run_id text PRIMARY KEY,
      seq bigint NOT NULL DEFAULT 0,
      execution_status text NOT NULL,
      run_status text NOT NULL,
      attempt_number int,
      created_at timestamptz NOT NULL DEFAULT now(),
      updated_at timestamptz NOT NULL DEFAULT now()
    ) WITH (${reloptions})`);
  await pool.query("CREATE EXTENSION IF NOT EXISTS pgstattuple");
  const BATCH = 100000;
  for (let off = 0; off < ACTIVE; off += BATCH) {
    const lim = Math.min(BATCH, ACTIVE - off);
    await pool.query(
      `INSERT INTO run_head (run_id, seq, execution_status, run_status)
       SELECT 'r' || g, 0, 'RUN_CREATED', 'PENDING' FROM generate_series($1::bigint, $2::bigint) g`,
      [off + 1, off + lim]
    );
  }
 }

 async function sample(pool) {
  const st = (await pool.query(
    `SELECT n_tup_upd, n_tup_hot_upd, n_dead_tup, n_live_tup, autovacuum_count
       FROM pg_stat_user_tables WHERE relname='run_head'`
  )).rows[0] || {};
  const sz = (await pool.query(
    `SELECT pg_table_size('run_head') heap, pg_indexes_size('run_head') idx`
  )).rows[0] || {};
  const tt = (await pool.query(
    `SELECT dead_tuple_percent dead, approx_free_percent free FROM pgstattuple_approx('run_head')`
  )).rows[0] || {};
  let idx = {};
  try { idx = (await pool.query(`SELECT avg_leaf_density, leaf_fragmentation FROM pgstatindex('run_head_pkey')`)).rows[0] || {}; } catch {}
  const age = (await pool.query(`SELECT age(relfrozenxid) a FROM pg_class WHERE relname='run_head'`)).rows[0] || {};
  return { ...st, ...sz, ...tt, ...idx, ...age };
 }

 async function main() {
  const pool = new Pool({ connectionString: CONN, max: CONCURRENCY + 2 });
  console.error(`[D] active=${fmt(ACTIVE)} dur=${DURATION_S}s conc=${CONCURRENCY} ff=${FILLFACTOR} sync_commit=${SYNC_COMMIT} commit_batch=${COMMIT_BATCH} reindex_every=${REINDEX_EVERY_S}s hold_txn=${HOLD_TXN} relopts="${AV_RELOPTS}"`);
  await setup(pool);
  console.error(`[D] preloaded ${fmt(ACTIVE)} rows`);

  // Optional: hold a long-open idle transaction to demonstrate xmin holdback.
  let holder;
  if (HOLD_TXN) {
    holder = new Client({ connectionString: CONN });
    await holder.connect();
    await holder.query("BEGIN");
    await holder.query("SELECT 1"); // pins a snapshot/xmin for the whole run
    console.error(`[D] HOLD_TXN: a long-open transaction is pinning xmin (vacuum should not be able to reclaim dead tuples created after now)`);
  }

  const entries = new Array(ACTIVE);
  for (let i = 0; i < ACTIVE; i++) {
    const parked = Math.random() < SLEEPER_PCT;
    entries[i] = { id: "r" + (i + 1), seq: 0, target: parked ? Infinity : randInt(TMIN, TMAX), parked };
  }

  const ops = { upd: 0, ins: 0, del: 0 };
  const start = Date.now();
  const deadline = start + DURATION_S * 1000;
  let lastUpd = 0, lastHot = 0, lastSampleOps = 0, lastT = 0;

  console.log("t_s,ops,upd_per_s,hot_ratio_window_pct,dead_pct,free_pct,heap_mb,idx_mb,idx_leaf_density,idx_fragmentation,n_dead_tup,autovac_count,xid_age");
  const sampler = setInterval(async () => {
    try {
      const t = Math.round((Date.now() - start) / 1000);
      const s = await sample(pool);
      const dUpd = num(s.n_tup_upd) - lastUpd, dHot = num(s.n_tup_hot_upd) - lastHot;
      const hotW = dUpd > 0 ? (dHot / dUpd) * 100 : 0;
      const upsPerS = lastT > 0 ? (ops.upd - lastSampleOps) / (t - lastT) : ops.upd / Math.max(1, t);
      lastUpd = num(s.n_tup_upd); lastHot = num(s.n_tup_hot_upd); lastSampleOps = ops.upd; lastT = t;
      console.log([
        t, ops.upd, Math.round(upsPerS), hotW.toFixed(1),
        num(s.dead).toFixed(2), num(s.free).toFixed(2),
        (num(s.heap) / 1e6).toFixed(1), (num(s.idx) / 1e6).toFixed(1),
        s.avg_leaf_density != null ? num(s.avg_leaf_density).toFixed(1) : "",
        s.leaf_fragmentation != null ? num(s.leaf_fragmentation).toFixed(1) : "",
        num(s.n_dead_tup), num(s.autovacuum_count), num(s.a),
      ].join(","));
    } catch (e) { console.error("sample error", e.message); }
  }, SAMPLE_S * 1000);

  let reindexer;
  if (REINDEX_EVERY_S > 0) {
    reindexer = setInterval(async () => {
      try {
        const before = num((await pool.query(`SELECT pg_indexes_size('run_head') i`)).rows[0].i);
        const t0 = Date.now();
        await pool.query("REINDEX INDEX CONCURRENTLY run_head_pkey");
        const after = num((await pool.query(`SELECT pg_indexes_size('run_head') i`)).rows[0].i);
        console.error(`[D] REINDEX pkey: ${(before / 1e6).toFixed(1)}MB -> ${(after / 1e6).toFixed(1)}MB in ${Date.now() - t0}ms`);
      } catch (e) { console.error("reindex error", e.message); }
    }, REINDEX_EVERY_S * 1000);
  }

  const updSQL = `UPDATE run_head SET seq=seq+1, execution_status='EXECUTING', run_status='EXECUTING', attempt_number=1, updated_at=now() WHERE run_id=$1 AND seq=$2 RETURNING seq`;

  async function worker(wid) {
    const client = await pool.connect();
    await client.query(`SET synchronous_commit = ${SYNC_COMMIT}`);
    let ptr = wid;
    try {
      while (Date.now() < deadline) {
        await client.query("BEGIN");
        for (let b = 0; b < COMMIT_BATCH && Date.now() < deadline; b++) {
          if (ptr >= ACTIVE) ptr = wid;
          const e = entries[ptr]; ptr += CONCURRENCY;
          if (!e || e.parked) continue;
          const r = await client.query(updSQL, [e.id, e.seq]);
          ops.upd++;
          if (r.rowCount === 1) e.seq = Number(r.rows[0].seq);
          if (e.seq >= e.target) {
            await client.query(`DELETE FROM run_head WHERE run_id=$1`, [e.id]); ops.del++;
            const nid = `w${wid}-${ops.ins}`;
            await client.query(`INSERT INTO run_head (run_id, seq, execution_status, run_status) VALUES ($1,0,'RUN_CREATED','PENDING')`, [nid]); ops.ins++;
            entries[ptr - CONCURRENCY] = { id: nid, seq: 0, target: randInt(TMIN, TMAX), parked: false };
          }
        }
        await client.query("COMMIT");
      }
    } finally { try { await client.query("ROLLBACK"); } catch {} client.release(); }
  }

  await Promise.all(Array.from({ length: CONCURRENCY }, (_, w) => worker(w)));
  clearInterval(sampler); if (reindexer) clearInterval(reindexer);
  const final = await sample(pool);
  const elapsed = (Date.now() - start) / 1000;
  if (holder) { await holder.query("ROLLBACK").catch(() => {}); await holder.end().catch(() => {}); }

  console.error(`\n===== D SUMMARY =====`);
  console.error(`elapsed:          ${elapsed.toFixed(0)}s`);
  console.error(`ops:              upd ${fmt(ops.upd)} (${fmt(ops.upd / elapsed)}/s)  ins ${fmt(ops.ins)}  del ${fmt(ops.del)}`);
  console.error(`final dead%:      ${num(final.dead).toFixed(2)}   free%: ${num(final.free).toFixed(2)}   n_dead_tup: ${fmt(num(final.n_dead_tup))}`);
  console.error(`final heap:       ${(num(final.heap) / 1e6).toFixed(1)} MB   index: ${(num(final.idx) / 1e6).toFixed(1)} MB`);
  console.error(`index leaf:       density ${final.avg_leaf_density != null ? num(final.avg_leaf_density).toFixed(1) : "?"}%   fragmentation ${final.leaf_fragmentation != null ? num(final.leaf_fragmentation).toFixed(1) : "?"}%`);
  console.error(`autovacuum runs:  ${num(final.autovacuum_count)}    final xid age: ${fmt(num(final.a))}`);
  console.error(HOLD_TXN
    ? `xmin test:        with a long-open txn held, dead tuples should be UNRECLAIMABLE (watch n_dead_tup climb without bound).`
    : `xmin test:        no long-open txn; sleepers are old ROWS only, which do NOT block vacuum.`);
  await pool.end();
 }
 main().catch((e) => { console.error(e); process.exit(1); });
metric	result
final seq	5,000 (= target)
distinct winners	5,000
duplicate seqs	0 (no forked transition)
gaps in 1..5000	0 (no lost transition)
conflicts / retries	42,363
verdict	PASS, the head-row CAS serialises transitions with no lock
scenario	HOT ratio	heap (start → end)	index (start → end)	dead% range	autovacuums	size/row
100k active, ff90, 60s	96.85%	9.5 → 9.7 MB (flat)	3.9 MB (flat)	0.5-1.6%	1	~136 B
1M active, ff90, 60s	90.97%	94.2 → 94.7 MB (flat)	39.0 MB (flat)	0.4%	0	~134 B
10k active, ff90, churn 120s	99.46%	1.0 → 1.4 MB (flat)	0.4 → 0.9 MB	2-8% sawtooth	10	~140 B
100k active, ff70, 60s	98.32%	12.4 MB (flat)	3.9 MB (flat)	0.4-1.5%	1	~164 B
check	MinIO	SeaweedFS	why it matters
read-after-write	OK	OK	the hot path doesn't depend on it, but cold reads do
list-after-write	OK	OK	"all states" / "since N" rely on strongly-consistent listing; both self-host backends provided it
batching 20:1	20x fewer PUTs, ~16x faster wall	20x fewer PUTs, ~19x faster wall	the PUT-cost lever, confirmed
cold "all states" (1,000-state run)	1 LIST page, ~240 ms	1 LIST page, ~170 ms	prefix listing is cheap at this size
cold "since N/2" (StartAfter)	499 keys, ~113 ms	499 keys, ~97 ms	the cursor read returns only the tail
GET fan-out of 1,000 objects	~2.7 s (~376 GET/s)	~2.3 s (~433 GET/s)	this is the cold-read cost, hence debug-only
	// Phase 0 POC harness for the run_head design.
	// A0: simplest schema (single table, delete-on-terminal, no partitioning).
	// Measures HOT-update ratio, bloat trend, size-vs-active, op latencies.
	// A2: CAS correctness under raw concurrent contention (no lock at all),
	// proving the head-row compare-and-swap serialises transitions by itself.
	//
	// Local-container run: throughput numbers are RELATIVE/shape only, not absolute.
	// HOT ratio, bloat trend, size-tracks-active, and CAS correctness are valid here.

	import { createRequire } from "module";
	const require = createRequire(import.meta.url);
	const { Pool } = require("/home/wes/projects/trigger/tech-test-wes-mason/node_modules/.pnpm/pg@8.16.3/node_modules/pg");

	const CONN = process.env.PG_URL \|\| "postgres://postgres:poc@localhost:55432/poc";
	const MODE = process.env.MODE \|\| "a0";
	const ACTIVE = parseInt(process.env.ACTIVE \|\| "100000", 10);
	const DURATION_S = parseInt(process.env.DURATION_S \|\| "60", 10);
	const CONCURRENCY = parseInt(process.env.CONCURRENCY \|\| "8", 10);
	const FILLFACTOR = parseInt(process.env.FILLFACTOR \|\| "90", 10);
	const SLEEPER_PCT = parseFloat(process.env.SLEEPER_PCT \|\| "0.02");
	const T_TARGET = parseInt(process.env.T_TARGET \|\| "5000", 10); // a2 rounds
	const TMIN = parseInt(process.env.TMIN \|\| "5", 10); // transitions per run before complete
	const TMAX = parseInt(process.env.TMAX \|\| "20", 10);

	const randInt = (lo, hi) => lo + Math.floor(Math.random() * (hi - lo + 1));
	const pct = (arr, p) => {
	if (arr.length === 0) return 0;
	const s = arr.slice().sort((a, b) => a - b);
	return s[Math.min(s.length - 1, Math.floor((p / 100) * s.length))];
	};
	const fmt = (n) => n.toLocaleString("en-US");

	async function setup(pool) {
	await pool.query("DROP TABLE IF EXISTS run_head");
	await pool.query(`CREATE TABLE run_head (
	run_id text PRIMARY KEY,
	seq bigint NOT NULL DEFAULT 0,
	execution_status text NOT NULL,
	run_status text NOT NULL,
	attempt_number int,
	created_at timestamptz NOT NULL DEFAULT now(),
	updated_at timestamptz NOT NULL DEFAULT now()
	) WITH (fillfactor = ${FILLFACTOR})`);
	await pool.query("CREATE EXTENSION IF NOT EXISTS pgstattuple");
	}

	async function statTable(pool) {
	const s = await pool.query(
	`SELECT n_tup_ins, n_tup_upd, n_tup_hot_upd, n_tup_del, n_live_tup, n_dead_tup,
	autovacuum_count, vacuum_count
	FROM pg_stat_user_tables WHERE relname = 'run_head'`
	);
	return s.rows[0] \|\| {};
	}

	async function sizeSample(pool) {
	const sz = await pool.query(
	`SELECT pg_table_size('run_head') AS heap,
	pg_indexes_size('run_head') AS idx,
	pg_total_relation_size('run_head') AS total`
	);
	const st = await pool.query(
	`SELECT tuple_count, dead_tuple_count, dead_tuple_percent, free_percent
	FROM pgstattuple('run_head')`
	);
	return { ...sz.rows[0], ...st.rows[0] };
	}

	async function preload(pool, n) {
	const t0 = Date.now();
	const BATCH = 50000;
	for (let off = 0; off < n; off += BATCH) {
	const lim = Math.min(BATCH, n - off);
	await pool.query(
	`INSERT INTO run_head (run_id, seq, execution_status, run_status)
	SELECT 'r' \|\| g, 0, 'RUN_CREATED', 'PENDING'
	FROM generate_series($1::bigint, $2::bigint) g`,
	[off + 1, off + lim]
	);
	}
	return Date.now() - t0;
	}

	async function runA0() {
	const pool = new Pool({ connectionString: CONN, max: CONCURRENCY + 2 });
	await setup(pool);
	console.log(`[A0] active=${fmt(ACTIVE)} concurrency=${CONCURRENCY} fillfactor=${FILLFACTOR} duration=${DURATION_S}s sleepers=${(SLEEPER_PCT * 100).toFixed(0)}%`);
	const loadMs = await preload(pool, ACTIVE);
	console.log(`[A0] preloaded ${fmt(ACTIVE)} rows in ${loadMs}ms`);

	// in-memory active set, sharded across workers; a fraction parked (sleepers)
	const entries = new Array(ACTIVE);
	for (let i = 0; i < ACTIVE; i++) {
	const parked = Math.random() < SLEEPER_PCT;
	entries[i] = { id: "r" + (i + 1), seq: 0, target: parked ? Infinity : randInt(TMIN, TMAX), parked };
	}

	const statBefore = await statTable(pool);
	const samples = [];
	const lat = { upd: [], ins: [], del: [] };
	let ops = { upd: 0, ins: 0, del: 0 };
	const deadline = Date.now() + DURATION_S * 1000;

	const sampler = setInterval(async () => {
	try { samples.push({ t: Math.round((Date.now() - (deadline - DURATION_S * 1000)) / 1000), ...(await sizeSample(pool)) }); } catch {}
	}, 5000);

	async function worker(wid) {
	const client = await pool.connect();
	let ptr = wid;
	try {
	while (Date.now() < deadline) {
	// round-robin across this worker's shard (stride = CONCURRENCY)
	if (ptr >= ACTIVE) ptr = wid;
	const e = entries[ptr];
	ptr += CONCURRENCY;
	if (!e \|\| e.parked) continue;

	let s = process.hrtime.bigint();
	const r = await client.query(
	`UPDATE run_head SET seq = seq + 1, execution_status = 'EXECUTING',
	run_status = 'EXECUTING', attempt_number = 1, updated_at = now()
	WHERE run_id = $1 AND seq = $2 RETURNING seq`,
	[e.id, e.seq]
	);
	lat.upd.push(Number(process.hrtime.bigint() - s) / 1e6);
	ops.upd++;
	if (r.rowCount === 1) e.seq = Number(r.rows[0].seq);

	if (e.seq >= e.target) {
	s = process.hrtime.bigint();
	await client.query(`DELETE FROM run_head WHERE run_id = $1`, [e.id]);
	lat.del.push(Number(process.hrtime.bigint() - s) / 1e6);
	ops.del++;
	const nid = `w${wid}-${ops.ins}`;
	s = process.hrtime.bigint();
	await client.query(
	`INSERT INTO run_head (run_id, seq, execution_status, run_status)
	VALUES ($1, 0, 'RUN_CREATED', 'PENDING')`,
	[nid]
	);
	lat.ins.push(Number(process.hrtime.bigint() - s) / 1e6);
	ops.ins++;
	entries[ptr - CONCURRENCY] = { id: nid, seq: 0, target: randInt(TMIN, TMAX), parked: false };
	}
	}
	} finally {
	client.release();
	}
	}

	const t0 = Date.now();
	await Promise.all(Array.from({ length: CONCURRENCY }, (_, w) => worker(w)));
	const elapsed = (Date.now() - t0) / 1000;
	clearInterval(sampler);
	const statAfter = await statTable(pool);
	const finalSize = await sizeSample(pool);

	const d = (k) => Number(statAfter[k] \|\| 0) - Number(statBefore[k] \|\| 0);
	const upd = d("n_tup_upd"), hot = d("n_tup_hot_upd");
	const totalOps = ops.upd + ops.ins + ops.del;

	console.log(`\n===== A0 RESULTS (active=${fmt(ACTIVE)}) =====`);
	console.log(`elapsed: ${elapsed.toFixed(1)}s`);
	console.log(`ops total: ${fmt(totalOps)} (${fmt(Math.round(totalOps / elapsed))}/s)`);
	console.log(` updates: ${fmt(ops.upd)} (${fmt(Math.round(ops.upd / elapsed))}/s)`);
	console.log(` inserts: ${fmt(ops.ins)}`);
	console.log(` deletes: ${fmt(ops.del)}`);
	console.log(`HOT-update ratio: ${upd > 0 ? ((hot / upd) * 100).toFixed(2) : "n/a"}% (hot ${fmt(hot)} / upd ${fmt(upd)}) <-- want ~100%`);
	console.log(`update latency ms: p50 ${pct(lat.upd, 50).toFixed(2)} p99 ${pct(lat.upd, 99).toFixed(2)} p999 ${pct(lat.upd, 99.9).toFixed(2)}`);
	console.log(`insert latency ms: p50 ${pct(lat.ins, 50).toFixed(2)} p99 ${pct(lat.ins, 99).toFixed(2)}`);
	console.log(`delete latency ms: p50 ${pct(lat.del, 50).toFixed(2)} p99 ${pct(lat.del, 99).toFixed(2)}`);
	console.log(`final live tuples: ${fmt(Number(statAfter.n_live_tup \|\| 0))} dead tuples: ${fmt(Number(statAfter.n_dead_tup \|\| 0))}`);
	console.log(`autovacuum runs: ${d("autovacuum_count")} (during test)`);
	console.log(`heap size: ${(finalSize.heap / 1e6).toFixed(1)} MB index: ${(finalSize.idx / 1e6).toFixed(1)} MB total: ${(finalSize.total / 1e6).toFixed(1)} MB`);
	console.log(`pgstattuple: live tuples ${fmt(Number(finalSize.tuple_count))} dead% ${Number(finalSize.dead_tuple_percent).toFixed(2)} free% ${Number(finalSize.free_percent).toFixed(2)}`);
	console.log(`\nsize / bloat over time (heapMB, idxMB, dead%, free%):`);
	for (const s of samples) {
	console.log(` t=${String(s.t).padStart(3)}s heap ${(s.heap / 1e6).toFixed(1)} idx ${(s.idx / 1e6).toFixed(1)} dead% ${Number(s.dead_tuple_percent).toFixed(2)} free% ${Number(s.free_percent).toFixed(2)}`);
	}
	await pool.end();
	}

	async function runA2() {
	const pool = new Pool({ connectionString: CONN, max: CONCURRENCY + 2 });
	await setup(pool);
	await pool.query(`INSERT INTO run_head (run_id, seq, execution_status, run_status) VALUES ('cas', 0, 'RUN_CREATED', 'PENDING')`);
	console.log(`[A2] CAS correctness: ${CONCURRENCY} concurrent writers racing one run, target ${fmt(T_TARGET)} transitions, NO lock`);

	const winners = new Map(); // seq -> count, to detect dupes
	let conflicts = 0;
	let done = false;

	async function writer() {
	const client = await pool.connect();
	try {
	while (!done) {
	const cur = await client.query(`SELECT seq FROM run_head WHERE run_id = 'cas'`);
	const seq = Number(cur.rows[0].seq);
	if (seq >= T_TARGET) { done = true; break; }
	const r = await client.query(
	`UPDATE run_head SET seq = seq + 1, updated_at = now()
	WHERE run_id = 'cas' AND seq = $1 RETURNING seq`,
	[seq]
	);
	if (r.rowCount === 1) {
	const ns = Number(r.rows[0].seq);
	winners.set(ns, (winners.get(ns) \|\| 0) + 1);
	if (ns >= T_TARGET) done = true;
	} else {
	conflicts++; // lost the race, retry
	}
	}
	} finally {
	client.release();
	}
	}

	const t0 = Date.now();
	await Promise.all(Array.from({ length: CONCURRENCY }, () => writer()));
	const elapsed = (Date.now() - t0) / 1000;

	const finalSeq = Number((await pool.query(`SELECT seq FROM run_head WHERE run_id = 'cas'`)).rows[0].seq);
	let dupes = 0, gaps = 0;
	for (const [, c] of winners) if (c > 1) dupes++;
	for (let i = 1; i <= T_TARGET; i++) if (!winners.has(i)) gaps++;

	console.log(`\n===== A2 RESULTS =====`);
	console.log(`final seq: ${fmt(finalSeq)} (target ${fmt(T_TARGET)})`);
	console.log(`distinct winners: ${fmt(winners.size)}`);
	console.log(`duplicate seqs: ${dupes} <-- MUST be 0 (no forked transition)`);
	console.log(`gaps in 1..target: ${gaps} <-- MUST be 0 (no lost transition)`);
	console.log(`conflicts (retries): ${fmt(conflicts)} (the CAS doing real work under contention)`);
	console.log(`result: ${finalSeq === T_TARGET && dupes === 0 && gaps === 0 ? "PASS - CAS serialises transitions with no lock" : "FAIL"}`);
	console.log(`elapsed: ${elapsed.toFixed(1)}s`);
	await pool.end();
	}

	(MODE === "a2" ? runA2() : runA0()).catch((e) => { console.error(e); process.exit(1); });
	// Phase 1 + Workstream C: object-storage harness for the snapshot body log.
	// Works against any S3-compatible endpoint (MinIO, SeaweedFS, real AWS S3).
	// B1 write latency/throughput for small objects across many run prefixes
	// B2 batching tradeoff (objects-per-PUT) -> request count + wall time
	// B3 cold reads: prefix LIST ("all states") and StartAfter ("since N") + GET fan-out
	// B4 read-after-write and list-after-write consistency
	//
	// Local MinIO/SeaweedFS numbers are RELATIVE/shape + correctness only.
	// Absolute latency/throughput needs real S3 (run this same harness with S3 env).

	import { createRequire } from "module";
	const require = createRequire(import.meta.url);
	const SDK = process.env.SDK_PATH \|\| "@aws-sdk/client-s3";
	const { S3Client, PutObjectCommand, GetObjectCommand, ListObjectsV2Command, CreateBucketCommand, HeadBucketCommand } = require(SDK);

	const BACKEND = process.env.BACKEND \|\| "minio";
	const presets = {
	minio: { endpoint: "http://localhost:9900", ak: "minioadmin", sk: "minioadmin" },
	seaweed: { endpoint: "http://localhost:8333", ak: "any", sk: "anysecret12345" },
	s3: { endpoint: process.env.ENDPOINT, ak: process.env.ACCESS_KEY, sk: process.env.SECRET_KEY },
	};
	const cfg = presets[BACKEND];
	const REGION = process.env.REGION \|\| "us-east-1";
	const BUCKET = process.env.BUCKET \|\| "snap-poc";
	const OBJ_SIZE = parseInt(process.env.OBJ_SIZE \|\| "4096", 10);
	const N_OBJECTS = parseInt(process.env.N_OBJECTS \|\| "3000", 10);
	const CONCURRENCY = parseInt(process.env.CONCURRENCY \|\| "32", 10);
	const PREFIXES = parseInt(process.env.PREFIXES \|\| "300", 10);
	const READ_PER_RUN = parseInt(process.env.READ_PER_RUN \|\| "1000", 10);
	const BATCH = parseInt(process.env.BATCH \|\| "20", 10);

	const fmt = (n) => Math.round(n).toLocaleString("en-US");
	const pctl = (arr, p) => { if (!arr.length) return 0; const s = arr.slice().sort((a, b) => a - b); return s[Math.min(s.length - 1, Math.floor((p / 100) * s.length))]; };
	const pad = (n) => String(n).padStart(8, "0");
	const body = Buffer.alloc(OBJ_SIZE, 7);

	const s3 = new S3Client({
	endpoint: cfg.endpoint, region: REGION, forcePathStyle: true,
	credentials: { accessKeyId: cfg.ak, secretAccessKey: cfg.sk },
	});

	async function runPool(items, concurrency, fn) {
	let i = 0; const lat = [];
	async function worker() {
	while (i < items.length) {
	const it = items[i++];
	const s = process.hrtime.bigint();
	await fn(it);
	lat.push(Number(process.hrtime.bigint() - s) / 1e6);
	}
	}
	await Promise.all(Array.from({ length: concurrency }, () => worker()));
	return lat;
	}

	async function ensureBucket() {
	try { await s3.send(new HeadBucketCommand({ Bucket: BUCKET })); }
	catch { try { await s3.send(new CreateBucketCommand({ Bucket: BUCKET })); } catch (e) { if (!/BucketAlreadyOwnedByYou\|BucketAlreadyExists/.test(String(e))) throw e; } }
	}

	const put = (Key, Body) => s3.send(new PutObjectCommand({ Bucket: BUCKET, Key, Body }));
	async function get(Key) { const r = await s3.send(new GetObjectCommand({ Bucket: BUCKET, Key })); return (await r.Body.transformToByteArray()).length; }

	async function listAll(prefix, startAfter) {
	let token, pages = 0, keys = 0; const out = [];
	do {
	// EncodingType url avoids list-XML deserialization breaking on special chars
	const r = await s3.send(new ListObjectsV2Command({ Bucket: BUCKET, Prefix: prefix, EncodingType: "url", ContinuationToken: token, StartAfter: token ? undefined : startAfter }));
	pages++; keys += r.KeyCount \|\| 0;
	for (const o of r.Contents \|\| []) out.push(decodeURIComponent(o.Key));
	token = r.IsTruncated ? r.NextContinuationToken : undefined;
	} while (token);
	return { pages, keys, out };
	}

	async function main() {
	console.log(`\n######## BACKEND=${BACKEND} (${cfg.endpoint}) bucket=${BUCKET} objSize=${OBJ_SIZE}B concurrency=${CONCURRENCY} ########`);
	await ensureBucket();

	// B4 consistency
	const probeKey = `cc/${Date.now()}/probe`;
	await put(probeKey, body);
	let raw = "n/a", law = "n/a";
	try { raw = (await get(probeKey)) === OBJ_SIZE ? "OK" : "MISMATCH"; } catch { raw = "FAIL"; }
	try { const l = await listAll(`cc/`); law = l.out.includes(probeKey) ? "OK" : "MISSING"; } catch { law = "FAIL"; }
	console.log(`B4 read-after-write: ${raw} list-after-write: ${law}`);

	// B1 write
	const keys = Array.from({ length: N_OBJECTS }, (_, i) => `runs/run${i % PREFIXES}/${pad(Math.floor(i / PREFIXES))}`);
	const t0 = Date.now();
	const wlat = await runPool(keys, CONCURRENCY, (k) => put(k, body));
	const welapsed = (Date.now() - t0) / 1000;
	console.log(`B1 write: ${fmt(N_OBJECTS)} objs in ${welapsed.toFixed(1)}s = ${fmt(N_OBJECTS / welapsed)} PUT/s`);
	console.log(` PUT latency ms: p50 ${pctl(wlat, 50).toFixed(1)} p99 ${pctl(wlat, 99).toFixed(1)} p999 ${pctl(wlat, 99.9).toFixed(1)}`);

	// B2 batching: same logical snapshots, one-per-object vs BATCH-per-object
	const G = Math.min(2000, N_OBJECTS);
	const unbatchedKeys = Array.from({ length: G }, (_, i) => `b2/unbatched/${pad(i)}`);
	let s = Date.now();
	await runPool(unbatchedKeys, CONCURRENCY, (k) => put(k, body));
	const ub = (Date.now() - s) / 1000;
	const batchedKeys = Array.from({ length: Math.ceil(G / BATCH) }, (_, i) => `b2/batched/${pad(i)}`);
	const bigBody = Buffer.alloc(OBJ_SIZE * BATCH, 7);
	s = Date.now();
	await runPool(batchedKeys, CONCURRENCY, (k) => put(k, bigBody));
	const bt = (Date.now() - s) / 1000;
	console.log(`B2 batching (${fmt(G)} snapshots): unbatched ${fmt(G)} PUTs in ${ub.toFixed(1)}s vs batched ${fmt(batchedKeys.length)} PUTs in ${bt.toFixed(1)}s (${BATCH}:1 -> ${(G / batchedKeys.length).toFixed(0)}x fewer requests, ${(ub / bt).toFixed(1)}x faster wall)`);

	// B3 cold reads: preload one run prefix with READ_PER_RUN objects
	const rprefix = `read-run/`;
	await runPool(Array.from({ length: READ_PER_RUN }, (_, i) => `${rprefix}${pad(i)}`), CONCURRENCY, (k) => put(k, body));
	s = Date.now(); const all = await listAll(rprefix); const listAllMs = Date.now() - s;
	const sinceFrom = `${rprefix}${pad(Math.floor(READ_PER_RUN / 2))}`;
	s = Date.now(); const since = await listAll(rprefix, sinceFrom); const sinceMs = Date.now() - s;
	s = Date.now(); await runPool(all.out, CONCURRENCY, (k) => get(k)); const getMs = Date.now() - s;
	console.log(`B3 cold reads (run with ${fmt(READ_PER_RUN)} states):`);
	console.log(` all states: LIST ${all.keys} keys in ${all.pages} page(s), ${listAllMs}ms \| since N/2: ${since.keys} keys in ${sinceMs}ms \| GET fan-out of all: ${getMs}ms (${fmt(all.out.length / (getMs / 1000))} GET/s)`);
	}
	main().catch((e) => { console.error("ERROR", e?.message \|\| e); process.exit(1); });
	// Workstream D: Postgres long-soak harness for the run_head design.
	// Decides the partitioning-vs-simple question by watching long-horizon dynamics
	// the short A0 runs can't: autovacuum keep-up, PK-index bloat trajectory,
	// xid-freeze age, and long-open-transaction (xmin) holdback.
	//
	// Rate, not hardware, is what makes bloat/vacuum dynamics extrapolate, so this
	// decouples from the laptop fsync ceiling via synchronous_commit=off + group commit.
	// Emits a CSV time series to stdout (redirect to a file and plot).
	//
	// Env:
	// ACTIVE=1000000 DURATION_S=3600 CONCURRENCY=16 FILLFACTOR=90
	// SYNC_COMMIT=off COMMIT_BATCH=10 TMIN=5 TMAX=20 SLEEPER_PCT=0.02
	// SAMPLE_S=30 REINDEX_EVERY_S=0 HOLD_TXN=0 (1 = open a long idle txn to test xmin holdback)
	// AV_RELOPTS="autovacuum_vacuum_scale_factor=0.02,autovacuum_vacuum_cost_limit=2000"

	import { createRequire } from "module";
	const require = createRequire(import.meta.url);
	const { Pool, Client } = require("/home/wes/projects/trigger/tech-test-wes-mason/node_modules/.pnpm/pg@8.16.3/node_modules/pg");

	const CONN = process.env.PG_URL \|\| "postgres://postgres:poc@localhost:55432/poc";
	const ACTIVE = parseInt(process.env.ACTIVE \|\| "1000000", 10);
	const DURATION_S = parseInt(process.env.DURATION_S \|\| "3600", 10);
	const CONCURRENCY = parseInt(process.env.CONCURRENCY \|\| "16", 10);
	const FILLFACTOR = parseInt(process.env.FILLFACTOR \|\| "90", 10);
	const SYNC_COMMIT = (process.env.SYNC_COMMIT \|\| "off").toLowerCase();
	const COMMIT_BATCH = parseInt(process.env.COMMIT_BATCH \|\| "10", 10);
	const TMIN = parseInt(process.env.TMIN \|\| "5", 10);
	const TMAX = parseInt(process.env.TMAX \|\| "20", 10);
	const SLEEPER_PCT = parseFloat(process.env.SLEEPER_PCT \|\| "0.02");
	const SAMPLE_S = parseInt(process.env.SAMPLE_S \|\| "30", 10);
	const REINDEX_EVERY_S = parseInt(process.env.REINDEX_EVERY_S \|\| "0", 10);
	const HOLD_TXN = process.env.HOLD_TXN === "1";
	const AV_RELOPTS = process.env.AV_RELOPTS \|\| "";

	const randInt = (lo, hi) => lo + Math.floor(Math.random() * (hi - lo + 1));
	const fmt = (n) => Math.round(n).toLocaleString("en-US");
	const num = (v) => Number(v \|\| 0);

	async function setup(pool) {
	await pool.query("DROP TABLE IF EXISTS run_head");
	const reloptions = [`fillfactor = ${FILLFACTOR}`, ...(AV_RELOPTS ? AV_RELOPTS.split(",").map((s) => s.trim()) : [])].join(", ");
	await pool.query(`CREATE TABLE run_head (
	run_id text PRIMARY KEY,
	seq bigint NOT NULL DEFAULT 0,
	execution_status text NOT NULL,
	run_status text NOT NULL,
	attempt_number int,
	created_at timestamptz NOT NULL DEFAULT now(),
	updated_at timestamptz NOT NULL DEFAULT now()
	) WITH (${reloptions})`);
	await pool.query("CREATE EXTENSION IF NOT EXISTS pgstattuple");
	const BATCH = 100000;
	for (let off = 0; off < ACTIVE; off += BATCH) {
	const lim = Math.min(BATCH, ACTIVE - off);
	await pool.query(
	`INSERT INTO run_head (run_id, seq, execution_status, run_status)
	SELECT 'r' \|\| g, 0, 'RUN_CREATED', 'PENDING' FROM generate_series($1::bigint, $2::bigint) g`,
	[off + 1, off + lim]
	);
	}
	}

	async function sample(pool) {
	const st = (await pool.query(
	`SELECT n_tup_upd, n_tup_hot_upd, n_dead_tup, n_live_tup, autovacuum_count
	FROM pg_stat_user_tables WHERE relname='run_head'`
	)).rows[0] \|\| {};
	const sz = (await pool.query(
	`SELECT pg_table_size('run_head') heap, pg_indexes_size('run_head') idx`
	)).rows[0] \|\| {};
	const tt = (await pool.query(
	`SELECT dead_tuple_percent dead, approx_free_percent free FROM pgstattuple_approx('run_head')`
	)).rows[0] \|\| {};
	let idx = {};
	try { idx = (await pool.query(`SELECT avg_leaf_density, leaf_fragmentation FROM pgstatindex('run_head_pkey')`)).rows[0] \|\| {}; } catch {}
	const age = (await pool.query(`SELECT age(relfrozenxid) a FROM pg_class WHERE relname='run_head'`)).rows[0] \|\| {};
	return { ...st, ...sz, ...tt, ...idx, ...age };
	}

	async function main() {
	const pool = new Pool({ connectionString: CONN, max: CONCURRENCY + 2 });
	console.error(`[D] active=${fmt(ACTIVE)} dur=${DURATION_S}s conc=${CONCURRENCY} ff=${FILLFACTOR} sync_commit=${SYNC_COMMIT} commit_batch=${COMMIT_BATCH} reindex_every=${REINDEX_EVERY_S}s hold_txn=${HOLD_TXN} relopts="${AV_RELOPTS}"`);
	await setup(pool);
	console.error(`[D] preloaded ${fmt(ACTIVE)} rows`);

	// Optional: hold a long-open idle transaction to demonstrate xmin holdback.
	let holder;
	if (HOLD_TXN) {
	holder = new Client({ connectionString: CONN });
	await holder.connect();
	await holder.query("BEGIN");
	await holder.query("SELECT 1"); // pins a snapshot/xmin for the whole run
	console.error(`[D] HOLD_TXN: a long-open transaction is pinning xmin (vacuum should not be able to reclaim dead tuples created after now)`);
	}

	const entries = new Array(ACTIVE);
	for (let i = 0; i < ACTIVE; i++) {
	const parked = Math.random() < SLEEPER_PCT;
	entries[i] = { id: "r" + (i + 1), seq: 0, target: parked ? Infinity : randInt(TMIN, TMAX), parked };
	}

	const ops = { upd: 0, ins: 0, del: 0 };
	const start = Date.now();
	const deadline = start + DURATION_S * 1000;
	let lastUpd = 0, lastHot = 0, lastSampleOps = 0, lastT = 0;

	console.log("t_s,ops,upd_per_s,hot_ratio_window_pct,dead_pct,free_pct,heap_mb,idx_mb,idx_leaf_density,idx_fragmentation,n_dead_tup,autovac_count,xid_age");
	const sampler = setInterval(async () => {
	try {
	const t = Math.round((Date.now() - start) / 1000);
	const s = await sample(pool);
	const dUpd = num(s.n_tup_upd) - lastUpd, dHot = num(s.n_tup_hot_upd) - lastHot;
	const hotW = dUpd > 0 ? (dHot / dUpd) * 100 : 0;
	const upsPerS = lastT > 0 ? (ops.upd - lastSampleOps) / (t - lastT) : ops.upd / Math.max(1, t);
	lastUpd = num(s.n_tup_upd); lastHot = num(s.n_tup_hot_upd); lastSampleOps = ops.upd; lastT = t;
	console.log([
	t, ops.upd, Math.round(upsPerS), hotW.toFixed(1),
	num(s.dead).toFixed(2), num(s.free).toFixed(2),
	(num(s.heap) / 1e6).toFixed(1), (num(s.idx) / 1e6).toFixed(1),
	s.avg_leaf_density != null ? num(s.avg_leaf_density).toFixed(1) : "",
	s.leaf_fragmentation != null ? num(s.leaf_fragmentation).toFixed(1) : "",
	num(s.n_dead_tup), num(s.autovacuum_count), num(s.a),
	].join(","));
	} catch (e) { console.error("sample error", e.message); }
	}, SAMPLE_S * 1000);

	let reindexer;
	if (REINDEX_EVERY_S > 0) {
	reindexer = setInterval(async () => {
	try {
	const before = num((await pool.query(`SELECT pg_indexes_size('run_head') i`)).rows[0].i);
	const t0 = Date.now();
	await pool.query("REINDEX INDEX CONCURRENTLY run_head_pkey");
	const after = num((await pool.query(`SELECT pg_indexes_size('run_head') i`)).rows[0].i);
	console.error(`[D] REINDEX pkey: ${(before / 1e6).toFixed(1)}MB -> ${(after / 1e6).toFixed(1)}MB in ${Date.now() - t0}ms`);
	} catch (e) { console.error("reindex error", e.message); }
	}, REINDEX_EVERY_S * 1000);
	}

	const updSQL = `UPDATE run_head SET seq=seq+1, execution_status='EXECUTING', run_status='EXECUTING', attempt_number=1, updated_at=now() WHERE run_id=$1 AND seq=$2 RETURNING seq`;

	async function worker(wid) {
	const client = await pool.connect();
	await client.query(`SET synchronous_commit = ${SYNC_COMMIT}`);
	let ptr = wid;
	try {
	while (Date.now() < deadline) {
	await client.query("BEGIN");
	for (let b = 0; b < COMMIT_BATCH && Date.now() < deadline; b++) {
	if (ptr >= ACTIVE) ptr = wid;
	const e = entries[ptr]; ptr += CONCURRENCY;
	if (!e \|\| e.parked) continue;
	const r = await client.query(updSQL, [e.id, e.seq]);
	ops.upd++;
	if (r.rowCount === 1) e.seq = Number(r.rows[0].seq);
	if (e.seq >= e.target) {
	await client.query(`DELETE FROM run_head WHERE run_id=$1`, [e.id]); ops.del++;
	const nid = `w${wid}-${ops.ins}`;
	await client.query(`INSERT INTO run_head (run_id, seq, execution_status, run_status) VALUES ($1,0,'RUN_CREATED','PENDING')`, [nid]); ops.ins++;
	entries[ptr - CONCURRENCY] = { id: nid, seq: 0, target: randInt(TMIN, TMAX), parked: false };
	}
	}
	await client.query("COMMIT");
	}
	} finally { try { await client.query("ROLLBACK"); } catch {} client.release(); }
	}

	await Promise.all(Array.from({ length: CONCURRENCY }, (_, w) => worker(w)));
	clearInterval(sampler); if (reindexer) clearInterval(reindexer);
	const final = await sample(pool);
	const elapsed = (Date.now() - start) / 1000;
	if (holder) { await holder.query("ROLLBACK").catch(() => {}); await holder.end().catch(() => {}); }

	console.error(`\n===== D SUMMARY =====`);
	console.error(`elapsed: ${elapsed.toFixed(0)}s`);
	console.error(`ops: upd ${fmt(ops.upd)} (${fmt(ops.upd / elapsed)}/s) ins ${fmt(ops.ins)} del ${fmt(ops.del)}`);
	console.error(`final dead%: ${num(final.dead).toFixed(2)} free%: ${num(final.free).toFixed(2)} n_dead_tup: ${fmt(num(final.n_dead_tup))}`);
	console.error(`final heap: ${(num(final.heap) / 1e6).toFixed(1)} MB index: ${(num(final.idx) / 1e6).toFixed(1)} MB`);
	console.error(`index leaf: density ${final.avg_leaf_density != null ? num(final.avg_leaf_density).toFixed(1) : "?"}% fragmentation ${final.leaf_fragmentation != null ? num(final.leaf_fragmentation).toFixed(1) : "?"}%`);
	console.error(`autovacuum runs: ${num(final.autovacuum_count)} final xid age: ${fmt(num(final.a))}`);
	console.error(HOLD_TXN
	? `xmin test: with a long-open txn held, dead tuples should be UNRECLAIMABLE (watch n_dead_tup climb without bound).`
	: `xmin test: no long-open txn; sleepers are old ROWS only, which do NOT block vacuum.`);
	await pool.end();
	}
	main().catch((e) => { console.error(e); process.exit(1); });