Support JSRPC for remote bindings (#10249)

* Support RPC bindings

Create yellow-kangaroos-bathe.md

fix checks

fix checks

Increase timeouts

more logging

static workers

static workers

pre-compile proxy server worker

* Mark cloudflare:* as external in CI tests

* address comments

Author: penalosa

.changeset/yellow-kangaroos-bathe.md

@@ -0,0 +1,10 @@
+---
+"miniflare": patch
+"wrangler": patch
+---
+
+Support JSRPC for remote bindings. This unlocks:
+- JSRPC over Service Bindings
+- JSRPC over Dispatch Namespace Bindings
+- Email
+- Pipelines

.gitignore

@@ -230,3 +230,4 @@ dist/**
 .env*
 !.env.example
 .node-cache/
+!vendor/jsrpc/dist

.prettierignore

@@ -49,3 +49,5 @@ fixtures/redirected-config-worker/build/*
 fixtures/redirected-config-worker-with-environments/build/*
 
 packages/vite-plugin-cloudflare/playground/**/*.d.ts
+
+vendor/jsrpc

packages/miniflare/package.json

@@ -60,6 +60,7 @@
 		"@ava/typescript": "^4.1.0",
 		"@cloudflare/cli": "workspace:*",
 		"@cloudflare/containers-shared": "workspace:*",
+		"@cloudflare/jsrpc": "link:../../vendor/jsrpc",
 		"@cloudflare/kv-asset-handler": "workspace:*",
 		"@cloudflare/workers-shared": "workspace:*",
 		"@cloudflare/workers-types": "catalog:default",

packages/miniflare/src/plugins/dispatch-namespace/index.ts

@@ -3,10 +3,8 @@ import LOCAL_DISPATCH_NAMESPACE from "worker:dispatch-namespace/dispatch-namespa
 import { z } from "zod";
 import { Worker_Binding } from "../../runtime";
 import {
-	getUserBindingServiceName,
 	Plugin,
 	ProxyNodeBinding,
-	remoteProxyClientWorker,
 	RemoteProxyConnectionString,
 } from "../shared";
 
@@ -35,20 +33,23 @@ export const DISPATCH_NAMESPACE_PLUGIN: Plugin<
 		const bindings = Object.entries(
 			options.dispatchNamespaces
 		).map<Worker_Binding>(([name, config]) => {
+			assert(
+				config.remoteProxyConnectionString,
+				"Dispatch Namespace bindings only support running remotely"
+			);
+
 			return {
 				name,
 				wrapped: {
 					moduleName: `${DISPATCH_NAMESPACE_PLUGIN_NAME}:local-dispatch-namespace`,
 					innerBindings: [
 						{
-							name: "fetcher",
-							service: {
-								name: getUserBindingServiceName(
-									DISPATCH_NAMESPACE_PLUGIN_NAME,
-									config.namespace,
-									config.remoteProxyConnectionString
-								),
-							},
+							name: "remoteProxyConnectionString",
+							text: config.remoteProxyConnectionString.href,
+						},
+						{
+							name: "binding",
+							text: name,
 						},
 					],
 				},
@@ -67,28 +68,8 @@ export const DISPATCH_NAMESPACE_PLUGIN: Plugin<
 			])
 		);
 	},
-	async getServices({ options }) {
-		if (!options.dispatchNamespaces) {
-			return [];
-		}
-
-		return Object.entries(options.dispatchNamespaces).map(([name, config]) => {
-			assert(
-				config.remoteProxyConnectionString,
-				"Dispatch Namespace bindings only support running remotely"
-			);
-			return {
-				name: getUserBindingServiceName(
-					DISPATCH_NAMESPACE_PLUGIN_NAME,
-					config.namespace,
-					config.remoteProxyConnectionString
-				),
-				worker: remoteProxyClientWorker(
-					config.remoteProxyConnectionString,
-					name
-				),
-			};
-		});
+	async getServices() {
+		return [];
 	},
 	getExtensions({ options }) {
 		if (!options.some((o) => o.dispatchNamespaces)) {

packages/miniflare/src/plugins/email/index.ts

@@ -5,13 +5,18 @@ import { Service, Worker_Binding } from "../../runtime";
 import {
 	getUserBindingServiceName,
 	Plugin,
+	remoteProxyClientWorker,
+	RemoteProxyConnectionString,
 	WORKER_BINDING_SERVICE_LOOPBACK,
 } from "../shared";
 
 // Define the mutually exclusive schema
 const EmailBindingOptionsSchema = z
 	.object({
 		name: z.string(),
+		remoteProxyConnectionString: z
+			.custom<RemoteProxyConnectionString>()
+			.optional(),
 	})
 	.and(
 		z.union([
@@ -53,10 +58,12 @@ export const EMAIL_PLUGIN: Plugin<typeof EmailOptionsSchema> = {
 
 		const sendEmailBindings = options.email.send_email;
 
-		return sendEmailBindings.map(({ name }) => ({
+		return sendEmailBindings.map(({ name, remoteProxyConnectionString }) => ({
 			name,
 			service: {
-				entrypoint: "SendEmailBinding",
+				entrypoint: remoteProxyConnectionString
+					? undefined
+					: "SendEmailBinding",
 				name: getUserBindingServiceName(SERVICE_SEND_EMAIL_WORKER_PREFIX, name),
 			},
 		}));
@@ -67,22 +74,25 @@ export const EMAIL_PLUGIN: Plugin<typeof EmailOptionsSchema> = {
 	async getServices(args) {
 		const services: Service[] = [];
 
-		for (const { name, ...config } of args.options.email?.send_email ?? []) {
+		for (const { name, remoteProxyConnectionString, ...config } of args.options
+			.email?.send_email ?? []) {
 			services.push({
 				name: getUserBindingServiceName(SERVICE_SEND_EMAIL_WORKER_PREFIX, name),
-				worker: {
-					compatibilityDate: "2025-03-17",
-					modules: [
-						{
-							name: "send_email.mjs",
-							esModule: SEND_EMAIL_BINDING(),
+				worker: remoteProxyConnectionString
+					? remoteProxyClientWorker(remoteProxyConnectionString, name)
+					: {
+							compatibilityDate: "2025-03-17",
+							modules: [
+								{
+									name: "send_email.mjs",
+									esModule: SEND_EMAIL_BINDING(),
+								},
+							],
+							bindings: [
+								...buildJsonBindings(config),
+								WORKER_BINDING_SERVICE_LOOPBACK,
+							],
 						},
-					],
-					bindings: [
-						...buildJsonBindings(config),
-						WORKER_BINDING_SERVICE_LOOPBACK,
-					],
-				},
 			});
 		}
 

packages/miniflare/src/plugins/pipelines/index.ts

@@ -1,10 +1,29 @@
 import SCRIPT_PIPELINE_OBJECT from "worker:pipelines/pipeline";
 import { z } from "zod";
 import { Service } from "../../runtime";
-import { namespaceKeys, Plugin, ProxyNodeBinding } from "../shared";
+import {
+	namespaceKeys,
+	Plugin,
+	ProxyNodeBinding,
+	remoteProxyClientWorker,
+	RemoteProxyConnectionString,
+} from "../shared";
 
 export const PipelineOptionsSchema = z.object({
-	pipelines: z.union([z.record(z.string()), z.string().array()]).optional(),
+	pipelines: z
+		.union([
+			z.record(z.string()),
+			z.string().array(),
+			z.record(
+				z.object({
+					pipeline: z.string(),
+					remoteProxyConnectionString: z
+						.custom<RemoteProxyConnectionString>()
+						.optional(),
+				})
+			),
+		])
+		.optional(),
 });
 
 export const PIPELINES_PLUGIN_NAME = "pipelines";
@@ -14,7 +33,7 @@ export const PIPELINE_PLUGIN: Plugin<typeof PipelineOptionsSchema> = {
 	options: PipelineOptionsSchema,
 	getBindings(options) {
 		const pipelines = bindingEntries(options.pipelines);
-		return pipelines.map<Service>(([name, id]) => ({
+		return pipelines.map<Service>(([name, { id }]) => ({
 			name,
 			service: { name: `${SERVICE_PIPELINE_PREFIX}:${id}` },
 		}));
@@ -29,18 +48,23 @@ export const PIPELINE_PLUGIN: Plugin<typeof PipelineOptionsSchema> = {
 		const pipelines = bindingEntries(options.pipelines);
 
 		const services = [];
-		for (const pipeline of pipelines) {
+		for (const [bindingName, pipeline] of pipelines) {
 			services.push({
-				name: `${SERVICE_PIPELINE_PREFIX}:${pipeline[1]}`,
-				worker: {
-					compatibilityDate: "2024-12-30",
-					modules: [
-						{
-							name: "pipeline.worker.js",
-							esModule: SCRIPT_PIPELINE_OBJECT(),
+				name: `${SERVICE_PIPELINE_PREFIX}:${pipeline.id}`,
+				worker: pipeline.remoteProxyConnectionString
+					? remoteProxyClientWorker(
+							pipeline.remoteProxyConnectionString,
+							bindingName
+						)
+					: {
+							compatibilityDate: "2024-12-30",
+							modules: [
+								{
+									name: "pipeline.worker.js",
+									esModule: SCRIPT_PIPELINE_OBJECT(),
+								},
+							],
 						},
-					],
-				},
 			});
 		}
 
@@ -50,16 +74,41 @@ export const PIPELINE_PLUGIN: Plugin<typeof PipelineOptionsSchema> = {
 
 function bindingEntries(
 	namespaces?:
-		| Record<string, { pipelineName: string }>
+		| Record<
+				string,
+				{
+					pipeline: string;
+					remoteProxyConnectionString?: RemoteProxyConnectionString;
+				}
+		  >
 		| string[]
 		| Record<string, string>
-): [bindingName: string, id: string][] {
+): [
+	bindingName: string,
+	{ id: string; remoteProxyConnectionString?: RemoteProxyConnectionString },
+][] {
 	if (Array.isArray(namespaces)) {
-		return namespaces.map((bindingName) => [bindingName, bindingName]);
+		return namespaces.map((bindingName) => [bindingName, { id: bindingName }]);
 	} else if (namespaces !== undefined) {
-		return Object.entries(namespaces).map(([name, opts]) => [
+		return (
+			Object.entries(namespaces) as [
+				string,
+				(
+					| string
+					| {
+							pipeline: string;
+							remoteProxyConnectionString?: RemoteProxyConnectionString;
+					  }
+				),
+			][]
+		).map(([name, opts]) => [
 			name,
-			typeof opts === "string" ? opts : opts.pipelineName,
+			typeof opts === "string"
+				? { id: opts }
+				: {
+						id: opts.pipeline,
+						remoteProxyConnectionString: opts.remoteProxyConnectionString,
+					},
 		]);
 	} else {
 		return [];

packages/miniflare/src/plugins/shared/index.ts

@@ -131,7 +131,7 @@ export class ProxyNodeBinding {
 }
 
 export function namespaceKeys(
-	namespaces?: Record<string, string | { id: string }> | string[]
+	namespaces?: Record<string, unknown> | string[]
 ): string[] {
 	if (Array.isArray(namespaces)) {
 		return namespaces;

packages/miniflare/src/workers/dispatch-namespace/dispatch-namespace.worker.ts

@@ -1,31 +1,48 @@
-interface Env {
-	fetcher: Fetcher;
-}
+import { newWebSocketRpcSession } from "@cloudflare/jsrpc";
+import { makeFetch } from "../shared/remote-bindings-utils";
 
-class LocalDispatchNamespace implements DispatchNamespace {
-	constructor(private env: Env) {}
-	get(
-		name: string,
-		args?: { [key: string]: any },
-		options?: DynamicDispatchOptions
-	): Fetcher {
-		return {
-			...this.env.fetcher,
-			fetch: (
-				input: RequestInfo | URL,
-				init?: RequestInit
-			): Promise<Response> => {
-				const request = new Request(input, init);
-				request.headers.set(
-					"MF-Dispatch-Namespace-Options",
-					JSON.stringify({ name, args, options })
-				);
-				return this.env.fetcher.fetch(request);
-			},
-		};
-	}
+interface Env {
+	remoteProxyConnectionString: string;
+	binding: string;
 }
 
 export default function (env: Env) {
-	return new LocalDispatchNamespace(env);
+	return {
+		get(
+			name: string,
+			args?: { [key: string]: any },
+			options?: DynamicDispatchOptions
+		): Fetcher {
+			const url = new URL(env.remoteProxyConnectionString);
+			url.protocol = "ws:";
+			url.searchParams.set("MF-Binding", env.binding);
+			url.searchParams.set(
+				"MF-Dispatch-Namespace-Options",
+				JSON.stringify({ name, args, options })
+			);
+			const stub = newWebSocketRpcSession(url.href);
+
+			return new Proxy(stub, {
+				get(_, p) {
+					// We don't want to wrap direct .fetch() calls on a customer worker in a JSRPC layer
+					// Instead, intercept accesses to the specific `fetch` key, and send them directly
+					if (p === "fetch") {
+						return makeFetch(
+							env.remoteProxyConnectionString,
+							env.binding,
+							new Headers({
+								"MF-Dispatch-Namespace-Options": JSON.stringify({
+									name,
+									args,
+									options,
+								}),
+							})
+						);
+					}
+
+					return Reflect.get(stub, p);
+				},
+			}) as Service;
+		},
+	} satisfies DispatchNamespace;
 }