订阅实例事件

使用 WorkflowInstance.subscribe() 接收事件，而无需轮询 status()。订阅会先投递你订阅之前已记录的事件。投递这些已保留事件后，订阅会在实例运行时等待新事件。

你可以在创建实例后立即订阅。若要稍后订阅，请使用 get() 检索实例。订阅在实例保留期内仍然可用。

订阅所有事件

export default {
	async fetch(_request, 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 });
	},
};

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<Env>;

当实例发出 workflow_completed、workflow_errored 或 workflow_terminated 时，订阅结束。在终端事件之后，后续每次 next() 调用都会返回 done: true。

过滤事件

设置 filter，将 next() 结果限制为特定事件类型。

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;
}

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;
}

即使过滤器排除了终端事件，订阅也会结束。在这种情况下，next() 会返回 done: true，且不包含该事件。

从游标恢复

每个事件都包含 eventId。要在远程过程调用 (RPC) 失败后恢复，请存储最后处理的事件 ID。然后将该 ID 作为 cursor 传入：

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);
}

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);
}

游标标识最后处理的事件。订阅从 eventId 大于该游标的第一个事件开始。

敏感输出

对于标记为敏感的步骤，step_completed 事件会将 output 设置为 "[REDACTED]"。

释放订阅

订阅会持有 Workers RPC 资源。释放订阅会停止事件投递、清除其状态并释放其资源。

使用 using 声明订阅，以便在作用域退出时自动释放，或在 finally 块中调用 subscription[Symbol.dispose]()。更多信息请参考 RPC 生命周期。

事件字段

公开类型定义展示了每个事件上可用的字段：

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" }
);

以下各节说明每种事件何时发出。

Workflow 生命周期事件

事件类型	发出时机
`workflow_queued`	实例进入执行队列
`workflow_started`	实例启动
`workflow_running`	实例开始或恢复执行
`workflow_paused`	实例暂停
`workflow_waiting_for_pause`	实例在暂停前等待当前工作完成
`workflow_waiting`	实例进入等待状态
`workflow_completed`	实例成功完成
`workflow_errored`	实例以错误结束
`workflow_terminated`	实例被终止

步骤与尝试事件

事件类型	发出时机
`step_started`	`step.do()` 调用开始
`step_completed`	`step.do()` 调用完成
`step_errored`	`step.do()` 调用出错
`attempt_started`	步骤尝试开始
`attempt_completed`	步骤尝试完成
`attempt_errored`	步骤尝试出错

休眠与等待事件

事件类型	发出时机
`sleep_started`	`step.sleep()` 或 `step.sleepUntil()` 调用开始
`sleep_completed`	休眠结束
`wait_started`	`step.waitForEvent()` 调用开始
`wait_completed`	匹配事件到达 `step.waitForEvent()`
`wait_timed_out`	`step.waitForEvent()` 调用超时

回滚事件

事件类型	发出时机
`rollback_started`	Workflow 开始回滚
`rollback_step_started`	回滚处理程序开始
`rollback_step_completed`	回滚处理程序完成
`rollback_step_errored`	回滚处理程序出错
`rollback_attempt_started`	回滚尝试开始
`rollback_attempt_completed`	回滚尝试完成
`rollback_attempt_errored`	回滚尝试出错
`rollback_completed`	所有必需的回滚处理程序完成
`rollback_errored`	回滚操作出错

有关方法签名和选项类型，请参考 WorkflowInstance.subscribe()。