Support Mixed Mode Dispatch Namespaces (#9245)

* Support Dispatch Namespace bindings

* Create tough-seals-whisper.md

* Update tough-seals-whisper.md

* address comments

Author: penalosa

.changeset/tough-seals-whisper.md

@@ -0,0 +1,6 @@
+---
+"miniflare": patch
+"wrangler": patch
+---
+
+Support Mixed Mode Dispatch Namespaces

fixtures/mixed-mode-node-test/tests/startMixedModeSession.test.js

@@ -170,4 +170,43 @@ baseDescribe("startMixedModeSession", () => {
 		await mixedModeSession.ready;
 		await mixedModeSession.dispose();
 	});
+
+	test("Dispatch Namespace mixed mode binding", async () => {
+		const mixedModeSession = await experimental_startMixedModeSession({
+			DISPATCH: {
+				type: "dispatch_namespace",
+				namespace: "mixed-mode-test-namespace",
+			},
+		});
+
+		const mf = new Miniflare({
+			compatibilityDate: "2025-01-01",
+			modules: true,
+			script: /* javascript */ `
+			export default {
+				async fetch(request, env) {
+					try{
+						const worker = env.DISPATCH.get("mixed-mode-test-customer-worker")
+					return Response.json({
+						"worker": await (await worker.fetch("http://example.com")).text(),
+					})}catch(e){console.log(e);return new Response(e)}
+				}
+			}
+		`,
+			dispatchNamespaces: {
+				DISPATCH: {
+					namespace: "mixed-mode-test-namespace",
+					mixedModeConnectionString: mixedModeSession.mixedModeConnectionString,
+				},
+			},
+		});
+		const response = await (
+			await mf.dispatchFetch("http://example.com")
+		).text();
+		assert.match(response, /Hello from customer worker/);
+		await mf.dispose();
+
+		await mixedModeSession.ready;
+		await mixedModeSession.dispose();
+	});
 });

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

@@ -0,0 +1,101 @@
+import assert from "node:assert";
+import LOCAL_DISPATCH_NAMESPACE from "worker:dispatch-namespace/dispatch-namespace";
+import { z } from "zod";
+import { Worker_Binding } from "../../runtime";
+import {
+	mixedModeClientWorker,
+	MixedModeConnectionString,
+	Plugin,
+	ProxyNodeBinding,
+} from "../shared";
+
+export const DispatchNamespaceOptionsSchema = z.object({
+	dispatchNamespaces: z
+		.record(
+			z.object({
+				namespace: z.string(),
+				mixedModeConnectionString: z
+					.custom<MixedModeConnectionString>()
+					.optional(),
+			})
+		)
+		.optional(),
+});
+
+export const DISPATCH_NAMESPACE_PLUGIN_NAME = "dispatch-namespace";
+
+export const DISPATCH_NAMESPACE_PLUGIN: Plugin<
+	typeof DispatchNamespaceOptionsSchema
+> = {
+	options: DispatchNamespaceOptionsSchema,
+	async getBindings(options) {
+		if (!options.dispatchNamespaces) {
+			return [];
+		}
+
+		const bindings = Object.entries(
+			options.dispatchNamespaces
+		).map<Worker_Binding>(([name, config]) => {
+			return {
+				name,
+				wrapped: {
+					moduleName: `${DISPATCH_NAMESPACE_PLUGIN_NAME}:local-dispatch-namespace`,
+					innerBindings: [
+						{
+							name: "fetcher",
+							service: {
+								name: `${DISPATCH_NAMESPACE_PLUGIN_NAME}:ns:${config.namespace}`,
+							},
+						},
+					],
+				},
+			};
+		});
+		return bindings;
+	},
+	getNodeBindings(options: z.infer<typeof DispatchNamespaceOptionsSchema>) {
+		if (!options.dispatchNamespaces) {
+			return {};
+		}
+		return Object.fromEntries(
+			Object.keys(options.dispatchNamespaces).map((name) => [
+				name,
+				new ProxyNodeBinding(),
+			])
+		);
+	},
+	async getServices({ options }) {
+		if (!options.dispatchNamespaces) {
+			return [];
+		}
+
+		return {
+			services: Object.entries(options.dispatchNamespaces).map(
+				([name, config]) => {
+					assert(
+						config.mixedModeConnectionString,
+						"Dispatch Namespace bindings only support Mixed Mode"
+					);
+					return {
+						name: `${DISPATCH_NAMESPACE_PLUGIN_NAME}:ns:${config.namespace}`,
+						worker: mixedModeClientWorker(
+							config.mixedModeConnectionString,
+							name
+						),
+					};
+				}
+			),
+			extensions: [
+				{
+					modules: [
+						{
+							name: `${DISPATCH_NAMESPACE_PLUGIN_NAME}:local-dispatch-namespace`,
+							esModule: LOCAL_DISPATCH_NAMESPACE(),
+							internal: true,
+						},
+					],
+				},
+			],
+		};
+	},
+};

packages/miniflare/src/plugins/index.ts

@@ -14,6 +14,10 @@ import {
 import { CACHE_PLUGIN, CACHE_PLUGIN_NAME } from "./cache";
 import { CORE_PLUGIN, CORE_PLUGIN_NAME } from "./core";
 import { D1_PLUGIN, D1_PLUGIN_NAME } from "./d1";
+import {
+	DISPATCH_NAMESPACE_PLUGIN,
+	DISPATCH_NAMESPACE_PLUGIN_NAME,
+} from "./dispatch-namespace";
 import { DURABLE_OBJECTS_PLUGIN, DURABLE_OBJECTS_PLUGIN_NAME } from "./do";
 import { EMAIL_PLUGIN, EMAIL_PLUGIN_NAME } from "./email";
 import { HYPERDRIVE_PLUGIN, HYPERDRIVE_PLUGIN_NAME } from "./hyperdrive";
@@ -43,6 +47,7 @@ export const PLUGINS = {
 	[ANALYTICS_ENGINE_PLUGIN_NAME]: ANALYTICS_ENGINE_PLUGIN,
 	[AI_PLUGIN_NAME]: AI_PLUGIN,
 	[BROWSER_RENDERING_PLUGIN_NAME]: BROWSER_RENDERING_PLUGIN,
+	[DISPATCH_NAMESPACE_PLUGIN_NAME]: DISPATCH_NAMESPACE_PLUGIN,
 };
 export type Plugins = typeof PLUGINS;
 
@@ -97,7 +102,8 @@ export type WorkerOptions = z.input<typeof CORE_PLUGIN.options> &
 	z.input<typeof SECRET_STORE_PLUGIN.options> &
 	z.input<typeof ANALYTICS_ENGINE_PLUGIN.options> &
 	z.input<typeof AI_PLUGIN.options> &
-	z.input<typeof BROWSER_RENDERING_PLUGIN.options>;
+	z.input<typeof BROWSER_RENDERING_PLUGIN.options> &
+	z.input<typeof DISPATCH_NAMESPACE_PLUGIN.options>;
 
 export type SharedOptions = z.input<typeof CORE_PLUGIN.sharedOptions> &
 	z.input<typeof CACHE_PLUGIN.sharedOptions> &
@@ -162,3 +168,4 @@ export * from "./email";
 export * from "./analytics-engine";
 export * from "./ai";
 export * from "./browser-rendering";
+export * from "./dispatch-namespace";

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

@@ -0,0 +1,31 @@
+interface Env {
+	fetcher: Fetcher;
+}
+
+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);
+			},
+		};
+	}
+}
+
+export default function (env: Env) {
+	return new LocalDispatchNamespace(env);
+}

packages/miniflare/src/workers/shared/mixed-mode-client.worker.ts

@@ -1,24 +1,26 @@
-export default {
-	async fetch(request, env) {
+import { WorkerEntrypoint } from "cloudflare:workers";
+
+export default class Client extends WorkerEntrypoint<{
+	mixedModeConnectionString: string;
+	binding: string;
+}> {
+	async fetch(request: Request) {
 		const proxiedHeaders = new Headers();
 		for (const [name, value] of request.headers) {
 			// The `Upgrade` header needs to be special-cased to prevent:
 			//   TypeError: Worker tried to return a WebSocket in a response to a request which did not contain the header "Upgrade: websocket"
-			if (name === "upgrade") {
+			if (name === "upgrade" || name.startsWith("MF-")) {
 				proxiedHeaders.set(name, value);
 			} else {
 				proxiedHeaders.set(`MF-Header-${name}`, value);
 			}
 		}
 		proxiedHeaders.set("MF-URL", request.url);
-		proxiedHeaders.set("MF-Binding", env.binding);
+		proxiedHeaders.set("MF-Binding", this.env.binding);
 		const req = new Request(request, {
 			headers: proxiedHeaders,
 		});
 
-		return fetch(env.mixedModeConnectionString, req);
-	},
-} satisfies ExportedHandler<{
-	mixedModeConnectionString: string;
-	binding: string;
-}>;
+		return fetch(this.env.mixedModeConnectionString, req);
+	}
+}

packages/wrangler/templates/mixedMode/proxyServerWorker/index.ts

@@ -13,7 +13,21 @@ export default {
 					originalHeaders.set(name, value);
 				}
 			}
-			return env[targetBinding].fetch(
+			let fetcher = env[targetBinding];
+
+			// Special case the Dispatch Namespace binding because it has a top-level synchronous .get() call
+			const dispatchNamespaceOptions = originalHeaders.get(
+				"MF-Dispatch-Namespace-Options"
+			);
+			if (dispatchNamespaceOptions) {
+				const { name, args, options } = JSON.parse(dispatchNamespaceOptions);
+				fetcher = (env[targetBinding] as DispatchNamespace).get(
+					name,
+					args,
+					options
+				);
+			}
+			return (fetcher as Fetcher).fetch(
 				request.headers.get("MF-URL")!,
 				new Request(request, {
 					redirect: "manual",
@@ -23,4 +37,4 @@ export default {
 		}
 		return new Response("Provide a binding", { status: 400 });
 	},
-} satisfies ExportedHandler<Record<string, Fetcher>>;
+} satisfies ExportedHandler<Record<string, Fetcher | DispatchNamespace>>;