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