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 AWS they run on Lambda durable functions: each run is one durable execution.

The same workflow code runs on Cloudflare too. See Effect workflows on Cloudflare.

Define a workflow

Workflow.make declares the workflow, and Workflows.layer implements it. Each step is an Activity: a named Effect whose result is stored the first time it completes.

src/orders.ts

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

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

export const PlaceOrderLive = Workflows.layer(PlaceOrder, ({ orderId }) =>
  Activity.make({
    name: "charge",
    success: Schema.String,
    execute: Effect.succeed(`charged ${orderId}`),
  }),
);

Host it in a workflow function

AWS.Lambda.WorkflowFunction is a durable function that runs workflows. Declare the workflows it runs, then deploy it with the Layer that implements them:

src/OrderWorkflows.ts

import * as AWS from "alchemy/AWS";
import { PlaceOrder, PlaceOrderLive } from "./orders.ts";

export class OrderWorkflows extends AWS.Lambda.WorkflowFunction<OrderWorkflows>()(
  "OrderWorkflows",
  { workflows: [PlaceOrder] },
) {}

export default OrderWorkflows.make(
  { main: import.meta.url, executionTimeout: "7 days" },
  PlaceOrderLive,
);

Each run is one durable execution of the function. executionTimeout caps how long one run may take in total, waits included, up to a year.

Host several workflows

One function can run many workflows. List them all, and merge their Layers:

src/OrderWorkflows.ts

import * as AWS from "alchemy/AWS";
import * as Layer from "effect/Layer";
import { PlaceOrder, PlaceOrderLive, RefundOrder, RefundOrderLive } from "./orders.ts";

export class OrderWorkflows extends AWS.Lambda.WorkflowFunction<OrderWorkflows>()(
  "OrderWorkflows",
  { workflows: [PlaceOrder, RefundOrder] },
) {}

export default OrderWorkflows.make(
  { main: import.meta.url, executionTimeout: "7 days" },
  Layer.mergeAll(PlaceOrderLive, RefundOrderLive),
);

The Layer must implement every workflow the function lists: leaving one out is a type error. A workflow’s child workflows run in the same function, so list them too.

Start a run from another function

AWS.Lambda.DurableFunctionEngine runs workflows on the functions that host them. Provide it to the calling function, and get a client with Workflows.RunWorkflow:

src/Api.ts

import * as AWS from "alchemy/AWS";
import * as Workflows from "alchemy/Workflows";
import * as Effect from "effect/Effect";
import * as HttpServerResponse from "effect/http/HttpServerResponse";
import { OrderWorkflows } from "./OrderWorkflows.ts";
import { PlaceOrder } from "./orders.ts";

export class Api extends AWS.Lambda.Function<Api>()("Api") {}

export default Api.make(
  { main: import.meta.url, functionUrl: true },
  Effect.gen(function* () {
    const placeOrder = yield* Workflows.RunWorkflow(PlaceOrder);
    return {
      fetch: Effect.gen(function* () {
        const run = yield* placeOrder.start({ orderId: "o-1" });
        return HttpServerResponse.text(run.executionId);
      }),
    };
  }).pipe(Effect.provide(AWS.Lambda.DurableFunctionEngine(OrderWorkflows))),
);

The engine grants the caller what it needs on OrderWorkflows: invoking it, and reading and signalling its durable executions. start returns as soon as the run is started. Starting again with the same key returns the run already started, so a retried request never starts a second one.

Running a workflow that none of the engine’s functions list is a type error.

Route across functions

Pass every function whose workflows the caller runs. Each run goes to the function that hosts its workflow:

Effect.provide(AWS.Lambda.DurableFunctionEngine(OrderWorkflows, BillingWorkflows))

Read a run

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

const run = placeOrder.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.

Sleep durably

DurableClock.sleep is a durable wait: the invocation ends, nothing is billed while it waits, and Lambda invokes the function again when it’s due.

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. Durable sleeps round up to whole seconds.

Wait for a signal

A DurableDeferred is a value a run waits for, delivered from outside. The run waits on a Lambda callback 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* approve.get(executionId).signal(Approval, "agent-7");

The engine finds the run’s callback in its execution history. A callback only exists once the run reaches its wait, so signal retries for up to a few minutes while the run gets there.

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 wakes a run that’s waiting on a signal, which then fails where it waits and runs its compensations. A run that’s busy in a step can only be stopped, with StopDurableExecution, and its compensations don’t run.

See what a run did

Every step is a named operation in the durable execution, so the Lambda console and GetDurableExecutionHistory show which steps ran, with their names (charge/1: the step and its attempt), results and timings.