feat(miniflare): Implement Assets RPC Proxy Worker behaviour (#8376)

CarmenPopoviciu · penalosa · web-flow · commit a25f060232bf · 2025-03-20T10:31:45.000+01:00
* feat(miniflare): Implement RPC Proxy Worker behaviour

This commit adds the appropriate behaviour for the `RpcProxyWorker`
used by Miniflare in Workers+Assets pipelines.

* Update fixtures/workers-with-assets-and-service-bindings/workerA/src/workerD.util.ts

Co-authored-by: Somhairle MacLeòid &lt;smacleod@cloudflare.com&gt;

* feedback fixes

---------

Co-authored-by: Somhairle MacLeòid &lt;smacleod@cloudflare.com&gt;
diff --git a/.changeset/real-jobs-own.md b/.changeset/real-jobs-own.md
@@ -0,0 +1,6 @@
+---
+"miniflare": patch
+"wrangler": patch
+---
+
+feat: Make local dev RPC behaviour on par with production for Workers with assets
diff --git a/fixtures/workers-with-assets-and-service-bindings/tests/index.test.ts b/fixtures/workers-with-assets-and-service-bindings/tests/index.test.ts
@@ -6,6 +6,10 @@ import { runWranglerDev } from "../../shared/src/run-wrangler-long-lived";
 
 const devCmds = [{ args: [] }, { args: ["--x-assets-rpc"] }];
 
+const failsIf = (condition: boolean) => {
+	return condition ? it.fails : it;
+};
+
 describe.each(devCmds)(
 	"[wrangler dev $args][Workers + Assets] Service bindings to Worker with assets",
 	({ args }) => {
@@ -54,8 +58,9 @@ describe.each(devCmds)(
 		describe("Workers running in separate wrangler dev sessions", () => {
 			describe("Service binding to default export", () => {
 				// this currently incorrectly returns the User Worker response
-				// instead of the Asset Worker response
-				it.fails(
+				// instead of the Asset Worker response, unless `--x-assets-rpc`
+				// is provided
+				failsIf(!args.length)(
 					"should return Asset Worker response for routes that serve static content",
 					async ({ expect }) => {
 						let response = await fetch(`http://${ipWorkerA}:${portWorkerA}`);
@@ -126,9 +131,10 @@ describe.each(devCmds)(
 
 			describe("Service binding to default entrypoint", () => {
 				// this currently incorrectly returns the User Worker response
-				// instead of the Asset Worker response
-				it.fails(
-					"should return Asset Worker response for fetch requestsfor routes that serve static content",
+				// instead of the Asset Worker response, unless `--x-assets-rpc`
+				// is provided
+				failsIf(!args.length)(
+					"should return Asset Worker response for fetch requests for routes that serve static content",
 					async ({ expect }) => {
 						let response = await fetch(`http://${ipWorkerA}:${portWorkerA}`);
 						let text = await response.text();
@@ -241,6 +247,20 @@ describe.each(devCmds)(
 						'env.NAMED_ENTRYPOINT.busyBee("🐝") response: Hello busy 🐝s from worker-d busyBee(bee)'
 					);
 				});
+
+				it("should support promise pipelining", async ({ expect }) => {
+					// fetch URL is irrelevant here. workerA will internally call
+					// the appropriate fns on the service binding instead
+					let response = await fetch(`http://${ipWorkerA}:${portWorkerA}`);
+					let text = await response.text();
+					expect(response.status).toBe(200);
+					expect(text).toContain(
+						`env.NAMED_ENTRYPOINT.foo("🐙").bar.buzz() response: You made it! 🐙`
+					);
+					expect(text).toContain(
+						`env.NAMED_ENTRYPOINT.newBeeCounter().value response: 2`
+					);
+				});
 			});
 		});
 	}
diff --git a/fixtures/workers-with-assets-and-service-bindings/vitest.config.mts b/fixtures/workers-with-assets-and-service-bindings/vitest.config.mts
@@ -0,0 +1,9 @@
+import { defineProject, mergeConfig } from "vitest/config";
+import configShared from "../../vitest.shared";
+
+export default mergeConfig(
+	configShared,
+	defineProject({
+		test: {},
+	})
+);
diff --git a/fixtures/workers-with-assets-and-service-bindings/workerA/src/index.ts b/fixtures/workers-with-assets-and-service-bindings/workerA/src/index.ts
@@ -24,6 +24,8 @@ export default {
 				`env.NAMED_ENTRYPOINT.fetch() response: ${workerDResponses.fetchResponse}\n` +
 				`env.NAMED_ENTRYPOINT.bee() response: ${workerDResponses.beeResult}\n` +
 				`env.NAMED_ENTRYPOINT.busyBee("🐝") response: ${workerDResponses.busyBeeResult}\n` +
+				`env.NAMED_ENTRYPOINT.foo("🐙").bar.buzz() response: ${workerDResponses.buzzResult}\n` +
+				`env.NAMED_ENTRYPOINT.newBeeCounter().value response: ${workerDResponses.beeCountResult}\n` +
 				`env.NAMED_ENTRYPOINT.scheduled() response: ${workerDResponses.scheduledResponse}\n\n`
 		);
 	},
diff --git a/fixtures/workers-with-assets-and-service-bindings/workerA/src/workerD.util.ts b/fixtures/workers-with-assets-and-service-bindings/workerA/src/workerD.util.ts
@@ -9,13 +9,26 @@ export async function getWorkerDResponses(request: Request, env) {
 	// test named functions with parameters
 	const busyBeeResult = await env.NAMED_ENTRYPOINT.busyBee("🐝");
 
+	// test nested functions + promise pipelining
+	const foo = env.NAMED_ENTRYPOINT.foo("🐙");
+	const buzzResult = await foo.bar.buzz();
+
+	// test RpcTarget + promise pipelining
+	const beeCounter = env.NAMED_ENTRYPOINT.newBeeCounter();
+	beeCounter.increment(1); // returns 1
+	beeCounter.increment(2); // returns 3
+	beeCounter.increment(-1); // returns 2
+	const beeCountResult = await beeCounter.value; // returns 2
+
 	// tests Cron Triggers
 	// Cron Triggers can only be defined on default exports, class-based or otherwise
 
 	return {
 		fetchResponse,
 		beeResult,
 		busyBeeResult,
+		buzzResult,
+		beeCountResult,
 		scheduledResponse:
 			"Not supported. Cron Triggers can only be defined on default exports.",
 	};
diff --git a/fixtures/workers-with-assets-and-service-bindings/workerB-with-default-export/src/index.ts b/fixtures/workers-with-assets-and-service-bindings/workerB-with-default-export/src/index.ts
@@ -22,6 +22,17 @@ export default {
 		return `Hello busy ${bee}s from worker-b busyBee(bee)`;
 	},
 
+	/*
+	 * Nested functions
+	 */
+	async foo(emoji: string) {
+		return {
+			bar: {
+				buzz: () => `You made it! ${emoji}`,
+			},
+		};
+	},
+
 	/*
 	 * Cron Triggers
 	 *
diff --git a/fixtures/workers-with-assets-and-service-bindings/workerC-with-default-entrypoint/src/index.ts b/fixtures/workers-with-assets-and-service-bindings/workerC-with-default-entrypoint/src/index.ts
@@ -1,4 +1,4 @@
-import { WorkerEntrypoint } from "cloudflare:workers";
+import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers";
 
 export default class extends WorkerEntrypoint {
 	/*
@@ -27,6 +27,28 @@ export default class extends WorkerEntrypoint {
 		return `Hello busy ${bee}s from worker-c busyBee(bee)`;
 	}
 
+	/*
+	 * Nested functions
+	 *
+	 * see https://developers.cloudflare.com/workers/runtime-apis/rpc/#promise-pipelining
+	 */
+	async foo(emoji: string) {
+		return {
+			bar: {
+				buzz: () => `You made it! ${emoji}`,
+			},
+		};
+	}
+
+	/*
+	 * Class instances
+	 *
+	 * see https://developers.cloudflare.com/workers/runtime-apis/rpc/#class-instances
+	 */
+	async newBeeCounter() {
+		return new BeeCounter();
+	}
+
 	/*
 	 * Cron Triggers
 	 *
@@ -39,3 +61,16 @@ export default class extends WorkerEntrypoint {
 		console.log("Hello from worker-c scheduled()");
 	}
 }
+
+class BeeCounter extends RpcTarget {
+	#value = 0;
+
+	increment(amount) {
+		this.#value += amount;
+		return this.#value;
+	}
+
+	get value() {
+		return this.#value;
+	}
+}
diff --git a/fixtures/workers-with-assets-and-service-bindings/workerD-with-named-entrypoint/src/index.ts b/fixtures/workers-with-assets-and-service-bindings/workerD-with-named-entrypoint/src/index.ts
@@ -1,4 +1,4 @@
-import { WorkerEntrypoint } from "cloudflare:workers";
+import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers";
 
 export class EntrypointD extends WorkerEntrypoint {
 	/*
@@ -22,6 +22,28 @@ export class EntrypointD extends WorkerEntrypoint {
 		return `Hello busy ${bee}s from worker-d busyBee(bee)`;
 	}
 
+	/*
+	 * Nested functions
+	 *
+	 * see https://developers.cloudflare.com/workers/runtime-apis/rpc/#promise-pipelining
+	 */
+	async foo(emoji: string) {
+		return {
+			bar: {
+				buzz: () => `You made it! ${emoji}`,
+			},
+		};
+	}
+
+	/*
+	 * Class instances
+	 *
+	 * see https://developers.cloudflare.com/workers/runtime-apis/rpc/#class-instances
+	 */
+	async newBeeCounter() {
+		return new BeeCounter();
+	}
+
 	/*
 	 * Cron Triggers/ Queues / etc can only be defined on a default
 	 * exports, class or non-class based
@@ -33,3 +55,16 @@ export default class extends WorkerEntrypoint {
 		return new Response("Hello from worker-d default entrypoint fetch()");
 	}
 }
+
+class BeeCounter extends RpcTarget {
+	#value = 0;
+
+	increment(amount) {
+		this.#value += amount;
+		return this.#value;
+	}
+
+	get value() {
+		return this.#value;
+	}
+}
diff --git a/packages/miniflare/src/index.ts b/packages/miniflare/src/index.ts
@@ -1343,13 +1343,24 @@ export class Miniflare {
 					directSocket.host,
 					directSocket.port
 				);
+				// check if Worker with assets with default export
+				// (class or non-class based)
+				const service =
+					this.#sharedOpts.core.unsafeEnableAssetsRpc &&
+					workerOpts.assets.assets &&
+					entrypoint === "default"
+						? {
+								name: `${RPC_PROXY_SERVICE_NAME}:${workerOpts.core.name}`,
+							}
+						: {
+								name: getUserServiceName(workerName),
+								entrypoint: entrypoint === "default" ? undefined : entrypoint,
+							};
+
 				sockets.push({
 					name,
 					address,
-					service: {
-						name: getUserServiceName(workerName),
-						entrypoint: entrypoint === "default" ? undefined : entrypoint,
-					},
+					service,
 					http: {
 						style: directSocket.proxy ? HttpOptions_Style.PROXY : undefined,
 						cfBlobHeader: CoreHeaders.CF_BLOB,
diff --git a/packages/miniflare/src/plugins/assets/index.ts b/packages/miniflare/src/plugins/assets/index.ts
@@ -252,6 +252,12 @@ export const ASSETS_PLUGIN: Plugin<typeof AssetsOptionsSchema> = {
 								name: `${ROUTER_SERVICE_NAME}:${id}`,
 							},
 						},
+						{
+							name: "USER_WORKER",
+							service: {
+								name: getUserServiceName(id),
+							},
+						},
 					],
 				},
 			};
diff --git a/packages/miniflare/src/workers/assets/rpc-proxy.worker.ts b/packages/miniflare/src/workers/assets/rpc-proxy.worker.ts
@@ -3,12 +3,53 @@ import type RouterWorker from "@cloudflare/workers-shared/asset-worker/src/index
 
 interface Env {
 	ROUTER_WORKER: Service<RouterWorker>;
+	USER_WORKER: Fetcher | Service;
 }
+
+/*
+ * The RPCProxyWorker is a Miniflare abstraction of the filter stage used in
+ * production Workers + Assets pipelines. It lives in front of the Router Worker,
+ * and is meant to redirect incoming `fetch` requests to the Router Worker, and
+ * all other RPC requests to the User Worker. It does so via a JSProxy mechanism.
+ * The RPCProxyWorker is specifically relevant in the context of
+ * Worker <> Worker+Assets service bindings.
+ *
+ * Please note that this conceptual pipeline architecture is applied only for
+ * Workers + Assets and default entrypoints, class or non-class based.
+ */
 export default class RPCProxyWorker extends WorkerEntrypoint<Env> {
 	async fetch(request: Request) {
-		// for now forward everything to the ROUTER WORKER
-		// this way we can wire this proxy Worker without
-		// changing the current assets behaviour
 		return this.env.ROUTER_WORKER.fetch(request);
 	}
+
+	constructor(ctx: ExecutionContext, env: Env) {
+		super(ctx, env);
+		/*
+		 * Create a proxy of the RPCProxyWorker instance
+		 *
+		 * see https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Proxy
+		 * see https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Reflect
+		 */
+		return new Proxy(this, {
+			get(target, prop) {
+				/*
+				 * If `prop` is defined on the RPCProxyWorker, then we are
+				 * intentionally meaning to intercept `env.USER_WORKER[prop]`
+				 * calls and override their return value . We should therefore
+				 * return whatever `RPCProxyWorker[prop]` returns
+				 */
+				if (Reflect.has(target, prop)) {
+					return Reflect.get(target, prop);
+				}
+
+				/*
+				 * Otherwise, forward to the USER_WORKER and return its response
+				 */
+				return function (...args: Array<unknown>) {
+					// @ts-expect-error
+					return Reflect.apply(target.env.USER_WORKER[prop], target, args);
+				};
+			},
+		});
+	}
 }
diff --git a/packages/wrangler/e2e/assets-multiworker.test.ts b/packages/wrangler/e2e/assets-multiworker.test.ts
