import { mkdtempSync } from "node:fs";import { tmpdir } from "node:os";import { join } from "node:path"; import { describe, expect, it } from "@effect/vitest";import { Effect } from "effect"; import { makeSqliteWorkflowRunner } from "./sqlite-workflow-runner";import type { DriveOutcome, WorkflowBindings, WorkflowBridge, WorkflowDriver,} from "../seams/workflow-runner"; // ---------------------------------------------------------------------------// Finding 3 regression: a run driven concurrently (start + signal racing) must// execute an unjournaled side-effecting `step.tool` EXACTLY once. Before the// single-driver lease, both drivers loaded the same "step not journaled" view// and both ran the tool, doubling its side effect.//// The stub driver drives one `step.tool("count")` per replay. The runner// services it: the FIRST driver to hold the lease executes the tool (bumping the// counter + journaling it); the second waits for the lease, re-drives, and finds// the step journaled -> replays without re-executing. So no matter how the race// interleaves, the counter ends at exactly 1.// --------------------------------------------------------------------------- const drivenTool: WorkflowDriver = { drive: (_input, bridge: WorkflowBridge): Effect.Effect<DriveOutcome, never> => Effect.gen(function* () { // One side-effecting step per replay. The runner runs the bound tool only // when the step is NOT yet journaled. const res = yield* bridge.call({ kind: "step.tool", step: "count", address: "count", args: {}, }); // A suspend/step-error result never happens for step.tool here; treat any // value as completion. const value = "value" in res ? res.value : undefined; return { status: "completed", output: value } satisfies DriveOutcome; }) as Effect.Effect<DriveOutcome, never>,}; describe("WorkflowRunner single-driver lease (concurrent drive)", () => { it("executes an unjournaled step.tool exactly once under concurrent start+signal", async () => { const dir = mkdtempSync(join(tmpdir(), "apps-wf-conc-")); const dbPath = join(dir, "journal.db"); let t = 1_000_000; const runner = makeSqliteWorkflowRunner({ path: dbPath, driver: drivenTool, clock: () => (t += 1), }); // A side-effecting bound tool that increments a shared counter each time it // actually runs. If the run double-drives the unjournaled step, this bumps // twice. let counter = 0; const bindings: WorkflowBindings = { runTool: async () => { counter += 1; // A small yield so a racing driver has a real window to interleave. await new Promise((r) => setTimeout(r, 5)); return counter; }, notify: async () => {}, }; const runId = "race-run"; // Fire start + signal concurrently, many times, all targeting the same run. // `start` is idempotent on runId (returns the existing run) and both paths // end up driving the run; the lease must still yield exactly one execution. const races: Promise<unknown>[] = []; for (let i = 0; i < 20; i++) { races.push( Effect.runPromise( runner .start( { scope: "s", workflow: "wf", snapshotId: "snap", entryPath: "workflows/wf.ts", input: {}, runId, }, bindings, ) .pipe(Effect.orElseSucceed(() => undefined)), ), ); races.push( Effect.runPromise( runner.signal(runId, "go", {}, bindings).pipe(Effect.orElseSucceed(() => undefined)), ), ); } await Promise.all(races); // The side effect ran exactly once. expect(counter).toBe(1); // And the step is journaled exactly once as completed. const steps = await Effect.runPromise(runner.listSteps(runId)); const countSteps = steps.filter((s) => s.name === "count"); expect(countSteps.length).toBe(1); expect(countSteps[0]?.status).toBe("completed"); await Effect.runPromise(runner.close()); }, 30_000);});
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { describe, expect, it } from "@effect/vitest";
import { Effect } from "effect";
import { makeSqliteWorkflowRunner } from "./sqlite-workflow-runner";
import type {
DriveOutcome,
WorkflowBindings,
WorkflowBridge,
WorkflowDriver,
} from "../seams/workflow-runner";
// ---------------------------------------------------------------------------
// Finding 3 regression: a run driven concurrently (start + signal racing) must
// execute an unjournaled side-effecting `step.tool` EXACTLY once. Before the
// single-driver lease, both drivers loaded the same "step not journaled" view
// and both ran the tool, doubling its side effect.
//
// The stub driver drives one `step.tool("count")` per replay. The runner
// services it: the FIRST driver to hold the lease executes the tool (bumping the
// counter + journaling it); the second waits for the lease, re-drives, and finds
// the step journaled -> replays without re-executing. So no matter how the race
// interleaves, the counter ends at exactly 1.
// ---------------------------------------------------------------------------
const drivenTool: WorkflowDriver = {
drive: (_input, bridge: WorkflowBridge): Effect.Effect<DriveOutcome, never> =>
Effect.gen(function* () {
// One side-effecting step per replay. The runner runs the bound tool only
// when the step is NOT yet journaled.
const res = yield* bridge.call({
kind: "step.tool",
step: "count",
address: "count",
args: {},
});
// A suspend/step-error result never happens for step.tool here; treat any
// value as completion.
const value = "value" in res ? res.value : undefined;
return { status: "completed", output: value } satisfies DriveOutcome;
}) as Effect.Effect<DriveOutcome, never>,
};
describe("WorkflowRunner single-driver lease (concurrent drive)", () => {
it("executes an unjournaled step.tool exactly once under concurrent start+signal", async () => {
const dir = mkdtempSync(join(tmpdir(), "apps-wf-conc-"));
const dbPath = join(dir, "journal.db");
let t = 1_000_000;
const runner = makeSqliteWorkflowRunner({
path: dbPath,
driver: drivenTool,
clock: () => (t += 1),
});
// A side-effecting bound tool that increments a shared counter each time it
// actually runs. If the run double-drives the unjournaled step, this bumps
// twice.
let counter = 0;
const bindings: WorkflowBindings = {
runTool: async () => {
counter += 1;
// A small yield so a racing driver has a real window to interleave.
await new Promise((r) => setTimeout(r, 5));
return counter;
},
notify: async () => {},
};
const runId = "race-run";
// Fire start + signal concurrently, many times, all targeting the same run.
// `start` is idempotent on runId (returns the existing run) and both paths
// end up driving the run; the lease must still yield exactly one execution.
const races: Promise<unknown>[] = [];
for (let i = 0; i < 20; i++) {
races.push(
Effect.runPromise(
runner
.start(
{
scope: "s",
workflow: "wf",
snapshotId: "snap",
entryPath: "workflows/wf.ts",
input: {},
runId,
},
bindings,
)
.pipe(Effect.orElseSucceed(() => undefined)),
),
);
races.push(
Effect.runPromise(
runner.signal(runId, "go", {}, bindings).pipe(Effect.orElseSucceed(() => undefined)),
),
);
}
await Promise.all(races);
// The side effect ran exactly once.
expect(counter).toBe(1);
// And the step is journaled exactly once as completed.
const steps = await Effect.runPromise(runner.listSteps(runId));
const countSteps = steps.filter((s) => s.name === "count");
expect(countSteps.length).toBe(1);
expect(countSteps[0]?.status).toBe("completed");
await Effect.runPromise(runner.close());
}, 30_000);
});