111 lines
4.9 KiB
TypeScript
111 lines
4.9 KiB
TypeScript
// 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 };
|
|
}
|