Repository Public
1/**2 * Soak: 50 runs × 2 worker PROCESSES, one of them SIGKILLed every few hundred ms for ~20 s and3 * restarted at once. Every run must complete with exactly the expected step results, and no step may4 * ever be executed by two processes at the same time (advisory-lock tripwire inside exec).5 */6import fs from 'node:fs';7import path from 'node:path';8import type { ChildProcess } from 'node:child_process';9import { afterAll, beforeAll, expect, it } from 'vitest';10import { createTestDb, exited, killAllChildren, newMission, rows, sleep, spawnFixture, waitFor, type TestDb } from './helpers.js';11import { expectedChain, SOAK_EFFECTS } from './fixtures/soak-logic.js';1213const RUNS = 50;14const CHAOS_MS = 20_000;1516let db: TestDb;17beforeAll(async () => {18 db = await createTestDb();19 await db.k.pool.query('CREATE TABLE soak_violations (run_id uuid, seq int, pid int, at timestamptz DEFAULT now())');20 await db.k.pool.query('CREATE TABLE soak_execs (run_id uuid, seq int, pid int, at timestamptz DEFAULT now())');21});22afterAll(async () => {23 await killAllChildren();24 await db.cleanup();25});2627it('50 runs survive ~20 s of random SIGKILLs on 2 workers: exact results, no concurrent execution of a step', async () => {28 const dir = path.join(db.tmp, 'soak');29 fs.mkdirSync(dir);30 const runs: { runId: string; missionId: string; seed: number }[] = [];31 for (let i = 0; i < RUNS; i++) {32 const seed = 1000 + i * 17;33 const m = await newMission(db, { workspaceAbs: dir, boundaries: String(seed) });34 runs.push({ runId: m.rootRunId, missionId: m.missionId, seed });35 }36 const slots: ChildProcess[] = [];37 const spawnSlot = (slot: number): void => {38 slots[slot] = spawnFixture('soak-worker.ts', { DATABASE_URL: db.url, SLOT: String(slot), LEASE_MS: '2000' });39 };40 spawnSlot(0);41 spawnSlot(1);4243 let kills = 0;44 const end = Date.now() + CHAOS_MS;45 while (Date.now() < end) {46 await sleep(250 + Math.floor(Math.random() * 500));47 const slot = Math.random() < 0.5 ? 0 : 1;48 const victim = slots[slot] as ChildProcess;49 victim.kill('SIGKILL');50 await exited(victim);51 kills++;52 spawnSlot(slot);53 }54 const unfinished = (await rows(db, `SELECT 1 FROM runs WHERE status <> 'done'`)).length;55 // Calm: let the two workers finish everything.56 await waitFor(async () => (await rows(db, `SELECT 1 FROM runs WHERE status <> 'done'`)).length === 0, 90_000, 'all runs done');57 // Graceful stop (SIGTERM) once both workers are up (their handler is installed before the loop starts).58 await waitFor(async () => {59 const ids = (await rows<{ id: string }>(db, 'SELECT id FROM workers')).map((w) => w.id);60 return slots.every((c, slot) => ids.includes(`soak-${slot}-${c.pid}`));61 }, 20_000, 'workers up');62 for (const c of slots) c.kill('SIGTERM');63 const exits = await Promise.all(slots.map(exited));64 expect(exits).toEqual([{ code: 0, signal: null }, { code: 0, signal: null }]);65 expect(kills).toBeGreaterThan(20);6667 // 1. Tripwire: no step was ever executed by two processes at once.68 expect(await rows(db, 'SELECT * FROM soak_violations')).toEqual([]);6970 // 2. Every run is done with exactly the expected step results.71 const execs = await rows<{ n: string }>(db, 'SELECT count(*) AS n FROM soak_execs');72 for (const r of runs) {73 const chain = expectedChain(r.seed);74 const st = await rows<{ seq: number; status: string; result: { v: number }; idem_key: string; effect: string }>(75 db, 'SELECT seq, status, result, idem_key, effect FROM steps WHERE run_id = $1 ORDER BY seq', [r.runId]);76 expect(st.map((s) => s.status)).toEqual(Array(SOAK_EFFECTS.length).fill('completed'));77 expect(st.map((s) => s.result.v)).toEqual(chain);78 const run = (await rows<{ status: string; result: { payload: { v: number } }; usage: { usd: number; turns: number } }>(db, 'SELECT status, result, usage FROM runs WHERE id = $1', [r.runId]))[0];79 expect(run?.status).toBe('done');80 expect(run?.result.payload.v).toBe(chain[chain.length - 1]);81 expect(run?.usage.usd).toBeCloseTo(0.001, 9); // the model_call was accounted exactly once82 expect(run?.usage.turns).toBe(1);83 // Idempotent writes: one file per step, named by idem_key, with the right content.84 for (const s of st.filter((x) => x.effect === 'idempotent_write')) {85 expect(fs.readFileSync(path.join(dir, s.idem_key), 'utf8')).toBe(String(s.result.v));86 }87 }88 expect(fs.readdirSync(dir)).toHaveLength(RUNS * SOAK_EFFECTS.filter((e) => e === 'idempotent_write').length);8990 // 3. Journal and events: nothing lost, nothing duplicated.91 const dup = await rows(db, `SELECT run_id, seq, kind, count(*) FROM events WHERE kind IN ('step.started','step.completed') GROUP BY 1,2,3 HAVING count(*) <> 1`);92 expect(dup).toEqual([]);93 const totals = (await rows<{ started: string; completed: string; done: string }>(94 db, `SELECT count(*) FILTER (WHERE kind = 'step.started') AS started, count(*) FILTER (WHERE kind = 'step.completed') AS completed,95 count(*) FILTER (WHERE kind = 'run.status' AND payload->>'status' = 'done') AS done FROM events`))[0];96 expect(totals).toEqual({ started: String(RUNS * 8), completed: String(RUNS * 8), done: String(RUNS) });97 const spent = (await rows<{ usd: string }>(db, 'SELECT sum(spent_usd) AS usd FROM missions'))[0];98 expect(Number(spent?.usd)).toBeCloseTo(RUNS * 0.001, 6);99 expect(Number((await rows<{ usd: string }>(db, 'SELECT usd FROM spend_days'))[0]?.usd)).toBeCloseTo(RUNS * 0.001, 6);100 // Crashes really interrupted work: some steps were executed more than once.101 expect(Number(execs[0]?.n)).toBeGreaterThan(RUNS * 8);102 const claims = (await rows<{ n: string; span: string }>(db, 'SELECT sum(fencing_token) AS n, round(EXTRACT(EPOCH FROM (max(ended_at) - min(created_at)))::numeric, 1) AS span FROM runs'))[0];103 process.stderr.write(`soak: ${kills} SIGKILLs, ${execs[0]?.n} step executions for ${RUNS * 8} steps, ${claims?.n} claims for ${RUNS} runs, ${unfinished} runs unfinished when the chaos stopped, last run ended after ${claims?.span} s\n`);104}, 180_000);105