Allow workers to be accessed directly via dedicated sockets

We'd like this for Wrangler, so we can use the same Miniflare
instance for both the regular proxy worker and the inspector proxy
worker. The inspector proxy worker needs to run on its own port as
`/json` and `/json/version` are well-known path names used by
`chrome://inspect`'s service discovery. Otherwise, we'd be able to
use `routes`. Note this is an unsafe option as it bypasses
Miniflare's entry worker, so won't include any request logging/
pretty errors.

Author: mrbbot

packages/miniflare/src/index.ts

@@ -53,8 +53,10 @@ import {
   QueuesError,
   R2_PLUGIN_NAME,
   ReplaceWorkersTypes,
+  SOCKET_ENTRY,
   SharedOptions,
   WorkerOptions,
+  getDirectSocketName,
   getGlobalServices,
   kProxyNodeBinding,
   maybeGetSitesManifestModule,
@@ -77,6 +79,7 @@ import {
   RuntimeOptions,
   Service,
   Socket,
+  SocketIdentifier,
   Worker_Binding,
   Worker_Module,
   serializeConfig,
@@ -98,6 +101,13 @@ import {
 } from "./workers";
 import { _formatZodError } from "./zod-format";
 
+const DEFAULT_HOST = "127.0.0.1";
+function getAccessibleHost(host: string) {
+  return host === "*" || host === "0.0.0.0" || host === "::"
+    ? "127.0.0.1"
+    : host;
+}
+
 // ===== `Miniflare` User Options =====
 export type MiniflareOptions = SharedOptions &
   (WorkerOptions | { workers: WorkerOptions[] });
@@ -534,6 +544,7 @@ export class Miniflare {
   #runtime?: Runtime;
   #removeRuntimeExitHook?: () => void;
   #runtimeEntryURL?: URL;
+  #socketPorts?: Map<SocketIdentifier, number>;
   #runtimeClient?: Client;
   #proxyClient?: ProxyClient;
 
@@ -575,11 +586,11 @@ export class Miniflare {
     }
 
     this.#log = this.#sharedOpts.core.log ?? new NoOpLog();
-    this.#host = this.#sharedOpts.core.host ?? "127.0.0.1";
-    this.#accessibleHost =
-      this.#host === "*" || this.#host === "0.0.0.0" || this.#host === "::"
-        ? "127.0.0.1"
-        : this.#host;
+    this.#host = this.#sharedOpts.core.host ?? DEFAULT_HOST;
+    // TODO: maybe remove `#accessibleHost` field, and just get whenever
+    //  constructing entry URL, then extract constructing entry URL into
+    //  function used `getUnsafeGetDirectURL()` too?
+    this.#accessibleHost = getAccessibleHost(this.#host);
 
     if (net.isIPv6(this.#accessibleHost)) {
       this.#accessibleHost = `[${this.#accessibleHost}]`;
@@ -965,6 +976,20 @@ export class Miniflare {
           }
         }
       }
+
+      // Allow additional sockets to be opened directly to specific workers,
+      // bypassing Miniflare's entry worker.
+      let { unsafeDirectHost, unsafeDirectPort } = workerOpts.core;
+      if (unsafeDirectHost !== undefined || unsafeDirectPort !== undefined) {
+        unsafeDirectHost ??= DEFAULT_HOST;
+        unsafeDirectPort ??= 0;
+        sockets.push({
+          name: getDirectSocketName(i),
+          address: `${unsafeDirectHost}:${unsafeDirectPort}`,
+          service: { name: getUserServiceName(workerName) },
+          http: {},
+        });
+      }
     }
 
     // For testing proxy client serialisation, add an API that just returns its
@@ -1027,28 +1052,46 @@ export class Miniflare {
     assert(this.#runtime !== undefined);
     const config = await this.#assembleConfig();
     const configBuffer = serializeConfig(config);
-    const maybePort = await this.#runtime.updateConfig(configBuffer, {
-      signal: this.#disposeController.signal,
-      entryPort: maybeApply(parseInt, this.#runtimeEntryURL?.port),
-    });
+
+    // Get all socket names we expect to get ports for
+    assert(config.sockets !== undefined);
+    const requiredSockets: SocketIdentifier[] = config.sockets.map(
+      ({ name }) => {
+        assert(name !== undefined);
+        return name;
+      }
+    );
+
+    const maybeSocketPorts = await this.#runtime.updateConfig(
+      configBuffer,
+      requiredSockets,
+      {
+        signal: this.#disposeController.signal,
+        entryPort: maybeApply(parseInt, this.#runtimeEntryURL?.port),
+      }
+    );
     if (this.#disposeController.signal.aborted) return;
-    if (maybePort === undefined) {
+    if (maybeSocketPorts === undefined) {
       throw new MiniflareCoreError(
         "ERR_RUNTIME_FAILURE",
         "The Workers runtime failed to start. " +
           "There is likely additional logging output above."
       );
     }
+    // Note: `updateConfig()` doesn't resolve until ports for all required
+    // sockets have been recorded. At this point, `maybeSocketPorts` contains
+    // all of `requiredSockets` as keys.
+    this.#socketPorts = maybeSocketPorts;
 
     const entrySocket = config.sockets?.[0];
     const secure = entrySocket !== undefined && "https" in entrySocket;
-
-    // noinspection HttpUrlsUsage
-    const previousEntry = this.#runtimeEntryURL;
+    const previousEntryURL = this.#runtimeEntryURL;
+    const entryPort = maybeSocketPorts.get(SOCKET_ENTRY);
+    assert(entryPort !== undefined);
     this.#runtimeEntryURL = new URL(
-      `${secure ? "https" : "http"}://${this.#accessibleHost}:${maybePort}`
+      `${secure ? "https" : "http"}://${this.#accessibleHost}:${entryPort}`
     );
-    if (previousEntry?.toString() !== this.#runtimeEntryURL.toString()) {
+    if (previousEntryURL?.toString() !== this.#runtimeEntryURL.toString()) {
       this.#runtimeClient = new Client(this.#runtimeEntryURL, {
         connect: { rejectUnauthorized: false },
       });
@@ -1072,7 +1115,7 @@ export class Miniflare {
 
       const host = net.isIPv6(this.#host) ? `[${this.#host}]` : this.#host;
       this.#log.info(
-        `${ready} on ${secure ? "https" : "http"}://${host}:${maybePort} `
+        `${ready} on ${secure ? "https" : "http"}://${host}:${entryPort} `
       );
 
       if (initial) {
@@ -1086,7 +1129,7 @@ export class Miniflare {
         }
 
         for (const h of hosts) {
-          this.#log.info(`- ${secure ? "https" : "http"}://${h}:${maybePort}`);
+          this.#log.info(`- ${secure ? "https" : "http"}://${h}:${entryPort}`);
         }
       }
 
@@ -1116,6 +1159,35 @@ export class Miniflare {
     return this.#waitForReady();
   }
 
+  async unsafeGetDirectURL(workerName?: string) {
+    this.#checkDisposed();
+    await this.ready;
+
+    // Get worker index and options from name, defaulting to entrypoint
+    const workerIndex = this.#findAndAssertWorkerIndex(workerName);
+    const workerOpts = this.#workerOpts[workerIndex];
+
+    // Try to get direct access port for worker
+    const socketName = getDirectSocketName(workerIndex);
+    // `#socketPorts` is assigned in `#assembleAndUpdateConfig()`, which is
+    // called by `#init()`, and `ready` doesn't resolve until `#init()` returns.
+    assert(this.#socketPorts !== undefined);
+    const maybePort = this.#socketPorts.get(socketName);
+    if (maybePort === undefined) {
+      const friendlyWorkerName =
+        workerName === undefined ? "entrypoint" : JSON.stringify(workerName);
+      throw new TypeError(
+        `Direct access disabled in ${friendlyWorkerName} worker`
+      );
+    }
+
+    // Construct accessible URL from configured host and port
+    const host = workerOpts.core.unsafeDirectHost ?? DEFAULT_HOST;
+    const accessibleHost = getAccessibleHost(host);
+    // noinspection HttpUrlsUsage
+    return new URL(`http://${accessibleHost}:${maybePort}`);
+  }
+
   #checkDisposed() {
     if (this.#disposeController.signal.aborted) {
       throw new MiniflareCoreError(
@@ -1212,24 +1284,29 @@ export class Miniflare {
     return this.#proxyClient;
   }
 
-  async getBindings<Env = Record<string, unknown>>(
-    workerName?: string
-  ): Promise<Env> {
-    const bindings: Record<string, unknown> = {};
-    const proxyClient = await this._getProxyClient();
-
-    // Find worker by name, defaulting to entrypoint worker if none specified
-    let workerOpts: PluginWorkerOptions | undefined;
+  #findAndAssertWorkerIndex(workerName?: string): number {
     if (workerName === undefined) {
-      workerOpts = this.#workerOpts[0];
+      return 0;
     } else {
-      workerOpts = this.#workerOpts.find(
+      const index = this.#workerOpts.findIndex(
         ({ core }) => (core.name ?? "") === workerName
       );
-      if (workerOpts === undefined) {
+      if (index === -1) {
         throw new TypeError(`${JSON.stringify(workerName)} worker not found`);
       }
+      return index;
     }
+  }
+
+  async getBindings<Env = Record<string, unknown>>(
+    workerName?: string
+  ): Promise<Env> {
+    const bindings: Record<string, unknown> = {};
+    const proxyClient = await this._getProxyClient();
+
+    // Find worker by name, defaulting to entrypoint worker if none specified
+    const workerIndex = this.#findAndAssertWorkerIndex(workerName);
+    const workerOpts = this.#workerOpts[workerIndex];
     workerName = workerOpts.core.name ?? "";
 
     // Populate bindings from each plugin

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

@@ -103,6 +103,8 @@ const CoreOptionsSchemaInput = z.intersection(
     fetchMock: z.instanceof(MockAgent).optional(),
 
     unsafeEphemeralDurableObjects: z.boolean().optional(),
+    unsafeDirectHost: z.string().optional(),
+    unsafeDirectPort: z.number().optional(),
   })
 );
 export const CoreOptionsSchema = CoreOptionsSchemaInput.transform((value) => {

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

@@ -7,6 +7,11 @@ import {
 import { CoreBindings, SharedBindings } from "../../workers";
 
 export const SOCKET_ENTRY = "entry";
+const SOCKET_DIRECT_PREFIX = "direct";
+
+export function getDirectSocketName(workerIndex: number) {
+  return `${SOCKET_DIRECT_PREFIX}:${workerIndex}`;
+}
 
 // Service looping back to Miniflare's Node.js process (for storage, etc)
 export const SERVICE_LOOPBACK = "loopback";

packages/miniflare/src/runtime/index.ts

@@ -17,22 +17,35 @@ const ControlMessageSchema = z.object({
   port: z.number(),
 });
 
-async function waitForPort(
-  socket: string,
+export type SocketIdentifier = string;
+
+async function waitForPorts(
+  requiredSockets: SocketIdentifier[],
   stream: Readable,
   options?: Abortable
-): Promise<number | undefined> {
+): Promise<Map<SocketIdentifier, number> | undefined> {
   if (options?.signal?.aborted) return;
   const lines = rl.createInterface(stream);
   // Calling `close()` will end the async iterator below and return undefined
   const abortListener = () => lines.close();
   options?.signal?.addEventListener("abort", abortListener, { once: true });
+  // We're going to be mutating `sockets`, so shallow copy it
+  requiredSockets = Array.from(requiredSockets);
+  const socketPorts = new Map<SocketIdentifier, number>();
   try {
     for await (const line of lines) {
       const message = ControlMessageSchema.safeParse(JSON.parse(line));
-      if (message.success && message.data.socket === socket) {
-        return message.data.port;
-      }
+      // If this was an unrecognised control message, ignore it
+      if (!message.success) continue;
+      const socket = message.data.socket;
+      const index = requiredSockets.indexOf(socket);
+      // If this wasn't a required socket, ignore it
+      if (index === -1) continue;
+      // Record the port of this socket
+      socketPorts.set(socket, message.data.port);
+      // Satisfy the requirement, if there are no more, return the ports map
+      requiredSockets.splice(index, 1);
+      if (requiredSockets.length === 0) return socketPorts;
     }
   } finally {
     options?.signal?.removeEventListener("abort", abortListener);
@@ -106,8 +119,9 @@ export class Runtime {
 
   async updateConfig(
     configBuffer: Buffer,
+    requiredSockets: SocketIdentifier[],
     options?: Abortable & Partial<Pick<RuntimeOptions, "entryPort">>
-  ): Promise<number | undefined> {
+  ): Promise<Map<SocketIdentifier, number /* port */> | undefined> {
     // 1. Stop existing process (if any) and wait for exit
     await this.dispose();
     // TODO: what happens if runtime crashes?
@@ -132,8 +146,8 @@ export class Runtime {
     runtimeProcess.stdin.write(configBuffer);
     runtimeProcess.stdin.end();
 
-    // 4. Wait for socket to start listening
-    return waitForPort(SOCKET_ENTRY, controlPipe, options);
+    // 4. Wait for sockets to start listening
+    return waitForPorts(requiredSockets, controlPipe, options);
   }
 
   dispose(): Awaitable<void> {

packages/miniflare/test/index.spec.ts

@@ -837,6 +837,51 @@ test("Miniflare: getBindings() and friends return bindings for different workers
   await t.throwsAsync(() => mf.getQueueProducer("BUCKET", "c"), expectations);
 });
 
+test("Miniflare: allows direct access to workers", async (t) => {
+  const mf = new Miniflare({
+    workers: [
+      {
+        name: "a",
+        script: `addEventListener("fetch", (e) => e.respondWith(new Response("a")))`,
+        unsafeDirectPort: 0,
+      },
+      {
+        routes: ["*/*"],
+        script: `addEventListener("fetch", (e) => e.respondWith(new Response("b")))`,
+      },
+      {
+        name: "c",
+        script: `addEventListener("fetch", (e) => e.respondWith(new Response("c")))`,
+        unsafeDirectHost: "127.0.0.1",
+      },
+    ],
+  });
+  t.teardown(() => mf.dispose());
+
+  // Check can access workers as usual
+  let res = await mf.dispatchFetch("http://localhost/");
+  t.is(await res.text(), "b");
+
+  // Check can access workers directly
+  // (`undefined` worker name should default to entrypoint, not unnamed worker)
+  const aURL = await mf.unsafeGetDirectURL();
+  const cURL = await mf.unsafeGetDirectURL("c");
+  res = await fetch(aURL);
+  t.is(await res.text(), "a");
+  res = await fetch(cURL);
+  t.is(await res.text(), "c");
+
+  // Can can only access configured for direct access
+  await t.throwsAsync(mf.unsafeGetDirectURL("d"), {
+    instanceOf: TypeError,
+    message: '"d" worker not found',
+  });
+  await t.throwsAsync(mf.unsafeGetDirectURL(""), {
+    instanceOf: TypeError,
+    message: 'Direct access disabled in "" worker',
+  });
+});
+
 // Only test `MINIFLARE_WORKERD_PATH` on Unix. The test uses a Node.js script
 // with a shebang, directly as the replacement `workerd` binary, which won't
 // work on Windows.