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