diff --git a/src/content/changelog/workflows/2026-09-11-instance-event-subscriptions.mdx b/src/content/changelog/workflows/2026-09-11-instance-event-subscriptions.mdx new file mode 100644 index 00000000000..0a7ab032e61 --- /dev/null +++ b/src/content/changelog/workflows/2026-09-11-instance-event-subscriptions.mdx @@ -0,0 +1,37 @@ +--- +title: Stream Workflow instance events in your Worker or via the API with .subscribe() +description: Workflow instances now support a .subscribe() method, which let Workers and applications consume events as instances run. +products: + - workflows + - workers +date: 2026-09-11 12:00:00 UTC +--- + +import { TypeScriptExample } from "~/components"; + +You can now stream Workflow instance events via `WorkflowInstance.subscribe()` and the `GET /subscribe` API endpoint. Workers and HTTP clients can react to [workflow](/workflows/build/events-and-parameters/) and [step](/workflows/build/step-context/#workflowstepcontext) events, including attempts, sleeps, waits, and rollbacks, without polling for instance status. + +A subscription first streams the entire event history of the Workflow instance. After streaming past events, the subscription waits for new events as the instance runs. You can use `filter` to receive only specific event types or `cursor` to start a subscription at a specific event. + +Use `.subscribe()` to update Workflow status in user-facing dashboards, send notifications when steps complete, or trigger follow-up work for specific events. + + + +```ts +const instance = await env.MY_WORKFLOW.get("report-123"); + +using subscription = await instance.subscribe(); + +while (true) { + const { value, done } = await subscription.next(); + if (done) { + break; + } + + console.log(value.type, value); +} +``` + + + +For event types, available fields, and subscription options, refer to [Subscribe to events](/workflows/build/subscribe-to-instance-events/). diff --git a/src/content/docs/workflows/build/subscribe-to-instance-events.mdx b/src/content/docs/workflows/build/subscribe-to-instance-events.mdx new file mode 100644 index 00000000000..35439d95845 --- /dev/null +++ b/src/content/docs/workflows/build/subscribe-to-instance-events.mdx @@ -0,0 +1,261 @@ +--- +title: Subscribe to instance events +description: Use Cloudflare Workflows subscribe method to receive historical and live instance events without polling status. +pcx_content_type: concept +sidebar: + order: 7 +products: + - workflows +--- + +import { TypeScriptExample } from "~/components"; + +Use `WorkflowInstance.subscribe()` to receive events without polling [`status()`](/workflows/build/workers-api/#status). A subscription first delivers events recorded before you subscribed. After delivering these retained events, the subscription waits for new events as the instance runs. + +You can subscribe immediately after creating an instance. To subscribe later, retrieve the instance with [`get()`](/workflows/build/workers-api/#get). Subscriptions remain available during the [instance retention period](/workflows/reference/limits/). + +## Subscribe to all events + + + +```ts +interface Env { + MY_WORKFLOW: Workflow; +} + +export default { + async fetch(_request: Request, env: Env) { + const instance = await env.MY_WORKFLOW.create({ + params: { reportId: "report-123" }, + }); + + using subscription = await instance.subscribe(); + + while (true) { + const result = await subscription.next(); + if (result.done) { + break; + } + + console.log(result.value.type, result.value); + } + + return Response.json({ instanceId: instance.id }); + }, +} satisfies ExportedHandler; +``` + + + +The subscription ends when the instance emits `workflow_completed`, `workflow_errored`, or `workflow_terminated`. After a terminal event, each later `next()` call returns `done: true`. + +## Filter events + +Set `filter` to limit `next()` results to specific event types. + + + +```ts +const instance = await env.MY_WORKFLOW.get("report-123"); + +using subscription = await instance.subscribe({ + filter: ["workflow_completed", "workflow_errored", "workflow_terminated"], +}); + +const result = await subscription.next(); +if (result.done) { + throw new Error("The instance ended without a matching event."); +} + +switch (result.value.type) { + case "workflow_completed": + console.log("Workflow output:", result.value.output); + break; + case "workflow_errored": + console.error("Workflow errored:", result.value.error); + break; + case "workflow_terminated": + console.log("Workflow terminated."); + break; +} +``` + + + +A subscription ends even when its filter excludes a terminal event. In that case, `next()` returns `done: true` without the event. + +## Resume from a cursor + +Each event includes an `eventId`. To resume after a remote procedure call (RPC) fails, store the last processed event ID. Then pass that ID as `cursor`: + + + +```ts +using subscription = await instance.subscribe({ + cursor: lastProcessedEventId, + filter: ["step_completed", "workflow_completed", "workflow_errored"], +}); + +while (true) { + const result = await subscription.next(); + if (result.done) { + break; + } + + await processEvent(result.value); + await saveLastProcessedEventId(result.value.eventId); +} +``` + + + +The cursor identifies the last processed event. The subscription starts with the first event whose `eventId` is greater than the cursor. + +## Sensitive outputs + +For steps marked as sensitive, the `step_completed` event sets `output` to `"[REDACTED]"`. + +## Dispose of a subscription + +A subscription holds a Workers RPC resource. Disposing the subscription stops event delivery, clears its state, and releases its resources. + +Declare the subscription with `using` for automatic disposal when the scope exits, or call `subscription[Symbol.dispose]()` in a `finally` block. For more information, refer to [RPC lifecycle](/workers/runtime-apis/rpc/lifecycle/). + +## Event fields + +The public type definition shows the fields available on each event: + + + +```ts +type WorkflowInstanceEvent = { + instanceId: string; + eventId: number; + timestamp: number; +} & ( + | { type: "workflow_queued" } + | { type: "workflow_started"; params?: unknown } + | { type: "workflow_running" } + | { type: "workflow_paused" } + | { type: "workflow_waiting_for_pause" } + | { type: "workflow_waiting" } + | { type: "workflow_completed"; output?: unknown } + | { type: "workflow_errored"; error: { name: string; message: string } } + | { type: "workflow_terminated" } + | { + type: "step_started"; + stepName: string; + config?: { + retries: { + limit: number; + delay: WorkflowSleepDuration | "[dynamic]"; + backoff?: "constant" | "linear" | "exponential"; + }; + timeout: WorkflowSleepDuration; + sensitive?: "output"; + }; + } + | { type: "step_completed"; stepName: string; output?: unknown } + | { type: "step_errored"; stepName: string } + | { type: "attempt_started"; stepName: string; attempt: number } + | { type: "attempt_completed"; stepName: string; attempt: number } + | { + type: "attempt_errored"; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { name: string; message: string }; + } + | { type: "sleep_started"; stepName: string; durationMs: number } + | { type: "sleep_completed"; stepName: string } + | { type: "wait_started"; stepName: string; eventType: string } + | { type: "wait_completed"; stepName: string } + | { type: "wait_timed_out"; stepName: string } + | { type: "rollback_started" } + | { + type: "rollback_step_started"; + stepName: string; + config?: { + retries: { + limit: number; + delay: WorkflowSleepDuration | "[dynamic]"; + backoff?: "constant" | "linear" | "exponential"; + }; + timeout: WorkflowSleepDuration; + sensitive?: "output"; + }; + } + | { type: "rollback_step_completed"; stepName: string } + | { + type: "rollback_step_errored"; + stepName: string; + error: { name: string; message: string }; + } + | { type: "rollback_attempt_started"; stepName: string; attempt: number } + | { type: "rollback_attempt_completed"; stepName: string; attempt: number } + | { + type: "rollback_attempt_errored"; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { name: string; message: string }; + } + | { type: "rollback_completed" } + | { type: "rollback_errored" } +); +``` + + + +The following sections describe when each event is emitted. + +### Workflow lifecycle events + +| Event type | Emitted when | +| ---------------------------- | ------------------------------------------------ | +| `workflow_queued` | The instance enters the execution queue | +| `workflow_started` | The instance starts | +| `workflow_running` | The instance starts or resumes execution | +| `workflow_paused` | The instance pauses | +| `workflow_waiting_for_pause` | The instance waits for current work before pause | +| `workflow_waiting` | The instance enters waiting state | +| `workflow_completed` | The instance completes successfully | +| `workflow_errored` | The instance ends with an error | +| `workflow_terminated` | The instance is terminated | + +### Step and attempt events + +| Event type | Emitted when | +| ------------------- | ---------------------------- | +| `step_started` | A `step.do()` call starts | +| `step_completed` | A `step.do()` call completes | +| `step_errored` | A `step.do()` call errors | +| `attempt_started` | A step attempt starts | +| `attempt_completed` | A step attempt completes | +| `attempt_errored` | A step attempt errors | + +### Sleep and wait events + +| Event type | Emitted when | +| ----------------- | --------------------------------------------------- | +| `sleep_started` | A `step.sleep()` or `step.sleepUntil()` call starts | +| `sleep_completed` | A sleep finishes | +| `wait_started` | A `step.waitForEvent()` call starts | +| `wait_completed` | A matching event reaches `step.waitForEvent()` | +| `wait_timed_out` | A `step.waitForEvent()` call times out | + +### Rollback events + +| Event type | Emitted when | +| ---------------------------- | --------------------------------------- | +| `rollback_started` | The Workflow starts a rollback | +| `rollback_step_started` | A rollback handler starts | +| `rollback_step_completed` | A rollback handler completes | +| `rollback_step_errored` | A rollback handler errors | +| `rollback_attempt_started` | A rollback attempt starts | +| `rollback_attempt_completed` | A rollback attempt completes | +| `rollback_attempt_errored` | A rollback attempt errors | +| `rollback_completed` | All required rollback handlers complete | +| `rollback_errored` | The rollback operation errors | + +For method signatures and option types, refer to [`WorkflowInstance.subscribe()`](/workflows/build/workers-api/#subscribe). diff --git a/src/content/docs/workflows/build/trigger-workflows.mdx b/src/content/docs/workflows/build/trigger-workflows.mdx index 4ce080558ca..0c2490dc31a 100644 --- a/src/content/docs/workflows/build/trigger-workflows.mdx +++ b/src/content/docs/workflows/build/trigger-workflows.mdx @@ -187,6 +187,8 @@ The possible values of status are as follows: If your Workflow registers rollback handlers on `step.do()`, inspect `rollback` after the instance finishes to see whether the compensating steps completed successfully. While rollback is actively running, the Workers API continues to return `status: "running"`. +To receive historical and live execution updates without polling, refer to [Subscribe to events](/workflows/build/subscribe-to-instance-events/). + ### Explicitly pause a Workflow You can explicitly pause a Workflow instance (and later resume it) by calling `pause` against a specific instance ID. diff --git a/src/content/docs/workflows/build/workers-api.mdx b/src/content/docs/workflows/build/workers-api.mdx index 88588f78f32..e171bb58299 100644 --- a/src/content/docs/workflows/build/workers-api.mdx +++ b/src/content/docs/workflows/build/workers-api.mdx @@ -154,7 +154,6 @@ export class MyWorkflow extends WorkflowEntrypoint { - `name` - the name of the step. - `duration` - the duration to sleep for, as a `number` in milliseconds or as a `WorkflowDuration`-compatible string. - Refer to the [documentation on sleeping and retrying](/workflows/build/sleeping-and-retrying/) to learn more about how Workflows are retried. - - step.sleepUntil(name: string, timestamp: Date | number): Promise<void> - `name` - the name of the step. - `timestamp` - a JavaScript `Date` object or milliseconds from the Unix epoch to sleep the Workflow instance until. @@ -610,6 +609,12 @@ declare abstract class WorkflowInstance { * Returns the current status of the instance. */ public status(): Promise; + /** + * Subscribe to events from this Workflow instance. + */ + public subscribe( + options?: WorkflowInstanceSubscribeOptions, + ): Promise; } ``` @@ -760,6 +765,55 @@ You can call `sendEvent` multiple times, setting the value of the `type` propert This allows you to wait for multiple events at once, or use `Promise.race` to wait for multiple events and allow the first event to progress the Workflow. +### subscribe + +Subscribe to historical and live execution events from a Workflow instance. + +- subscribe(options?: WorkflowInstanceSubscribeOptions): Promise<WorkflowInstanceSubscription> + - `options` - optional properties that set the starting cursor and filter event types. + +The returned subscription provides a `next()` method. Each call returns the next matching `WorkflowInstanceEvent` or waits for one. The subscription ends when the instance completes, errors, or terminates. + +#### WorkflowInstanceSubscribeOptions + +```ts +interface WorkflowInstanceSubscribeOptions { + /** + * The event ID after which to start. + */ + cursor?: number; + /** + * Emit only events with one of these types. + */ + filter?: WorkflowInstanceEventType[]; +} + +type WorkflowInstanceEventType = WorkflowInstanceEvent["type"]; +``` + +Call `subscribe()` without options to receive all events. Use `cursor` and `filter` to control which events the subscription returns: + + + +```ts +const instance = await env.MY_WORKFLOW.get("abc-123"); + +// Subscribe to all events. +using allEvents = await instance.subscribe(); + +// Subscribe to events after event ID 100. +using eventsAfterCursor = await instance.subscribe({ cursor: 100 }); + +// Subscribe to selected event types. +using filteredEvents = await instance.subscribe({ + filter: ["workflow_completed", "workflow_errored"], +}); +``` + + + +For event types, filtering behavior, and cursor usage, refer to [Subscribe to events](/workflows/build/subscribe-to-instance-events/). + ### InstanceStatus Details the status of a Workflow instance.