docs(spikes): OpenWorkflow and Pi SDK compatibility spikes + ADR-002/003 (W0-B)
Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
rootandClaude Fable 5.1 committed
1 parent
8b90d396b2
commit
ec023a8405
16 files changed
+2884
No files matched your search
@@ -0,0 +1,64 @@
|
||||
import { makeClient, initBiz, db } from "./common.ts";
|
||||
import { sql } from "kysely";
|
||||
await initBiz();
|
||||
const { ow, backend, acquire, flaky, slow, burst } = await makeClient();
|
||||
const [cmd, ...args] = process.argv.slice(2);
|
||||
const out = (x: unknown) => console.log(JSON.stringify(x, null, 0));
|
||||
switch (cmd) {
|
||||
case "start-acquire": {
|
||||
const [downloadId, key, copyMs, retention] = args;
|
||||
const h = await acquire.run({ downloadId, copyMs: Number(copyMs ?? 0), retention: retention ?? "3s" },
|
||||
key ? { idempotencyKey: key } : undefined);
|
||||
out({ runId: h.workflowRun.id, status: h.workflowRun.status, version: h.workflowRun.version });
|
||||
break;
|
||||
}
|
||||
case "start-flaky": {
|
||||
const h = await flaky.run({ failTimes: Number(args[0]), maxAttempts: Number(args[1]) });
|
||||
out({ runId: h.workflowRun.id });
|
||||
break;
|
||||
}
|
||||
case "start-slow": { const h = await slow.run({}); out({ runId: h.workflowRun.id }); break; }
|
||||
case "start-burst": {
|
||||
const ids = [];
|
||||
for (let i = 0; i < Number(args[0]); i++) ids.push((await burst.run({ i })).workflowRun.id);
|
||||
out({ count: ids.length });
|
||||
break;
|
||||
}
|
||||
case "signal": {
|
||||
const r = await ow.sendSignal({ signal: `download-done:${args[0]}`, data: { ok: true }, idempotencyKey: args[1] });
|
||||
out(r);
|
||||
break;
|
||||
}
|
||||
case "cancel": { await ow.cancelWorkflowRun(args[0]); out({ canceled: args[0] }); break; }
|
||||
case "get": {
|
||||
const r = await backend.getWorkflowRun({ workflowRunId: args[0] });
|
||||
out({ status: r?.status, version: r?.version, attempts: r?.attempts, error: r?.error, output: r?.output });
|
||||
break;
|
||||
}
|
||||
case "steps": {
|
||||
const r = await backend.listStepAttempts({ workflowRunId: args[0], limit: 100 } as any);
|
||||
for (const s of [...r.data].sort((a,b)=>+a.createdAt-+b.createdAt)) console.log(s.stepName.padEnd(18), s.kind.padEnd(12), s.status, s.error ? JSON.stringify(s.error.message) : "");
|
||||
break;
|
||||
}
|
||||
case "ledger": {
|
||||
const rows = await db.selectFrom("biz.side_effect_ledger").selectAll()
|
||||
.$if(!!args[0], (q) => q.where("run_id", "=", args[0])).orderBy("id").execute();
|
||||
for (const r of rows) console.log(r.run_id.slice(0, 8), r.step.padEnd(16), r.worker.padEnd(6), r.note);
|
||||
break;
|
||||
}
|
||||
case "burst-dups": {
|
||||
const rows = await sql<any>`SELECT count(*) total, count(DISTINCT run_id) runs,
|
||||
count(*) FILTER (WHERE worker='A') a, count(*) FILTER (WHERE worker='B') b
|
||||
FROM biz.side_effect_ledger WHERE step='work'`.execute(db);
|
||||
out(rows.rows[0]);
|
||||
break;
|
||||
}
|
||||
case "tables": {
|
||||
const r = await sql<any>`SELECT table_schema, table_name FROM information_schema.tables
|
||||
WHERE table_schema IN ('openworkflow','biz') ORDER BY 1,2`.execute(db);
|
||||
for (const t of r.rows) console.log(`${t.table_schema}.${t.table_name}`);
|
||||
break;
|
||||
}
|
||||
}
|
||||
await backend.stop();
|
||||
await db.destroy();
|
||||
@@ -0,0 +1,110 @@
|
||||
// Shared: DB URLs, OpenWorkflow client, Kysely business DB (schema "biz"), workflow definitions.
|
||||
import { OpenWorkflow } from "openworkflow";
|
||||
import { BackendPostgres } from "openworkflow/postgres";
|
||||
import { Kysely, PostgresDialect, sql } from "kysely";
|
||||
import pg from "pg";
|
||||
|
||||
export const PG_URL = process.env.PG_URL ?? "postgresql://postgres:[email protected]:55434/postgres";
|
||||
export const WF_VERSION = process.env.WF_VERSION ?? "v1";
|
||||
export const TAG = process.env.TAG ?? `pid${process.pid}`;
|
||||
|
||||
export const db = new Kysely<any>({
|
||||
dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: PG_URL, max: 4 }) }),
|
||||
});
|
||||
|
||||
export async function initBiz() {
|
||||
await sql`CREATE SCHEMA IF NOT EXISTS biz`.execute(db);
|
||||
await sql`CREATE TABLE IF NOT EXISTS biz.side_effect_ledger (
|
||||
id bigserial PRIMARY KEY, run_id text, step text, worker text, pid int, note text,
|
||||
at timestamptz DEFAULT now())`.execute(db);
|
||||
}
|
||||
|
||||
export async function ledger(runId: string, step: string, note = "") {
|
||||
await db.insertInto("biz.side_effect_ledger")
|
||||
.values({ run_id: runId, step, worker: TAG, pid: process.pid, note }).execute();
|
||||
console.log(`[${TAG}] run=${runId.slice(0, 8)} step=${step} ${note}`);
|
||||
}
|
||||
|
||||
const wait = (ms: number) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
export async function makeClient() {
|
||||
// connect() runs migrations by default (runMigrations: true), schema default "openworkflow".
|
||||
const backend = await BackendPostgres.connect(PG_URL, { schema: "openworkflow" });
|
||||
const ow = new OpenWorkflow({ backend });
|
||||
|
||||
// Main vertical slice: submit download -> wait signal -> copy -> register -> retention sleep -> cleanup
|
||||
const acquire = ow.defineWorkflow(
|
||||
{ name: "acquire", version: WF_VERSION },
|
||||
async ({ input, step, run, version }) => {
|
||||
const { downloadId, copyMs = 0, retention = "3s" } = input as any;
|
||||
const sub = await step.run({ name: "submit-download" }, async () => {
|
||||
await ledger(run.id, "submit-download", `version=${version}`);
|
||||
return { torrentHash: `hash-${downloadId}` };
|
||||
});
|
||||
const done = await step.waitForSignal<{ ok: boolean }>({
|
||||
name: "wait-download", signal: `download-done:${downloadId}`, timeout: "10m",
|
||||
});
|
||||
if (!done) throw new Error("download wait timed out");
|
||||
const copied = await step.run({ name: "copy-file" }, async () => {
|
||||
await ledger(run.id, "copy-file", "start");
|
||||
await wait(copyMs);
|
||||
await ledger(run.id, "copy-file", "end");
|
||||
return { dest: `/lib/${downloadId}.mkv` };
|
||||
});
|
||||
if (process.env.EXTRA_STEP === "1") {
|
||||
// EXTRA_STEP=1 simulates a code change between copy and register (tests code upgrade semantics)
|
||||
await step.run({ name: "v2-extra-verify" }, async () => { await ledger(run.id, "v2-extra-verify"); return 1; });
|
||||
}
|
||||
await step.run({ name: "register" }, async () => { await ledger(run.id, "register", `version=${version}`); return 1; });
|
||||
await step.sleep("retention", retention);
|
||||
await step.run({ name: "cleanup" }, async () => { await ledger(run.id, "cleanup"); return 1; });
|
||||
return { sub, copied, signal: done.data };
|
||||
},
|
||||
);
|
||||
|
||||
// Retry test: step fails `failTimes` times then succeeds; step maximumAttempts = maxAttempts
|
||||
const flaky = ow.defineWorkflow(
|
||||
{ name: "flaky", version: WF_VERSION },
|
||||
async ({ input, step, run }) => {
|
||||
const { failTimes, maxAttempts } = input as any;
|
||||
await step.run({ name: "before" }, async () => { await ledger(run.id, "before"); return 1; });
|
||||
return await step.run(
|
||||
{ name: "flaky-step", retryPolicy: { maximumAttempts: maxAttempts, initialInterval: "200ms", maximumInterval: "1s" } },
|
||||
async () => {
|
||||
const n = await db.selectFrom("biz.side_effect_ledger").select(db.fn.countAll().as("c"))
|
||||
.where("run_id", "=", run.id).where("step", "=", "flaky-step").executeTakeFirst();
|
||||
const attempt = Number(n?.c ?? 0) + 1;
|
||||
await ledger(run.id, "flaky-step", `attempt=${attempt}`);
|
||||
if (attempt <= failTimes) throw new Error(`boom attempt ${attempt}`);
|
||||
return { attempt };
|
||||
},
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
// Cancel test: long step, then a follow-up step
|
||||
const slow = ow.defineWorkflow(
|
||||
{ name: "slow", version: WF_VERSION },
|
||||
async ({ step, run }) => {
|
||||
await step.run({ name: "long" }, async () => {
|
||||
await ledger(run.id, "long", "start");
|
||||
await wait(8000);
|
||||
await ledger(run.id, "long", "end (still ran after cancel?)");
|
||||
return 1;
|
||||
});
|
||||
await step.run({ name: "after-long" }, async () => { await ledger(run.id, "after-long"); return 1; });
|
||||
return "done";
|
||||
},
|
||||
);
|
||||
|
||||
// Concurrency test: many short runs
|
||||
const burst = ow.defineWorkflow(
|
||||
{ name: "burst", version: WF_VERSION },
|
||||
async ({ step, run }) => {
|
||||
await step.run({ name: "work" }, async () => { await wait(300); await ledger(run.id, "work"); return 1; });
|
||||
return 1;
|
||||
},
|
||||
);
|
||||
|
||||
return { ow, backend, acquire, flaky, slow, burst };
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
import { makeClient, initBiz, TAG, WF_VERSION } from "./common.ts";
|
||||
await initBiz();
|
||||
const { ow } = await makeClient();
|
||||
const worker = ow.newWorker({ concurrency: Number(process.env.CONCURRENCY ?? 4) });
|
||||
await worker.start();
|
||||
if (process.env.PIDFILE) (await import("node:fs")).writeFileSync(process.env.PIDFILE, String(process.pid));
|
||||
console.log(`[${TAG}] worker started pid=${process.pid} version=${WF_VERSION}`);
|
||||
process.on("SIGTERM", async () => { await worker.stop(); process.exit(0); });
|
||||
Reference in new issue
Block a user