Example: a Workflow waiting for an event

An expense approval runs as a Cloudflare Workflow. Submitting an expense starts an instance, which waits up to a week for the manager's decision; a later request delivers the decision to it as an event. Once approved, the instance pays the employee through a payouts API, inside step.do, so that a restart of the Workflow does not pay again.

Run the check

From examples/workflow-approval in the repository:

pnpm install
pnpm check

The scenario in expenses.sim.ts submits one expense and then approves it. The Workflow runs on a model of the engine: the wait suspends the instance until the event arrives, each step attempt can crash before it runs or after it ran, and a failed step is retried on the engine's schedule in virtual time. The check reports:

VIOLATED  expenses  (seed expenses)

Invariant paid-once is violated: Fetch:payouts POST happened twice for key "exp-1" (seq 15 and 22)

Schedule (run seed ...), events kept: req-1, req-2
  crash_after at workflow:APPROVALS:exp-1#1 effect #3 (Workflow:APPROVALS step_result)

The payout step ran, and the engine restarted before its result was saved, so the step ran again and paid a second time. A step is retried the same way when its call fails after the far side acted (with --seed example the check finds that path: the payouts API's answer lost). Steps make a Workflow's code run once per completed step, not once per effect; an ordinary test, where every step completes the first time, cannot tell the two apart.

The fix

worker.fixed.ts sends the payout with an idempotency key derived from the expense, which the payouts API deduplicates by (the scenario's external declares that guarantee with dedupeBy), so a step that runs again is answered with the first payout. pnpm check:fixed reports BOUNDED.

Using it in your own project

workflow(binding, className) names the Workflow and the class its instances run; workflowRows reads the instances at the end of the run. An outbound call is answered from the scenario's external declarations: declare each host your Workflow reaches, with dedupeBy when the service promises to deduplicate.

The code

The complete project is examples/workflow-approval in the repository.

worker.tsexpenses.sim.tsworker.fixed.ts

/**
 * Expense approval as a Cloudflare Workflow. Submitting an expense starts an
 * instance, which waits up to a week for the manager's decision, delivered
 * later as an event; once approved, a step pays the employee through a
 * payouts API. The payout is a step, so the Workflow does not repeat it
 * after a restart. Run `eutaxy check ./expenses.sim.ts` to see that a step
 * is not enough.
 */

import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers";

export interface Env {
  readonly APPROVALS: Workflow<Expense>;
}

export interface Expense {
  readonly expenseId: string;
  readonly employee: string;
  readonly amountCents: number;
}

export interface Decision {
  readonly approved: boolean;
  readonly by: string;
}

/** Pays the expense out; `headers` lets a caller add its own (the fix adds an idempotency key). */
export async function payOut(
  expense: Expense,
  headers: Record<string, string> = {},
): Promise<{ payoutId: string }> {
  const res = await fetch("https://payouts.example/payouts", {
    method: "POST",
    headers: { "content-type": "application/json", ...headers },
    body: JSON.stringify(expense),
  });
  if (!res.ok) throw new Error(`payout failed with ${res.status}`);
  return (await res.json()) as { payoutId: string };
}

export class ExpenseApproval extends WorkflowEntrypoint<Env, Expense> {
  override async run(event: WorkflowEvent<Expense>, step: WorkflowStep) {
    const expense = event.payload;
    const decision = await step.waitForEvent<Decision>("manager decision", {
      type: "decision",
      timeout: "7 days",
    });
    if (!decision.payload.approved) return { expenseId: expense.expenseId, paid: false };
    const payout = await step.do("pay out", () => this.pay(expense));
    return { expenseId: expense.expenseId, paid: true, payoutId: payout.payoutId };
  }

  protected pay(expense: Expense): Promise<{ payoutId: string }> {
    return payOut(expense);
  }
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    if (request.method === "POST" && url.pathname === "/expenses") {
      const expense = (await request.json()) as Expense;
      await env.APPROVALS.create({ id: expense.expenseId, params: expense });
      return Response.json({ submitted: expense.expenseId }, { status: 202 });
    }
    const decided = /^\/expenses\/([\w-]+)\/decision$/.exec(url.pathname);
    if (request.method === "POST" && decided?.[1]) {
      const decision = (await request.json()) as Decision;
      const instance = await env.APPROVALS.get(decided[1]);
      await instance.sendEvent({ type: "decision", payload: decision });
      return Response.json({ decided: decided[1] });
    }
    return new Response("not found", { status: 404 });
  },
} satisfies ExportedHandler<Env>;

import {
  atMostOncePerKey,
  defineScenario,
  external,
  fetchCall,
  http,
  unique,
  workflow,
  workflowRows,
} from "eutaxy/cloudflare";

export default defineScenario({
  name: "expenses",
  worker: { entry: "./worker.ts", config: "./wrangler.jsonc" },
  setup: [workflow("APPROVALS", "ExpenseApproval")],
  // An employee submits an expense; later the manager approves it.
  drive: http([
    { path: "/expenses", body: { expenseId: "exp-1", employee: "e-7", amountCents: 4200 } },
    { path: "/expenses/exp-1/decision", body: { approved: true, by: "m-1" } },
  ]),
  externals: [
    // The payouts API deduplicates by Idempotency-Key, as payment APIs do; the planted Workflow sends none.
    external("payouts", "payouts.example", {
      dedupeBy: "idempotency-key",
      responses: [{ method: "POST", path: "/payouts", status: 201, body: { payoutId: "po-1" } }],
    }),
  ],
  invariants: [
    // An approved expense is paid once.
    atMostOncePerKey("paid-once", fetchCall("payouts", "POST"), "args.body.expenseId"),
    // And it has one approval run.
    unique("one-run-per-expense", workflowRows("APPROVALS"), "params.expenseId"),
  ],
  bounds: { maxFaultsPerRun: 1, maxRedeliveries: 2, maxConcurrentInvocations: 1, budgetMs: 4000 },
});

/**
 * The same Workflow, made safe to retry: the payout carries the expense id
 * as its idempotency key, which the payouts API deduplicates by, so a step
 * that runs again pays nothing more. `eutaxy check ./expenses.fixed.sim.ts`
 * reports BOUNDED.
 */

import { type Expense, ExpenseApproval as PlantedApproval, payOut } from "./worker.ts";

export { default } from "./worker.ts";

export class ExpenseApproval extends PlantedApproval {
  protected override pay(expense: Expense): Promise<{ payoutId: string }> {
    return payOut(expense, { "idempotency-key": `expense:${expense.expenseId}` });
  }
}

Example: a Workflow waiting for an event ​

Run the check ​

The fix ​

Using it in your own project ​

The code ​

Example: a Workflow waiting for an event

Run the check

The fix

Using it in your own project

The code