From aeb83ff65f40d8a53f26b07d9992d19b794c8089 Mon Sep 17 00:00:00 2001 From: Caio Nogueira Date: Wed, 12 Aug 2026 17:00:55 +0100 Subject: [PATCH] WOR-1491: make workflows_bindings_rpc obsolete WOR-1537 added a compatibility-gate bypass for private inner fetchers when constructing wrapped bindings. This allows Workflow bindings on old compatibility dates to resolve their JSRPC methods. Use JSRPC unconditionally and remove the legacy HTTP transport. Keep the workflows_bindings_rpc flag name for configuration validation, but make it a no-op. Cover old dates in workerd and edgeworker tests. --- .../internal/test/workflows/BUILD.bazel | 1 - .../test/workflows/workflows-api-rpc-test.js | 138 ------------------ .../test/workflows/workflows-api-test.js | 70 ++++++--- .../test/workflows/workflows-api-test.wd-test | 31 +--- .../internal/test/workflows/workflows-mock.js | 127 ++++------------ src/cloudflare/internal/workflows-api.ts | 127 ++-------------- src/workerd/io/compatibility-date.capnp | 6 +- 7 files changed, 104 insertions(+), 396 deletions(-) delete mode 100644 src/cloudflare/internal/test/workflows/workflows-api-rpc-test.js diff --git a/src/cloudflare/internal/test/workflows/BUILD.bazel b/src/cloudflare/internal/test/workflows/BUILD.bazel index bd7d1bcc9ac..f582955f9f0 100644 --- a/src/cloudflare/internal/test/workflows/BUILD.bazel +++ b/src/cloudflare/internal/test/workflows/BUILD.bazel @@ -2,6 +2,5 @@ load("//:build/wd_test.bzl", "wd_test") wd_test( src = "workflows-api-test.wd-test", - args = ["--experimental"], data = glob(["*.js"]), ) diff --git a/src/cloudflare/internal/test/workflows/workflows-api-rpc-test.js b/src/cloudflare/internal/test/workflows/workflows-api-rpc-test.js deleted file mode 100644 index 6c5124a0518..00000000000 --- a/src/cloudflare/internal/test/workflows/workflows-api-rpc-test.js +++ /dev/null @@ -1,138 +0,0 @@ -// Copyright (c) 2024 Cloudflare, Inc. -// Licensed under the Apache 2.0 license found in the LICENSE file or at: -// https://opensource.org/licenses/Apache-2.0 - -import * as assert from 'node:assert'; - -async function lastRestartBody(env, id) { - return env.mock.lastRestart(id); -} - -export const tests = { - async test(_, env) { - { - const instance = await env.workflow.create({ - id: 'foo', - payload: { bar: 'baz' }, - }); - assert.deepStrictEqual(instance.id, 'foo'); - } - - { - const instance = await env.workflow.get('bar'); - assert.deepStrictEqual(instance.id, 'bar'); - } - - { - const instances = await env.workflow.createBatch([ - { id: 'foo', payload: { bar: 'baz' } }, - { id: 'bar', payload: { bar: 'baz' } }, - ]); - assert.deepStrictEqual(instances[0].id, 'foo'); - assert.deepStrictEqual(instances[1].id, 'bar'); - } - - { - const result = await env.workflow.deleteBatch([ - 'delete-rpc-1', - 'missing-delete', - 'delete-rpc-1', - ]); - assert.deepStrictEqual(result, { - deleted: [{ id: 'delete-rpc-1' }, { id: 'delete-rpc-1' }], - errors: [ - { - id: 'missing-delete', - code: 10400, - message: 'workflows.api.error.instance.not_found', - }, - ], - }); - } - - { - const instance = await env.workflow.get('inst'); - await instance.pause(); - await instance.resume(); - await instance.terminate(); - await instance.delete(); - await instance.sendEvent({ - type: 'my-event', - payload: { hello: 'world' }, - }); - } - - { - const instance = await env.workflow.get('status-rpc'); - const status = await instance.status(); - assert.deepStrictEqual(status.status, 'running'); - assert.strictEqual(status.transport, 'rpc'); - assert.strictEqual(status.output, 'status-rpc'); - } - - { - for (const method of ['get', 'create', 'createBatch', 'deleteBatch']) { - assert.strictEqual(typeof env.workflow[method], 'function'); - } - - const fromGet = await env.workflow.get('a'); - const fromCreate = await env.workflow.create({ id: 'b' }); - const [fromBatch] = await env.workflow.createBatch([{ id: 'c' }]); - - const proto = Object.getPrototypeOf(fromGet); - assert.strictEqual(Object.getPrototypeOf(fromCreate), proto); - assert.strictEqual(Object.getPrototypeOf(fromBatch), proto); - - for (const method of [ - 'pause', - 'resume', - 'terminate', - 'restart', - 'delete', - 'status', - 'sendEvent', - ]) { - assert.strictEqual(typeof fromGet[method], 'function'); - } - } - - { - await assert.rejects(env.workflow.get('throw'), { - message: 'workflow instance not found', - }); - } - }, - - async testRestartNoOptions(_, env) { - const instance = await env.workflow.get('rpc-restart-basic'); - await instance.restart(); - - const body = await lastRestartBody(env, 'rpc-restart-basic'); - assert.deepStrictEqual(body.id, 'rpc-restart-basic'); - assert.strictEqual(body.from, undefined); - }, - - async testRestartFromStepNameOnly(_, env) { - const instance = await env.workflow.get('rpc-restart-step'); - await instance.restart({ from: { name: 'fetch data' } }); - - const body = await lastRestartBody(env, 'rpc-restart-step'); - assert.deepStrictEqual(body.id, 'rpc-restart-step'); - assert.deepStrictEqual(body.from, { name: 'fetch data' }); - }, - - async testRestartFromStepAllOptions(_, env) { - const instance = await env.workflow.get('rpc-restart-full'); - await instance.restart({ - from: { name: 'process item', count: 3, type: 'do' }, - }); - - const body = await lastRestartBody(env, 'rpc-restart-full'); - assert.deepStrictEqual(body.id, 'rpc-restart-full'); - assert.deepStrictEqual(body.from, { - name: 'process item', - count: 3, - type: 'do', - }); - }, -}; diff --git a/src/cloudflare/internal/test/workflows/workflows-api-test.js b/src/cloudflare/internal/test/workflows/workflows-api-test.js index ce9acd21b92..222f07caa20 100644 --- a/src/cloudflare/internal/test/workflows/workflows-api-test.js +++ b/src/cloudflare/internal/test/workflows/workflows-api-test.js @@ -4,6 +4,11 @@ import * as assert from 'node:assert'; +// 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. + +// Introspection goes over `env.mock.fetch()` rather than JSRPC on purpose: `env.mock` is an +// ordinary service binding, so its RPC wildcard stays gated at this worker's compatibility date. async function getLastRestartBody(env, id) { const res = await env.mock.fetch('http://placeholder/last-restart', { method: 'POST', @@ -12,7 +17,7 @@ async function getLastRestartBody(env, id) { return (await res.json()).result; } -export const tests = { +export const workflowsApi = { async test(_, env) { { // Test create instance @@ -45,27 +50,14 @@ export const tests = { assert.deepStrictEqual(instances[1].id, 'bar'); } - { - const instance = await env.workflow.get('status-http'); - const status = await instance.status(); - assert.deepStrictEqual(status.status, 'running'); - assert.strictEqual(status.transport, 'http'); - } - - { - // Test delete hits the /delete endpoint without throwing. - const instance = await env.workflow.get('delete-http'); - await instance.delete(); - } - { const result = await env.workflow.deleteBatch([ - 'delete-http-1', + 'delete-1', 'missing-delete', - 'delete-http-1', + 'delete-1', ]); assert.deepStrictEqual(result, { - deleted: [{ id: 'delete-http-1' }, { id: 'delete-http-1' }], + deleted: [{ id: 'delete-1' }, { id: 'delete-1' }], errors: [ { id: 'missing-delete', @@ -76,6 +68,25 @@ export const tests = { }); } + { + const instance = await env.workflow.get('inst'); + await instance.pause(); + await instance.resume(); + await instance.terminate(); + await instance.delete(); + await instance.sendEvent({ + type: 'my-event', + payload: { hello: 'world' }, + }); + } + + { + const instance = await env.workflow.get('status-1'); + const status = await instance.status(); + assert.deepStrictEqual(status.status, 'running'); + assert.strictEqual(status.output, 'status-1'); + } + { for (const method of ['get', 'create', 'createBatch', 'deleteBatch']) { assert.strictEqual(typeof env.workflow[method], 'function'); @@ -101,9 +112,24 @@ export const tests = { assert.strictEqual(typeof fromGet[method], 'function'); } } + + { + // The binding keeps its ungated inner fetcher inaccessible to user code. + // Instances returned by the binding omit the fetcher too. + assert.strictEqual(env.workflow.fetcher, undefined); + assert.strictEqual((await env.workflow.get('d')).fetcher, undefined); + } + + { + await assert.rejects(env.workflow.get('throw'), { + message: 'workflow instance not found', + }); + } }, +}; - async testRestartNoOptions(_, env) { +export const restartNoOptions = { + async test(_, env) { const instance = await env.workflow.get('restart-basic'); await instance.restart(); @@ -111,8 +137,10 @@ export const tests = { assert.deepStrictEqual(body.id, 'restart-basic'); assert.strictEqual(body.from, undefined); }, +}; - async testRestartFromStepNameOnly(_, env) { +export const restartFromStepNameOnly = { + async test(_, env) { const instance = await env.workflow.get('restart-step'); await instance.restart({ from: { name: 'fetch data' } }); @@ -120,8 +148,10 @@ export const tests = { assert.deepStrictEqual(body.id, 'restart-step'); assert.deepStrictEqual(body.from, { name: 'fetch data' }); }, +}; - async testRestartFromStepAllOptions(_, env) { +export const restartFromStepAllOptions = { + async test(_, env) { const instance = await env.workflow.get('restart-full'); await instance.restart({ from: { name: 'process item', count: 3, type: 'do' }, diff --git a/src/cloudflare/internal/test/workflows/workflows-api-test.wd-test b/src/cloudflare/internal/test/workflows/workflows-api-test.wd-test index 4b6b03854d3..18e5c3cdf5f 100644 --- a/src/cloudflare/internal/test/workflows/workflows-api-test.wd-test +++ b/src/cloudflare/internal/test/workflows/workflows-api-test.wd-test @@ -1,5 +1,8 @@ using Workerd = import "/workerd/workerd.capnp"; +# No `rpc` or `experimental` flag here on purpose. wd_test runs the default variant at compatibility +# date 2000-01-01, so this covers the binding working while `fetcher_rpc` is off. + const unitTests :Workerd.Config = ( services = [ ( name = "workflows-api-test", @@ -26,35 +29,9 @@ const unitTests :Workerd.Config = ( ], ) ), - ( name = "workflows-api-rpc-test", - worker = ( - modules = [ - (name = "worker", esModule = embed "workflows-api-rpc-test.js") - ], - compatibilityFlags = ["nodejs_compat", "workflows_bindings_rpc", - "experimental", "service_binding_extra_handlers", - "rpc"], - 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 = ["experimental", "nodejs_compat"], + compatibilityFlags = ["nodejs_compat"], modules = [ (name = "worker", esModule = embed "workflows-mock.js") ], diff --git a/src/cloudflare/internal/test/workflows/workflows-mock.js b/src/cloudflare/internal/test/workflows/workflows-mock.js index 2e7b28ba144..ea062589f06 100644 --- a/src/cloudflare/internal/test/workflows/workflows-mock.js +++ b/src/cloudflare/internal/test/workflows/workflows-mock.js @@ -9,106 +9,35 @@ const restartBodies = new Map(); const THROW_ID = 'throw'; const MISSING_DELETE_ID = 'missing-delete'; -function getInstance(id) { - if (id === THROW_ID) { - throw new Error('workflow instance not found'); - } - return { id }; -} - -function createInstance(options) { - return { id: options?.id }; -} - -function createBatchInstances(options) { - return options.map((val) => ({ id: val.id })); -} - -function deleteBatchInstances(options) { - return { - deleted: options.instances - .filter((id) => id !== MISSING_DELETE_ID) - .map((id) => ({ id })), - errors: options.instances - .filter((id) => id === MISSING_DELETE_ID) - .map((id) => ({ - id, - code: 10400, - message: 'workflows.api.error.instance.not_found', - })), - }; -} - -function instanceStatus(id, transport) { - return { status: 'running', output: id, transport }; -} - -async function handleHttp(request) { - const data = await request.json(); - const reqUrl = new URL(request.url); - - if (request.method !== 'POST') { - return Response.json({ success: false }, { status: 500 }); - } - - try { - switch (reqUrl.pathname) { - case '/get': - return Response.json({ result: getInstance(data.id) }, { status: 200 }); - case '/create': - return Response.json({ result: createInstance(data) }, { status: 201 }); - case '/createBatch': - return Response.json( - { result: createBatchInstances(data) }, - { status: 201 } - ); - case '/deleteBatch': - return Response.json( - { result: deleteBatchInstances(data) }, - { status: 200 } - ); - case '/pause': - case '/resume': - case '/terminate': - case '/delete': - case '/send-event': - return Response.json({ result: null }, { status: 200 }); - case '/restart': - restartBodies.set(data.id, data); - return Response.json({ result: null }, { status: 200 }); - case '/status': - return Response.json( - { result: instanceStatus(data.id, 'http') }, - { status: 200 } - ); - case '/last-restart': - return Response.json( - { result: restartBodies.get(data.id) ?? null }, - { status: 200 } - ); - default: - return Response.json({ success: false }, { status: 404 }); - } - } catch (err) { - return Response.json({ error: { message: err.message } }, { status: 500 }); - } -} - export default class WorkflowsMock extends WorkerEntrypoint { async getInstance(id) { - return getInstance(id); + if (id === THROW_ID) { + throw new Error('workflow instance not found'); + } + return { id }; } async create(options) { - return createInstance(options); + return { id: options?.id }; } async createBatch(options) { - return createBatchInstances(options); + return options.map((val) => ({ id: val.id })); } async deleteBatch(options) { - return deleteBatchInstances(options); + return { + deleted: options.instances + .filter((id) => id !== MISSING_DELETE_ID) + .map((id) => ({ id })), + errors: options.instances + .filter((id) => id === MISSING_DELETE_ID) + .map((id) => ({ + id, + code: 10400, + message: 'workflows.api.error.instance.not_found', + })), + }; } async deleteInstance(_id) {} @@ -124,16 +53,24 @@ export default class WorkflowsMock extends WorkerEntrypoint { } async status(id) { - return instanceStatus(id, 'rpc'); + return { status: 'running', output: id }; } async sendEvent(_id, _event) {} - async lastRestart(id) { - return restartBodies.get(id) ?? null; - } - + // 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) { - return handleHttp(request); + const data = await request.json(); + const pathname = new URL(request.url).pathname; + + switch (pathname) { + case '/last-restart': + return Response.json({ result: restartBodies.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 4ecada4c92f..d7f736825dd 100644 --- a/src/cloudflare/internal/workflows-api.ts +++ b/src/cloudflare/internal/workflows-api.ts @@ -11,16 +11,12 @@ export class NonRetryableError extends Error { } } -const workflowsBindingsRpc = - !!Cloudflare.compatibilityFlags['workflows_bindings_rpc']; - type WorkflowBatchDeleteResult = { deleted: { id: string }[]; errors: { id: string; code: number; message: string }[]; }; interface Fetcher { - fetch: typeof fetch; getInstance(id: string): Promise<{ id: string }>; deleteInstance(id: string): Promise; create(options?: WorkflowInstanceCreateOptions): Promise<{ id: string }>; @@ -45,32 +41,6 @@ interface Fetcher { ): Promise; } -async function callFetcher( - fetcher: Fetcher, - path: string, - body: object -): Promise { - const res = await fetcher.fetch(`http://workflow-binding.local${path}`, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - 'X-Version': '1', - }, - body: JSON.stringify(body), - }); - - const response = (await res.json()) as { - result: T; - error?: WorkflowError; - }; - - if (res.ok) { - return response.result; - } else { - throw new Error(response.error?.message); - } -} - class InstanceImpl implements WorkflowInstance { readonly #fetcher: Fetcher; readonly id: string; @@ -81,66 +51,29 @@ class InstanceImpl implements WorkflowInstance { } async pause(): Promise { - if (workflowsBindingsRpc) { - await this.#fetcher.pause(this.id); - return; - } - await callFetcher(this.#fetcher, '/pause', { - id: this.id, - }); + await this.#fetcher.pause(this.id); } + async resume(): Promise { - if (workflowsBindingsRpc) { - await this.#fetcher.resume(this.id); - return; - } - await callFetcher(this.#fetcher, '/resume', { - id: this.id, - }); + await this.#fetcher.resume(this.id); } async terminate(options?: WorkflowInstanceTerminateOptions): Promise { - if (workflowsBindingsRpc) { - await this.#fetcher.terminate(this.id, options); - return; - } - await callFetcher(this.#fetcher, '/terminate', { - id: this.id, - ...(options?.rollback === true ? { rollback: true } : {}), - }); + await this.#fetcher.terminate(this.id, options); } async restart(options?: WorkflowInstanceRestartOptions): Promise { - if (workflowsBindingsRpc) { - await this.#fetcher.restart(this.id, options); - return; - } - await callFetcher(this.#fetcher, '/restart', { - ...options, - id: this.id, - }); + await this.#fetcher.restart(this.id, options); } async delete(): Promise { - if (workflowsBindingsRpc) { - // deleteInstance, not delete: avoids colliding with the built-in Fetcher.delete(url), which - // is still present on compatibility dates before `fetcher_no_get_put_delete`. - await this.#fetcher.deleteInstance(this.id); - return; - } - await callFetcher(this.#fetcher, '/delete', { - id: this.id, - }); + // deleteInstance, not delete: avoids colliding with the built-in Fetcher.delete(url), which + // is still present on compatibility dates before `fetcher_no_get_put_delete`. + await this.#fetcher.deleteInstance(this.id); } async status(): Promise { - if (workflowsBindingsRpc) { - return await this.#fetcher.status(this.id); - } - const result = await callFetcher(this.#fetcher, '/status', { - id: this.id, - }); - return result; + return await this.#fetcher.status(this.id); } async sendEvent({ @@ -150,15 +83,7 @@ class InstanceImpl implements WorkflowInstance { type: string; payload: unknown; }): Promise { - if (workflowsBindingsRpc) { - await this.#fetcher.sendEvent(this.id, { type, payload }); - return; - } - await callFetcher(this.#fetcher, '/send-event', { - type, - payload, - id: this.id, - }); + await this.#fetcher.sendEvent(this.id, { type, payload }); } } @@ -171,12 +96,9 @@ class WorkflowImpl extends wrappedBinding.WrappedBinding { } async get(id: string): Promise { - const result = workflowsBindingsRpc - ? // getInstance, not get: avoids colliding with the built-in Fetcher.get(url). - await this.#fetcher.getInstance(id) - : await callFetcher<{ - id: string; - }>(this.#fetcher, '/get', { id }); + // getInstance, not get: avoids colliding with the built-in Fetcher.get(url), which is still + // present on compatibility dates before `fetcher_no_get_put_delete`. + const result = await this.#fetcher.getInstance(id); return new InstanceImpl(result.id, this.#fetcher); } @@ -184,11 +106,7 @@ class WorkflowImpl extends wrappedBinding.WrappedBinding { async create( options?: WorkflowInstanceCreateOptions ): Promise { - const result = workflowsBindingsRpc - ? await this.#fetcher.create(options) - : await callFetcher<{ - id: string; - }>(this.#fetcher, '/create', options ?? {}); + const result = await this.#fetcher.create(options); return new InstanceImpl(result.id, this.#fetcher); } @@ -196,26 +114,13 @@ class WorkflowImpl extends wrappedBinding.WrappedBinding { async createBatch( options: WorkflowInstanceCreateOptions[] ): Promise { - const results = workflowsBindingsRpc - ? await this.#fetcher.createBatch(options) - : await callFetcher< - { - id: string; - }[] - >(this.#fetcher, '/createBatch', options); + const results = await this.#fetcher.createBatch(options); return results.map((result) => new InstanceImpl(result.id, this.#fetcher)); } async deleteBatch(instanceIds: string[]): Promise { - const options = { instances: instanceIds }; - return workflowsBindingsRpc - ? await this.#fetcher.deleteBatch(options) - : await callFetcher( - this.#fetcher, - '/deleteBatch', - options - ); + return await this.#fetcher.deleteBatch({ instances: instanceIds }); } } diff --git a/src/workerd/io/compatibility-date.capnp b/src/workerd/io/compatibility-date.capnp index a2087d79bbf..1ac5aa6220e 100644 --- a/src/workerd/io/compatibility-date.capnp +++ b/src/workerd/io/compatibility-date.capnp @@ -1615,10 +1615,8 @@ struct CompatibilityFlags @0x8f8c1b68151b6cef { workflowsBindingsRpc @182 :Bool $compatEnableFlag("workflows_bindings_rpc") $experimental; - # When enabled, the `env.WORKFLOW` binding (cloudflare-internal:workflows-api) - # dispatches its methods as JSRPC calls on the inner fetcher instead of HTTP - # requests against the binding-shim worker. Without the flag the legacy HTTP - # transport is used. + # Obsolete flag. Has no effect: the `env.WORKFLOW` binding always dispatches its methods as + # JSRPC calls on the inner fetcher. Still accepted so configs which set it keep validating. typeScriptImplementedStreams @183 :Bool $compatEnableFlag("typescript_implemented_streams")