workflows: add experimental subscribe support

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"]),
+)

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")
+        ],
+      )
+    )
+  ]
+);

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

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

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<void>;
+  subscribe(
+    id: string,
+    options?: WorkflowInstanceSubscribeOptions
+  ): Promise<WorkflowInstanceSubscription>;
 }
 
 class InstanceImpl implements WorkflowInstance {
@@ -85,6 +92,22 @@ class InstanceImpl implements WorkflowInstance {
   }): Promise<void> {
     await this.#fetcher.sendEvent(this.id, { type, payload });
   }
+
+  async subscribe(
+    options?: WorkflowInstanceSubscribeOptions
+  ): Promise<WorkflowInstanceSubscription> {
+    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 {

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<IteratorResult<WorkflowInstanceEvent, undefined>>;
+  return(
+    value?: unknown
+  ): Promise<IteratorResult<WorkflowInstanceEvent, unknown>>;
+  throw(
+    error?: unknown
+  ): Promise<IteratorResult<WorkflowInstanceEvent, unknown>>;
+  [Symbol.asyncIterator](): WorkflowInstanceSubscription;
+}
+
 declare abstract class WorkflowInstance {
   id: string;
 
@@ -190,4 +259,8 @@ declare abstract class WorkflowInstance {
     type: string;
     payload: unknown;
   }): Promise<void>;
+
+  subscribe(
+    options?: WorkflowInstanceSubscribeOptions
+  ): Promise<WorkflowInstanceSubscription>;
 }

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