Effect workflows

An Effect workflow is a durable process written with effect/workflow: a series of steps whose results are stored, so a run that’s interrupted resumes after its last finished step, and a run can wait hours or days for a timer or a signal. On Cloudflare they run on Durable Objects (Cloudflare.Workflows.DurableObjectEngine) or on Cloudflare Workflows (Cloudflare.Workflows.NativeEngine), with the same workflow code.

Reach for Effect workflows when you want the same workflow code to run anywhere Effect’s WorkflowEngine does: on Cloudflare, on AWS Lambda, or in memory in tests. To write against Cloudflare Workflows’ own step API instead, see Workflows.

Define a workflow

Workflow.make declares the workflow: its name, the payload it starts with, its result, and the key that identifies one run.

src/Greet.ts

import * as Schema from "effect/Schema";
import { Workflow } from "effect/workflow";

export const Greet = Workflow.make("Greet", {
  payload: { id: Schema.String, name: Schema.String },
  success: Schema.String,
  idempotencyKey: ({ id }) => id,
});

Starting a workflow again with the same key returns the run already started, so a retried request never starts a second one.

Implement its steps

Workflows.layer implements the workflow. Each step is an Activity: a named Effect whose result is stored the first time it completes. When the run replays, a finished step returns its stored result instead of running again.

src/Greet.ts

import * as Workflows from "alchemy/Workflows";
import * as Effect from "effect/Effect";
import { Activity } from "effect/workflow";

export const GreetLive = Workflows.layer(Greet, ({ name }) =>
  Activity.make({
    name: "greet",
    success: Schema.String,
    execute: Effect.succeed(`hello ${name}`),
  }),
);

A step that returns a value or fails with a typed error declares success and error schemas, because its result is stored as JSON.

Run it on Durable Objects

Cloudflare.Workflows.DurableObjectEngine is the engine. Provide it under the workflow Layers, and provide both to the Worker:

src/worker.ts

import * as Cloudflare from "alchemy/Cloudflare";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import { GreetLive } from "./Greet.ts";

export default class AppWorker extends Cloudflare.Worker<AppWorker>()(
  "AppWorker",
  { main: import.meta.url },
  Effect.gen(function* () {
    return {};
  }).pipe(
    Effect.provide(GreetLive.pipe(Layer.provideMerge(Cloudflare.Workflows.DurableObjectEngine))),
  ),
) {}

Each workflow gets its own Durable Object class on the Worker, named after it (GreetWorkflow), and deploying the Worker deploys them.

Start a run

Workflows.RunWorkflow gives you a client for the workflow. Resolve it during init, and start runs from your handlers:

src/worker.ts

import * as Cloudflare from "alchemy/Cloudflare";
import * as Workflows from "alchemy/Workflows";
import * as Effect from "effect/Effect";
import * as HttpServerResponse from "effect/http/HttpServerResponse";
import * as Layer from "effect/Layer";
import { Greet, GreetLive } from "./Greet.ts";

export default class AppWorker extends Cloudflare.Worker<AppWorker>()(
  "AppWorker",
  { main: import.meta.url },
  Effect.gen(function* () {
    const greet = yield* Workflows.RunWorkflow(Greet);
    return {
      fetch: Effect.gen(function* () {
        const run = yield* greet.start({ id: "g-1", name: "sam" });
        return HttpServerResponse.text(run.executionId);
      }),
    };
  }).pipe(
    Effect.provide(GreetLive.pipe(Layer.provideMerge(Cloudflare.Workflows.DurableObjectEngine))),
  ),
) {}

start returns as soon as the run is started, with a handle to it. Starting again with the same key returns the run already started, so a retried request never starts a second one.

RunWorkflow only compiles where the workflow is implemented: running a workflow with no Workflows.layer in the Worker is a type error.

Read a run

Store the run’s executionId, and get the run back with get:

const run = greet.get(executionId);
const status = yield* run.status;

status is NotStarted, Running, Completed with its result, Failed with its error, Interrupted, or Died with a defect. run.result waits for the run to finish, which suits tests and scripts more than requests.

Or run it on Cloudflare Workflows

Cloudflare.Workflows.NativeEngine runs the same workflow Layers on Cloudflare Workflows instead: one Workflow class, NativeEngine, runs them all, and each run is one Workflow instance. Swap the engine and nothing else changes:

Effect.provide(GreetLive.pipe(Layer.provideMerge(Cloudflare.Workflows.NativeEngine)))

Steps run with step.do, sleeps with step.sleep, and signals are Workflow events, so runs show up in the Cloudflare dashboard step by step. Cloudflare’s limits apply: steps per instance, the size of each step’s result, and a year at most for a wait.

Sleep durably

DurableClock.sleep waits on a Durable Object alarm or a Workflow sleep, so the wait survives restarts and costs nothing while it lasts:

yield* DurableClock.sleep({ name: "cool-off", duration: "1 hour" });

Sleeps of a minute or less run in memory by default. Pass inMemoryThreshold: 0 to make a short sleep durable too.

Wait for a signal

A DurableDeferred is a value a run waits for, delivered from outside: an approval, a webhook, a person clicking a link. The run suspends until it arrives.

export const Approval = DurableDeferred.make("approval", { success: Schema.String });

const approvedBy = yield* DurableDeferred.await(Approval);

Send it with the run’s signal:

yield* greet.get(executionId).signal(Approval, "agent-7");

Undo a step when a later one fails

withCompensation attaches what to do if the run fails after the step succeeded. Compensations run last step first.

yield* Activity.make({ name: "reserve", execute: reserve(order) }).pipe(
  Checkout.withCompensation(() => Activity.make({ name: "release", execute: release(order) })),
);

run.interrupt stops a run where it waits, and its compensations run too.

Call other Durable Objects from a step

Steps often call another Durable Object’s methods. Resolve the namespace when the Layer builds, and call it from a step:

export const RecordLive = Layer.unwrap(
  Effect.gen(function* () {
    const journals = yield* Journal;
    return Workflows.layer(Record, ({ id }) =>
      Activity.make({ name: "append", execute: journals.getByName(id).append("done") }),
    );
  }),
);

Those methods are runtime-only: they need RuntimeContext, which init doesn’t have. Steps only run inside a run, which has one, so Workflows.layer doesn’t pass that requirement on to the Worker.

How a run executes on Durable Objects

A run is one Durable Object instance, named by its execution id, with its journal in that instance’s SQLite: the payload, each finished step’s result, completed signals and pending timers. Every wake-up (a start, a signal, a timer, an interrupt) schedules one durable alarm callback that replays the run, so runs never overlap, and a run that fails or is cut short is retried.

Under alchemy dev the same Durable Objects run locally in workerd, so workflows, sleeps and signals behave as they do when deployed.