Workflows

Some work doesn’t fit in one request. Welcome a new user, wait a day, ask them for feedback, wait for the answer, then record it. A workflow is Cloudflare’s tool for that: a body whose durable work is named steps, each done once and kept, that can sleep for months and wait for events.

In forge-effect, a workflow is a class, like every other resource. Its payload, its result, its error and its events are schemas:

import { Schema } from 'effect';

import { Kv, Workflow } from '@forge-effect/infra';

export class Users extends Kv.Namespace<Users>()('users') {}

export class UserMissing extends Schema.TaggedError<UserMissing>()('UserMissing', { userId: Schema.String }) {}

export class Onboarding extends Workflow.Make<Onboarding>()('onboarding', {
  payload: { userId: Schema.String },
  success: Schema.Struct({ name: Schema.String, score: Schema.Finite, at: Schema.Date }),
  error: UserMissing,
  events: { survey: Schema.Struct({ score: Schema.Finite }) },
  idempotencyKey: ({ userId }) => `onboarding-${userId}`,
}) {}

This page builds that onboarding workflow: one Worker runs it, another starts it and sends it the survey. It’s the app in test/fixtures/onboard, which has run live on Cloudflare.

Declare it

Each option above says one thing:

payload: what an instance starts with. Fields, as here, or any schema.
success: what it ends with. Left out, it ends with nothing.
error: how it fails, in a way you plan for. Left out, it never fails that way.
events: what it can wait for, by type, each with its schema.
idempotencyKey: an instance’s id, from its payload. With it, one user gets one onboarding however many times you start it.

Put it in a file of its own, onboarding.workflow.ts, with only schemas in it. The Worker that runs the workflow and the Worker that starts it both import this file, and never each other.

Write the body

implement is the body: an Effect, given the payload. Its durable work goes in steps:

import { Clock, Effect, Option, Schema } from 'effect';

import { Worker, Workflow } from '@forge-effect/infra';

import { Onboarding, UserMissing, Users } from './onboarding.workflow.ts';

// The user's name, or UserMissing.
const loadUser = (userId: string) =>
  Effect.gen(function* () {
    const name = yield* Effect.orDie((yield* Users).get(userId));
    return yield* Option.match(name, {
      onNone: () => Effect.fail(new UserMissing({ userId })),
      onSome: Effect.succeed,
    });
  });

// Keeps the score, and says when.
const record = (userId: string, score: number) =>
  Effect.gen(function* () {
    yield* Effect.orDie((yield* Users).put(`survey:${userId}`, String(score)));
    return new Date(yield* Clock.currentTimeMillis);
  });

export const onboarding = Onboarding.implement(({ userId }) =>
  Effect.gen(function* () {
    const name = yield* Workflow.run('load user', loadUser(userId), {
      error: UserMissing,
      fatal: () => true, // a missing user won't turn up on a retry
      retries: { limit: 3, delay: '1 second' },
    });
    yield* Workflow.sleep('a moment', '2 seconds');
    const { score } = yield* Onboarding.waitFor('survey', { timeout: '1 hour' }).pipe(
      Effect.catchTag('WaitTimedOut', () => Effect.succeed({ score: 0 })), // no answer: count it as 0
    );
    const at = yield* Workflow.run('record', record(userId, score), { success: Schema.Date });
    return { name, score, at };
  }),
);

export default Worker.make('jobs', { workflows: [onboarding] }).bind(Users);

The Worker jobs hosts the workflow: workflows: [onboarding]. It binds Users, because the steps use it. The compiler checks that, as it does for any handler.

Steps

Workflow.run(name, effect, options) is Cloudflare’s step.do. The effect runs once, its result is kept under the name, and from then on the kept result is returned instead of running it again. The options:

success: the result’s schema. The result is kept as JSON and decoded on the way back, so Schema.Date gives you a Date, not a string. Without one, the result is kept as Cloudflare copies it.
error: the failure’s schema. A failure it declares fails the step, and then the body, with that error, typed. Without one, a failure is a defect.
fatal: which declared failures not to retry. Here, a missing user stays missing, so UserMissing fails at once.
retries and timeout: Cloudflare’s own, with durations written the Effect way ('1 second').
rollback: how to undo the step, if the instance fails for good (below), with rollbackRetries and rollbackTimeout for the undo itself.

So in load user, a KV hiccup (a defect, from Effect.orDie) is retried up to three times, a second apart, and a missing user isn’t retried at all.

Sleeping and waiting

Workflow.sleep(name, duration) waits. It can wait for months; nothing of yours runs while it does.
Workflow.sleepUntil(name, date) waits until a moment.
Onboarding.waitFor(type, { timeout }) waits for an event of a type the workflow declares, and gives you its payload, decoded by the event’s schema. If none comes in time, it fails with WaitTimedOut (Cloudflare’s default timeout is 24 hours).

Each of these has a name too, as steps do.

Rolling back

A workflow that reserves a room and charges a card shouldn’t leave either behind when a later step fails for good. Give each step its undo:

const room = yield* Workflow.run('reserve a room', reserve(guest), {
  success: Room,
  rollback: (kept, cause) => Option.match(kept, { onNone: () => Effect.void, onSome: (r) => release(r.id) }),
});
const charge = yield* Workflow.run('charge the card', chargeCard(guest, room), {
  success: Charge,
  rollback: (kept) => Option.match(kept, { onNone: () => Effect.void, onSome: (c) => refund(c.id) }),
  rollbackRetries: { limit: 3, delay: '30 seconds', backoff: 'linear' },
});
yield* Workflow.run('email the guest', email(guest)); // fails for good: refund, then release

When the instance fails for good, after a step’s retries, or at once for a fatal failure, Cloudflare runs the undos of the steps that started, newest first. The failing step’s own undo comes first, if it has one. Each undo gets the step’s result, decoded by its success schema, or None if the step failed before it had one. It also gets cause.message, why the instance failed (a declared failure, as JSON). An undo is work, as a step’s is: it has the Worker’s services, it’s retried as rollbackRetries says, and it should be safe to run twice. One that still fails stops the rest.

instance.rollBack does the same by hand: it terminates the instance and runs its undos, with cause.message “Instance terminated during rollback”. While they run, its status is RollingBack. Measured: an instance in a 90-second sleep was undone and Terminated within seconds; one in an hour-long sleep stayed RollingBack, no undo run, for over an hour, past the end of its sleep (its Worker was redeployed meanwhile, which may be why). Roll back an instance that sleeps for long with care, and check its status.

In tests, Workflow.test runs the undos too, and says which ran, in order, as rollbacks. infra dev runs them as Cloudflare does.

Replay: why services live in steps

Here’s the one idea that makes workflows different from other code.

When an instance wakes from a sleep or a wait, Cloudflare doesn’t resume it where it was. It runs the body again from the start. Each step it has already done returns its kept result at once, so the body races through to where it left off, and carries on.

That means code outside a step runs every time. If it reads the clock or the database there, it can see something different on the second run, and decide on different steps. That’s a bug that only appears in production, days later. So forge-effect doesn’t allow it: the Worker’s services are only available inside a step. Try reading Users outside one:

export const onboarding = Onboarding.implement(({ userId }) =>
  Effect.gen(function* () {
    // Read outside a step: it runs again on every replay.
    const name = yield* Effect.orDie((yield* Users).get(userId));
    // …
  }),
);

export default Worker.make('jobs', { workflows: [onboarding] }).bind(Users);

The bind doesn’t compile (its type error says 'services outside a step'), and infra check puts it plainly:

error workflow-service-outside-step  …/jobs.ts:20:70
  a workflow's body uses Users outside a step, where it runs again on each replay
  fix: use Users in a step: Workflow.run('name', effect)

The same goes for anything that changes between runs, even without a service. To wait until tomorrow, work out “tomorrow” in a step, so it’s kept:

const tomorrow = yield* Workflow.run('when', Effect.map(Clock.currentTimeMillis, (now) => new Date(now + 86_400_000)), { success: Schema.Date });
yield* Workflow.sleepUntil('until tomorrow', tomorrow);

Two more rules follow from replay: a step’s name is what its result is kept by, so a name used twice in one run is a defect; and the body has to decide the same steps each time. Workflow.test checks that second one for you, below.

Start it, and steer it

Another Worker binds Onboarding and gets its client. Here’s the fixture’s api, trimmed:

const fetch = Effect.gen(function* () {
  const request = yield* Worker.Request;
  const url = new URL(request.url);
  const [, verb, id] = url.pathname.split('/');
  const onboarding = yield* Onboarding;
  switch (verb) {
    case 'start':
      return Response.json({ id: (yield* onboarding.create({ userId: id! })).id }, { status: 201 });
    case 'survey':
      yield* (yield* onboarding.get(id!)).send('survey', { score: Number(url.searchParams.get('score')) });
      return new Response(null, { status: 202 });
    case 'status':
      return Response.json(yield* (yield* onboarding.get(id!)).status);
    // …
  }
});

export default Worker.make('api', { fetch }).bind(Onboarding, Users);

create(payload) starts an instance, its payload encoded by the schema. Its id comes from idempotencyKey, so a second create for the same user gets the first instance back. create(payload, { id }) names one yourself.
get(id) finds an instance.
send(type, payload) sends it an event. The type must be one the workflow declares, and the payload must fit its schema, or it doesn’t compile.
status says where it is, as a tagged union: Running, Paused, Waiting and so on; Complete with its output decoded (so at is a Date again); Errored with its declared error, as an Option of a UserMissing.
pause, resume, terminate, and restart, which starts it again from the top, its steps’ results forgotten; and rollBack, which terminates it and runs its steps’ undos (Rolling back).

const instance = yield* (yield* Onboarding).get(id);
yield* instance.pause;
yield* instance.resume;
yield* instance.restart; // from the top, its steps' results forgotten
yield* instance.terminate;
yield* instance.rollBack; // terminated, its steps undone, newest first

Test it

Workflow.test runs a body in memory, as Cloudflare would: sleeps pass at once, each wait takes the next event of its type, and then it runs the whole thing again, with every step kept, the way a replay does. You provide what the steps use as Layers:

test('onboards Ada', async () => {
  const ran = await Effect.runPromise(
    Workflow.test(onboarding, { userId: 'u1' }, { events: [{ type: 'survey', payload: { score: 9 } }] }).pipe(
      Effect.provide(Kv.memory(Users, [['u1', 'Ada']])), // what the steps use
    ),
  );
  assert.ok(Exit.isSuccess(ran.exit));
  assert.equal((ran.exit.value as { name: string }).name, 'Ada');
  assert.deepEqual(ran.steps, ['load user', 'a moment', 'wait for survey', 'record']);
  assert.deepEqual(ran.sleeps, ['a moment']);
});

You get how it ended (exit), the steps it did in order, and what it slept on. The replay is the clever part. Here’s a body that decides its step outside one, from the workflow tests:

class Coin extends Workflow.Make<Coin>()('coin', { payload: { n: Schema.Finite } }) {}
let flips = 0;
const flaky = Coin.implement(() =>
  Effect.gen(function* () {
    // Read outside a step: different on replay.
    const side = flips++ % 2 === 0 ? 'heads' : 'tails';
    yield* Workflow.run(`flip ${side}`, Effect.void);
  }),
);

It passes the first time and fails on the replay, naming the problem:

on replay, the body's step 1 was "flip tails", where it was "flip heads": what decides its steps has to be the same each time, so put it in a step

That’s the production-only bug from above, caught in a unit test.

Schedules, and the instance’s Event

A workflow can start itself on a schedule. The body’s second argument is the Event: the instance’s id, when it was created, and, for one a schedule started, the cron and when it was due. Here’s the fixture’s Digest, which runs every minute:

export class Digest extends Workflow.Make<Digest>()('digest', {
  payload: { note: Schema.optional(Schema.String) },
  success: Schema.String,
  schedules: ['* * * * *'],
  limits: { steps: 100 },
  concurrency: { limit: 2 },
  defaultRetention: { successRetention: '1 day', errorRetention: '3 days' },
}) {}

// What started it, a cron or a create, noted in KV.
export const digest = Digest.implement(({ note }, event) =>
  Workflow.run(
    'note',
    Effect.gen(function* () {
      const started = Option.match(event.schedule, {
        onNone: () => note ?? 'by hand',
        onSome: ({ cron, scheduledTime }) => `${cron} at ${scheduledTime.toISOString()}`,
      });
      yield* (yield* Users).put('digest:last', started);
      return started;
    }).pipe(Effect.orDie),
  ),
);

The settings are Cloudflare’s, by Cloudflare’s names:

schedules: crons, in UTC, each starting an instance. They’re checked as a Worker’s crons are (cron-invalid).
limits.steps: the most steps one instance may run (Cloudflare’s default is 10,000).
concurrency.limit: how many instances run at once; the rest queue.
defaultRetention: how long finished instances are kept, as durations.

A scheduled instance has no payload, so its body gets {}. That means a workflow with schedules needs a payload that takes {} (its fields optional, like note here). One that doesn’t is refused where it’s declared:

the workflow nightly has schedules, which start instances with no payload, so its payload has to take {}: make its fields optional

To test a scheduled run, pass the cron: Workflow.test(digest, undefined, { schedule: { cron: '* * * * *' } }).

Deploying

One Worker of the app hosts each workflow; others bind it. infra graph shows both sides:

$ pnpm infra graph test/fixtures/onboard/app.ts
onboard-jobs (workflow)
  runs the workflow onboarding as Onboarding
  runs the workflow digest as Digest
  USERS: kv onboard-users
onboard-api (fetch)
  ONBOARDING: workflow onboard-onboarding
  USERS: kv onboard-users

A few things happen for you when you ship:

Order. The host deploys before the Workers that bind it, so an instance started at once finds the workflow.
Registration. Cloudflare creates a workflow apart from its Worker, so ship registers it after the upload, as <app>-<name> (onboard-onboarding), with its settings, and records it as the app’s. Each ship registers it again, so a schedule you remove is removed.
Teardown. destroy and prune delete it. A workflow of that name the app didn’t make stops the ship (name-taken).
Locally. infra dev runs the app’s own workflows in workerd: start, sleep, wait and complete, all on your machine.

A workflow another app hosts is bound with from, once that app has deployed it:

export default Worker.make('web', { fetch }).bind(Onboarding.from('onboard'));

The caller’s plan checks it’s there first. See Workers calling Workers for how from works across apps.

Good to know

Checked live. The onboard fixture shipped; an instance started, slept, took its event and completed, its output decoded. One for a missing user errored with UserMissing, decoded. A second start for the same user got the first. The every-minute Digest started within the minute with its cron in the Event, and a ship without the schedule removed it. The destroy left no workflow behind.
A waiting instance reads Running. While an instance waits for an event, Cloudflare (and infra dev) report it as Running, not Waiting. Don’t poll for Waiting.
Early events are kept. An event sent before the instance reaches its waitFor is kept for it, so there’s no race.
Not yet: rollback steps, retention per instance, instance lifecycle events to a queue (Cloudflare’s event subscriptions), and schedules under infra dev, where a schedule doesn’t fire.

Durable Objects: state with an address, for when you need one thing that many requests talk to.
Queues: for work that’s one step, not many.