PoC: multi-worker wrangler dev

fixtures/multi-worker/README.md

@@ -0,0 +1,11 @@
+# multi-worker
+
+The aim of this fixture is to use `wrangler dev` for multiple workers in the same project.
+
+## Usage
+
+```sh
+npx wrangler dev --x-dev-env \
+    -c packages/worker-a/wrangler.toml \
+    -c packages/worker-b/wrangler.toml
+```

fixtures/multi-worker/package.json

@@ -0,0 +1,9 @@
+{
+	"name": "multi-worker",
+	"workspaces": [
+		"packages/*"
+	],
+	"devDependencies": {
+		"wrangler": "workspace:*"
+	}
+}

fixtures/multi-worker/packages/worker-a/package.json

@@ -0,0 +1,15 @@
+{
+	"name": "worker-a",
+	"private": true,
+	"scripts": {
+		"deploy": "wrangler deploy",
+		"start": "wrangler dev --x-dev-env"
+	},
+	"devDependencies": {
+		"@cloudflare/workers-types": "^4.20240909.0",
+		"wrangler": "workspace:*"
+	},
+	"volta": {
+		"extends": "../../package.json"
+	}
+}

fixtures/multi-worker/packages/worker-a/src/index.ts

@@ -0,0 +1,21 @@
+export default {
+	async fetch(
+		request: Request,
+		env: Env,
+		ctx: ExecutionContext
+	): Promise<Response> {
+		const result = await env.WORKER_B.hello();
+
+		return new Response(
+            `Worker A called Worker B: ${result}`
+        );
+	},
+};
+
+
+
+
+
+export interface Env {
+	WORKER_B: { hello(): Promise<string> };
+}

fixtures/multi-worker/packages/worker-a/tsconfig.json

@@ -0,0 +1,13 @@
+{
+	"compilerOptions": {
+		"target": "es2021",
+		"lib": ["es2021"],
+		"module": "es2022",
+		"types": ["@cloudflare/workers-types/experimental"],
+		"noEmit": true,
+		"isolatedModules": true,
+		"forceConsistentCasingInFileNames": true,
+		"strict": true,
+		"skipLibCheck": true
+	}
+}

fixtures/multi-worker/packages/worker-a/wrangler.toml

@@ -0,0 +1,7 @@
+name = "worker-a"
+main = "src/index.ts"
+compatibility_date = "2024-09-09"
+
+services = [
+  { binding = "WORKER_B", service = "worker-b", entrypoint = "default" }
+]

fixtures/multi-worker/packages/worker-b/package.json

@@ -0,0 +1,15 @@
+{
+	"name": "worker-b",
+	"private": true,
+	"scripts": {
+		"deploy": "wrangler deploy",
+		"start": "wrangler dev --x-dev-env"
+	},
+	"devDependencies": {
+		"@cloudflare/workers-types": "^4.20240909.0",
+		"wrangler": "workspace:*"
+	},
+	"volta": {
+		"extends": "../../package.json"
+	}
+}

fixtures/multi-worker/packages/worker-b/src/index.ts

@@ -0,0 +1,7 @@
+import { WorkerEntrypoint } from "cloudflare:workers";
+
+export default class extends WorkerEntrypoint {
+	async hello() {
+		return "Hello World from Worker B!";
+	}
+}

fixtures/multi-worker/packages/worker-b/tsconfig.json

@@ -0,0 +1,13 @@
+{
+	"compilerOptions": {
+		"target": "es2021",
+		"lib": ["es2021"],
+		"module": "es2022",
+		"types": ["@cloudflare/workers-types/experimental"],
+		"noEmit": true,
+		"isolatedModules": true,
+		"forceConsistentCasingInFileNames": true,
+		"strict": true,
+		"skipLibCheck": true
+	}
+}

fixtures/multi-worker/packages/worker-b/wrangler.toml

@@ -0,0 +1,5 @@
+name = "worker-b"
+main = "src/index.ts"
+compatibility_date = "2024-09-09"
+
+dev.port = 8788

packages/wrangler/src/api/dev.ts

@@ -176,7 +176,8 @@ export async function unstable_dev(
 		onReady: (address, port, proxyData) => {
 			readyResolve({ address, port, proxyData });
 		},
-		config: options?.config,
+		// @ts-expect-error who cares
+		config: options?.config === undefined ? undefined : [options.config],
 		env: options?.env,
 		processEntrypoint,
 		additionalModules,

packages/wrangler/src/api/startDevWorker/LocalRuntimeController.ts

@@ -6,6 +6,7 @@ import * as MF from "../../dev/miniflare";
 import { logger } from "../../logger";
 import { RuntimeController } from "./BaseController";
 import { castErrorCause } from "./events";
+import { ProxyControllerLogger } from "./ProxyController";
 import { convertBindingsToCfWorkerInitBindings } from "./utils";
 import type { WorkerEntrypointsDefinition } from "../../dev-registry";
 import type {
@@ -146,12 +147,19 @@ export class LocalRuntimeController extends RuntimeController {
 
 	async #onBundleComplete(data: BundleCompleteEvent, id: number) {
 		try {
-			const { options, internalObjects, entrypointNames } =
+			// eslint-disable-next-line prefer-const
+			let { options, internalObjects, entrypointNames } =
 				await MF.buildMiniflareOptions(
 					this.#log,
 					await convertToConfigBundle(data),
 					this.#proxyToUserWorkerAuthenticationSecret
 				);
+			options = {
+				...options,
+				log: new ProxyControllerLogger(MF.castLogLevel(logger.loggerLevel), {
+					prefix: `wrangler-${data.config.name}`,
+				}),
+			};
 			options.liveReload = false; // TODO: set in buildMiniflareOptions once old code path is removed
 			if (this.#mf === undefined) {
 				logger.log(chalk.dim("⎔ Starting local server..."));

packages/wrangler/src/api/startDevWorker/ProxyController.ts

@@ -3,6 +3,7 @@ import { randomUUID } from "node:crypto";
 import events from "node:events";
 import path from "node:path";
 import { LogLevel, Miniflare, Mutex, Response } from "miniflare";
+import { fetch } from "undici";
 import inspectorProxyWorkerPath from "worker:startDevWorker/InspectorProxyWorker";
 import proxyWorkerPath from "worker:startDevWorker/ProxyWorker";
 import WebSocket from "ws";
@@ -106,6 +107,7 @@ export class ProxyController extends Controller<ProxyControllerEventMap> {
 					},
 					bindings: {
 						PROXY_CONTROLLER_AUTH_SECRET: this.secret,
+						NAME: this.latestConfig.name ?? "",
 					},
 
 					// no need to use file-system, so don't
@@ -158,7 +160,7 @@ export class ProxyController extends Controller<ProxyControllerEventMap> {
 			log: new ProxyControllerLogger(castLogLevel(logger.loggerLevel), {
 				prefix:
 					// if debugging, log requests with specic ProxyWorker prefix
-					logger.loggerLevel === "debug" ? "wrangler-ProxyWorker" : "wrangler",
+					`wrangler-ProxyWorker-${this.latestConfig.name}`,
 			}),
 			handleRuntimeStdio,
 			liveReload: false,
@@ -286,20 +288,22 @@ export class ProxyController extends Controller<ProxyControllerEventMap> {
 
 		try {
 			await this.runtimeMessageMutex.runWith(async () => {
-				const { proxyWorker } = await this.ready.promise;
+				const { proxyWorker, url } = await this.ready.promise;
 
 				const ready = await proxyWorker.ready.catch(() => undefined);
 				if (!ready) {
 					return;
 				}
 
-				return proxyWorker.dispatchFetch(
-					`http://dummy/cdn-cgi/ProxyWorker/${message.type}`,
-					{
-						headers: { Authorization: this.secret },
-						cf: { hostMetadata: message },
-					}
+				const proxyWorkerUrl = new URL(
+					`/cdn-cgi/ProxyWorker/${message.type}`,
+					url
 				);
+				return fetch(proxyWorkerUrl, {
+					method: "POST",
+					headers: { Authorization: this.secret },
+					body: JSON.stringify(message),
+				});
 			});
 		} catch (cause) {
 			if (this._torndown) {
@@ -317,7 +321,7 @@ export class ProxyController extends Controller<ProxyControllerEventMap> {
 				error
 			);
 
-			throw error;
+			// throw error;
 		}
 	}
 	async sendMessageToInspectorProxyWorker(

packages/wrangler/src/api/startDevWorker/index.ts

@@ -1,4 +1,8 @@
+import assert from "node:assert";
+import { updateDevEnvRegistry } from "../../dev";
+import { serializeWorkerRegistryDefinition } from "../../dev/local";
 import { DevEnv } from "./DevEnv";
+import type { WorkerDefinition } from "../../dev-registry";
 import type { StartDevWorkerInput, Worker } from "./types";
 
 export { DevEnv };
@@ -12,3 +16,81 @@ export async function startWorker(
 
 	return devEnv.startWorker(options);
 }
+
+export async function startMultiWorker(
+	optionsArray: StartDevWorkerInput[],
+	devEnv0: DevEnv
+): Promise<DevEnv[]> {
+	const workerRegistry = new Map<string, WorkerDefinition>();
+	let prevRegistry: Record<string, WorkerDefinition> = {};
+	async function updateWorkerRegistry(
+		name: string,
+		definition: WorkerDefinition
+	) {
+		workerRegistry.set(name, definition);
+
+		if (!devEnvs) {
+			return;
+		}
+
+		const nextRegistry = Object.fromEntries(workerRegistry);
+
+		if (JSON.stringify(prevRegistry) !== JSON.stringify(nextRegistry)) {
+			prevRegistry = nextRegistry;
+			await Promise.all(
+				devEnvs.map(async (devEnv) => {
+					await updateDevEnvRegistry(
+						devEnv,
+						Object.fromEntries(workerRegistry)
+					);
+				})
+			);
+		}
+	}
+
+	const devEnvs = await Promise.all(
+		optionsArray.map(async (options, workerIndex) => {
+			const devEnv = workerIndex === 0 ? devEnv0 : new DevEnv();
+
+			devEnv.runtimes.forEach((runtime) => {
+				runtime.on("reloadComplete", async (reloadEvent) => {
+					if (!reloadEvent.config.dev?.remote) {
+						const { url } = await devEnv.proxy.ready.promise;
+						const { name } = reloadEvent.config;
+						assert(name); // default value "multi-worker-n" is defined below
+
+						const definition = serializeWorkerRegistryDefinition(
+							url,
+							name,
+							reloadEvent.proxyData.internalDurableObjects,
+							reloadEvent.proxyData.entrypointAddresses
+						);
+
+						if (definition) {
+							await updateWorkerRegistry(name, definition);
+						}
+					}
+				});
+			});
+
+			await devEnv.config.set({
+				// name: `multi-worker-${workerIndex + 1}`,
+				...options,
+				dev: {
+					...options.dev,
+					remote: false,
+					inspector: { port: 0 },
+					server: {
+						...options.dev?.server,
+						// port: options.dev?.server?.port ?? 0,
+						// hostname: "localhost",
+					},
+				},
+			});
+
+			return devEnv;
+		})
+	);
+
+	return devEnvs;
+}