feat: add Workflows support

Cargo.toml

@@ -75,7 +75,7 @@ web-sys = { version = "0.3.102", features = [
     "WritableStream",
     "WritableStreamDefaultWriter",
 ] }
-worker = { version = "0.8.5", path = "worker", features = ["queue", "d1", "axum", "timezone"] }
+worker = { version = "0.8.5", path = "worker", features = ["queue", "d1", "axum", "timezone", "workflow"] }
 worker-codegen = { path = "worker-codegen", version = "0.2.0" }
 worker-macros = { version = "0.8.5", path = "worker-macros", features = ["queue"] }
 worker-sys = { version = "0.8.5", path = "worker-sys", features = ["d1", "queue"] }

examples/workflow/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "workflow-example"
+version = "0.1.0"
+edition = "2021"
+
+[package.metadata.release]
+release = false
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+worker = { path = "../../worker", features = ["workflow"] }

examples/workflow/README.md

@@ -0,0 +1,69 @@
+# Workflows on Cloudflare Workers
+
+Durable, multi-step [Cloudflare Workflows](https://developers.cloudflare.com/workflows/) written in Rust with the `#[workflow]` macro and `worker::WorkflowEntrypoint`.
+
+`MyWorkflow` (in `src/lib.rs`) runs a small pipeline and exercises most of the Workflows surface:
+
+- **Durable steps.** `step.do_(...)` and `step.do_with_config(...)` persist their return values and replay them instead of re-running.
+- **Retries & backoff.** `StepConfig` / `RetryConfig` set a per-step `limit`, `delay`, and `Backoff` strategy. The `send-notification` step fails randomly to show retries in action.
+- **Non-retryable errors.** `NonRetryableError` fails the instance immediately (used when the email is invalid).
+- **Sleeping.** `step.sleep(...)` hibernates the instance (10 seconds here) without burning CPU.
+- **Saga rollback.** `step.do_with_rollback(...)` attaches a compensation handler that the runtime invokes if a later step fails. See `reserve-inventory`.
+- **Cron triggers.** When started from a `schedules` cron (see `wrangler.toml`), `event.schedule` and `event.workflow_name` are populated.
+
+A `fetch` handler is included so you can drive the workflow over HTTP.
+
+## Routes
+
+| Route | Description |
+|---|---|
+| `POST /workflow` | Create an instance. Body: `{ "email": "...", "name": "..." }`. Returns the instance `id`. |
+| `GET /workflow/{id}` | Get an instance's status, output, and error. |
+| `POST /workflow/{id}/pause` | Pause a running instance. |
+| `POST /workflow/{id}/resume` | Resume a paused instance. |
+
+## Running locally
+
+```sh
+npx wrangler dev
+```
+
+`wrangler dev` runs the `[build]` command from `wrangler.toml`, which invokes `worker-build`. If you don't have it, install it first with `cargo install worker-build` (and point the build command at it, or drop the `../../target/release/` prefix).
+
+> Saga rollbacks are a recent Workflows feature. Local development needs a current `wrangler`/`workerd`; an older toolchain will run the steps but skip the rollback.
+
+### Walkthrough
+
+1. Create an instance:
+
+   ```sh
+   curl -X POST http://localhost:8787/workflow \
+     -H 'Content-Type: application/json' \
+     -d '{"email":"[email protected]","name":"Ada"}'
+   # → {"id":"<instance-id>","message":"Workflow created"}
+   ```
+
+2. Poll its status (it sleeps for 10s mid-run, so give it a moment):
+
+   ```sh
+   curl http://localhost:8787/workflow/<instance-id>
+   # → {"id":"...","status":"Running","error":null,"output":null}
+   # ...later...
+   # → {"id":"...","status":"Complete","output":{"message":"Workflow completed for Ada","steps_completed":4}}
+   ```
+
+   Passing an email without an `@` makes the `validate-params` step throw a
+   `NonRetryableError`, so the instance ends up `Errored` instead.
+
+3. Pause / resume while it's running (e.g. during the 10s sleep):
+
+   ```sh
+   curl -X POST http://localhost:8787/workflow/<instance-id>/pause
+   curl -X POST http://localhost:8787/workflow/<instance-id>/resume
+   ```
+
+## Deploying
+
+```sh
+npx wrangler deploy
+```

examples/workflow/src/lib.rs

@@ -0,0 +1,225 @@
+use serde::{Deserialize, Serialize};
+use worker::*;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MyParams {
+    pub email: String,
+    pub name: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MyOutput {
+    pub message: String,
+    pub steps_completed: u32,
+}
+
+#[workflow]
+pub struct MyWorkflow {
+    #[allow(dead_code)]
+    env: Env,
+}
+
+impl WorkflowEntrypoint for MyWorkflow {
+    type Input = MyParams;
+    type Output = MyOutput;
+
+    fn new(_ctx: Context, env: Env) -> Self {
+        Self { env }
+    }
+
+    async fn run(&self, event: WorkflowEvent<MyParams>, step: WorkflowStep) -> Result<MyOutput> {
+        console_log!(
+            "Workflow '{}' started with instance ID: {}",
+            event.workflow_name,
+            event.instance_id
+        );
+
+        // When triggered by a cron schedule (configured via `schedules` on the
+        // workflow binding in wrangler.toml), `event.schedule` is populated.
+        if let Some(schedule) = &event.schedule {
+            console_log!("Triggered by cron schedule: {}", schedule.cron);
+        }
+
+        let params = event.payload;
+
+        let email_for_validation = params.email.clone();
+        step.do_with_config(
+            "validate-params",
+            StepConfig {
+                retries: Some(RetryConfig {
+                    limit: 3,
+                    delay: "1 second".into(),
+                    backoff: None,
+                }),
+                timeout: None,
+                sensitive: None,
+            },
+            move |ctx| {
+                let email = email_for_validation.clone();
+                async move {
+                    console_log!(
+                        "step '{}' attempt {}/{}",
+                        ctx.step.name,
+                        ctx.attempt,
+                        ctx.config.retries.limit
+                    );
+                    if !email.contains('@') {
+                        return Err(NonRetryableError::new("invalid email address").into());
+                    }
+                    Ok(serde_json::json!({ "valid": true }))
+                }
+            },
+        )
+        .await?;
+
+        let name_for_step1 = params.name.clone();
+        let step1_result = step
+            .do_("initial-processing", move |_ctx| {
+                let name = name_for_step1.clone();
+                async move {
+                    console_log!("Processing for user: {}", name);
+                    Ok(serde_json::json!({
+                        "processed": true,
+                        "user": name
+                    }))
+                }
+            })
+            .await?;
+
+        console_log!("Step 1 completed: {:?}", step1_result);
+
+        console_log!("Step 2: Sleeping for 10 seconds...");
+        step.sleep("wait-for-processing", "10 seconds").await?;
+
+        let email_for_step3 = params.email.clone();
+        let notification_result = step
+            .do_with_config(
+                "send-notification",
+                StepConfig {
+                    retries: Some(RetryConfig {
+                        limit: 3,
+                        delay: "5 seconds".into(),
+                        backoff: Some(Backoff::Exponential),
+                    }),
+                    timeout: Some("1 minute".into()),
+                    sensitive: None,
+                },
+                move |_ctx| {
+                    let email = email_for_step3.clone();
+                    async move {
+                        console_log!("Sending notification to: {}", email);
+                        if js_sys::Math::random() < 0.5 {
+                            return Err("notification service temporarily unavailable".into());
+                        }
+                        Ok(serde_json::json!({
+                            "notification_sent": true,
+                            "email": email
+                        }))
+                    }
+                },
+            )
+            .await?;
+
+        console_log!("Step 3 completed: {:?}", notification_result);
+
+        // Step 4: saga-style rollback. If a later step fails, the runtime
+        // invokes the rollback handler (in reverse order of completion) to undo
+        // this step's side effects. The handler receives the step's output, the
+        // original step context, and the error that triggered the rollback.
+        let reservation = step
+            .do_with_rollback(
+                "reserve-inventory",
+                |_ctx| async move {
+                    console_log!("Reserving inventory");
+                    Ok(serde_json::json!({ "reservation_id": "resv_123" }))
+                },
+                Rollback::new(
+                    |rollback: WorkflowRollbackContext<serde_json::Value>| async move {
+                        if let Some(output) = &rollback.output {
+                            console_log!(
+                                "Rolling back '{}' (caused by: {}) — releasing {}",
+                                rollback.ctx.step.name,
+                                rollback.error.message,
+                                output
+                            );
+                        }
+                        Ok(())
+                    },
+                ),
+            )
+            .await?;
+
+        console_log!("Step 4 completed: {:?}", reservation);
+
+        Ok(MyOutput {
+            message: format!("Workflow completed for {}", params.name),
+            steps_completed: 4,
+        })
+    }
+}
+
+#[event(fetch)]
+async fn fetch(mut req: Request, env: Env, _ctx: Context) -> Result<Response> {
+    let url = req.url()?;
+    let path = url.path();
+    let workflow = env.workflow("MY_WORKFLOW")?;
+
+    match (req.method(), path) {
+        (Method::Post, "/workflow") => {
+            let params: MyParams = req.json().await?;
+
+            let instance = workflow
+                .create(CreateOptions {
+                    params: Some(params),
+                    ..Default::default()
+                })
+                .await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id(),
+                "message": "Workflow created"
+            }))
+        }
+
+        (Method::Get, path) if path.starts_with("/workflow/") => {
+            let id = path.trim_start_matches("/workflow/");
+            let instance = workflow.get(id).await?;
+            let status = instance.status().await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id(),
+                "status": format!("{:?}", status.status),
+                "error": status.error,
+                "output": status.output
+            }))
+        }
+
+        (Method::Post, path) if path.starts_with("/workflow/") && path.ends_with("/pause") => {
+            let id = path
+                .trim_start_matches("/workflow/")
+                .trim_end_matches("/pause");
+            let instance = workflow.get(id).await?;
+            instance.pause().await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id(),
+                "message": "Workflow paused"
+            }))
+        }
+
+        (Method::Post, path) if path.starts_with("/workflow/") && path.ends_with("/resume") => {
+            let id = path
+                .trim_start_matches("/workflow/")
+                .trim_end_matches("/resume");
+            let instance = workflow.get(id).await?;
+            instance.resume().await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id(),
+                "message": "Workflow resumed"
+            }))
+        }
+
+        _ => Response::error("Not Found", 404),
+    }
+}

examples/workflow/wrangler.toml

@@ -0,0 +1,16 @@
+name = "workflow-example"
+main = "build/worker/shim.mjs"
+compatibility_date = "2026-04-21"
+
+[build]
+# For development: use local worker-build binary
+# For production: command = "cargo install -q worker-build && worker-build --release"
+command = "RUSTFLAGS='--cfg=web_sys_unstable_apis' ../../target/release/worker-build --release"
+
+[[workflows]]
+name = "my-workflow"
+binding = "MY_WORKFLOW"
+class_name = "MyWorkflow"
+# Optionally trigger this workflow on a cron schedule. When triggered this way,
+# `WorkflowEvent::schedule` is populated with the matching cron expression.
+# schedules = ["0 * * * *", "*/15 * * * *"]