ch21-workflows

對應 21. Workflows：可持久執行的多步驟流程

在 GitHub 上檢視·4 個檔案·10.7 KB

取得並執行

這個範例可以獨立 clone 執行，不依賴其他章節。

npx degit mrsuner/cloudflare-serverless-platform-manuals/examples/ch21-workflows
cd ch21-workflows
npm install

可用指令

npm run dev	# wrangler dev
npm run typecheck	# tsc --noEmit
npm run cf-typegen	# wrangler types --env-interface CloudflareBindings

說明

ch21 — Workflows

Probe project for chapter 21. Every route exists to surface a runtime behaviour that the official docs either omit or state incorrectly.

Run

npm install
npx wrangler types
npx wrangler dev --port 9023

Routes

Route	What it shows
/start?mode=<mode>&id=<id>[&retention=1]	Create an instance. retention=1 passes the undocumented retention create option.
/status?id=	instance.status(). Locally this includes __LOCAL_DEV_STEP_OUTPUTS, an array of every memoised step.do output.
/send?id=&type=	instance.sendEvent(). Omit type to send the matching approve event.
/control?id=&op=pause|resume|terminate|restart[&rollback=1][&from=<step>]	Lifecycle ops + Object.getOwnPropertyNames of the instance prototype.
/batch?n=&id=	createBatch. With id, produces deterministic ids <id>-0..n so you can re-run and observe idempotency.
/binding	Prototype of the Workflow binding, including the unsafe* introspection API.
/trace	Module-scope trace + side-effect log (deliberate anti-pattern; see below).
/clear	Reset the module-scope logs.

Modes

Mode	Exercises
ok	Baseline. Dumps WorkflowEvent keys and the WorkflowStepContext (step, attempt, config) including the runtime default retry policy.
retry-then-ok	Branches on ctx.attempt, not on module state. Three attempts ~1.09s apart with backoff: "constant".
dynamic-delay	retries.delay as a function of {ctx, error} — present in the types, absent from the docs.
nonretryable	NonRetryableError → terminal status: "errored" with error.name === "WorkflowFatalError".
sleep	step.sleep + step.sleepUntil. Locally status() reports running, not waiting.
wait	waitForEvent returns the envelope {payload, type, timestamp}, not the raw payload.
wait-timeout	3s timeout → plain Error: Execution timed out after 3000ms. No dedicated error class.
rollback	step.do(name, fn, { rollback }) saga compensation, fired automatically when a later step fails.
sensitive	sensitive: "output" persists the literal string "[REDACTED]".

Reproducing the key findings

# Runtime default retry policy (docs say delay: 10000; runtime says 1000)
curl -s "localhost:9023/start?mode=ok&id=a" >/dev/null && sleep 2
curl -s "localhost:9023/status?id=a" | jq '.ok.output.shape.ctx.config'

# waitForEvent envelope
curl -s "localhost:9023/start?mode=wait&id=b" >/dev/null && sleep 1
curl -s "localhost:9023/send?id=b" >/dev/null && sleep 2
curl -s "localhost:9023/status?id=b" | jq '.ok.output.received.keys'

# sendEvent with a mismatched type succeeds silently and the instance keeps waiting
curl -s "localhost:9023/start?mode=wait&id=c" >/dev/null && sleep 1
curl -s "localhost:9023/send?id=c&type=typo"
curl -s "localhost:9023/status?id=c" | jq '.ok.status'   # still "running"

# rollback fires automatically
curl -s "localhost:9023/clear" >/dev/null
curl -s "localhost:9023/start?mode=rollback&id=d" >/dev/null && sleep 3
curl -s "localhost:9023/trace" | jq '.sideEffects'        # ["charged","refunded"]

# sensitive output is persisted as "[REDACTED]"
curl -s "localhost:9023/start?mode=sensitive&id=e" >/dev/null && sleep 7
curl -s "localhost:9023/status?id=e" | jq '.ok.__LOCAL_DEV_STEP_OUTPUTS[2]'

# get() on an unknown id throws
curl -s "localhost:9023/status?id=nope" | jq

# restart from a specific step keeps earlier cached results
curl -s "localhost:9023/start?mode=ok&id=f" >/dev/null && sleep 2
curl -s "localhost:9023/status?id=f" | jq '.ok.__LOCAL_DEV_STEP_OUTPUTS[-1]'
curl -s "localhost:9023/control?id=f&op=restart&from=stamp" >/dev/null && sleep 3
curl -s "localhost:9023/status?id=f" | jq '.ok.__LOCAL_DEV_STEP_OUTPUTS[-1]'   # different

Deliberate anti-patterns in this code

src/index.ts keeps a module-scope trace array and a sideEffects array. Both violate rule 3 of the rules of Workflows (“do not rely on state outside of a step”). They are here so the demo can print ordering, and so you can observe the failure mode directly: run /batch?n=101 and then start a fresh instance — the module state is shared across every instance in the isolate. A Workflow instance is not a Durable Object; it gets no memory isolation.

Local dev caveats

• wrangler dev does not hibernate and replay run() across a step.sleep. The single most important correctness property of Workflows is therefore the one thing you cannot verify locally.
• status() never reports waiting locally.
• Duplicate create({id}) does not throw locally; the docs and the type annotation both say it does in production.
• createBatch accepts more than the documented 100 entries locally. Calling /batch?n=101 repeatedly can crash the dev server.
• pause() prints Uncaught Error: Aborting engine: User called pause to the dev server log. That is the abort mechanism working, not a bug.

原始碼

• wrangler.jsonc
• package.json
• src/index.ts
• tsconfig.json

wrangler.jsonc

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "ch21-workflows",
  "main": "src/index.ts",
  "compatibility_date": "2026-07-24",
  "observability": { "enabled": true },
  "workflows": [
    {
      "name": "ch21-report",
      "binding": "REPORT",
      "class_name": "ReportWorkflow",
      // Per-workflow step ceiling. NOT the same key as the top-level Worker
      // `limits` (cpu_ms / subrequests). Paid default 10_000, max 25_000.
      "limits": { "steps": 25000 }
      // Cron-triggered instances. Wrangler does NOT validate the expression
      // locally -- "not a cron" passes `deploy --dry-run` without a warning.
      // "schedules": ["0 * * * *"]
    }
  ]
}

package.json

{ "name": "ch21-workflows", "private": true, "type": "module",
  "scripts": { "dev": "wrangler dev", "typecheck": "tsc --noEmit",
    "cf-typegen": "wrangler types --env-interface CloudflareBindings" },
  "devDependencies": { "typescript": "^5.9.0", "wrangler": "^4.114.0" } }

src/index.ts

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

type Mode =
  | "ok"
  | "retry-then-ok"
  | "nonretryable"
  | "sleep"
  | "wait"
  | "wait-timeout"
  | "rollback"
  | "sensitive"
  | "dynamic-delay";

type Params = { tenantId: string; mode?: Mode };

// Module-scope log. This is DELIBERATELY an anti-pattern (rules of workflows #3:
// "do not rely on state outside of a step"). It exists so the demo can print
// ordering, and so we can show that it is shared across every instance in the
// isolate. Never use module state to make workflow decisions.
const trace: string[] = [];
const log = (s: string) => void trace.push(`${new Date().toISOString().slice(11, 23)} ${s}`);

// Side effects that a rollback handler will undo.
const sideEffects: string[] = [];

export class ReportWorkflow extends WorkflowEntrypoint<CloudflareBindings, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep): Promise<unknown> {
    const mode = event.payload.mode ?? "ok";
    log(`run start mode=${mode} instanceId=${event.instanceId}`);

    // What is actually on the event, the step object, and the step context?
    const shape = await step.do("inspect", async (ctx) => ({
      eventKeys: Object.keys(event),
      payload: event.payload,
      instanceId: event.instanceId,
      workflowName: event.workflowName,
      timestampIsDate: event.timestamp instanceof Date,
      hasSchedule: "schedule" in event,
      schedule: event.schedule ?? null,
      stepProto: Object.getOwnPropertyNames(Object.getPrototypeOf(step)),
      // The step callback receives a context. This is the ONLY correct way to
      // know which attempt you are on -- module state is not durable.
      ctxKeys: Object.keys(ctx),
      ctx,
    }));

    // A step with an explicit retry policy. Note we branch on ctx.attempt,
    // not on a module-scope counter.
    const retried = await step.do(
      "flaky",
      { retries: { limit: 3, delay: "1 second", backoff: "constant" }, timeout: "30 seconds" },
      async (ctx) => {
        log(`flaky attempt=${ctx.attempt} name=${ctx.step.name} count=${ctx.step.count}`);
        if (mode === "retry-then-ok" && ctx.attempt < 3) throw new Error("transient");
        return { attempt: ctx.attempt, config: ctx.config, step: ctx.step };
      },
    );

    // Retry delay can be a FUNCTION of the error. Not in the docs; only in the types.
    if (mode === "dynamic-delay") {
      await step.do(
        "dynamic",
        {
          retries: {
            limit: 4,
            delay: ({ ctx, error }) => {
              log(`delayFn attempt=${ctx.attempt} error=${error.message}`);
              return `${ctx.attempt} seconds`;
            },
            backoff: "constant",
          },
        },
        async (ctx) => {
          if (ctx.attempt < 3) throw new Error(`boom-${ctx.attempt}`);
          return { settledOn: ctx.attempt };
        },
      );
    }

    // sensitive: "output" -- undocumented WorkflowStepConfig field.
    // We read the step's value BEFORE and AFTER a sleep, because a sleep forces
    // the engine to hibernate and replay: the second read comes from storage.
    let sensitive: unknown = null;
    if (mode === "sensitive") {
      const secret = await step.do("secret", { sensitive: "output" }, async () => {
        log("secret step body ran");
        return { token: "sk-live-do-not-log" };
      });
      log(`secret in memory: ${JSON.stringify(secret)}`);
      // Sleeping forces the engine to hibernate. When run() replays, step.do("secret")
      // no longer executes its body -- it returns whatever was persisted.
      await step.sleep("hibernate", "3 seconds");
      sensitive = await step.do("observe", async () => ({ secretAfterReplay: secret }));
    }

    // Rollback handlers: compensating actions, the saga pattern.
    if (mode === "rollback") {
      await step.do(
        "charge",
        async () => {
          sideEffects.push("charged");
          return { chargeId: "ch_1" };
        },
        {
          rollback: async ({ ctx, error, output }) => {
            log(`rollback charge step=${ctx.step.name} err=${error.message} out=${JSON.stringify(output)}`);
            sideEffects.push("refunded");
          },
        },
      );
      await step.do("ship", async () => {
        throw new NonRetryableError("warehouse is on fire");
      });
    }

    if (mode === "nonretryable") {
      await step.do("fatal", async () => {
        log("fatal step");
        throw new NonRetryableError("this must not be retried");
      });
    }

    if (mode === "sleep") {
      log("sleeping");
      await step.sleep("nap", "3 seconds");
      log("woke up");
      await step.sleepUntil("until", Date.now() + 2000);
      log("woke up again");
    }

    let received: unknown = null;
    if (mode === "wait" || mode === "wait-timeout") {
      log("waiting for event");
      const timeout = mode === "wait-timeout" ? "3 seconds" : "30 seconds";
      try {
        const ev = await step.waitForEvent<{ by: string }>("approval", { type: "approve", timeout });
        received = { keys: Object.keys(ev), value: ev, timestampIsDate: ev.timestamp instanceof Date };
        log(`got event: ${JSON.stringify(ev)}`);
      } catch (e) {
        received = { threwName: (e as Error).name, threw: String(e).slice(0, 200) };
        log(`waitForEvent threw: ${String(e).slice(0, 120)}`);
      }
    }

    // Steps are memoised: this value is stable across retries of LATER steps.
    const stamped = await step.do("stamp", async () => Date.now());

    log("run end");
    return {
      shape,
      retried,
      received,
      sensitive,
      stamped,
      sideEffects: [...sideEffects],
      trace: [...trace],
    };
  }
}

const t = async (fn: () => Promise<unknown>): Promise<unknown> => {
  try {
    return { ok: (await fn()) ?? null };
  } catch (e) {
    return { threwName: (e as Error).name, threw: String(e).slice(0, 220) };
  }
};

export default {
  async fetch(request: Request, env: CloudflareBindings): Promise<Response> {
    const url = new URL(request.url);
    const q = url.searchParams;
    const wf = env.REPORT;

    switch (url.pathname) {
      case "/start": {
        const mode = (q.get("mode") ?? "ok") as Mode;
        const id = q.get("id") ?? undefined;
        return Response.json(
          await t(async () => {
            const inst = await wf.create({
              id,
              params: { tenantId: "t1", mode },
              // Undocumented on the docs site, present in the generated types.
              ...(q.get("retention")
                ? { retention: { successRetention: "1 hour", errorRetention: "2 hours" } }
                : {}),
            });
            return { id: inst.id, status: await inst.status() };
          }),
        );
      }

      case "/status":
        return Response.json(
          await t(async () => {
            const inst = await wf.get(q.get("id")!);
            return await inst.status();
          }),
        );

      case "/send": {
        const inst = await wf.get(q.get("id")!);
        return Response.json({
          sent: await t(() =>
            inst.sendEvent({ type: q.get("type") ?? "approve", payload: { by: "alice" } }),
          ),
        });
      }

      case "/control": {
        const op = q.get("op") ?? "status";
        const out: Record<string, unknown> = {};
        const inst = await wf.get(q.get("id")!);
        if (op === "pause") out.pause = await t(() => inst.pause());
        if (op === "resume") out.resume = await t(() => inst.resume());
        if (op === "terminate")
          out.terminate = await t(() =>
            inst.terminate(q.get("rollback") ? { rollback: true } : undefined),
          );
        if (op === "restart")
          out.restart = await t(() =>
            inst.restart(q.get("from") ? { from: { name: q.get("from")! } } : undefined),
          );
        out.status = await inst.status();
        out.instanceProto = Object.getOwnPropertyNames(Object.getPrototypeOf(inst));
        return Response.json(out);
      }

      case "/batch": {
        const n = Number(q.get("n") ?? 3);
        const id = q.get("id");
        return Response.json(
          await t(async () => {
            const created = await wf.createBatch(
              Array.from({ length: n }, (_, i) => ({
                ...(id ? { id: `${id}-${i}` } : {}),
                params: { tenantId: "t1", mode: "ok" as const },
              })),
            );
            return { count: created.length, ids: created.map((c) => c.id) };
          }),
        );
      }

      case "/binding":
        return Response.json({
          workflowProto: Object.getOwnPropertyNames(Object.getPrototypeOf(wf)),
          bindingName: await t(async () =>
            (wf as unknown as { unsafeGetBindingName(): string }).unsafeGetBindingName(),
          ),
        });

      case "/trace":
        return Response.json({ trace, sideEffects });
      case "/clear":
        trace.length = 0;
        sideEffects.length = 0;
        return Response.json({ ok: true });

      default:
        return new Response(
          "/start?mode=ok|retry-then-ok|nonretryable|sleep|wait|wait-timeout|rollback|sensitive|dynamic-delay&id=\n" +
            "/status?id=  /send?id=&type=  /control?id=&op=pause|resume|terminate|restart[&rollback=1][&from=step]\n" +
            "/batch?n=&id=  /binding  /trace  /clear\n",
          { status: 404 },
        );
    }
  },
} satisfies ExportedHandler<CloudflareBindings>;

tsconfig.json

{ "compilerOptions": { "target": "esnext", "lib": ["esnext"], "module": "esnext",
  "moduleResolution": "bundler", "types": ["./worker-configuration.d.ts"],
  "strict": true, "skipLibCheck": true, "noEmit": true, "isolatedModules": true },
  "include": ["src/**/*.ts", "worker-configuration.d.ts"] }