Repository Public
1/** Run logic of the soak test: 8 chained steps; every exec holds an advisory-lock tripwire. */2import fs from 'node:fs';3import path from 'node:path';4import type pg from 'pg';5import type { EffectClass } from '@aos/shared';6import type { RunLogic } from '../../src/index.js';78export const SOAK_EFFECTS: EffectClass[] = ['pure', 'read', 'idempotent_write', 'pure', 'model_call', 'read', 'idempotent_write', 'pure'];9export const next = (prev: number, i: number): number => (prev * 31 + i + 7) % 1_000_003;10export function expectedChain(seed: number): number[] {11 const out: number[] = [];12 let v = seed;13 for (let i = 0; i < SOAK_EFFECTS.length; i++) out.push((v = next(v, i)));14 return out;15}1617function lockKey(runId: string, seq: number): [number, number] {18 let h = 0;19 for (const ch of runId) h = (h * 33 + ch.charCodeAt(0)) | 0;20 return [h, seq];21}2223/**24 * Tripwire: a session advisory lock on (run, seq) held for the whole exec. If another process is25 * executing the same step at the same time the try-lock fails and a violation row is written.26 * A SIGKILLed process drops its connection, which releases the lock.27 */28export function makeSoakLogic(tripPool: pg.Pool): RunLogic {29 return async (ctx) => {30 const seed = Number(ctx.contract.boundaries); // the test passes the seed through the contract31 let v = seed;32 for (let i = 0; i < SOAK_EFFECTS.length; i++) {33 const effect = SOAK_EFFECTS[i] as EffectClass;34 const prev = v;35 const r = await ctx.step({ kind: 'soak', name: `k${i}`, effect, request: { i, prev } }, async (idem) => {36 const [a, b] = lockKey(ctx.runId, i);37 const c = await tripPool.connect();38 try {39 const got = (await c.query('SELECT pg_try_advisory_lock($1, $2) AS ok', [a, b])).rows[0] as { ok: boolean };40 if (!got.ok) await c.query('INSERT INTO soak_violations (run_id, seq, pid) VALUES ($1, $2, $3)', [ctx.runId, i, process.pid]);41 await c.query('INSERT INTO soak_execs (run_id, seq, pid) VALUES ($1, $2, $3)', [ctx.runId, i, process.pid]);42 await new Promise((res) => setTimeout(res, 80 + ((i * 37 + seed) % 120)));43 if (effect === 'idempotent_write') fs.writeFileSync(path.join(ctx.workspaceAbs, idem), String(next(prev, i)));44 if (got.ok) await c.query('SELECT pg_advisory_unlock($1, $2)', [a, b]);45 } finally {46 c.release();47 }48 return { result: { v: next(prev, i) }, usage: effect === 'model_call' ? { usd: 0.001 } : undefined };49 });50 v = r.v;51 }52 return { status: 'done', summary: 'soak', payload: { v }, artifacts: [], evidence: [], openQuestions: [] };53 };54}55