From 5f6f60588e18a2289c053f50966af3c28dcc6616 Mon Sep 17 00:00:00 2001 From: rcorreia Date: Fri, 21 Aug 2026 17:33:01 +0100 Subject: [PATCH 1/3] Add workflows_instance_subscribe experimental compatibility flag --- src/workerd/io/compatibility-date.capnp | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/workerd/io/compatibility-date.capnp b/src/workerd/io/compatibility-date.capnp index fb7b6a74516..2dccce86499 100644 --- a/src/workerd/io/compatibility-date.capnp +++ b/src/workerd/io/compatibility-date.capnp @@ -1657,4 +1657,9 @@ struct CompatibilityFlags @0x8f8c1b68151b6cef { $experimental $pythonSnapshotRelease; # Enables Python Workers using Pyodide 314.0.5. + + workflowsInstanceSubscribe @188 :Bool + $compatEnableFlag("workflows_instance_subscribe") + $experimental; + # Enables the experimental WorkflowInstance.subscribe() API. } From c9f81d2dcdc425e6f8b275557ac9b9cc2c36ffbe Mon Sep 17 00:00:00 2001 From: rcorreia Date: Fri, 21 Aug 2026 17:34:51 +0100 Subject: [PATCH 2/3] Add workflows instance subscribe fetcher function and tests --- .../internal/test/workflows/BUILD.bazel | 6 ++ .../workflows-api-subscribe-test.wd-test | 42 ++++++++++ .../test/workflows/workflows-api-test.js | 77 +++++++++++++++++++ .../internal/test/workflows/workflows-mock.js | 47 ++++++++++- src/cloudflare/internal/workflows-api.ts | 23 ++++++ src/cloudflare/internal/workflows.d.ts | 73 ++++++++++++++++++ 6 files changed, 267 insertions(+), 1 deletion(-) create mode 100644 src/cloudflare/internal/test/workflows/workflows-api-subscribe-test.wd-test diff --git a/src/cloudflare/internal/test/workflows/BUILD.bazel b/src/cloudflare/internal/test/workflows/BUILD.bazel index f582955f9f0..a756bd01aa3 100644 --- a/src/cloudflare/internal/test/workflows/BUILD.bazel +++ b/src/cloudflare/internal/test/workflows/BUILD.bazel @@ -4,3 +4,9 @@ wd_test( src = "workflows-api-test.wd-test", data = glob(["*.js"]), ) + +wd_test( + src = "workflows-api-subscribe-test.wd-test", + args = ["--experimental"], + data = glob(["*.js"]), +) diff --git a/src/cloudflare/internal/test/workflows/workflows-api-subscribe-test.wd-test b/src/cloudflare/internal/test/workflows/workflows-api-subscribe-test.wd-test new file mode 100644 index 00000000000..04315f3c5ea --- /dev/null +++ b/src/cloudflare/internal/test/workflows/workflows-api-subscribe-test.wd-test @@ -0,0 +1,42 @@ +using Workerd = import "/workerd/workerd.capnp"; + +# The workflows_instance_subscribe flag enables WorkflowInstance.subscribe(). No `rpc` flag is set +# so this also verifies that the wrapped binding's inner fetcher remains capable of making JSRPC +# calls while `fetcher_rpc` is off. + +const unitTests :Workerd.Config = ( + services = [ + ( name = "workflows-api-test", + worker = ( + modules = [ + (name = "worker", esModule = embed "workflows-api-test.js") + ], + compatibilityFlags = ["nodejs_compat", "workflows_instance_subscribe"], + bindings = [ + ( + name = "workflow", + wrapped = ( + moduleName = "cloudflare-internal:workflows-api", + innerBindings = [( + name = "fetcher", + service = "workflows-mock" + )], + ) + ), + ( + name = "mock", + service = "workflows-mock" + ) + ], + ) + ), + ( name = "workflows-mock", + worker = ( + compatibilityFlags = ["nodejs_compat"], + modules = [ + (name = "worker", esModule = embed "workflows-mock.js") + ], + ) + ) + ] +); diff --git a/src/cloudflare/internal/test/workflows/workflows-api-test.js b/src/cloudflare/internal/test/workflows/workflows-api-test.js index 222f07caa20..cca37bfbe3a 100644 --- a/src/cloudflare/internal/test/workflows/workflows-api-test.js +++ b/src/cloudflare/internal/test/workflows/workflows-api-test.js @@ -4,6 +4,9 @@ import * as assert from 'node:assert'; +const workflowsInstanceSubscribeEnabled = + !!Cloudflare.compatibilityFlags['workflows_instance_subscribe']; + // Every test is its own export: `workerd test` runs the `test()` handler of each entrypoint, so // extra methods hung off a single exported object would silently never run. @@ -17,6 +20,14 @@ async function getLastRestartBody(env, id) { return (await res.json()).result; } +async function getLastSubscribeOptions(env, id) { + const res = await env.mock.fetch('http://placeholder/last-subscribe', { + method: 'POST', + body: JSON.stringify({ id }), + }); + return (await res.json()).result; +} + export const workflowsApi = { async test(_, env) { { @@ -108,6 +119,7 @@ export const workflowsApi = { 'delete', 'status', 'sendEvent', + 'subscribe', ]) { assert.strictEqual(typeof fromGet[method], 'function'); } @@ -128,6 +140,71 @@ export const workflowsApi = { }, }; +export const subscribeNoOptions = { + async test(_, env) { + const instance = await env.workflow.get('subscribe-basic'); + if (!workflowsInstanceSubscribeEnabled) { + await assert.rejects(instance.subscribe(), { + message: + 'WorkflowInstance.subscribe() requires the workflows_instance_subscribe compatibility flag. Enable workflows_instance_subscribe before calling subscribe().', + }); + return; + } + + using subscription = await instance.subscribe(); + + assert.strictEqual(subscription[Symbol.asyncIterator](), subscription); + + const events = []; + for await (const event of subscription) { + events.push(event); + } + assert.deepStrictEqual(events, [ + { + instanceId: 'subscribe-basic', + eventId: 0, + timestamp: 0, + type: 'workflow_completed', + output: 'done', + }, + ]); + assert.strictEqual( + await getLastSubscribeOptions(env, 'subscribe-basic'), + null + ); + }, +}; + +export const subscribeAllOptions = { + async test(_, env) { + const instance = await env.workflow.get('subscribe-full'); + if (!workflowsInstanceSubscribeEnabled) { + await assert.rejects( + instance.subscribe({ + cursor: 1, + filter: ['workflow_queued', 'workflow_completed'], + }), + { + message: + 'WorkflowInstance.subscribe() requires the workflows_instance_subscribe compatibility flag. Enable workflows_instance_subscribe before calling subscribe().', + } + ); + return; + } + + using subscription = await instance.subscribe({ + cursor: 1, + filter: ['workflow_queued', 'workflow_completed'], + }); + + assert.strictEqual(subscription[Symbol.asyncIterator](), subscription); + assert.deepStrictEqual( + await getLastSubscribeOptions(env, 'subscribe-full'), + { cursor: 1, filter: ['workflow_queued', 'workflow_completed'] } + ); + }, +}; + export const restartNoOptions = { async test(_, env) { const instance = await env.workflow.get('restart-basic'); diff --git a/src/cloudflare/internal/test/workflows/workflows-mock.js b/src/cloudflare/internal/test/workflows/workflows-mock.js index ea062589f06..11297796435 100644 --- a/src/cloudflare/internal/test/workflows/workflows-mock.js +++ b/src/cloudflare/internal/test/workflows/workflows-mock.js @@ -2,13 +2,42 @@ // Licensed under the Apache 2.0 license found in the LICENSE file or at: // https://opensource.org/licenses/Apache-2.0 -import { WorkerEntrypoint } from 'cloudflare:workers'; +import { RpcTarget, WorkerEntrypoint } from 'cloudflare:workers'; const restartBodies = new Map(); +const subscribeOptions = new Map(); const THROW_ID = 'throw'; const MISSING_DELETE_ID = 'missing-delete'; +class SubscriptionMock extends RpcTarget { + #events; + #closed = false; + + constructor(events) { + super(); + this.#events = events; + } + + async next() { + if (this.#closed || this.#events.length === 0) { + this.#closed = true; + return { done: true, value: undefined }; + } + return { done: false, value: this.#events.shift() }; + } + + async return(value) { + this.#closed = true; + return { done: true, value }; + } + + async throw(error) { + this.#closed = true; + throw error; + } +} + export default class WorkflowsMock extends WorkerEntrypoint { async getInstance(id) { if (id === THROW_ID) { @@ -58,6 +87,20 @@ export default class WorkflowsMock extends WorkerEntrypoint { async sendEvent(_id, _event) {} + async subscribe(id, options) { + subscribeOptions.set(id, options ?? null); + + return new SubscriptionMock([ + { + instanceId: id, + eventId: 0, + timestamp: 0, + type: 'workflow_completed', + output: 'done', + }, + ]); + } + // Introspection only. The binding itself never uses fetch(), but the test worker's own compat // date leaves RPC gated on `env.mock`, so it reaches these records over HTTP instead. async fetch(request) { @@ -67,6 +110,8 @@ export default class WorkflowsMock extends WorkerEntrypoint { switch (pathname) { case '/last-restart': return Response.json({ result: restartBodies.get(data.id) ?? null }); + case '/last-subscribe': + return Response.json({ result: subscribeOptions.get(data.id) ?? null }); default: throw new Error( `unexpected HTTP request to the workflows mock: ${pathname}` diff --git a/src/cloudflare/internal/workflows-api.ts b/src/cloudflare/internal/workflows-api.ts index d7f736825dd..a7c0209eedd 100644 --- a/src/cloudflare/internal/workflows-api.ts +++ b/src/cloudflare/internal/workflows-api.ts @@ -4,6 +4,9 @@ import wrappedBinding from 'cloudflare-internal:wrapped-binding'; +const workflowsInstanceSubscribeEnabled = + !!Cloudflare.compatibilityFlags['workflows_instance_subscribe']; + export class NonRetryableError extends Error { constructor(message: string, name = 'NonRetryableError') { super(message); @@ -39,6 +42,10 @@ interface Fetcher { id: string, event: { type: string; payload: unknown } ): Promise; + subscribe( + id: string, + options?: WorkflowInstanceSubscribeOptions + ): Promise; } class InstanceImpl implements WorkflowInstance { @@ -85,6 +92,22 @@ class InstanceImpl implements WorkflowInstance { }): Promise { await this.#fetcher.sendEvent(this.id, { type, payload }); } + + async subscribe( + options?: WorkflowInstanceSubscribeOptions + ): Promise { + if (!workflowsInstanceSubscribeEnabled) { + throw new Error( + 'WorkflowInstance.subscribe() requires the workflows_instance_subscribe compatibility flag. Enable workflows_instance_subscribe before calling subscribe().' + ); + } + + const subscription = await this.#fetcher.subscribe(this.id, options); + Object.defineProperty(subscription, Symbol.asyncIterator, { + value: () => subscription, + }); + return subscription; + } } class WorkflowImpl extends wrappedBinding.WrappedBinding { diff --git a/src/cloudflare/internal/workflows.d.ts b/src/cloudflare/internal/workflows.d.ts index ce431131dfc..2e70479700c 100644 --- a/src/cloudflare/internal/workflows.d.ts +++ b/src/cloudflare/internal/workflows.d.ts @@ -149,6 +149,75 @@ interface WorkflowInstanceRestartOptions { }; } +type WorkflowInstanceEventCommon = { + instanceId: string; + eventId: number; + timestamp: number; +}; + +type WorkflowInstanceEvent = WorkflowInstanceEventCommon & + ( + | { type: 'workflow_queued' } + | { type: 'workflow_started'; params?: unknown } + | { type: 'workflow_completed'; output?: unknown } + | { type: 'workflow_failed'; error: { name: string; message: string } } + | { type: 'workflow_terminated' } + | { type: 'step_started'; stepName: string } + | { type: 'step_completed'; stepName: string; output?: unknown } + | { type: 'step_failed'; stepName: string } + | { type: 'attempt_started'; stepName: string; attempt: number } + | { type: 'attempt_completed'; stepName: string; attempt: number } + | { + type: 'attempt_failed'; + 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 } + | { type: 'rollback_step_completed'; stepName: string } + | { + type: 'rollback_step_failed'; + 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_failed'; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { name: string; message: string }; + } + | { type: 'rollback_completed' } + | { type: 'rollback_failed' } + ); + +type WorkflowInstanceEventType = WorkflowInstanceEvent['type']; + +type WorkflowInstanceSubscribeOptions = { + cursor?: number; + filter?: WorkflowInstanceEventType[]; +}; + +interface WorkflowInstanceSubscription extends Disposable { + next(): Promise>; + return( + value?: unknown + ): Promise>; + throw( + error?: unknown + ): Promise>; + [Symbol.asyncIterator](): WorkflowInstanceSubscription; +} + declare abstract class WorkflowInstance { id: string; @@ -190,4 +259,8 @@ declare abstract class WorkflowInstance { type: string; payload: unknown; }): Promise; + + subscribe( + options?: WorkflowInstanceSubscribeOptions + ): Promise; } From d0546dc856846c152b42eb3ce0c16c44d26e3abd Mon Sep 17 00:00:00 2001 From: rcorreia Date: Fri, 21 Aug 2026 18:12:13 +0100 Subject: [PATCH 3/3] Add workflows instance subscribe type definitions --- types/defines/workflows.d.ts | 84 ++++++++++ .../experimental/index.d.ts | 158 ++++++++++++++++++ .../generated-snapshot/experimental/index.ts | 158 ++++++++++++++++++ types/generated-snapshot/index.d.ts | 158 ++++++++++++++++++ types/generated-snapshot/index.ts | 158 ++++++++++++++++++ types/test/types/rpc.ts | 153 +++++++++++++++++ 6 files changed, 869 insertions(+) diff --git a/types/defines/workflows.d.ts b/types/defines/workflows.d.ts index 8b575ed1fa4..f3ad3cc081f 100644 --- a/types/defines/workflows.d.ts +++ b/types/defines/workflows.d.ts @@ -159,6 +159,82 @@ interface WorkflowInstanceRestartOptions { }; } +/** An event emitted during the execution of the Workflow instance. */ +type WorkflowInstanceEvent = { + /** The ID of the Workflow instance that emitted the event. */ + instanceId: string; + /** The event ID. */ + eventId: number; + /** The event timestamp. */ + timestamp: number; +} & + ( + | { type: 'workflow_queued' } + | { type: 'workflow_started'; params?: unknown } + | { type: 'workflow_completed'; output?: unknown } + | { type: 'workflow_failed'; error: { name: string; message: string } } + | { type: 'workflow_terminated' } + | { type: 'step_started'; stepName: string } + | { type: 'step_completed'; stepName: string; output?: unknown } + | { type: 'step_failed'; stepName: string } + | { type: 'attempt_started'; stepName: string; attempt: number } + | { type: 'attempt_completed'; stepName: string; attempt: number } + | { + type: 'attempt_failed'; + 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 } + | { type: 'rollback_step_completed'; stepName: string } + | { + type: 'rollback_step_failed'; + 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_failed'; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { name: string; message: string }; + } + | { type: 'rollback_completed' } + | { type: 'rollback_failed' } + ); + +/** A Workflow event type accepted by a subscription filter. */ +type WorkflowInstanceEventType = WorkflowInstanceEvent['type']; + +/** Options controlling a Workflow instance event subscription. */ +type WorkflowInstanceSubscribeOptions = { + /** The event cursor from which to start the subscription from. */ + cursor?: number; + /** Emit only events with one of these types. */ + filter?: WorkflowInstanceEventType[]; +}; + +/** A instance of a subscription to a specific workflow instance. */ +interface WorkflowInstanceSubscription extends Disposable { + next(): Promise>; + return( + value?: unknown + ): Promise>; + throw( + error?: unknown + ): Promise>; + [Symbol.asyncIterator](): WorkflowInstanceSubscription; +} + declare abstract class WorkflowInstance { public id: string; @@ -205,4 +281,12 @@ declare abstract class WorkflowInstance { type: string; payload: unknown; }): Promise; + + /** + * Subscribe to execution events from this Workflow instance. + * @param options Options controlling the starting cursor and event type filter. + */ + public subscribe( + options?: WorkflowInstanceSubscribeOptions + ): Promise; } diff --git a/types/generated-snapshot/experimental/index.d.ts b/types/generated-snapshot/experimental/index.d.ts index 768fd5206e2..ba314db838a 100755 --- a/types/generated-snapshot/experimental/index.d.ts +++ b/types/generated-snapshot/experimental/index.d.ts @@ -17473,6 +17473,157 @@ interface WorkflowInstanceRestartOptions { type?: "do" | "sleep" | "waitForEvent"; }; } +/** An event emitted during the execution of the Workflow instance. */ +type WorkflowInstanceEvent = { + /** The ID of the Workflow instance that emitted the event. */ + instanceId: string; + /** The event ID. */ + eventId: number; + /** The event timestamp. */ + timestamp: number; +} & ( + | { + type: "workflow_queued"; + } + | { + type: "workflow_started"; + params?: unknown; + } + | { + type: "workflow_completed"; + output?: unknown; + } + | { + type: "workflow_failed"; + error: { + name: string; + message: string; + }; + } + | { + type: "workflow_terminated"; + } + | { + type: "step_started"; + stepName: string; + } + | { + type: "step_completed"; + stepName: string; + output?: unknown; + } + | { + type: "step_failed"; + stepName: string; + } + | { + type: "attempt_started"; + stepName: string; + attempt: number; + } + | { + type: "attempt_completed"; + stepName: string; + attempt: number; + } + | { + type: "attempt_failed"; + 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; + } + | { + type: "rollback_step_completed"; + stepName: string; + } + | { + type: "rollback_step_failed"; + 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_failed"; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { + name: string; + message: string; + }; + } + | { + type: "rollback_completed"; + } + | { + type: "rollback_failed"; + } +); +/** A Workflow event type accepted by a subscription filter. */ +type WorkflowInstanceEventType = WorkflowInstanceEvent["type"]; +/** Options controlling a Workflow instance event subscription. */ +type WorkflowInstanceSubscribeOptions = { + /** The event cursor from which to start the subscription from. */ + cursor?: number; + /** Emit only events with one of these types. */ + filter?: WorkflowInstanceEventType[]; +}; +/** A instance of a subscription to a specific workflow instance. */ +interface WorkflowInstanceSubscription extends Disposable { + next(): Promise>; + return( + value?: unknown, + ): Promise>; + throw( + error?: unknown, + ): Promise>; + [Symbol.asyncIterator](): WorkflowInstanceSubscription; +} declare abstract class WorkflowInstance { public id: string; /** @@ -17512,4 +17663,11 @@ declare abstract class WorkflowInstance { type: string; payload: unknown; }): Promise; + /** + * Subscribe to execution events from this Workflow instance. + * @param options Options controlling the starting cursor and event type filter. + */ + public subscribe( + options?: WorkflowInstanceSubscribeOptions, + ): Promise; } diff --git a/types/generated-snapshot/experimental/index.ts b/types/generated-snapshot/experimental/index.ts index 13f54e10e9d..7dd3a1fbfbc 100755 --- a/types/generated-snapshot/experimental/index.ts +++ b/types/generated-snapshot/experimental/index.ts @@ -17422,6 +17422,157 @@ export interface WorkflowInstanceRestartOptions { type?: "do" | "sleep" | "waitForEvent"; }; } +/** An event emitted during the execution of the Workflow instance. */ +export type WorkflowInstanceEvent = { + /** The ID of the Workflow instance that emitted the event. */ + instanceId: string; + /** The event ID. */ + eventId: number; + /** The event timestamp. */ + timestamp: number; +} & ( + | { + type: "workflow_queued"; + } + | { + type: "workflow_started"; + params?: unknown; + } + | { + type: "workflow_completed"; + output?: unknown; + } + | { + type: "workflow_failed"; + error: { + name: string; + message: string; + }; + } + | { + type: "workflow_terminated"; + } + | { + type: "step_started"; + stepName: string; + } + | { + type: "step_completed"; + stepName: string; + output?: unknown; + } + | { + type: "step_failed"; + stepName: string; + } + | { + type: "attempt_started"; + stepName: string; + attempt: number; + } + | { + type: "attempt_completed"; + stepName: string; + attempt: number; + } + | { + type: "attempt_failed"; + 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; + } + | { + type: "rollback_step_completed"; + stepName: string; + } + | { + type: "rollback_step_failed"; + 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_failed"; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { + name: string; + message: string; + }; + } + | { + type: "rollback_completed"; + } + | { + type: "rollback_failed"; + } +); +/** A Workflow event type accepted by a subscription filter. */ +export type WorkflowInstanceEventType = WorkflowInstanceEvent["type"]; +/** Options controlling a Workflow instance event subscription. */ +export type WorkflowInstanceSubscribeOptions = { + /** The event cursor from which to start the subscription from. */ + cursor?: number; + /** Emit only events with one of these types. */ + filter?: WorkflowInstanceEventType[]; +}; +/** A instance of a subscription to a specific workflow instance. */ +export interface WorkflowInstanceSubscription extends Disposable { + next(): Promise>; + return( + value?: unknown, + ): Promise>; + throw( + error?: unknown, + ): Promise>; + [Symbol.asyncIterator](): WorkflowInstanceSubscription; +} export declare abstract class WorkflowInstance { public id: string; /** @@ -17461,4 +17612,11 @@ export declare abstract class WorkflowInstance { type: string; payload: unknown; }): Promise; + /** + * Subscribe to execution events from this Workflow instance. + * @param options Options controlling the starting cursor and event type filter. + */ + public subscribe( + options?: WorkflowInstanceSubscribeOptions, + ): Promise; } diff --git a/types/generated-snapshot/index.d.ts b/types/generated-snapshot/index.d.ts index a67b42aea2e..1eeddb879e5 100755 --- a/types/generated-snapshot/index.d.ts +++ b/types/generated-snapshot/index.d.ts @@ -17198,6 +17198,157 @@ interface WorkflowInstanceRestartOptions { type?: "do" | "sleep" | "waitForEvent"; }; } +/** An event emitted during the execution of the Workflow instance. */ +type WorkflowInstanceEvent = { + /** The ID of the Workflow instance that emitted the event. */ + instanceId: string; + /** The event ID. */ + eventId: number; + /** The event timestamp. */ + timestamp: number; +} & ( + | { + type: "workflow_queued"; + } + | { + type: "workflow_started"; + params?: unknown; + } + | { + type: "workflow_completed"; + output?: unknown; + } + | { + type: "workflow_failed"; + error: { + name: string; + message: string; + }; + } + | { + type: "workflow_terminated"; + } + | { + type: "step_started"; + stepName: string; + } + | { + type: "step_completed"; + stepName: string; + output?: unknown; + } + | { + type: "step_failed"; + stepName: string; + } + | { + type: "attempt_started"; + stepName: string; + attempt: number; + } + | { + type: "attempt_completed"; + stepName: string; + attempt: number; + } + | { + type: "attempt_failed"; + 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; + } + | { + type: "rollback_step_completed"; + stepName: string; + } + | { + type: "rollback_step_failed"; + 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_failed"; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { + name: string; + message: string; + }; + } + | { + type: "rollback_completed"; + } + | { + type: "rollback_failed"; + } +); +/** A Workflow event type accepted by a subscription filter. */ +type WorkflowInstanceEventType = WorkflowInstanceEvent["type"]; +/** Options controlling a Workflow instance event subscription. */ +type WorkflowInstanceSubscribeOptions = { + /** The event cursor from which to start the subscription from. */ + cursor?: number; + /** Emit only events with one of these types. */ + filter?: WorkflowInstanceEventType[]; +}; +/** A instance of a subscription to a specific workflow instance. */ +interface WorkflowInstanceSubscription extends Disposable { + next(): Promise>; + return( + value?: unknown, + ): Promise>; + throw( + error?: unknown, + ): Promise>; + [Symbol.asyncIterator](): WorkflowInstanceSubscription; +} declare abstract class WorkflowInstance { public id: string; /** @@ -17237,4 +17388,11 @@ declare abstract class WorkflowInstance { type: string; payload: unknown; }): Promise; + /** + * Subscribe to execution events from this Workflow instance. + * @param options Options controlling the starting cursor and event type filter. + */ + public subscribe( + options?: WorkflowInstanceSubscribeOptions, + ): Promise; } diff --git a/types/generated-snapshot/index.ts b/types/generated-snapshot/index.ts index c3d5b9f18cc..c02df60e69f 100755 --- a/types/generated-snapshot/index.ts +++ b/types/generated-snapshot/index.ts @@ -17147,6 +17147,157 @@ export interface WorkflowInstanceRestartOptions { type?: "do" | "sleep" | "waitForEvent"; }; } +/** An event emitted during the execution of the Workflow instance. */ +export type WorkflowInstanceEvent = { + /** The ID of the Workflow instance that emitted the event. */ + instanceId: string; + /** The event ID. */ + eventId: number; + /** The event timestamp. */ + timestamp: number; +} & ( + | { + type: "workflow_queued"; + } + | { + type: "workflow_started"; + params?: unknown; + } + | { + type: "workflow_completed"; + output?: unknown; + } + | { + type: "workflow_failed"; + error: { + name: string; + message: string; + }; + } + | { + type: "workflow_terminated"; + } + | { + type: "step_started"; + stepName: string; + } + | { + type: "step_completed"; + stepName: string; + output?: unknown; + } + | { + type: "step_failed"; + stepName: string; + } + | { + type: "attempt_started"; + stepName: string; + attempt: number; + } + | { + type: "attempt_completed"; + stepName: string; + attempt: number; + } + | { + type: "attempt_failed"; + 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; + } + | { + type: "rollback_step_completed"; + stepName: string; + } + | { + type: "rollback_step_failed"; + 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_failed"; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: { + name: string; + message: string; + }; + } + | { + type: "rollback_completed"; + } + | { + type: "rollback_failed"; + } +); +/** A Workflow event type accepted by a subscription filter. */ +export type WorkflowInstanceEventType = WorkflowInstanceEvent["type"]; +/** Options controlling a Workflow instance event subscription. */ +export type WorkflowInstanceSubscribeOptions = { + /** The event cursor from which to start the subscription from. */ + cursor?: number; + /** Emit only events with one of these types. */ + filter?: WorkflowInstanceEventType[]; +}; +/** A instance of a subscription to a specific workflow instance. */ +export interface WorkflowInstanceSubscription extends Disposable { + next(): Promise>; + return( + value?: unknown, + ): Promise>; + throw( + error?: unknown, + ): Promise>; + [Symbol.asyncIterator](): WorkflowInstanceSubscription; +} export declare abstract class WorkflowInstance { public id: string; /** @@ -17186,4 +17337,11 @@ export declare abstract class WorkflowInstance { type: string; payload: unknown; }): Promise; + /** + * Subscribe to execution events from this Workflow instance. + * @param options Options controlling the starting cursor and event type filter. + */ + public subscribe( + options?: WorkflowInstanceSubscribeOptions, + ): Promise; } diff --git a/types/test/types/rpc.ts b/types/test/types/rpc.ts index 0e50de33087..80a13752c9a 100644 --- a/types/test/types/rpc.ts +++ b/types/test/types/rpc.ts @@ -1329,3 +1329,156 @@ expectTypeOf().toEqualTypeOf<{ code: number; message: string; }>(); + +expectTypeOf( + workflowInstance.subscribe({ + cursor: 1, + filter: ['workflow_queued', 'attempt_failed'], + }) +).toEqualTypeOf>(); + +declare const workflowSubscription: WorkflowInstanceSubscription; +expectTypeOf( + workflowSubscription[Symbol.asyncIterator]() +).toEqualTypeOf(); +expectTypeOf(workflowSubscription.next()).toEqualTypeOf< + Promise> +>(); +expectTypeOf(workflowSubscription.return()).toEqualTypeOf< + Promise> +>(); +expectTypeOf(workflowSubscription.throw()).toEqualTypeOf< + Promise> +>(); +expectTypeOf(workflowSubscription[Symbol.dispose]()).toEqualTypeOf(); + +type WorkflowEventBase = { + instanceId: string; + eventId: number; + timestamp: number; +}; +type WorkflowEventOfType = Extract< + WorkflowInstanceEvent, + {type: T} +>; +type WorkflowEventError = {name: string; message: string}; + +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'workflow_queued'} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'workflow_started'; params?: unknown} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'workflow_completed'; output?: unknown} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'workflow_failed'; error: WorkflowEventError} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'workflow_terminated'} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'step_started'; stepName: string} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'step_completed'; + stepName: string; + output?: unknown; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'step_failed'; stepName: string} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'attempt_started'; + stepName: string; + attempt: number; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'attempt_completed'; + stepName: string; + attempt: number; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'attempt_failed'; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: WorkflowEventError; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'sleep_started'; + stepName: string; + durationMs: number; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'sleep_completed'; stepName: string} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'wait_started'; + stepName: string; + eventType: string; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'wait_completed'; stepName: string} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'wait_timed_out'; stepName: string} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'rollback_started'} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'rollback_step_started'; stepName: string} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'rollback_step_completed'; stepName: string} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'rollback_step_failed'; + stepName: string; + error: WorkflowEventError; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'rollback_attempt_started'; + stepName: string; + attempt: number; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'rollback_attempt_completed'; + stepName: string; + attempt: number; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & { + type: 'rollback_attempt_failed'; + stepName: string; + attempt: number; + retryDelayMs?: number; + error: WorkflowEventError; + } +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'rollback_completed'} +>(); +expectTypeOf>().toEqualTypeOf< + WorkflowEventBase & {type: 'rollback_failed'} +>();