From df849c8302b6deb1d24b73274e1a58e436e53cb2 Mon Sep 17 00:00:00 2001 From: Edmund Hung Date: Mon, 19 Oct 2026 21:35:24 +0100 Subject: [PATCH 1/4] [hyperdrive] Support remote bindings through the shared session With remote: true, application code still runs locally and uses an ordinary database driver, but its connections reach the deployed Hyperdrive binding. For example, pg.Client can use env.DB.connectionString and run SELECT 1 without knowing about remote bindings. Miniflare implements this shared path; Wrangler and Vite only pass through the configuration and resolve local fallbacks. The local Hyperdrive binding needs a TCP destination, whereas remote bindings provide remote calls (RPC) and streams over WebSocket. A bridge listening on 127.0.0.1 joins those interfaces without changing the driver or workerd. The existing local remote-session gateway forwards its WebSocket to a proxy Worker on Cloudflare, which holds the real Hyperdrive binding. Neither the bridge nor the proxy interprets SQL: they carry the driver's normal database traffic. During startup, the bridge opens a persistent RPC session and asks the proxy Worker for the binding's connection string. Those details configure the local binding before the local runtime starts. remoteProxyConnectionString names the gateway URL, not the database URL. When a driver connects, the bridge calls connect() on the same remote handle and copies bytes between the local socket and the returned streams. Three driver connections get three independent database sockets, not one shared database connection. Startup retries are bounded. A failed bridge is retained across unchanged reloads so one unavailable binding does not delay every file save. Restart development or change its target to retry. Losing an established session still allows a fresh session on the next reload. Configuration assembly awaits startup before rolling back, so shutdown cancellation is sufficient; no per-update abort signal is needed. The shared handle matters because credentials from remote binding instance A may not authenticate a socket opened by instance B. Fetching a connection string separately from opening sockets can cross that boundary. One session owns both operations instead of seeding credentials from a throwaway connection. - `ProxyServerWorker.ts`: The remote adapter exposes the native binding's connection string and a fresh connect() stream pair through the same handle. - `hyperdrive/index.ts`: Worker bindings route their synthetic database address to the TCP bridge. Node bindings instead expose its loopback address and port; both receive the session's user, password, and database together. - `hyperdrive-proxy.ts`: Each remote listener owns its session and accepted clients. An unchanged target reuses it. A successful reload retires unused listeners without interrupting existing connections; a failed reload rolls back new candidates. Shutdown cancels startup and closes active and retired resources. The existing local database/TLS proxy remains a separate connection path. - `remote-affinity.spec.ts`: Different credentials for each simulated session make accidental cross-session reuse fail. Worker and Node clients exercise reloads, database ID changes, reconnection, and another binding's failure without retrying that unchanged failed binding on each update. Co-authored-by: Henry Jang --- .../hyperdrive-remote-bindings-local-dev.md | 21 + packages/config/src/__tests__/convert.test.ts | 17 + packages/config/src/bindings.ts | 2 +- packages/config/src/convert.ts | 1 + packages/config/src/schema.ts | 4 +- packages/miniflare/src/config/schema.ts | 17 +- packages/miniflare/src/config/v4-convert.ts | 7 +- packages/miniflare/src/config/v4-schema.ts | 20 +- packages/miniflare/src/index.ts | 21 +- .../plugins/hyperdrive/hyperdrive-proxy.ts | 405 +++++++++++++++++- .../miniflare/src/plugins/hyperdrive/index.ts | 222 ++++++++-- .../miniflare/src/plugins/shared/index.ts | 13 +- packages/miniflare/test/config/schema.spec.ts | 25 +- .../miniflare/test/config/v4-convert.spec.ts | 23 + .../miniflare/test/config/v4-schema.spec.ts | 10 + .../test/plugins/analytics-sql/index.spec.ts | 6 +- .../test/plugins/hyperdrive/helpers.ts | 26 ++ .../test/plugins/hyperdrive/index.spec.ts | 21 +- .../test/plugins/hyperdrive/lifecycle.spec.ts | 10 + .../hyperdrive/remote-affinity.spec.ts | 163 +++++++ .../hyperdrive/remote-lifecycle.spec.ts | 88 ++++ .../plugins/hyperdrive/remote-startup.spec.ts | 233 ++++++++++ .../shared/remote-bindings-connect.spec.ts | 135 ++++++ .../remoteBindings/ProxyServerWorker.ts | 13 + .../src/__tests__/local-env.spec.ts | 21 + .../src/__tests__/remote-bindings.spec.ts | 12 +- .../vite-plugin-cloudflare/src/local-env.ts | 8 +- .../src/miniflare-options.ts | 11 +- .../src/remote-bindings.ts | 7 + .../src/config/binding-local-support.ts | 2 +- .../workers-utils/src/config/environment.ts | 5 + .../workers-utils/src/config/validation.ts | 8 + packages/workers-utils/src/print-bindings.ts | 6 +- packages/workers-utils/src/worker.ts | 5 + packages/wrangler/src/__tests__/dev.test.ts | 35 ++ .../dev/miniflare-hyperdrive.test.ts | 100 +++++ packages/wrangler/src/dev.ts | 5 +- packages/wrangler/src/dev/miniflare/index.ts | 50 ++- 38 files changed, 1672 insertions(+), 106 deletions(-) create mode 100644 .changeset/hyperdrive-remote-bindings-local-dev.md create mode 100644 packages/miniflare/test/plugins/hyperdrive/helpers.ts create mode 100644 packages/miniflare/test/plugins/hyperdrive/remote-affinity.spec.ts create mode 100644 packages/miniflare/test/plugins/hyperdrive/remote-lifecycle.spec.ts create mode 100644 packages/miniflare/test/plugins/hyperdrive/remote-startup.spec.ts create mode 100644 packages/wrangler/src/__tests__/dev/miniflare-hyperdrive.test.ts diff --git a/.changeset/hyperdrive-remote-bindings-local-dev.md b/.changeset/hyperdrive-remote-bindings-local-dev.md new file mode 100644 index 00000000000..16529b58f3b --- /dev/null +++ b/.changeset/hyperdrive-remote-bindings-local-dev.md @@ -0,0 +1,21 @@ +--- +"@cloudflare/config": minor +"miniflare": minor +"wrangler": minor +"@cloudflare/vite-plugin": minor +"@cloudflare/workers-utils": minor +--- + +Support remote Hyperdrive bindings in local development + +```jsonc +{ + "hyperdrive": [ + { + "binding": "HYPERDRIVE", + "id": "", + "remote": true, + }, + ], +} +``` diff --git a/packages/config/src/__tests__/convert.test.ts b/packages/config/src/__tests__/convert.test.ts index 45ff29a6757..7e9446f476b 100644 --- a/packages/config/src/__tests__/convert.test.ts +++ b/packages/config/src/__tests__/convert.test.ts @@ -394,6 +394,23 @@ describe("convertToWranglerConfig", () => { ]); }); + it("maps remote Hyperdrive without a local connection string", ({ + expect, + }) => { + const result = convertToWranglerConfig({ + worker: { + ...baseWorker, + env: { + HD: { type: "hyperdrive", id: "h-1", dev: { remote: true } }, + }, + }, + containers: [], + }); + expect(result.hyperdrive).toEqual([ + { binding: "HD", id: "h-1", remote: true }, + ]); + }); + it("maps pipeline.name to stream", ({ expect }) => { const result = convertToWranglerConfig({ worker: { diff --git a/packages/config/src/bindings.ts b/packages/config/src/bindings.ts index 90dfea13cf4..a5c38e3ff3f 100644 --- a/packages/config/src/bindings.ts +++ b/packages/config/src/bindings.ts @@ -253,7 +253,7 @@ export interface HyperdriveBindingOptions { /** The ID of the Hyperdrive configuration. */ id: string; /** Options that only apply during local development. */ - dev?: { + dev?: BindingDevOptions & { /** The database connection string used during local development. */ connectionString?: string; }; diff --git a/packages/config/src/convert.ts b/packages/config/src/convert.ts index 19e02c74cd7..b641f56fa1c 100644 --- a/packages/config/src/convert.ts +++ b/packages/config/src/convert.ts @@ -543,6 +543,7 @@ function convertBindingsAndAssets( omitUndefined({ binding: name, id: binding.id, + remote: binding.dev?.remote, localConnectionString: binding.dev?.connectionString, }) ); diff --git a/packages/config/src/schema.ts b/packages/config/src/schema.ts index f9ae5b57edd..541b03522c8 100644 --- a/packages/config/src/schema.ts +++ b/packages/config/src/schema.ts @@ -99,7 +99,9 @@ export const FlagshipBindingSchema = z.strictObject({ export const HyperdriveBindingSchema = z.strictObject({ type: z.literal("hyperdrive"), id: z.string(), - dev: z.strictObject({ connectionString: z.string().optional() }).optional(), + dev: RemoteBindingDevSchema.extend({ + connectionString: z.string().optional(), + }).optional(), }); export const KnownBindingSchema = z.discriminatedUnion("type", [ diff --git a/packages/miniflare/src/config/schema.ts b/packages/miniflare/src/config/schema.ts index 9666de6fcf8..a5388d481a8 100644 --- a/packages/miniflare/src/config/schema.ts +++ b/packages/miniflare/src/config/schema.ts @@ -181,8 +181,23 @@ const MiniflareBrowserBindingSchema = BrowserBindingSchema.extend({ headful: z.boolean().optional(), }); +/** + * Remote Hyperdrive bindings get their connection string from the proxy session. + */ const MiniflareHyperdriveBindingSchema = HyperdriveBindingSchema.extend({ - dev: z.strictObject({ connectionString: z.string() }), + dev: z.strictObject({ + connectionString: z.string().optional(), + remote: z.boolean().optional(), + }), +}).superRefine((value, ctx) => { + if (!value.dev.remote && value.dev.connectionString === undefined) { + ctx.addIssue({ + code: "custom", + path: ["dev", "connectionString"], + message: + "A Hyperdrive binding requires a connection string unless it sets 'remote': true.", + }); + } }); /** diff --git a/packages/miniflare/src/config/v4-convert.ts b/packages/miniflare/src/config/v4-convert.ts index 734cd06a2d4..78317dbb54c 100644 --- a/packages/miniflare/src/config/v4-convert.ts +++ b/packages/miniflare/src/config/v4-convert.ts @@ -704,10 +704,13 @@ function addProductBindings( env[name] = { type: "analytics-engine-dataset", name: binding.dataset }; } for (const [name, value] of Object.entries(worker.hyperdrives ?? {})) { + const isRemoteEntry = typeof value === "object" && !(value instanceof URL); env[name] = { type: "hyperdrive", - id: name, - dev: { connectionString: String(value) }, + id: isRemoteEntry && value.id ? value.id : name, + dev: isRemoteEntry + ? { remote: isRemote(value.remoteProxyConnectionString) } + : { connectionString: String(value) }, }; } for (const [name, binding] of Object.entries(worker.ratelimits ?? {})) { diff --git a/packages/miniflare/src/config/v4-schema.ts b/packages/miniflare/src/config/v4-schema.ts index 01c32bac455..069334f65c8 100644 --- a/packages/miniflare/src/config/v4-schema.ts +++ b/packages/miniflare/src/config/v4-schema.ts @@ -278,7 +278,7 @@ const V4QueueConsumerOptionsSchema = z.object({ retryDelay: V4QueueMessageDelaySchema, }); -const V4HyperdriveSchema = z +const V4HyperdriveConnectionSchema = z .union([z.url(), z.instanceof(URL)]) .superRefine((value, ctx) => { const url = typeof value === "string" ? new URL(value) : value; @@ -323,6 +323,14 @@ const V4HyperdriveSchema = z } }); +const V4HyperdriveSchema = z.union([ + V4HyperdriveConnectionSchema, + z.object({ + id: z.string().optional(), + remoteProxyConnectionString: RemoteProxyConnectionStringSchema, + }), +]); + const V4PipelineSchema = z.union([ z.string(), z.object({ @@ -867,7 +875,15 @@ export type V4WorkerOptionsShape = { }) > | string[]; - hyperdrives?: Record; + hyperdrives?: Record< + string, + | string + | URL + | { + id?: string; + remoteProxyConnectionString: RemoteProxyConnectionString; + } + >; ratelimits?: Record< string, { namespace_id: string; simple: { limit: number; period?: 10 | 60 } } diff --git a/packages/miniflare/src/index.ts b/packages/miniflare/src/index.ts index 6f0148dfa55..7bc74a97657 100644 --- a/packages/miniflare/src/index.ts +++ b/packages/miniflare/src/index.ts @@ -879,7 +879,7 @@ export class Miniflare { readonly #initPromise: Promise; // Aborted when dispose() is called - readonly #disposeController: AbortController; + readonly #disposeController = new AbortController(); #loopbackServer?: StoppableServer; #loopbackHost?: string; readonly #webSocketServer: WebSocketServer; @@ -891,7 +891,7 @@ export class Miniflare { #containerPrivilegesCache = new ContainerPrivilegesCache(); #hyperdriveProxyController: HyperdriveProxyController = - new HyperdriveProxyController(); + new HyperdriveProxyController(this.#disposeController.signal); // Set the first time the remote-bindings proxy CLIENT worker reports a // Cloudflare Access 403 block on a request to the remote-bindings proxy @@ -1028,7 +1028,6 @@ export class Miniflare { this.#devRegistry.dispose(); }); - this.#disposeController = new AbortController(); this.#runtimeMutex = new Mutex(); this.#initPromise = this.#runtimeMutex .runWith(async () => { @@ -2675,6 +2674,14 @@ export class Miniflare { this.#hyperdriveProxyController.beginUpdate(); try { await this.#assembleAndUpdateConfigInternal(reusePorts); + } catch (error) { + if ( + !this.#disposeController.signal.aborted || + !(error instanceof Error) || + error.name !== "AbortError" + ) { + throw error; + } } finally { // If assembly or runtime startup failed, keep the old listeners and // close only the new candidates. A successful update already committed. @@ -3433,7 +3440,10 @@ export class Miniflare { // Populate bindings from each plugin for (const [key, plugin] of this.#mergedPluginEntries) { - const pluginBindings = await plugin.getNodeBindings(workerOpts); + const pluginBindings = await plugin.getNodeBindings(workerOpts, { + hyperdriveProxyController: this.#hyperdriveProxyController, + workerIndex, + }); for (const [name, binding] of Object.entries(pluginBindings)) { if (binding instanceof ProxyNodeBinding) { const proxyBindingName = getProxyBindingName(key, workerName, name); @@ -3955,8 +3965,7 @@ export class Miniflare { // Unregister workers from dev registry and stop the file watcher await this.#devRegistry.dispose(); - // shutdown hyperdrive proxies if any exist - await this.#hyperdriveProxyController.dispose(); + this.#hyperdriveProxyController.dispose(); // Remove from instance registry as last step in disposal, preserving the // existing behavior when an earlier cleanup operation fails. diff --git a/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts b/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts index 315863bb6e1..c6f707dfa7e 100644 --- a/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts +++ b/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts @@ -1,7 +1,17 @@ import fs from "node:fs"; import net from "node:net"; +import { Readable, Writable } from "node:stream"; +import { pipeline } from "node:stream/promises"; +import { setTimeout as delay } from "node:timers/promises"; import tls from "node:tls"; +import { newWebSocketRpcSession } from "capnweb"; +import WebSocket from "ws"; import type { Log } from "../../shared"; +import type { RpcStub } from "capnweb"; + +const REMOTE_HYPERDRIVE_CONNECTION_ATTEMPTS = 3; +const REMOTE_HYPERDRIVE_CONNECTION_TIMEOUT_MS = 10_000; +const REMOTE_HYPERDRIVE_RETRY_BACKOFF_MS = 500; export interface HyperdriveProxyConfig { // Name of the Hyperdrive binding @@ -29,12 +39,36 @@ export const POSTGRES_SSL_REQUEST_PACKET = Buffer.from([ 0x00, 0x00, 0x00, 0x08, 0x04, 0xd2, 0x16, 0x2f, ]); +interface ConnectionOwner { + track(socket: T): T; + assertOpen(): void; + close(): void; +} + interface TlsConfig { rejectUnauthorized: boolean; ca?: Buffer; checkServerIdentity?: () => undefined; } +interface RemoteHyperdrive { + getConnectionString(): string; + connect(): { + readable: ReadableStream; + writable: WritableStream; + }; +} + +interface ProxyServer { + server: net.Server; + port: number; + connections: Set; + session?: { + stub: RpcStub; + connectionString: string; + }; +} + /** * Builds TLS configuration based on sslmode and sslrootcert settings. * @@ -112,12 +146,91 @@ function buildTlsConnectionOptions( */ export class HyperdriveProxyController { // Map a binding and target to the listening proxy server and port - #servers = new Map(); + #servers = new Map(); #starting = new Map>(); - #update?: { created: Set }; + #retiredServers = new Set(); + #remoteBindings = new Map(); + #abortController = new AbortController(); + #signal: AbortSignal; + #update?: { + created: Set; + remoteBindings: Map; + }; #disposed = false; log?: Log; + constructor(signal?: AbortSignal) { + this.#signal = signal + ? AbortSignal.any([this.#abortController.signal, signal]) + : this.#abortController.signal; + } + + #createOwner(entry: ProxyServer, clientSocket: net.Socket): ConnectionOwner { + const sockets = new Set(); + let closed = false; + const owner: ConnectionOwner = { + track(socket) { + if (closed) { + socket.destroy(); + return socket; + } + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + return socket; + }, + assertOpen() { + if (closed || clientSocket.destroyed || clientSocket.readableEnded) { + throw new Error("Hyperdrive connection closed"); + } + }, + close: () => { + if (closed) { + return; + } + closed = true; + for (const socket of sockets) { + socket.destroy(); + } + entry.connections.delete(owner); + if ( + entry.connections.size === 0 && + this.#retiredServers.delete(entry) + ) { + entry.session?.stub[Symbol.dispose](); + } + }, + }; + entry.connections.add(owner); + owner.track(clientSocket); + clientSocket.once("close", owner.close); + return owner; + } + + /** + * Stops a listener accepting new connections and forgets it, leaving any + * connection it already accepted to finish on its own. Used when a reload + * stops using a listener; a query mid-flight should not fail because the + * config around it changed. + */ + #retireServer(key: string): void { + const entry = this.#servers.get(key); + if (entry === undefined) { + return; + } + entry.server.close(); + this.#servers.delete(key); + if (entry.connections.size > 0) { + this.#retiredServers.add(entry); + } else { + entry.session?.stub[Symbol.dispose](); + } + for (const [name, active] of this.#remoteBindings) { + if (active === entry) { + this.#remoteBindings.delete(name); + } + } + } + /** * Creates a proxy server for a Hyperdrive binding. * @@ -163,21 +276,40 @@ export class HyperdriveProxyController { ): Promise { const { name, targetHost, targetPort, scheme, sslmode, sslrootcert } = config; - const server = net.createServer((clientSocket) => { + const entry: ProxyServer = { + server: net.createServer(), + port: 0, + connections: new Set(), + }; + entry.server.on("connection", (clientSocket) => { clientSocket.setNoDelay(true); - this.#handleConnection( + const owner = this.#createOwner(entry, clientSocket); + clientSocket.once("end", owner.close); + clientSocket.once("error", owner.close); + void this.#handleConnection( clientSocket, targetHost, Number.parseInt(targetPort), scheme, sslmode, sslrootcert - ); + ).catch(() => owner.close()); }); + return this.#startServer(key, entry, "proxy server", name, update); + } + + async #startServer( + key: string, + entry: ProxyServer, + description: string, + name: string, + update?: { created: Set } + ): Promise { + const { server } = entry; server.on("error", (err) => { this.log?.error( new Error( - `Hyperdrive proxy server error for binding "${name}": ${err.message}` + `Hyperdrive ${description} error for binding "${name}": ${err.message}` ) ); }); @@ -197,11 +329,22 @@ export class HyperdriveProxyController { server.close(); throw new Error("Hyperdrive proxy configuration update was interrupted"); } - this.#servers.set(key, { server, port }); + entry.port = port; + this.#servers.set(key, entry); update?.created.add(key); return port; } + /** Port of the local TCP bridge for a remote binding, if one is running. */ + getRemoteBridgePort(name: string): number | undefined { + return this.#remoteBindings.get(name)?.port; + } + + /** Returns credentials from the session that opens this binding's sockets. */ + getRemoteConnectionString(name: string): string | undefined { + return this.#remoteBindings.get(name)?.session?.connectionString; + } + /** * Handles a connection from workerd by hyperdrive magic hostname "randomUUID.hyperdrive.local" * @@ -271,12 +414,223 @@ export class HyperdriveProxyController { dbSocket.pipe(clientSocket); } + /** + * Creates a local TCP bridge for a *remote* Hyperdrive binding. + * + * The bridge is a plain `net.Server` on `127.0.0.1:`; the binding's + * `external.tcp` designator points at it, so workerd stays unmodified — + * pointing a Hyperdrive designator at a Worker service crashes workerd + * (SIGSEGV). Each inbound TCP connection is relayed, byte-for-byte, over a + * shared RPC session that owns both the credentials and the edge binding. + * Each client gets its own socket without moving to another Worker instance. + * + * Bridges share `createProxyServer`'s lifecycle: one is reused across + * reloads while its target is unchanged, and `commitUpdate()` retires it + * once a reload stops using it — including when the binding is removed or + * switched to local. + * + * @returns the local bridge port the designator should target. + */ + async createRemoteTcpBridge(config: { + // Unique per-worker key for this bridge (see `getHyperdriveServiceName`), + // used to track the listener across reloads. + name: string; + // The binding name as the Worker declares it. This is what the edge relay + // dispatches on, so it must stay unqualified. + bindingName: string; + bindingId: string; + // The remote proxy connection string (a local URL that upgrades to a + // WebSocket relaying to the edge Hyperdrive binding). + remoteProxyConnectionString: URL; + }): Promise { + if (this.#disposed) { + throw new Error("Hyperdrive proxy controller has been disposed"); + } + const { name, bindingName, bindingId, remoteProxyConnectionString } = + config; + const wsUrl = new URL(remoteProxyConnectionString.href); + wsUrl.protocol = wsUrl.protocol === "https:" ? "wss:" : "ws:"; + wsUrl.searchParams.set("MF-Binding", bindingName); + const wsHref = wsUrl.href; + const key = JSON.stringify([ + "remote", + name, + bindingName, + bindingId, + wsHref, + ]); + const update = this.#update; + const existing = this.#servers.get(key); + if (existing !== undefined) { + this.#remoteBindings.set(name, existing); + return existing.port; + } + const starting = this.#starting.get(key); + if (starting !== undefined) { + return starting; + } + const promise = this.#startRemoteTcpBridge( + key, + name, + bindingName, + wsHref, + update + ); + this.#starting.set(key, promise); + try { + return await promise; + } finally { + this.#starting.delete(key); + } + } + + async #startRemoteTcpBridge( + key: string, + name: string, + bindingName: string, + wsHref: string, + update?: { created: Set } + ): Promise { + const session = await this.#connectRemoteHyperdrive( + wsHref, + bindingName, + this.#signal + ); + const entry: ProxyServer = { + server: net.createServer(), + port: 0, + connections: new Set(), + session, + }; + entry.server.on("connection", (clientSocket) => { + clientSocket.setNoDelay(true); + this.#createOwner(entry, clientSocket); + if (session) { + void this.#handleRemoteBridgeConnection(clientSocket, session.stub); + } else { + clientSocket.destroy(); + } + }); + const port = await this.#startServer( + key, + entry, + "remote bridge", + name, + update + ).catch((error: unknown) => { + session?.stub[Symbol.dispose](); + throw error; + }); + this.#remoteBindings.set(name, entry); + session?.stub.onRpcBroken(() => { + if (!this.#disposed && this.#servers.get(key) === entry) { + this.#retireServer(key); + this.log?.warn( + `Remote Hyperdrive connection for binding "${bindingName}" was lost. Reload local development to reconnect.` + ); + } + }); + return port; + } + + async #connectRemoteHyperdrive( + wsHref: string, + bindingName: string, + signal: AbortSignal + ) { + for ( + let attempt = 1; + attempt <= REMOTE_HYPERDRIVE_CONNECTION_ATTEMPTS; + attempt++ + ) { + signal.throwIfAborted(); + const websocket = new WebSocket(wsHref, { + headers: process.env.CF_TRACE_ID + ? { "cf-trace-id": process.env.CF_TRACE_ID } + : undefined, + }); + const stub = newWebSocketRpcSession( + websocket as unknown as globalThis.WebSocket + ); + function abortConnection() { + stub[Symbol.dispose](); + websocket.terminate(); + } + signal.addEventListener("abort", abortConnection, { once: true }); + const timeout = setTimeout( + () => websocket.terminate(), + REMOTE_HYPERDRIVE_CONNECTION_TIMEOUT_MS + ); + try { + const connectionString = await stub.getConnectionString(); + signal.throwIfAborted(); + if (typeof connectionString !== "string") { + throw new Error("Missing Hyperdrive connection string"); + } + return { stub, connectionString }; + } catch { + stub[Symbol.dispose](); + signal.throwIfAborted(); + } finally { + clearTimeout(timeout); + signal.removeEventListener("abort", abortConnection); + } + if (attempt < REMOTE_HYPERDRIVE_CONNECTION_ATTEMPTS) { + await delay(REMOTE_HYPERDRIVE_RETRY_BACKOFF_MS * attempt, undefined, { + signal, + }); + } + } + this.log?.warn( + `Unable to connect remote Hyperdrive binding "${bindingName}". Other bindings are still available. Restart local development to retry.` + ); + } + + /** + * Relays an independent TCP stream through the credential-owning RPC session. + */ + async #handleRemoteBridgeConnection( + clientSocket: net.Socket, + stub: RpcStub + ): Promise { + clientSocket.pause(); + function destroyClient() { + clientSocket.destroy(); + } + clientSocket.on("error", destroyClient); + try { + const socket = await stub.connect(); + try { + if (clientSocket.destroyed) { + await Promise.allSettled([ + socket.readable.cancel(), + socket.writable.abort(), + ]); + return; + } + await Promise.all([ + pipeline(clientSocket, Writable.fromWeb(socket.writable)), + pipeline(Readable.fromWeb(socket.readable), clientSocket), + ]); + } finally { + socket[Symbol.dispose](); + } + } catch { + clientSocket.destroy(); + } finally { + clientSocket.off("error", destroyClient); + } + } + /** Begins tracking proxy servers created for the next runtime config. */ beginUpdate(): void { if (this.#update !== undefined) { throw new Error("Hyperdrive proxy configuration update already started"); } - this.#update = { created: new Set() }; + this.#update = { + created: new Set(), + remoteBindings: new Map(this.#remoteBindings), + }; } /** Stops listeners omitted by the successfully installed runtime config. */ @@ -286,15 +640,14 @@ export class HyperdriveProxyController { return; } this.#update = undefined; - for (const [key, { server, port }] of this.#servers) { + for (const [key, { port }] of this.#servers) { if (!activeAddresses.has(`127.0.0.1:${port}`)) { - server.close(); - this.#servers.delete(key); + this.#retireServer(key); } } } - /** Discards new listeners when the runtime config could not be installed. */ + /** Discards new listeners and restores addresses after a failed config update. */ abortUpdate(): void { const update = this.#update; if (update === undefined) { @@ -302,24 +655,32 @@ export class HyperdriveProxyController { } this.#update = undefined; for (const key of update.created) { - const entry = this.#servers.get(key); - entry?.server.close(); - this.#servers.delete(key); + this.#retireServer(key); } + const activeServers = new Set(this.#servers.values()); + this.#remoteBindings = new Map( + Array.from(update.remoteBindings).filter(([, entry]) => + activeServers.has(entry) + ) + ); } /** Disposes of the proxy servers when shutting down the worker.*/ dispose(): void { this.#disposed = true; + this.#abortController.abort(); this.#update = undefined; - // Stop accepting new connections on each proxy server. We don't await - // server.close() because net.Server waits for all existing connections - // to end before calling the callback, and lingering TCP sockets (e.g. - // from in-progress TLS negotiation) could block dispose indefinitely. - for (const { server } of this.#servers.values()) { - server.close(); - } + const entries = [...this.#servers.values(), ...this.#retiredServers]; this.#servers.clear(); + this.#retiredServers.clear(); + this.#remoteBindings.clear(); + for (const entry of entries) { + entry.server.close(); + entry.session?.stub[Symbol.dispose](); + for (const owner of entry.connections) { + owner.close(); + } + } } } diff --git a/packages/miniflare/src/plugins/hyperdrive/index.ts b/packages/miniflare/src/plugins/hyperdrive/index.ts index c0851da8d86..31696825c44 100644 --- a/packages/miniflare/src/plugins/hyperdrive/index.ts +++ b/packages/miniflare/src/plugins/hyperdrive/index.ts @@ -1,20 +1,68 @@ import assert from "node:assert"; import { z } from "zod"; -import { getEnvBindingsOfType, ProxyNodeBinding } from "../shared"; +import { + getEnvBindingsOfType, + getRemoteProxyConnectionString, + ProxyNodeBinding, +} from "../shared"; +import type { ParsedDevConfig } from "../../config/schema"; import type { Worker_Binding } from "../../runtime"; import type { ParsedMiniflareWorkerConfig } from "../shared"; -import type { Plugin } from "../shared"; +import type { Plugin, RemoteProxyConnectionString } from "../shared"; export const HYPERDRIVE_PLUGIN_NAME = "hyperdrive"; -/** Resolves `hyperdrive` env bindings, parsing each connection string. */ +/** + * Service (and proxy/bridge) name for a Hyperdrive binding. + * + * Namespaced by worker as well as binding name: in a multi-worker setup it is + * common for several workers to bind Hyperdrive under the same name (`DB`), and + * an unqualified name would make them collide on one service and one bridge. + */ +export function getHyperdriveServiceName( + workerIndex: number, + bindingName: string +) { + return `${HYPERDRIVE_PLUGIN_NAME}:${workerIndex}:${bindingName}`; +} + +/** + * Placeholder connection string used to synthesise the local Hyperdrive binding + * instead of an unused local connection string. workerd still needs a + * scheme/database/user/password to build the magic `connectionString` it exposes + * to the Worker; getServices replaces these fields with credentials from the + * live remote session before installing the runtime configuration. + */ +const REMOTE_PLACEHOLDER_URL = new URL( + "mysql://user:password@hyperdrive.local:3306/database" +); + +/** + * Resolves `hyperdrive` env bindings, parsing each connection string. + * + * A binding with `dev.remote: true` (and a remote proxy session) is reached + * through a local TCP bridge relaying to the edge. Its `dev.connectionString` + * is ignored; the placeholder is replaced with the session credentials. + */ function getHyperdrives( - config: ParsedMiniflareWorkerConfig -): [name: string, url: URL][] { - return getEnvBindingsOfType(config, "hyperdrive").map(([name, binding]) => [ - name, - HyperdriveSchema.parse(binding.dev.connectionString), - ]); + config: ParsedMiniflareWorkerConfig, + dev: ParsedDevConfig | undefined +): [ + name: string, + url: URL, + remoteProxyConnectionString: RemoteProxyConnectionString | undefined, + id: string, +][] { + return getEnvBindingsOfType(config, "hyperdrive").map(([name, binding]) => { + const remoteProxyConnectionString = getRemoteProxyConnectionString( + binding, + dev + ); + const url = remoteProxyConnectionString + ? REMOTE_PLACEHOLDER_URL + : HyperdriveSchema.parse(binding.dev.connectionString); + return [name, url, remoteProxyConnectionString, binding.id]; + }); } function hasPostgresProtocol(url: URL) { @@ -93,46 +141,130 @@ export const HyperdriveSchema = z export const HYPERDRIVE_PLUGIN: Plugin = { bindingTypeDescription: "Hyperdrive", - getBindings(options) { - return getHyperdrives(options.config).map(([name, url]) => { - const database = url.pathname.replace("/", ""); - const scheme = url.protocol.replace(":", ""); - return { - name, - hyperdrive: { - designator: { - name: `${HYPERDRIVE_PLUGIN_NAME}:${name}`, + getBindings(options, _sharedOptions, workerIndex) { + return getHyperdrives(options.config, options.dev).map( + ([name, url]) => { + const database = url.pathname.replace("/", ""); + const scheme = url.protocol.replace(":", ""); + return { + name, + hyperdrive: { + // Both local and remote bindings use the per-binding + // `hyperdrive::` external.tcp designator. For remote bindings + // that service points at the local TCP bridge (see `getServices`), + // which relays to the edge. workerd is unmodified either way — + // pointing a Hyperdrive designator at a Worker service SIGSEGVs. + designator: { + name: getHyperdriveServiceName(workerIndex, name), + }, + database: decodeURIComponent(database), + user: decodeURIComponent(url.username), + password: decodeURIComponent(url.password), + scheme, }, - database: decodeURIComponent(database), - user: decodeURIComponent(url.username), - password: decodeURIComponent(url.password), - scheme, - }, - }; - }); + }; + } + ); }, - getNodeBindings(options) { + getNodeBindings(options, { hyperdriveProxyController, workerIndex }) { return Object.fromEntries( - getHyperdrives(options.config).map(([name, url]) => { - const connectionOverrides: Record = { - connectionString: `${url}`, - port: Number.parseInt(url.port), - host: url.hostname, - }; - const proxyNodeBinding = new ProxyNodeBinding({ - get(target, prop) { - return prop in connectionOverrides - ? connectionOverrides[prop] - : target[prop]; - }, - }); - return [name, proxyNodeBinding]; - }) + getHyperdrives(options.config, options.dev).map( + ([name, url, remoteProxyConnectionString]) => { + // A remote binding's connection string points at a + // `*.hyperdrive.local` host, which only resolves inside workerd via + // the binding's designator. Node has no such resolver, so hand it the + // local TCP bridge instead — same credentials, an address it can dial. + const bridgePort = remoteProxyConnectionString + ? hyperdriveProxyController.getRemoteBridgePort( + getHyperdriveServiceName(workerIndex, name) + ) + : undefined; + const nodeUrl = new URL( + hyperdriveProxyController.getRemoteConnectionString( + getHyperdriveServiceName(workerIndex, name) + ) ?? url + ); + if (bridgePort !== undefined) { + nodeUrl.hostname = "127.0.0.1"; + nodeUrl.port = String(bridgePort); + } + // Every credential field has to come from the same source as the + // address. workerd's binding carries the values for its own + // connection, so leaving any of them un-overridden mixes two sets + // of credentials and Hyperdrive rejects the login. + const connectionOverrides: Record = + { + connectionString: `${nodeUrl}`, + port: Number.parseInt(nodeUrl.port), + host: nodeUrl.hostname, + user: decodeURIComponent(nodeUrl.username), + password: decodeURIComponent(nodeUrl.password), + database: decodeURIComponent(nodeUrl.pathname.replace("/", "")), + }; + const proxyNodeBinding = new ProxyNodeBinding({ + get(target, prop) { + return prop in connectionOverrides + ? connectionOverrides[prop] + : target[prop]; + }, + }); + return [name, proxyNodeBinding]; + } + ) ); }, - async getServices({ options, hyperdriveProxyController }) { + async getServices({ + options, + workerIndex, + workerBindings, + hyperdriveProxyController, + }) { const services = []; - for (const [name, url] of getHyperdrives(options.config)) { + for (const [name, url, remoteProxyConnectionString, id] of getHyperdrives( + options.config, + options.dev + )) { + // Remote bindings: stand up a local TCP bridge that relays `connect()` + // bytes to the edge Hyperdrive binding over a WebSocket, and point the + // `hyperdrive::` external.tcp designator at it. This keeps workerd + // unmodified (the Worker-service designator path SIGSEGVs). + if (remoteProxyConnectionString) { + const bridgePort = + await hyperdriveProxyController.createRemoteTcpBridge({ + name: getHyperdriveServiceName(workerIndex, name), + bindingName: name, + bindingId: id, + remoteProxyConnectionString, + }); + const connectionString = + hyperdriveProxyController.getRemoteConnectionString( + getHyperdriveServiceName(workerIndex, name) + ); + const binding = workerBindings.find((binding) => binding.name === name); + if ( + connectionString && + binding && + "hyperdrive" in binding && + binding.hyperdrive + ) { + const remoteUrl = HyperdriveSchema.parse(connectionString); + binding.hyperdrive.database = decodeURIComponent( + remoteUrl.pathname.replace(/^\//, "") + ); + binding.hyperdrive.user = decodeURIComponent(remoteUrl.username); + binding.hyperdrive.password = decodeURIComponent(remoteUrl.password); + binding.hyperdrive.scheme = remoteUrl.protocol.replace(":", ""); + } + services.push({ + name: getHyperdriveServiceName(workerIndex, name), + external: { + address: `127.0.0.1:${bridgePort}`, + tcp: {}, + }, + }); + continue; + } + const scheme = url.protocol.replace(":", ""); const sslmode = parseSslMode(url, scheme); const targetPort = getPort(url); @@ -147,7 +279,7 @@ export const HYPERDRIVE_PLUGIN: Plugin = { // SSL modes (require, prefer) need the proxy to handle // TLS negotiation with the target database const proxyPort = await hyperdriveProxyController.createProxyServer({ - name, + name: getHyperdriveServiceName(workerIndex, name), targetHost: url.hostname, targetPort, scheme, @@ -158,7 +290,7 @@ export const HYPERDRIVE_PLUGIN: Plugin = { } services.push({ - name: `${HYPERDRIVE_PLUGIN_NAME}:${name}`, + name: getHyperdriveServiceName(workerIndex, name), external: { address, tcp: {}, diff --git a/packages/miniflare/src/plugins/shared/index.ts b/packages/miniflare/src/plugins/shared/index.ts index 9c0658a0c3b..b31f09e7cbc 100644 --- a/packages/miniflare/src/plugins/shared/index.ts +++ b/packages/miniflare/src/plugins/shared/index.ts @@ -95,6 +95,16 @@ export interface ServicesExtensions { extensions: Extension[]; } +/** + * Extra state a plugin may need when building its Node-side bindings. Kept + * separate from `ParsedWorkerOptions` because it is runtime state owned by the + * Miniflare instance rather than configuration. + */ +export interface PluginNodeBindingsContext { + hyperdriveProxyController: HyperdriveProxyController; + workerIndex: number; +} + /** * Every plugin receives the full parsed per-worker `WorkerOptions` and filters * its own bindings out of `options.config.env` / `options.config.exports` / @@ -108,7 +118,8 @@ export interface Plugin { workerIndex: number ): Awaitable; getNodeBindings( - options: ParsedWorkerOptions + options: ParsedWorkerOptions, + context: PluginNodeBindingsContext ): Awaitable>; getServices( options: PluginServicesOptions diff --git a/packages/miniflare/test/config/schema.spec.ts b/packages/miniflare/test/config/schema.spec.ts index 86b0bf8aa00..dba2cc7a617 100644 --- a/packages/miniflare/test/config/schema.spec.ts +++ b/packages/miniflare/test/config/schema.spec.ts @@ -193,7 +193,8 @@ describe("MiniflareWorkerConfigSchema", () => { expect(result.error.issues).toEqual([ expect.objectContaining({ path: ["env", "HYPERDRIVE", "dev", "connectionString"], - message: "Invalid input: expected string, received undefined", + message: + "A Hyperdrive binding requires a connection string unless it sets 'remote': true.", }), ]); } @@ -248,6 +249,28 @@ describe("MiniflareWorkerConfigSchema", () => { } }); + test("allows a remote Hyperdrive binding without dev.connectionString", ({ + expect, + }) => { + expect( + MiniflareWorkerConfigSchema.parse({ + name: "api", + compatibilityDate: "2026-01-01", + env: { + HYPERDRIVE: { + type: "hyperdrive", + id: "hyperdrive", + dev: { remote: true }, + }, + }, + }).env?.HYPERDRIVE + ).toEqual({ + type: "hyperdrive", + id: "hyperdrive", + dev: { remote: true }, + }); + }); + test("strips tombstoned durable object exports", ({ expect }) => { const parsed = MiniflareWorkerConfigSchema.parse({ name: "api", diff --git a/packages/miniflare/test/config/v4-convert.spec.ts b/packages/miniflare/test/config/v4-convert.spec.ts index d121d3da45d..0cde82182a8 100644 --- a/packages/miniflare/test/config/v4-convert.spec.ts +++ b/packages/miniflare/test/config/v4-convert.spec.ts @@ -4,6 +4,29 @@ import { convertV4MiniflareOptions } from "../../src/config/v4-convert"; import type { RemoteProxyConnectionString } from "../../src/plugins/shared"; describe("convertV4MiniflareOptions", () => { + test.for([undefined, "database-id"])( + "preserves remote Hyperdrive IDs, falling back to the binding name when absent: %s", + (id, { expect }) => { + const converted = convertV4MiniflareOptions({ + script: "export default {};", + modules: true, + hyperdrives: { + DB: { + id, + remoteProxyConnectionString: new URL( + "http://127.0.0.1:8787" + ) as RemoteProxyConnectionString, + }, + }, + }); + expect(converted.workers[0].config.env?.DB).toEqual({ + type: "hyperdrive", + id: id ?? "DB", + dev: { remote: true }, + }); + } + ); + test("preserves opaque K2 stream IDs through v4 parsing and conversion", ({ expect, }) => { diff --git a/packages/miniflare/test/config/v4-schema.spec.ts b/packages/miniflare/test/config/v4-schema.spec.ts index 3f14575e0fe..fec4dd502bf 100644 --- a/packages/miniflare/test/config/v4-schema.spec.ts +++ b/packages/miniflare/test/config/v4-schema.spec.ts @@ -2,6 +2,16 @@ import { describe, test } from "vitest"; import { V4MiniflareOptionsSchema } from "../../src/config/v4-schema"; describe("V4MiniflareOptionsSchema", () => { + test("requires a remote proxy for Hyperdrive object entries", ({ + expect, + }) => { + const result = V4MiniflareOptionsSchema.safeParse({ + script: "export default {}", + hyperdrives: { DB: {} }, + }); + + expect(result.success).toBe(false); + }); test("parses a single-worker v4-shaped config", ({ expect }) => { const parsed = V4MiniflareOptionsSchema.parse({ rootPath: "./project", diff --git a/packages/miniflare/test/plugins/analytics-sql/index.spec.ts b/packages/miniflare/test/plugins/analytics-sql/index.spec.ts index 33dda48bfdc..d52d70ddc9e 100644 --- a/packages/miniflare/test/plugins/analytics-sql/index.spec.ts +++ b/packages/miniflare/test/plugins/analytics-sql/index.spec.ts @@ -28,8 +28,10 @@ describe("Analytics SQL plugin", () => { }); test("creates Node proxy bindings", async ({ expect }) => { - const bindings = - await ANALYTICS_SQL_PLUGIN.getNodeBindings(workerOptions()); + const bindings = await ANALYTICS_SQL_PLUGIN.getNodeBindings( + workerOptions(), + {} as Parameters[1] + ); expect(Object.keys(bindings)).toEqual(["ANALYTICS"]); expect(bindings.ANALYTICS).toBeInstanceOf(ProxyNodeBinding); diff --git a/packages/miniflare/test/plugins/hyperdrive/helpers.ts b/packages/miniflare/test/plugins/hyperdrive/helpers.ts new file mode 100644 index 00000000000..06619c389e6 --- /dev/null +++ b/packages/miniflare/test/plugins/hyperdrive/helpers.ts @@ -0,0 +1,26 @@ +import assert from "node:assert"; +import { once } from "node:events"; +import { onTestFinished } from "vitest"; +import { WebSocketServer } from "ws"; +import type { RemoteProxyConnectionString } from "miniflare"; + +export async function useRemoteProxyServer() { + const edge = new WebSocketServer({ port: 0, host: "127.0.0.1" }); + await once(edge, "listening"); + const address = edge.address(); + assert(address !== null && typeof address !== "string"); + onTestFinished(async () => { + for (const websocket of edge.clients) { + websocket.terminate(); + } + await new Promise((resolve, reject) => + edge.close((error) => (error ? reject(error) : resolve())) + ); + }); + return { + edge, + url: new URL( + `http://127.0.0.1:${address.port}` + ) as RemoteProxyConnectionString, + }; +} diff --git a/packages/miniflare/test/plugins/hyperdrive/index.spec.ts b/packages/miniflare/test/plugins/hyperdrive/index.spec.ts index 2686e7d35fb..6def4a086db 100644 --- a/packages/miniflare/test/plugins/hyperdrive/index.spec.ts +++ b/packages/miniflare/test/plugins/hyperdrive/index.spec.ts @@ -248,6 +248,7 @@ describe("proxy server creation", () => { ), }, }, + workerIndex: 0, hyperdriveProxyController: controller, } as unknown as Parameters[0]; } @@ -265,7 +266,7 @@ describe("proxy server creation", () => { expect(controller.createProxyServer).not.toHaveBeenCalled(); expect(services).toEqual([ { - name: "hyperdrive:DB", + name: "hyperdrive:0:DB", external: { address: "db.example.com:5432", tcp: {} }, }, ]); @@ -288,7 +289,7 @@ describe("proxy server creation", () => { expect(controller.createProxyServer).not.toHaveBeenCalled(); expect(services).toEqual([ { - name: "hyperdrive:DB", + name: "hyperdrive:0:DB", external: { address: "db.example.com:5432", tcp: {} }, }, ]); @@ -309,7 +310,7 @@ describe("proxy server creation", () => { expect(controller.createProxyServer).not.toHaveBeenCalled(); expect(services).toEqual([ { - name: "hyperdrive:DB", + name: "hyperdrive:0:DB", external: { address: "[::1]:5432", tcp: {} }, }, ]); @@ -330,7 +331,7 @@ describe("proxy server creation", () => { expect(controller.createProxyServer).not.toHaveBeenCalled(); expect(services).toEqual([ { - name: "hyperdrive:DB", + name: "hyperdrive:0:DB", external: { address: "db.example.com:3306", tcp: {} }, }, ]); @@ -350,7 +351,7 @@ describe("proxy server creation", () => { ); expect(controller.createProxyServer).toHaveBeenCalledOnce(); expect(controller.createProxyServer).toHaveBeenCalledWith({ - name: "DB", + name: "hyperdrive:0:DB", targetHost: "db.example.com", targetPort: "5432", scheme: "postgres", @@ -358,7 +359,7 @@ describe("proxy server creation", () => { }); expect(services).toEqual([ { - name: "hyperdrive:DB", + name: "hyperdrive:0:DB", external: { address: "127.0.0.1:12345", tcp: {} }, }, ]); @@ -378,7 +379,7 @@ describe("proxy server creation", () => { ); expect(controller.createProxyServer).toHaveBeenCalledOnce(); expect(controller.createProxyServer).toHaveBeenCalledWith({ - name: "DB", + name: "hyperdrive:0:DB", targetHost: "db.example.com", targetPort: "5432", scheme: "postgres", @@ -386,7 +387,7 @@ describe("proxy server creation", () => { }); expect(services).toEqual([ { - name: "hyperdrive:DB", + name: "hyperdrive:0:DB", external: { address: "127.0.0.1:12345", tcp: {} }, }, ]); @@ -406,7 +407,7 @@ describe("proxy server creation", () => { ); expect(controller.createProxyServer).toHaveBeenCalledOnce(); expect(controller.createProxyServer).toHaveBeenCalledWith({ - name: "DB", + name: "hyperdrive:0:DB", targetHost: "db.example.com", targetPort: "3306", scheme: "mysql", @@ -414,7 +415,7 @@ describe("proxy server creation", () => { }); expect(services).toEqual([ { - name: "hyperdrive:DB", + name: "hyperdrive:0:DB", external: { address: "127.0.0.1:12345", tcp: {} }, }, ]); diff --git a/packages/miniflare/test/plugins/hyperdrive/lifecycle.spec.ts b/packages/miniflare/test/plugins/hyperdrive/lifecycle.spec.ts index 594db25d607..649c71d335f 100644 --- a/packages/miniflare/test/plugins/hyperdrive/lifecycle.spec.ts +++ b/packages/miniflare/test/plugins/hyperdrive/lifecycle.spec.ts @@ -103,6 +103,7 @@ test("duplicate Hyperdrive binding names do not retain unused listeners", async }) => { const first = "postgresql://user:password@127.0.0.1:5432/db?sslmode=require"; const second = "postgresql://user:password@127.0.0.1:5433/db?sslmode=require"; + const before = listeningServers(); const mf = new Miniflare({ workers: [worker("first", first), worker("second")], }); @@ -112,8 +113,17 @@ test("duplicate Hyperdrive binding names do not retain unused listeners", async await mf.setOptions({ workers: [worker("first", first), worker("second", second)], }); + await vi.waitFor(() => expect(listeningServers()).toBe(active + 1)); + await mf.setOptions({ + workers: [worker("first"), worker("second", second)], + }); await vi.waitFor(() => expect(listeningServers()).toBe(active)); + await mf.setOptions({ + workers: [worker("first"), worker("second")], + }); + await vi.waitFor(() => expect(listeningServers()).toBe(active - 1)); } finally { await mf.dispose(); } + await vi.waitFor(() => expect(listeningServers()).toBe(before)); }); diff --git a/packages/miniflare/test/plugins/hyperdrive/remote-affinity.spec.ts b/packages/miniflare/test/plugins/hyperdrive/remote-affinity.spec.ts new file mode 100644 index 00000000000..2b7fe06d49d --- /dev/null +++ b/packages/miniflare/test/plugins/hyperdrive/remote-affinity.spec.ts @@ -0,0 +1,163 @@ +import { once } from "node:events"; +import net from "node:net"; +import { newWebSocketRpcSession } from "capnweb"; +import { LogLevel, Miniflare } from "miniflare"; +import { test, vi } from "vitest"; +import { singleModuleManifest, TestLog } from "../../test-shared"; +import { useRemoteProxyServer } from "./helpers"; +import type { Hyperdrive } from "@cloudflare/workers-types"; +import type { MiniflareOptions } from "miniflare"; + +const script = ` +import { connect } from "cloudflare:sockets"; +export default { + async fetch(request, env) { + const socket = connect({ hostname: env.DB.host, port: env.DB.port }); + const writer = socket.writable.getWriter(); + await writer.write(new TextEncoder().encode( + env.DB.user + ":" + env.DB.password + ":" + env.DB.database + "\\n" + )); + writer.releaseLock(); + return new Response(socket.readable); + } +}; +`; + +test("Worker and Node clients use their session credentials across reloads and reconnects despite another binding failing", async ({ + expect, +}) => { + const { edge, url } = await useRemoteProxyServer(); + let sessions = 0; + let connections = 0; + let failures = 0; + edge.on("connection", (websocket, request) => { + const broken = + new URL(request.url ?? "/", "http://localhost").searchParams.get( + "MF-Binding" + ) === "BROKEN"; + const password = broken ? "private-password" : `password-${++sessions}`; + newWebSocketRpcSession(websocket as unknown as WebSocket, { + getConnectionString() { + if (broken) { + failures++; + throw new Error("private-password"); + } + return `mysql://remote-user:${password}@hyperdrive.local:3306/remote-db`; + }, + connect() { + connections++; + let controller: ReadableStreamDefaultController; + let input = ""; + return { + readable: new ReadableStream({ + start(streamController) { + controller = streamController; + }, + }), + writable: new WritableStream({ + write(chunk) { + input += new TextDecoder().decode(chunk); + if (input.includes("\n")) { + if (input !== `remote-user:${password}:remote-db\n`) { + controller.error(new Error("Wrong session credentials")); + throw new Error("Wrong session credentials"); + } + controller.enqueue(new TextEncoder().encode("authenticated")); + controller.close(); + } + }, + }), + }; + }, + }); + }); + const log = new TestLog(); + function getOptions(id: string): MiniflareOptions { + return { + log, + workers: [ + { + config: { + name: "main", + compatibilityDate: "2026-09-01", + manifest: singleModuleManifest(script), + env: { + DB: { + type: "hyperdrive", + id, + dev: { + remote: true, + connectionString: "unused-invalid-local", + }, + }, + BROKEN: { + type: "hyperdrive", + id: "broken", + dev: { remote: true }, + }, + }, + }, + dev: { + remoteProxyConnectionString: url, + }, + }, + ], + }; + } + const mf = new Miniflare(getOptions("db")); + try { + await mf.ready; + for (const mode of ["initial", "reload", "changed", "reconnect"]) { + if (mode === "reconnect") { + for (const websocket of edge.clients) { + websocket.terminate(); + } + await vi.waitFor(() => + expect(log.logsAtLevel(LogLevel.WARN).join("\n")).toContain( + "was lost" + ) + ); + } + if (mode !== "initial") { + await mf.setOptions(getOptions(mode === "reload" ? "db" : "next-db")); + } + const responses = await Promise.all( + Array.from({ length: 3 }, async () => { + const response = await mf.dispatchFetch("http://localhost/"); + return response.text(); + }) + ); + expect(responses).toEqual(Array(3).fill("authenticated")); + const { DB } = await mf.getBindings<{ DB: Hyperdrive }>(); + expect(DB.user).toBe("remote-user"); + expect(DB.password).toBe( + mode === "reconnect" + ? "password-3" + : mode === "changed" + ? "password-2" + : "password-1" + ); + expect(DB.database).toBe("remote-db"); + expect(DB.host).toBe("127.0.0.1"); + const client = net.connect(DB.port, DB.host); + try { + client.write(`${DB.user}:${DB.password}:${DB.database}\n`); + const [chunk] = await once(client, "data"); + expect(chunk.toString()).toBe("authenticated"); + } finally { + client.destroy(); + } + } + expect(sessions).toBe(3); + expect(connections).toBe(16); + expect(failures).toBe(3); + expect(log.logsAtLevel(LogLevel.WARN).join("\n")).toContain( + 'binding "BROKEN"' + ); + expect(log.logsAtLevel(LogLevel.WARN).join("\n")).not.toContain( + "private-password" + ); + } finally { + await mf.dispose(); + } +}); diff --git a/packages/miniflare/test/plugins/hyperdrive/remote-lifecycle.spec.ts b/packages/miniflare/test/plugins/hyperdrive/remote-lifecycle.spec.ts new file mode 100644 index 00000000000..f97a7410df5 --- /dev/null +++ b/packages/miniflare/test/plugins/hyperdrive/remote-lifecycle.spec.ts @@ -0,0 +1,88 @@ +import assert from "node:assert"; +import { once } from "node:events"; +import net from "node:net"; +import { newWebSocketRpcSession } from "capnweb"; +import { test, vi } from "vitest"; +import { HyperdriveProxyController } from "../../../src/plugins/hyperdrive/hyperdrive-proxy"; +import { useRemoteProxyServer } from "./helpers"; + +test.for(["changed", "aborted", "reconnected"])( + "disposes live connections from a %s remote listener", + async (state, { expect }) => { + const { edge, url } = await useRemoteProxyServer(); + edge.on("connection", (socket) => { + newWebSocketRpcSession(socket as unknown as WebSocket, { + getConnectionString() { + return "mysql://user:password@hyperdrive.local:3306/database"; + }, + connect() { + return { + readable: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("ready")); + }, + }), + writable: new WritableStream(), + }; + }, + }); + }); + const controller = new HyperdriveProxyController(); + let client: net.Socket | undefined; + try { + const config = { + name: "hyperdrive:0:DB", + bindingName: "DB", + bindingId: "db", + remoteProxyConnectionString: url, + }; + const port = await controller.createRemoteTcpBridge(config); + client = net.connect(port, "127.0.0.1"); + await once(client, "data"); + expect(edge.clients.size).toBe(1); + controller.beginUpdate(); + const activeAddresses = new Set(); + if (state !== "reconnected") { + const nextPort = await controller.createRemoteTcpBridge({ + ...config, + remoteProxyConnectionString: new URL(`${url.href}?next`), + }); + activeAddresses.add(`127.0.0.1:${nextPort}`); + } + if (state === "aborted") { + controller.abortUpdate(); + expect(controller.getRemoteBridgePort(config.name)).toBe(port); + } else { + controller.commitUpdate(activeAddresses); + } + expect(client.destroyed).toBe(false); + await vi.waitFor(() => + expect(edge.clients.size).toBe(state === "changed" ? 2 : 1) + ); + if (state === "reconnected") { + const retiredSession = Array.from(edge.clients)[0]; + assert(retiredSession !== undefined); + controller.beginUpdate(); + const nextPort = await controller.createRemoteTcpBridge(config); + controller.commitUpdate(new Set([`127.0.0.1:${nextPort}`])); + expect(client.destroyed).toBe(false); + expect(edge.clients.size).toBe(2); + retiredSession.terminate(); + await vi.waitFor(() => expect(client?.destroyed).toBe(true)); + await vi.waitFor(() => expect(edge.clients.size).toBe(1)); + expect(controller.getRemoteBridgePort(config.name)).toBe(nextPort); + client = net.connect(nextPort, "127.0.0.1"); + await once(client, "data"); + } + controller.dispose(); + await vi.waitFor(() => expect(client?.destroyed).toBe(true)); + await vi.waitFor(() => expect(edge.clients.size).toBe(0), { + timeout: 3_000, + }); + expect(controller.getRemoteBridgePort(config.name)).toBeUndefined(); + } finally { + client?.destroy(); + controller.dispose(); + } + } +); diff --git a/packages/miniflare/test/plugins/hyperdrive/remote-startup.spec.ts b/packages/miniflare/test/plugins/hyperdrive/remote-startup.spec.ts new file mode 100644 index 00000000000..7add4a6c5c3 --- /dev/null +++ b/packages/miniflare/test/plugins/hyperdrive/remote-startup.spec.ts @@ -0,0 +1,233 @@ +import { once } from "node:events"; +import net from "node:net"; +import { setTimeout as delay } from "node:timers/promises"; +import { newWebSocketRpcSession } from "capnweb"; +import { Miniflare } from "miniflare"; +import { test, vi } from "vitest"; +import { HyperdriveProxyController } from "../../../src/plugins/hyperdrive/hyperdrive-proxy"; +import { singleModuleManifest } from "../../test-shared"; +import { useRemoteProxyServer } from "./helpers"; + +test("disposal cancels startup during retry backoff", async ({ expect }) => { + const { edge, url } = await useRemoteProxyServer(); + const controller = new HyperdriveProxyController(); + const closed = Promise.withResolvers(); + let attempts = 0; + edge.on("connection", (websocket) => { + attempts++; + websocket.once("close", () => closed.resolve()); + newWebSocketRpcSession(websocket as unknown as WebSocket, { + getConnectionString() { + throw new Error("retry startup"); + }, + }); + }); + try { + const starting = controller.createRemoteTcpBridge({ + name: "hyperdrive:0:DB", + bindingName: "DB", + bindingId: "db", + remoteProxyConnectionString: url, + }); + const rejected = expect(starting).rejects.toMatchObject({ + name: "AbortError", + }); + await closed.promise; + await delay(25); + controller.dispose(); + await rejected; + await vi.waitFor(() => expect(edge.clients.size).toBe(0)); + await delay(600); + expect(attempts).toBe(1); + expect(controller.getRemoteBridgePort("hyperdrive:0:DB")).toBeUndefined(); + } finally { + controller.dispose(); + } +}); + +test("Miniflare disposal cancels pending Hyperdrive initialization", async ({ + expect, +}) => { + const { edge, url } = await useRemoteProxyServer(); + const requested = Promise.withResolvers(); + edge.on("connection", (websocket) => { + newWebSocketRpcSession(websocket as unknown as WebSocket, { + getConnectionString() { + requested.resolve(); + return new Promise(() => {}); + }, + }); + }); + const mf = new Miniflare({ + workers: [ + { + config: { + name: "test", + compatibilityDate: "2026-09-01", + manifest: singleModuleManifest("export default {}"), + env: { DB: { type: "hyperdrive", id: "db", dev: { remote: true } } }, + }, + dev: { + remoteProxyConnectionString: url, + }, + }, + ], + }); + const ready = expect(mf.ready).rejects.toMatchObject({ + code: "ERR_DISPOSED", + }); + try { + await requested.promise; + await mf.dispose(); + await ready; + await vi.waitFor(() => expect(edge.clients.size).toBe(0)); + } finally { + await mf.dispose(); + } +}); + +test("an early client reset releases its pending socket without breaking the session", async ({ + expect, +}) => { + const { edge, url } = await useRemoteProxyServer(); + const requested = Promise.withResolvers(); + const release = Promise.withResolvers(); + const aborted = vi.fn(); + let connections = 0; + edge.on("connection", (websocket) => { + newWebSocketRpcSession(websocket as unknown as WebSocket, { + getConnectionString() { + return "mysql://user:password@hyperdrive.local:3306/database"; + }, + async connect() { + if (++connections === 1) { + requested.resolve(); + await release.promise; + let readableController: + | ReadableStreamDefaultController + | undefined; + return { + readable: new ReadableStream({ + start(streamController) { + readableController = streamController; + }, + }), + writable: new WritableStream({ + abort() { + aborted(); + readableController?.close(); + }, + }), + }; + } + return { + readable: new ReadableStream({ + start(streamController) { + streamController.enqueue(new TextEncoder().encode("ready")); + streamController.close(); + }, + }), + writable: new WritableStream(), + }; + }, + }); + }); + const controller = new HyperdriveProxyController(); + const createServer = vi.spyOn(net, "createServer"); + const accepted = Promise.withResolvers(); + let client: net.Socket | undefined; + let nextClient: net.Socket | undefined; + try { + const port = await controller.createRemoteTcpBridge({ + name: "hyperdrive:0:DB", + bindingName: "DB", + bindingId: "db", + remoteProxyConnectionString: url, + }); + const server = createServer.mock.results[0].value as net.Server; + server.once("connection", accepted.resolve); + client = net.connect(port, "127.0.0.1"); + client.on("error", vi.fn()); + await once(client, "connect"); + client.write("hello"); + await requested.promise; + const acceptedSocket = await accepted.promise; + const closed = new Promise((resolve) => + acceptedSocket.once("close", () => resolve()) + ); + client.resetAndDestroy(); + await closed; + release.resolve(); + await vi.waitFor(() => expect(aborted).toHaveBeenCalledOnce()); + nextClient = net.connect(port, "127.0.0.1"); + const [chunk] = await once(nextClient, "data"); + expect(chunk.toString()).toBe("ready"); + expect(edge.clients.size).toBe(1); + } finally { + release.resolve(); + client?.destroy(); + nextClient?.destroy(); + controller.dispose(); + createServer.mockRestore(); + } +}); + +test("retries a failed startup connection and forwards CF_TRACE_ID", async ({ + expect, +}) => { + const traceId = "hyperdrive-startup-trace"; + vi.stubEnv("CF_TRACE_ID", traceId); + const { edge, url } = await useRemoteProxyServer(); + const controller = new HyperdriveProxyController(); + const receivedTraceIds: (string | string[] | undefined)[] = []; + const connectionString = + "mysql://user:password@hyperdrive.local:3306/database"; + let client: net.Socket | undefined; + try { + edge.on("connection", (socket, request) => { + receivedTraceIds.push(request.headers["cf-trace-id"]); + if (receivedTraceIds.length === 1) { + socket.terminate(); + return; + } + newWebSocketRpcSession(socket as unknown as WebSocket, { + getConnectionString() { + return connectionString; + }, + connect() { + return { + readable: new ReadableStream({ + start(readableController) { + readableController.enqueue(new TextEncoder().encode("ready")); + readableController.close(); + }, + }), + writable: new WritableStream(), + }; + }, + }); + }); + const bindingName = "hyperdrive:0:DB"; + const port = await controller.createRemoteTcpBridge({ + name: bindingName, + bindingName: "DB", + bindingId: "db", + remoteProxyConnectionString: url, + }); + expect(controller.getRemoteConnectionString(bindingName)).toBe( + connectionString + ); + expect(controller.getRemoteBridgePort(bindingName)).toBe(port); + client = net.connect(port, "127.0.0.1"); + const [chunk] = await once(client, "data"); + expect(chunk.toString()).toBe("ready"); + expect(receivedTraceIds).toEqual([traceId, traceId]); + expect(edge.clients.size).toBe(1); + controller.dispose(); + await vi.waitFor(() => expect(edge.clients.size).toBe(0)); + } finally { + client?.destroy(); + controller.dispose(); + vi.unstubAllEnvs(); + } +}); diff --git a/packages/miniflare/test/plugins/shared/remote-bindings-connect.spec.ts b/packages/miniflare/test/plugins/shared/remote-bindings-connect.spec.ts index a5537390d0f..bfa551d61b1 100644 --- a/packages/miniflare/test/plugins/shared/remote-bindings-connect.spec.ts +++ b/packages/miniflare/test/plugins/shared/remote-bindings-connect.spec.ts @@ -1,3 +1,6 @@ +import assert from "node:assert"; +import { once } from "node:events"; +import net from "node:net"; import path from "node:path"; // The relay helper under unit test. It is the same implementation bundled into // miniflare's dist and mirrored (byte-for-byte, comments aside) into the edge @@ -14,6 +17,7 @@ import { VPC_SERVICES_PLUGIN, } from "miniflare"; import { beforeAll, describe, test } from "vitest"; +import { HyperdriveProxyController } from "../../../src/plugins/hyperdrive/hyperdrive-proxy"; import { singleModuleManifest, useDispose } from "../../test-shared"; import type { RemoteProxyConnectionString } from "miniflare"; @@ -770,3 +774,134 @@ describe("VPC_SERVICES plugin: raw TCP opt-in", () => { expect(serviceList[0].worker?.compatibilityFlags).toEqual(["experimental"]); }); }); + +// Reads from a raw TCP socket until the accumulated output contains `sentinel`, +// or rejects after `timeoutMs`. Used to observe bytes coming back through the +// Hyperdrive remote bridge. +function readFromSocket( + socket: net.Socket, + sentinel: string, + timeoutMs = 10_000 +): Promise { + return new Promise((resolve, reject) => { + let out = ""; + const cleanup = () => { + clearTimeout(timer); + socket.off("data", onData); + socket.off("error", onError); + }; + const timer = setTimeout(() => { + cleanup(); + reject(new Error(`timed out waiting for ${JSON.stringify(sentinel)}`)); + }, timeoutMs); + const onData = (chunk: Buffer) => { + out += chunk.toString("utf8"); + if (out.includes(sentinel)) { + cleanup(); + resolve(out); + } + }; + const onError = (err: Error) => { + cleanup(); + reject(err); + }; + socket.on("data", onData); + socket.on("error", onError); + }); +} + +describe("Hyperdrive remote binding: local TCP bridge", () => { + test.for(["client disconnect", "database reset"])( + "relays bytes between a local TCP client and the edge binding (%s)", + async (closeMode, { expect }) => { + const sockets = new Set(); + const socketErrors: Error[] = []; + const database = net.createServer((socket) => { + sockets.add(socket); + socket.on("error", (error) => socketErrors.push(error)); + socket.once("close", () => sockets.delete(socket)); + socket.pipe(socket); + }); + database.listen(0, "127.0.0.1"); + await once(database, "listening"); + const address = database.address(); + assert(address !== null && typeof address !== "string"); + const edge = new Miniflare({ + workers: [ + { + config: { + name: "proxy-server", + compatibilityDate: COMPAT_DATE, + compatibilityFlags: ["experimental"], + manifest: { + mainModule: "ProxyServerWorker.js", + modules: { + "ProxyServerWorker.js": { + type: "esm", + contents: proxyServerBundle, + }, + }, + }, + env: { + HYPERDRIVE: { + type: "hyperdrive", + id: "db", + dev: { + connectionString: `mysql://user:password@127.0.0.1:${address.port}/database`, + }, + }, + }, + }, + }, + ], + }); + useDispose(edge); + const edgeUrl = await edge.ready; + + const controller = new HyperdriveProxyController(); + try { + const bridgePort = await controller.createRemoteTcpBridge({ + bindingId: "db", + name: "hyperdrive:0:HYPERDRIVE", + bindingName: "HYPERDRIVE", + remoteProxyConnectionString: edgeUrl, + }); + + const socket = net.connect(bridgePort, "127.0.0.1"); + socket.on("error", (error) => socketErrors.push(error)); + try { + await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); + }); + socket.write("PING\n"); + const received = await readFromSocket(socket, "PING\n"); + expect(received).toContain("PING\n"); + if (closeMode === "database reset") { + const closed = new Promise((resolve) => + socket.once("close", () => resolve()) + ); + for (const databaseSocket of sockets) { + databaseSocket.resetAndDestroy(); + } + await closed; + } + } finally { + socket.destroy(); + } + } finally { + controller.dispose(); + await edge.dispose(); + for (const socket of sockets) { + socket.destroy(); + } + await new Promise((resolve, reject) => + database.close((error) => (error ? reject(error) : resolve())) + ); + for (const error of socketErrors) { + expect(error).toMatchObject({ code: "ECONNRESET" }); + } + } + } + ); +}); diff --git a/packages/remote-bindings/templates/remoteBindings/ProxyServerWorker.ts b/packages/remote-bindings/templates/remoteBindings/ProxyServerWorker.ts index 055bfeda91d..cd37f48e130 100644 --- a/packages/remote-bindings/templates/remoteBindings/ProxyServerWorker.ts +++ b/packages/remote-bindings/templates/remoteBindings/ProxyServerWorker.ts @@ -48,6 +48,19 @@ function getExposedJSRPCBinding(request: Request, env: Env) { throw new BindingNotFoundError(bindingName); } + if (targetBinding.constructor.name === "Hyperdrive") { + const hyperdrive = targetBinding as Hyperdrive; + return { + getConnectionString() { + return hyperdrive.connectionString; + }, + connect() { + const socket = hyperdrive.connect(); + return { readable: socket.readable, writable: socket.writable }; + }, + }; + } + if (targetBinding.constructor.name === "SendEmail") { return { async send(e: SendEmailInput) { diff --git a/packages/vite-plugin-cloudflare/src/__tests__/local-env.spec.ts b/packages/vite-plugin-cloudflare/src/__tests__/local-env.spec.ts index 6834b6fbfed..e11d577d27e 100644 --- a/packages/vite-plugin-cloudflare/src/__tests__/local-env.spec.ts +++ b/packages/vite-plugin-cloudflare/src/__tests__/local-env.spec.ts @@ -43,5 +43,26 @@ describe("local bindings", () => { expect(() => resolveLocalBindings(workerConfig.env, {})).toThrow( 'Hyperdrive binding "HYPERDRIVE" must define dev.connectionString for local development.' ); + expect(() => + resolveLocalBindings( + { + HYPERDRIVE: { ...workerConfig.env.HYPERDRIVE, dev: { remote: true } }, + }, + {}, + undefined, + false + ) + ).toThrow( + 'Hyperdrive binding "HYPERDRIVE" must define dev.connectionString' + ); + }); + + test("does not require local credentials for remote Hyperdrive", ({ + expect, + }) => { + const binding = { ...workerConfig.env.HYPERDRIVE, dev: { remote: true } }; + expect(resolveLocalBindings({ HYPERDRIVE: binding }, {}).bindings).toEqual({ + HYPERDRIVE: binding, + }); }); }); diff --git a/packages/vite-plugin-cloudflare/src/__tests__/remote-bindings.spec.ts b/packages/vite-plugin-cloudflare/src/__tests__/remote-bindings.spec.ts index a67185078d5..a22c316e2c6 100644 --- a/packages/vite-plugin-cloudflare/src/__tests__/remote-bindings.spec.ts +++ b/packages/vite-plugin-cloudflare/src/__tests__/remote-bindings.spec.ts @@ -2,7 +2,7 @@ import { describe, test } from "vitest"; import { getRemoteBindings } from "../remote-bindings"; describe("getRemoteBindings", () => { - test("maps bindings that always use remote resources", ({ expect }) => { + test("maps supported remote bindings", ({ expect }) => { expect( getRemoteBindings({ env: { @@ -14,6 +14,11 @@ describe("getRemoteBindings", () => { type: "k2", stream: "stream-id", }, + HYPERDRIVE: { + type: "hyperdrive", + id: "hyperdrive-id", + dev: { remote: true }, + }, }, }) ).toEqual({ @@ -26,6 +31,11 @@ describe("getRemoteBindings", () => { stream: "stream-id", remote: undefined, }, + HYPERDRIVE: { + type: "hyperdrive", + id: "hyperdrive-id", + remote: true, + }, }); }); diff --git a/packages/vite-plugin-cloudflare/src/local-env.ts b/packages/vite-plugin-cloudflare/src/local-env.ts index 4cd52c7de18..1fd218af770 100644 --- a/packages/vite-plugin-cloudflare/src/local-env.ts +++ b/packages/vite-plugin-cloudflare/src/local-env.ts @@ -17,12 +17,14 @@ export interface ResolvedLocalBindings { * @param configuredBindings The Worker's configured bindings. * @param localEnv Values resolved from `.env` files and the process environment. * @param devVars Values from the selected `.dev.vars` file, which are used exclusively when provided. + * @param remoteBindings Whether remote binding sessions are enabled. * @returns The complete bindings and any secrets missing a local value. */ export function resolveLocalBindings( configuredBindings: ParsedInputWorkerConfig["env"], localEnv: Record, - devVars?: Record + devVars?: Record, + remoteBindings = true ): ResolvedLocalBindings { const bindings: MiniflareEnv = {}; const missingSecrets: string[] = []; @@ -30,6 +32,10 @@ export function resolveLocalBindings( for (const [name, binding] of Object.entries(configuredBindings ?? {})) { if (binding.type === "hyperdrive") { + if (binding.dev?.remote && remoteBindings) { + bindings[name] = { ...binding, dev: binding.dev }; + continue; + } const connectionString = localEnv[`CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_${name}`] || binding.dev?.connectionString; diff --git a/packages/vite-plugin-cloudflare/src/miniflare-options.ts b/packages/vite-plugin-cloudflare/src/miniflare-options.ts index 237ac662ec3..b2ec8bb606f 100644 --- a/packages/vite-plugin-cloudflare/src/miniflare-options.ts +++ b/packages/vite-plugin-cloudflare/src/miniflare-options.ts @@ -259,7 +259,8 @@ export async function getDevMiniflareOptions( const resolvedLocalBindingsByWorkerName = resolveLocalBindingsByWorkerName( ctx.allWorkerConfigs, resolvedPluginConfig.localEnv, - resolvedPluginConfig.devVars + resolvedPluginConfig.devVars, + resolvedPluginConfig.remoteBindings ); warnForMissingSecrets( resolvedLocalBindingsByWorkerName, @@ -808,7 +809,8 @@ export async function getPreviewMiniflareOptions( const resolvedLocalBindingsByWorkerName = resolveLocalBindingsByWorkerName( ctx.allWorkerConfigs, resolvedPluginConfig.localEnv, - resolvedPluginConfig.devVars + resolvedPluginConfig.devVars, + resolvedPluginConfig.remoteBindings ); warnForMissingSecrets( resolvedLocalBindingsByWorkerName, @@ -987,7 +989,8 @@ export function createMiniflareExports(options: { function resolveLocalBindingsByWorkerName( workerConfigs: Array, localEnv: LoadedEnv, - devVars: Record | undefined + devVars: Record | undefined, + remoteBindings: boolean ): Map { const resolvedLocalBindingsByWorkerName = new Map< string, @@ -996,7 +999,7 @@ function resolveLocalBindingsByWorkerName( for (const config of workerConfigs) { resolvedLocalBindingsByWorkerName.set( config.name, - resolveLocalBindings(config.env, localEnv.values, devVars) + resolveLocalBindings(config.env, localEnv.values, devVars, remoteBindings) ); } return resolvedLocalBindingsByWorkerName; diff --git a/packages/vite-plugin-cloudflare/src/remote-bindings.ts b/packages/vite-plugin-cloudflare/src/remote-bindings.ts index 6a656abaeb8..e6617ed8d82 100644 --- a/packages/vite-plugin-cloudflare/src/remote-bindings.ts +++ b/packages/vite-plugin-cloudflare/src/remote-bindings.ts @@ -85,6 +85,13 @@ export function getRemoteBindings( remote: binding.dev?.remote, }; break; + case "hyperdrive": + bindings[name] = { + type: "hyperdrive", + id: binding.id, + remote: binding.dev?.remote, + }; + break; case "images": bindings[name] = { type: "images", remote: binding.dev?.remote }; break; diff --git a/packages/workers-utils/src/config/binding-local-support.ts b/packages/workers-utils/src/config/binding-local-support.ts index 724bb7f2619..29a791dd44a 100644 --- a/packages/workers-utils/src/config/binding-local-support.ts +++ b/packages/workers-utils/src/config/binding-local-support.ts @@ -38,7 +38,6 @@ const BINDING_LOCAL_SUPPORT: Record< assets: "local-only", unsafe_hello_world: "local-only", durable_object_namespace: "local-only", - hyperdrive: "local-only", fetcher: "local-only", analytics_engine: "local-only", secrets_store_secret: "local-only", @@ -48,6 +47,7 @@ const BINDING_LOCAL_SUPPORT: Record< kv_namespace: "local-and-remote", r2_bucket: "local-and-remote", d1: "local-and-remote", + hyperdrive: "local-and-remote", workflow: "local-only", browser: "local-and-remote", images: "local-and-remote", diff --git a/packages/workers-utils/src/config/environment.ts b/packages/workers-utils/src/config/environment.ts index fcb65f7569f..2d0c7914597 100644 --- a/packages/workers-utils/src/config/environment.ts +++ b/packages/workers-utils/src/config/environment.ts @@ -1435,6 +1435,11 @@ export interface EnvironmentNonInheritable { id: string; /** The local database connection string for `wrangler dev` */ localConnectionString?: string; + /** + * Connect to the deployed Hyperdrive configuration during local development + * instead of a local database. + */ + remote?: boolean; }[]; /** diff --git a/packages/workers-utils/src/config/validation.ts b/packages/workers-utils/src/config/validation.ts index e77bfe9b2b5..51618553e3c 100644 --- a/packages/workers-utils/src/config/validation.ts +++ b/packages/workers-utils/src/config/validation.ts @@ -5322,10 +5322,18 @@ const validateHyperdriveBinding: ValidatorFn = (diagnostics, field, value) => { isValid = false; } + // `remote: true` opts the binding into connecting to the deployed Hyperdrive + // configuration at the edge during local dev (via the remote-bindings proxy). + // When set, a `localConnectionString` is optional. + if (!isRemoteValid(value, field, diagnostics)) { + isValid = false; + } + validateAdditionalProperties(diagnostics, field, Object.keys(value), [ "binding", "id", "localConnectionString", + "remote", ]); return isValid; diff --git a/packages/workers-utils/src/print-bindings.ts b/packages/workers-utils/src/print-bindings.ts index f07de4ce3bb..cef114a5cd0 100644 --- a/packages/workers-utils/src/print-bindings.ts +++ b/packages/workers-utils/src/print-bindings.ts @@ -364,12 +364,14 @@ export function printBindings( if (hyperdrive.length > 0) { output.push( - ...hyperdrive.map(({ binding, id }) => { + ...hyperdrive.map(({ binding, id, remote }) => { return { name: binding, type: getBindingTypeFriendlyName("hyperdrive"), value: id, - mode: getMode({ isSimulatedLocally: true }), + mode: getMode({ + isSimulatedLocally: context.remoteBindingsDisabled || !remote, + }), }; }) ); diff --git a/packages/workers-utils/src/worker.ts b/packages/workers-utils/src/worker.ts index 357713ca89e..c68fcf7a173 100644 --- a/packages/workers-utils/src/worker.ts +++ b/packages/workers-utils/src/worker.ts @@ -320,6 +320,11 @@ export interface CfHyperdrive { binding: string; id: string; localConnectionString?: string; + /** + * Connect to the deployed Hyperdrive configuration during local development + * instead of a local database. + */ + remote?: boolean; } export interface CfDevPluginCfg { diff --git a/packages/wrangler/src/__tests__/dev.test.ts b/packages/wrangler/src/__tests__/dev.test.ts index cf64c2282e0..0cfccca343d 100644 --- a/packages/wrangler/src/__tests__/dev.test.ts +++ b/packages/wrangler/src/__tests__/dev.test.ts @@ -2693,6 +2693,41 @@ describe("wrangler dev", { concurrent: false }, () => { }); }); + describe("hyperdrive bindings", () => { + it("should report a `remote: true` Hyperdrive binding as remote", async ({ + expect, + }) => { + writeWranglerConfig({ + hyperdrive: [ + { + binding: "HYPERDRIVE_REMOTE", + id: "remote-id", + remote: true, + }, + { + binding: "HYPERDRIVE_LOCAL", + id: "local-id", + localConnectionString: + "postgres://user:pass@localhost:5432/database", + }, + ], + }); + fs.writeFileSync("index.js", `export default {};`); + await runWranglerUntilConfig("dev index.js"); + expect(std.out).toMatchInlineSnapshot(` + " + ⛅️ wrangler x.x.x + ────────────────── + Your Worker has access to the following bindings: + Binding Resource Mode + env.HYPERDRIVE_REMOTE (remote-id) Hyperdrive Config remote + env.HYPERDRIVE_LOCAL (local-id) Hyperdrive Config local + + " + `); + }); + }); + describe("print bindings", () => { it("should print bindings", async ({ expect }) => { writeWranglerConfig({ diff --git a/packages/wrangler/src/__tests__/dev/miniflare-hyperdrive.test.ts b/packages/wrangler/src/__tests__/dev/miniflare-hyperdrive.test.ts new file mode 100644 index 00000000000..11cc9fed008 --- /dev/null +++ b/packages/wrangler/src/__tests__/dev/miniflare-hyperdrive.test.ts @@ -0,0 +1,100 @@ +import { describe, it } from "vitest"; +import { buildMiniflareBindingOptions } from "../../dev/miniflare"; +import { mockConsoleMethods } from "../helpers/mock-console"; +import type { Binding } from "@cloudflare/workers-utils"; +import type { RemoteProxyConnectionString } from "miniflare"; + +const remoteProxyConnectionString = new URL( + "http://localhost:52222/" +) as RemoteProxyConnectionString; + +function buildHyperdriveOptions( + binding: Extract, + connectionString?: RemoteProxyConnectionString +) { + const { bindingOptions } = buildMiniflareBindingOptions( + { + name: "test-worker", + complianceRegion: undefined, + bindings: { HYPERDRIVE: binding }, + queueConsumers: undefined, + migrations: undefined, + exports: undefined, + tails: [], + streamingTails: [], + enableContainers: false, + }, + connectionString + ); + return bindingOptions.hyperdrives; +} + +describe("hyperdrive bindings in local dev", () => { + const std = mockConsoleMethods(); + + it("passes the local connection string through for a local binding", ({ + expect, + }) => { + expect( + buildHyperdriveOptions({ + type: "hyperdrive", + id: "hyperdrive-id", + localConnectionString: "postgres://user:pass@localhost:5432/db", + }) + ).toEqual({ HYPERDRIVE: "postgres://user:pass@localhost:5432/db" }); + expect(std.warn).toBe(""); + }); + + it.for([undefined, "unused-invalid-local"])( + "ignores local credentials for a remote binding (%s)", + (localConnectionString, { expect }) => { + expect( + buildHyperdriveOptions( + { + type: "hyperdrive", + id: "hyperdrive-id", + remote: true, + localConnectionString, + }, + remoteProxyConnectionString + ) + ).toEqual({ + HYPERDRIVE: { + id: "hyperdrive-id", + remoteProxyConnectionString, + }, + }); + expect(std.warn).toBe(""); + } + ); + + it("explains how to fix a remote binding that has neither a session nor a local database", ({ + expect, + }) => { + expect(() => + buildHyperdriveOptions({ + type: "hyperdrive", + id: "hyperdrive-id", + remote: true, + }) + ).toThrowErrorMatchingInlineSnapshot( + `[Error: The Hyperdrive binding "HYPERDRIVE" is configured with "remote": true, but no remote connection could be established, and it has no local database to fall back to. Please make sure you are authenticated (run \`wrangler login\`) and connected to the internet so that Wrangler can reach your Hyperdrive configuration, or set the value of the 'CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_HYPERDRIVE' variable or "HYPERDRIVE"'s "localConnectionString" to a local Postgres connection string.]` + ); + }); + + it("warns when a remote binding without a session falls back to its local database", ({ + expect, + }) => { + expect( + buildHyperdriveOptions({ + type: "hyperdrive", + id: "hyperdrive-id", + remote: true, + localConnectionString: "postgres://user:pass@localhost:5432/db", + }) + ).toEqual({ HYPERDRIVE: "postgres://user:pass@localhost:5432/db" }); + expect(std.warn).toContain( + `The Hyperdrive binding "HYPERDRIVE" is configured with "remote": true, but no remote connection could be established. Falling back to its "localConnectionString".` + ); + }); +}); diff --git a/packages/wrangler/src/dev.ts b/packages/wrangler/src/dev.ts index 0875a2999e1..4067bd271b1 100644 --- a/packages/wrangler/src/dev.ts +++ b/packages/wrangler/src/dev.ts @@ -501,9 +501,12 @@ function applyHyperdriveEnvVars(config: Config, local: boolean): void { connectionStringFromEnv = process.env[varName]; } - // only require a local connection string in the wrangler file or the env if not using dev --remote + // only require a local connection string in the wrangler file or the env + // if not using dev --remote, and not opted into a remote Hyperdrive + // binding (which tunnels to the edge instead of a local database) if ( local && + !hyperdrive.remote && connectionStringFromEnv === undefined && hyperdrive.localConnectionString === undefined ) { diff --git a/packages/wrangler/src/dev/miniflare/index.ts b/packages/wrangler/src/dev/miniflare/index.ts index da6c613fc7d..0c26afb1f02 100644 --- a/packages/wrangler/src/dev/miniflare/index.ts +++ b/packages/wrangler/src/dev/miniflare/index.ts @@ -356,7 +356,49 @@ function pipelineEntry( throw new Error("Pipeline must have either a stream"); } } -function hyperdriveEntry(hyperdrive: CfHyperdrive): [string, string] { +function hyperdriveEntry( + hyperdrive: CfHyperdrive, + remoteProxyConnectionString?: RemoteProxyConnectionString +): + | [string, string] + | [ + string, + { + id: string; + remoteProxyConnectionString: RemoteProxyConnectionString; + }, + ] { + if (hyperdrive.remote && remoteProxyConnectionString) { + return [ + hyperdrive.binding, + { + id: hyperdrive.id, + remoteProxyConnectionString, + }, + ]; + } + + // Opted into a remote binding, but no remote proxy session is available + // (not logged in, offline, or remote bindings turned off). There is nothing + // to tunnel to, so fall back to the local database if one was configured and + // otherwise explain how to get either half working — without this the empty + // connection string below fails miniflare's URL validation with an opaque + // error. + if (hyperdrive.remote) { + if (hyperdrive.localConnectionString === undefined) { + throw new UserError( + `The Hyperdrive binding "${hyperdrive.binding}" is configured with "remote": true, but no remote connection could be established, and it has no local database to fall back to. Please make sure you are authenticated (run \`wrangler login\`) and connected to the internet so that Wrangler can reach your Hyperdrive configuration, or set the value of the 'CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_${hyperdrive.binding}' variable or "${hyperdrive.binding}"'s "localConnectionString" to a local Postgres connection string.`, + { + telemetryMessage: + "hyperdrive remote binding session and local connection string missing", + } + ); + } + logger.once.warn( + `The Hyperdrive binding "${hyperdrive.binding}" is configured with "remote": true, but no remote connection could be established. Falling back to its "localConnectionString".` + ); + } + return [hyperdrive.binding, hyperdrive.localConnectionString ?? ""]; } function workflowEntry({ @@ -905,7 +947,11 @@ export function buildMiniflareBindingOptions( ]; }) ), - hyperdrives: Object.fromEntries(hyperdrives.map(hyperdriveEntry)), + hyperdrives: Object.fromEntries( + hyperdrives.map((hyperdrive) => + hyperdriveEntry(hyperdrive, remoteProxyConnectionString) + ) + ), analyticsEngineDatasets: Object.fromEntries( analyticsEngineDatasets.map((binding) => [ binding.binding, From 70ee65ab01063fb394c0175a2fbb77e3efd812c1 Mon Sep 17 00:00:00 2001 From: Edmund Hung Date: Mon, 19 Oct 2026 21:35:24 +0100 Subject: [PATCH 2/4] [miniflare] Close local Hyperdrive sockets during interrupted setup The existing local TLS proxy can still be waiting for a database greeting or TLS handshake when its client disconnects. Closing the listener alone does not close that outbound connection, especially after a reload retires the listener. Give each accepted client ownership of its outbound TCP, TLS, and fallback sockets before waiting for them. Client disconnects and shutdown close the whole connection. A closed peer rejects pending reads and handshake waits, and cannot trigger a new plaintext fallback after its client has gone away. PostgreSQL startup bytes may arrive while the proxy waits for the database's SSL response. Consume a bounded startup buffer so client EOF remains visible, then replay those bytes in order once TLS or plaintext fallback is ready. - `proxy.spec.ts`: Controlled pauses at the greeting, SSL response, and TLS handshake check cleanup before negotiation completes. Delayed SSL responses also check that buffering preserves startup bytes for TLS and fallback paths. Co-authored-by: Henry Jang --- .../plugins/hyperdrive/hyperdrive-proxy.ts | 242 +++++++++++------- .../test/plugins/hyperdrive/proxy.spec.ts | 239 ++++++++++++++++- 2 files changed, 387 insertions(+), 94 deletions(-) diff --git a/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts b/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts index c6f707dfa7e..ba210a12799 100644 --- a/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts +++ b/packages/miniflare/src/plugins/hyperdrive/hyperdrive-proxy.ts @@ -292,6 +292,7 @@ export class HyperdriveProxyController { Number.parseInt(targetPort), scheme, sslmode, + owner, sslrootcert ).catch(() => owner.close()); }); @@ -361,10 +362,14 @@ export class HyperdriveProxyController { targetPort: number, scheme: string, sslmode: string, + owner: ConnectionOwner, sslrootcert?: string ) { - // Connect to real database - const dbSocket = net.connect({ host: targetHost, port: targetPort }); + // Track outbound sockets before any connect or TLS wait can yield. + const dbSocket = owner.track( + net.connect({ host: targetHost, port: targetPort }) + ); + dbSocket.on("error", () => {}); dbSocket.setNoDelay(true); const sslmodeRequire = sslmode === "require"; const sslmodePrefer = sslmode === "prefer"; @@ -387,7 +392,8 @@ export class HyperdriveProxyController { targetHost, targetPort, sslmodeRequire || sslmodeStrict, - tlsConfig + tlsConfig, + owner ); } else if (scheme === schemes.mysql) { return await handleMySQLTlsConnection( @@ -396,22 +402,20 @@ export class HyperdriveProxyController { targetHost, targetPort, sslmodeRequire || sslmodeStrict, - tlsConfig + tlsConfig, + owner ); } } catch (e) { - if (sslmodeRequire || sslmodeStrict) { - // Write error to client so worker can read it + if (!clientSocket.destroyed) { clientSocket.write(`${e}\n`); clientSocket.end(); - dbSocket.destroy(); - return; } + dbSocket.destroy(); + return; } } - // Pipe plain tcp sockets - clientSocket.pipe(dbSocket); - dbSocket.pipe(clientSocket); + setupTLSConnection(clientSocket, dbSocket); } /** @@ -691,45 +695,74 @@ async function handlePostgresTlsConnection( targetHost: string, targetPort: number, sslmodeRequire: boolean, - tlsConfig: TlsConfig + tlsConfig: TlsConfig, + owner: ConnectionOwner ) { - // Send Postgres sslrequest bytes - await writeAsync(dbSocket, POSTGRES_SSL_REQUEST_PACKET); - - const response = await readAsync(dbSocket); - // Read first byte ssl flag - const sslResponseFlag = response.toString("utf8", 0, 1); - if (sslResponseFlag === "S") { - const tlsOptions = buildTlsConnectionOptions( - dbSocket, - targetHost, - tlsConfig - ); - try { - const tlsSocket = await tlsConnect(tlsOptions); - setupTLSConnection(clientSocket, tlsSocket); + // Consume pending startup bytes so EOF can cancel negotiation. Bound this + // buffer rather than allowing a stalled database to accumulate client input. + const startupChunks: Buffer[] = []; + let startupLength = 0; + const collectStartup = (chunk: Buffer) => { + startupLength += chunk.length; + if (startupLength > 64 * 1024) { + owner.close(); return; - } catch (e) { - if (sslmodeRequire) { - throw e; + } + startupChunks.push(chunk); + }; + const releaseStartup = () => { + clientSocket.pause(); + clientSocket.off("data", collectStartup); + owner.assertOpen(); + if (startupLength > 0) { + clientSocket.unshift(Buffer.concat(startupChunks)); + } + }; + clientSocket.on("data", collectStartup); + clientSocket.resume(); + try { + // Send Postgres sslrequest bytes + await writeAsync(dbSocket, POSTGRES_SSL_REQUEST_PACKET); + + const response = await readAsync(dbSocket); + // Read first byte ssl flag + const sslResponseFlag = response.toString("utf8", 0, 1); + if (sslResponseFlag === "S") { + const tlsOptions = buildTlsConnectionOptions( + dbSocket, + targetHost, + tlsConfig + ); + try { + const tlsSocket = await tlsConnect(tlsOptions, owner); + releaseStartup(); + setupTLSConnection(clientSocket, tlsSocket); + return; + } catch (e) { + if (sslmodeRequire) { + throw e; + } + // Fallback to plain TCP connection } - // Fallback to plain TCP connection } + // Server doesn't support SSL but client requires it + if (sslmodeRequire) { + throw Error("Server does not support SSL, but client requires SSL."); + } + // fallback to plain TCP + dbSocket.destroy(); + const newDbSocket = await createPlainTCPConnection( + targetHost, + targetPort, + clientSocket, + owner + ); + releaseStartup(); + // Pipe plain TCP sockets + setupTLSConnection(clientSocket, newDbSocket); + } finally { + clientSocket.off("data", collectStartup); } - // Server doesn't support SSL but client requires it - if (sslmodeRequire) { - throw Error("Server does not support SSL, but client requires SSL."); - } - // fallback to plain TCP - dbSocket.destroy(); - const newDbSocket = await createPlainTCPConnection( - targetHost, - targetPort, - clientSocket - ); - // Pipe plain TCP sockets - clientSocket.pipe(newDbSocket); - newDbSocket.pipe(clientSocket); } /** Handles MySQL TLS connection */ @@ -739,7 +772,8 @@ async function handleMySQLTlsConnection( targetHost: string, targetPort: number, sslmodeRequire: boolean, - tlsConfig: TlsConfig + tlsConfig: TlsConfig, + owner: ConnectionOwner ) { const initPacketChunk = await readAsync(dbSocket); // Little-endian parse payload header length @@ -798,7 +832,7 @@ async function handleMySQLTlsConnection( ); try { - const tlsSocket = await tlsConnect(tlsOptions); + const tlsSocket = await tlsConnect(tlsOptions, owner); // Send original init packet to client (with SSL capability) const fullInitPacket = Buffer.concat([ @@ -837,52 +871,37 @@ async function handleMySQLTlsConnection( const newDbSocket = await createPlainTCPConnection( targetHost, targetPort, - clientSocket + clientSocket, + owner ); // Pipe plain TCP sockets - clientSocket.pipe(newDbSocket); - newDbSocket.pipe(clientSocket); + setupTLSConnection(clientSocket, newDbSocket); return; } -/** Create and wait for plain TCP connection */ +/** Creates a fallback socket only while the client still owns the connection. */ async function createPlainTCPConnection( targetHost: string, targetPort: number, - clientSocket: net.Socket + clientSocket: net.Socket, + owner: ConnectionOwner ): Promise { - const dbSocket = net.connect({ host: targetHost, port: targetPort }); + owner.assertOpen(); + const dbSocket = owner.track( + net.connect({ host: targetHost, port: targetPort }) + ); dbSocket.setNoDelay(true); - - // Wait for connection to be established - await new Promise((resolve, reject) => { - const handleConnect = () => { - dbSocket.off("error", handleError); - resolve(); - }; - - const handleError = (err: Error) => { - dbSocket.off("data", handleConnect); - reject(err); - }; - - dbSocket.once("connect", handleConnect); - dbSocket.once("error", handleError); - }); - - // Set up error handler - dbSocket.on("error", () => { - clientSocket.destroy(); - }); - + dbSocket.on("error", () => clientSocket.destroy()); + await waitForSocket(dbSocket, "connect"); + owner.assertOpen(); return dbSocket; } /** Set up TLS connection with pipes and error handlers */ function setupTLSConnection( clientSocket: net.Socket, - tlsSocket: tls.TLSSocket + tlsSocket: net.Socket ): void { // Set up error handler for runtime TLS errors tlsSocket.on("error", () => { @@ -890,6 +909,7 @@ function setupTLSConnection( tlsSocket.destroy(); }); + tlsSocket.once("close", () => clientSocket.destroy()); // Pipe sockets clientSocket.pipe(tlsSocket); tlsSocket.pipe(clientSocket); @@ -906,43 +926,81 @@ function writeAsync( }); } -/** Ready buffer from socket and return as a promise helper function */ +/** Reads the next chunk, rejecting when the peer closes before sending it. */ function readAsync(socket: net.Socket): Promise> { return new Promise((resolve, reject) => { - const handleData = (data: Buffer) => { + const cleanup = () => { + socket.off("data", handleData); socket.off("error", handleError); + socket.off("close", handleClose); + socket.off("end", handleClose); + }; + const handleData = (data: Buffer) => { + cleanup(); resolve(data); }; - const handleError = (err: Error) => { - socket.off("data", handleData); + cleanup(); reject(err); }; - + const handleClose = () => + handleError(new Error("Hyperdrive connection closed")); + if (socket.destroyed || socket.readableEnded) { + handleClose(); + return; + } socket.once("data", handleData); socket.once("error", handleError); + socket.once("close", handleClose); + socket.once("end", handleClose); }); } -function tlsConnect(options: tls.ConnectionOptions): Promise { +/** Waits for TCP/TLS readiness without retaining listeners after settlement. */ +function waitForSocket( + socket: net.Socket, + event: "connect" | "secureConnect" +): Promise { return new Promise((resolve, reject) => { - const socket = tls.connect(options); - - const handleSecureConnect = () => { + const cleanup = () => { + socket.off(event, handleReady); socket.off("error", handleError); - resolve(socket); + socket.off("close", handleClose); + socket.off("end", handleClose); }; - - const handleError = (err: Error) => { - socket.off("secureConnect", handleSecureConnect); - reject(err); + const handleReady = () => { + cleanup(); + resolve(); }; - - socket.once("secureConnect", handleSecureConnect); + const handleError = (error: Error) => { + cleanup(); + reject(error); + }; + const handleClose = () => + handleError(new Error("Hyperdrive connection closed")); + if (socket.destroyed || socket.readableEnded) { + handleClose(); + return; + } + socket.once(event, handleReady); socket.once("error", handleError); + socket.once("close", handleClose); + socket.once("end", handleClose); }); } +async function tlsConnect( + options: tls.ConnectionOptions, + owner: ConnectionOwner +): Promise { + owner.assertOpen(); + const socket = owner.track(tls.connect(options)); + socket.on("error", () => {}); + await waitForSocket(socket, "secureConnect"); + owner.assertOpen(); + return socket; +} + /** MySQL helper that skips through reading the init packet * until we get to the ssl capability flag, only need to look at lower 16 bits * docs: https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_connection_phase_packets_protocol_handshake_v10.html diff --git a/packages/miniflare/test/plugins/hyperdrive/proxy.spec.ts b/packages/miniflare/test/plugins/hyperdrive/proxy.spec.ts index 24fde871718..c26dac5cc30 100644 --- a/packages/miniflare/test/plugins/hyperdrive/proxy.spec.ts +++ b/packages/miniflare/test/plugins/hyperdrive/proxy.spec.ts @@ -1,12 +1,21 @@ import childProcess from "node:child_process"; +import { once } from "node:events"; import fs from "node:fs"; import net from "node:net"; import os from "node:os"; import path from "node:path"; import tls from "node:tls"; import { removeDirSync } from "@cloudflare/workers-utils"; -import { Miniflare } from "miniflare"; -import { afterAll, beforeAll, describe, test, vi } from "vitest"; +import { DeferredPromise, Miniflare } from "miniflare"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + test, + vi, +} from "vitest"; import { HyperdriveProxyController, POSTGRES_SSL_REQUEST_PACKET, @@ -902,3 +911,229 @@ test("retiring a proxy stops new connections but preserves an active one", async database.close(); } }); + +describe("local Hyperdrive connection cleanup", () => { + let database: net.Server; + let controller: HyperdriveProxyController; + let client: net.Socket | undefined; + let connections: net.Socket[]; + let secureSockets: tls.TLSSocket[]; + let tmpDir: string; + let certs: TestCerts; + + beforeAll(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "hyperdrive-cleanup-")); + certs = generateCerts(tmpDir); + }); + afterAll(() => removeDirSync(tmpDir)); + beforeEach(() => { + controller = new HyperdriveProxyController(); + client = undefined; + connections = []; + secureSockets = []; + database = net.createServer((socket) => { + connections.push(socket); + socket.on("error", () => {}); + }); + }); + afterEach(() => { + client?.destroy(); + controller.dispose(); + for (const socket of [...connections, ...secureSockets]) { + socket.destroy(); + } + database.close(); + }); + + async function connect(scheme = "postgres", sslmode = "prefer") { + await new Promise((resolve) => + database.listen(0, "127.0.0.1", resolve) + ); + const address = database.address(); + if (address === null || typeof address === "string") { + throw new Error("Expected TCP port"); + } + const port = await controller.createProxyServer({ + name: "cleanup", + targetHost: "127.0.0.1", + targetPort: String(address.port), + scheme, + sslmode, + }); + client = net.connect(port, "127.0.0.1"); + client.on("error", () => {}); + await once(client, "connect"); + return client; + } + + test.for([ + { scheme: "postgres", stage: "SSL response" }, + { scheme: "postgres", stage: "TLS handshake" }, + { scheme: "mysql", stage: "greeting" }, + ])( + "dispose closes local outbound sockets during $scheme $stage", + async ({ scheme, stage }, { expect }) => { + const waiting = new DeferredPromise(); + database.on("connection", (socket) => { + if (scheme === "mysql") { + waiting.resolve(socket); + } else { + socket.once("data", () => { + if (stage === "TLS handshake") { + socket.write("S"); + socket.once("data", () => waiting.resolve(socket)); + } else { + waiting.resolve(socket); + } + }); + } + }); + const client = await connect(scheme, "require"); + const accepted = await waiting; + const closed = Promise.all([ + once(client, "close"), + once(accepted, "close"), + ]); + controller.beginUpdate(); + controller.commitUpdate(new Set()); + controller.dispose(); + controller.dispose(); + await closed; + expect(client.destroyed).toBe(true); + expect(accepted.destroyed).toBe(true); + } + ); + + test("client close cancels PostgreSQL negotiation without starting a fallback", async ({ + expect, + }) => { + const waiting = new DeferredPromise(); + database.on("connection", (socket) => { + socket.once("data", () => waiting.resolve(socket)); + }); + const client = await connect(); + const accepted = await waiting; + const closed = once(accepted, "close"); + client.destroy(); + await closed; + expect(connections).toHaveLength(1); + }); + + test.for(["CLIENT STARTUP", "x".repeat(64 * 1024 + 1)])( + "PostgreSQL pending SSL closes on buffered FIN or oversized startup (%#)", + async (input, { expect }) => { + const waiting = new DeferredPromise(); + database.on("connection", (socket) => { + socket.once("data", () => waiting.resolve(socket)); + }); + const client = await connect(); + const accepted = await waiting; + const closed = Promise.all( + [client, accepted].map( + (socket) => + new Promise((resolve) => + socket.once("close", () => resolve()) + ) + ) + ); + if (input.length > 64 * 1024) { + client.write(input); + } else { + client.end(input); + } + await closed; + expect(connections).toHaveLength(1); + expect(accepted.destroyed).toBe(true); + } + ); + + test.for(["S", "N"] as const)( + "PostgreSQL replays startup bytes sent before a delayed %s response in order", + async (sslResponse, { expect }) => { + const responseGate = new DeferredPromise(); + const sslRequestSeen = new DeferredPromise(); + let carried = ""; + database.on("connection", (socket) => { + if (connections.length > 1) { + socket.on("data", (data) => { + carried += data.toString(); + }); + return; + } + socket.once("readable", () => { + socket.read(POSTGRES_SSL_REQUEST_PACKET.length); + sslRequestSeen.resolve(); + void responseGate.then(() => { + socket.write(sslResponse); + if (sslResponse === "S") { + const secure = new tls.TLSSocket(socket, { + isServer: true, + key: certs.localhost.key, + cert: certs.localhost.cert, + }); + secureSockets.push(secure); + secure.on("error", () => {}); + secure.on("data", (data) => { + carried += data.toString(); + }); + } + }); + }); + }); + const client = await connect(); + await sslRequestSeen; + client.write("STARTUP-1|"); + client.write("STARTUP-2|"); + await vi.waitFor(() => expect(connections).toHaveLength(1)); + responseGate.resolve(); + await vi.waitFor(() => expect(carried).toBe("STARTUP-1|STARTUP-2|")); + client.write("AFTER|"); + await vi.waitFor(() => + expect(carried).toBe("STARTUP-1|STARTUP-2|AFTER|") + ); + expect(connections).toHaveLength(sslResponse === "S" ? 1 : 2); + } + ); + + test("MySQL auth half-close cancels TLS without opening a prefer fallback", async ({ + expect, + }) => { + const payload = Buffer.concat([ + Buffer.from([10]), + Buffer.from("test\0"), + Buffer.alloc(4), + Buffer.alloc(8), + Buffer.from([0, 0, 8]), + ]); + const greeting = Buffer.concat([ + Buffer.from([payload.length, 0, 0, 0]), + payload, + ]); + database.on("connection", (socket) => { + socket.write(greeting); + socket.once("readable", () => { + socket.read(36); + const secure = new tls.TLSSocket(socket, { + isServer: true, + key: certs.localhost.key, + cert: certs.localhost.cert, + }); + secureSockets.push(secure); + secure.on("error", () => {}); + secure.on("data", () => {}); + }); + }); + const client = await connect("mysql"); + await once(client, "data"); + const dbClosed = Promise.all( + secureSockets.map( + (socket) => + new Promise((resolve) => socket.once("close", () => resolve())) + ) + ); + const clientClosed = once(client, "close"); + client.end(); + await Promise.all([dbClosed, clientClosed]); + expect(connections).toHaveLength(1); + }); +}); From d1bde640826e6fbe6970bcc555b454a01f873fc8 Mon Sep 17 00:00:00 2001 From: Edmund Hung Date: Mon, 19 Oct 2026 21:35:24 +0100 Subject: [PATCH 3/4] [remote-bindings] Preserve startup errors during cleanup Failed startup already tears down its development environment, but a teardown error can replace the failure the caller needs to diagnose. Preserve the original error both when startup throws and when it fails before becoming ready, even if cleanup also rejects. For a started session, repeated dispose() calls share one teardown promise, including its rejection. Cleanup is attempted once rather than starting overlapping teardown operations. Co-authored-by: Henry Jang --- .../src/start-remote-proxy-session.test.ts | 80 ++++++++++++++++--- .../src/start-remote-proxy-session.ts | 19 ++++- 2 files changed, 85 insertions(+), 14 deletions(-) diff --git a/packages/remote-bindings/src/start-remote-proxy-session.test.ts b/packages/remote-bindings/src/start-remote-proxy-session.test.ts index d3d0cfa247e..428fb27b414 100644 --- a/packages/remote-bindings/src/start-remote-proxy-session.test.ts +++ b/packages/remote-bindings/src/start-remote-proxy-session.test.ts @@ -1,23 +1,33 @@ import { APIError, UserError } from "@cloudflare/workers-utils"; -import { describe, it, vi } from "vitest"; +import { afterEach, describe, it, vi } from "vitest"; import { startRemoteProxySession } from "./start-remote-proxy-session"; import { RemoteSessionAuthenticationError } from "./utils/remote"; import type { RemoteBindingsLogger } from "./logger"; import type { ErrorEvent } from "./startDevWorker/events"; // The error that the mocked `DevEnv` emits asynchronously from `start()`. -const mockDevEnv = vi.hoisted(() => ({ error: undefined as unknown })); - -// Replace `DevEnv` with a stub that never becomes ready and instead emits a -// configurable error event just after `start()`, so we can exercise the async -// error-handling path in `startRemoteProxySession` without any real Miniflare. +const mockDevEnv = vi.hoisted(() => ({ + error: undefined as unknown, + startError: undefined as unknown, + teardown: vi.fn(async () => {}), +})); + +// Replace `DevEnv` with a configurable stub so we can exercise startup errors +// and cleanup without any real Miniflare. vi.mock("./startDevWorker/DevEnv", async () => { const { EventEmitter } = await import("node:events"); class DevEnv extends EventEmitter { proxy = { - // Never resolves — the startup race is always settled by the error event. - localServerReady: { promise: new Promise(() => {}) }, - ready: { promise: new Promise(() => {}) }, + localServerReady: { + // With an error configured, the error event settles startup instead. + promise: + mockDevEnv.error === undefined + ? Promise.resolve() + : new Promise(() => {}), + }, + ready: { + promise: Promise.resolve({ url: new URL("http://localhost:8787") }), + }, runtimeMessageMutex: { drained: async () => {} }, }; constructor() { @@ -26,10 +36,15 @@ vi.mock("./startDevWorker/DevEnv", async () => { this.on("error", () => {}); } start() { - queueMicrotask(() => this.emit("error", mockDevEnv.error)); + if (mockDevEnv.startError !== undefined) { + throw mockDevEnv.startError; + } + if (mockDevEnv.error !== undefined) { + queueMicrotask(() => this.emit("error", mockDevEnv.error)); + } } update() {} - async teardown() {} + teardown = mockDevEnv.teardown; } return { DevEnv }; }); @@ -81,6 +96,49 @@ describe("startRemoteProxySession error handling", () => { }); }); + afterEach(() => { + mockDevEnv.error = undefined; + mockDevEnv.startError = undefined; + mockDevEnv.teardown.mockReset(); + mockDevEnv.teardown.mockResolvedValue(undefined); + }); + + it("disposes only once even when teardown rejects", async ({ expect }) => { + const error = new Error("cleanup failure"); + mockDevEnv.teardown.mockRejectedValueOnce(error); + const session = await startRemoteProxySession( + {}, + { cliDisplayName: "Test CLI", logger: createTestLogger() } + ); + const first = session.dispose(); + const second = session.dispose(); + expect(second).toBe(first); + await expect(first).rejects.toBe(error); + expect(session.dispose()).toBe(first); + expect(mockDevEnv.teardown).toHaveBeenCalledOnce(); + }); + + it.for(["start", "async"])( + "preserves the %s startup error when cleanup fails", + async (stage, { expect }) => { + const original = new UserError("original startup failure", { + telemetryMessage: "test startup error", + }); + if (stage === "start") { + mockDevEnv.startError = original; + } else { + mockDevEnv.error = original; + } + mockDevEnv.teardown.mockRejectedValueOnce(new Error("cleanup failure")); + await expect( + startRemoteProxySession( + {}, + { cliDisplayName: "Test CLI", logger: createTestLogger() } + ) + ).rejects.toBe(original); + expect(mockDevEnv.teardown).toHaveBeenCalledOnce(); + } + ); // Regression: a `UserError` emitted asynchronously (e.g. the Miniflare // user-error case in `DevEnv.handleErrorEvent`) used to fall through to being // wrapped in a generic `Error`, which makes Wrangler treat it as an diff --git a/packages/remote-bindings/src/start-remote-proxy-session.ts b/packages/remote-bindings/src/start-remote-proxy-session.ts index 08e1fb869dd..aa4d9d78daa 100644 --- a/packages/remote-bindings/src/start-remote-proxy-session.ts +++ b/packages/remote-bindings/src/start-remote-proxy-session.ts @@ -139,7 +139,11 @@ export async function startRemoteProxySession( devEnv = new DevEnv(workerConfig); devEnv.start(); } catch (startWorkerError: unknown) { - await devEnv?.teardown(); + try { + await devEnv?.teardown(); + } catch { + // Preserve the original startup error if cleanup also fails. + } if (startWorkerError instanceof UserError) { throw startWorkerError; } @@ -193,11 +197,16 @@ export async function startRemoteProxySession( remoteProxyConnectionString = (await devEnv.proxy.ready.promise) .url as RemoteProxyConnectionString; } catch (error) { - await devEnv.teardown(); + try { + await devEnv.teardown(); + } catch { + // Preserve the original startup error if cleanup also fails. + } throw error; } finally { devEnv.removeListener("error", onStartupError); } + let disposePromise: Promise | undefined; const updateBindings = async ( newBindings: StartDevWorkerInput["bindings"] ) => { @@ -225,7 +234,11 @@ export async function startRemoteProxySession( ready: Promise.resolve(), remoteProxyConnectionString, updateBindings, - dispose: () => devEnv.teardown(), + dispose: () => { + // Share the same cleanup attempt across calls, even if it fails. + disposePromise ??= devEnv.teardown(); + return disposePromise; + }, }; } From 3c9053f24332dedb2eb7ebae6920bf2df053f6b7 Mon Sep 17 00:00:00 2001 From: Edmund Hung Date: Mon, 19 Oct 2026 21:35:25 +0100 Subject: [PATCH 4/4] [wrangler] Query remote Hyperdrive databases in E2E tests A database greeting or a separately fetched connection string does not prove that the local binding can authenticate through its bridge. Exercise actual SQL queries against provisioned MySQL and PostgreSQL Hyperdrive resources using ordinary database drivers. The drivers and PostgreSQL types are test-only dependencies, not additions to the runtime bundle. - `hyperdrive-remote.test.ts`: Each protocol runs concurrent queries through wrangler dev and a Node-side query through getPlatformProxy(). An invalid local connection string confirms remote mode does not use the local fallback. - `miniflare-remote-resources.test.ts`: A direct Miniflare consumer configures the remote binding and runs a MySQL query before and after setOptions(). This exercises the shared implementation independently of Wrangler's dev adapter. Co-authored-by: Henry Jang --- .../remote-binding/hyperdrive-remote.test.ts | 102 ++++++++++++++++++ .../miniflare-remote-resources.test.ts | 64 +++++++++++ .../workers/hyperdrive-postgres-query.js | 23 ++++ .../workers/hyperdrive-query.js | 23 ++++ packages/wrangler/package.json | 3 + pnpm-lock.yaml | 9 ++ 6 files changed, 224 insertions(+) create mode 100644 packages/wrangler/e2e/remote-binding/hyperdrive-remote.test.ts create mode 100644 packages/wrangler/e2e/remote-binding/workers/hyperdrive-postgres-query.js create mode 100644 packages/wrangler/e2e/remote-binding/workers/hyperdrive-query.js diff --git a/packages/wrangler/e2e/remote-binding/hyperdrive-remote.test.ts b/packages/wrangler/e2e/remote-binding/hyperdrive-remote.test.ts new file mode 100644 index 00000000000..051ad1323b4 --- /dev/null +++ b/packages/wrangler/e2e/remote-binding/hyperdrive-remote.test.ts @@ -0,0 +1,102 @@ +import path from "node:path"; +import { createConnection } from "mysql2/promise"; +import pg from "pg"; +import { fetch } from "undici"; +import { describe, it } from "vitest"; +import { CLOUDFLARE_ACCOUNT_ID } from "../helpers/account-id"; +import { + importWrangler, + WranglerE2ETestHelper, +} from "../helpers/e2e-wrangler-test"; +import { generateResourceName } from "../helpers/generate-resource-name"; +import type { Hyperdrive } from "@cloudflare/workers-types"; + +const { getPlatformProxy } = await importWrangler(); + +describe.skipIf(!CLOUDFLARE_ACCOUNT_ID)("remote Hyperdrive consumers", () => { + it.for(["mysql", "postgresql"] as const)( + "authenticates and queries through Wrangler and getPlatformProxy (%s)", + async (scheme, { expect }) => { + const helper = new WranglerE2ETestHelper(); + const { id } = await helper.hyperdrive(false, scheme); + const driver = scheme === "mysql" ? "mysql2/promise" : "pg"; + const scriptPath = + scheme === "mysql" + ? "hyperdrive-query.js" + : "hyperdrive-postgres-query.js"; + await helper.seed(path.resolve(__dirname, "./workers")); + await helper.seed({ + "wrangler.json": JSON.stringify({ + name: generateResourceName("hyperdrive-query"), + main: scriptPath, + compatibility_date: "2026-09-01", + compatibility_flags: ["nodejs_compat"], + alias: { [driver]: require.resolve(driver) }, + hyperdrive: [ + { + binding: "HYPERDRIVE_BINDING", + id, + remote: true, + localConnectionString: "unused-invalid-local", + }, + ], + }), + }); + const worker = helper.runLongLived("wrangler dev"); + try { + const { url } = await worker.waitForReady(); + await Promise.all( + Array.from({ length: 3 }, async () => { + const response = await fetch(url); + expect(response.status).toBe(200); + expect(await response.json()).toEqual([ + { remote_hyperdrive_probe: 1 }, + ]); + }) + ); + const platform = await getPlatformProxy<{ + HYPERDRIVE_BINDING: Hyperdrive; + }>({ + configPath: path.join(helper.tmpPath, "wrangler.json"), + remoteBindings: true, + }); + try { + const binding = platform.env.HYPERDRIVE_BINDING; + const connectionOptions = { + host: binding.host, + port: binding.port, + user: binding.user, + password: binding.password, + database: binding.database, + }; + if (scheme === "mysql") { + const connection = await createConnection(connectionOptions); + try { + const [rows] = await connection.query( + "SELECT 1 AS remote_hyperdrive_probe" + ); + expect(rows).toEqual([{ remote_hyperdrive_probe: 1 }]); + } finally { + await connection.end(); + } + } else { + const client = new pg.Client(connectionOptions); + try { + await client.connect(); + const { rows } = await client.query( + "SELECT 1 AS remote_hyperdrive_probe" + ); + expect(rows).toEqual([{ remote_hyperdrive_probe: 1 }]); + } finally { + await client.end(); + } + } + } finally { + await platform.dispose(); + } + } finally { + await worker.stop(); + } + } + ); +}); diff --git a/packages/wrangler/e2e/remote-binding/miniflare-remote-resources.test.ts b/packages/wrangler/e2e/remote-binding/miniflare-remote-resources.test.ts index 5fb03623e69..6fa5bf6c043 100644 --- a/packages/wrangler/e2e/remote-binding/miniflare-remote-resources.test.ts +++ b/packages/wrangler/e2e/remote-binding/miniflare-remote-resources.test.ts @@ -743,6 +743,70 @@ if (!CLOUDFLARE_ACCOUNT_ID) { 45_000 ); } + + it("queries remote Hyperdrive before and after a Miniflare reload", async ({ + expect, + }) => { + const { id } = await helper.hyperdrive(false, "mysql"); + const session = await startRemoteProxySession({ + HYPERDRIVE_BINDING: { type: "hyperdrive", id }, + }); + try { + await helper.seed({ + "hyperdrive-query.jsonc": JSON.stringify({ + name: "hyperdrive-query", + compatibility_date: "2026-09-01", + compatibility_flags: ["nodejs_compat"], + alias: { "mysql2/promise": require.resolve("mysql2/promise") }, + }), + }); + await helper.run( + "wrangler deploy hyperdrive-query.js --config hyperdrive-query.jsonc --dry-run --outdir hyperdrive-query-output" + ); + const queryOptions: MiniflareOptions = { + workers: [ + { + config: { + name: "hyperdrive-query", + compatibilityDate: "2026-09-01", + compatibilityFlags: ["nodejs_compat"], + manifest: createManifest(helper.tmpPath, [ + "hyperdrive-query-output/hyperdrive-query.js", + ]), + env: { + HYPERDRIVE_BINDING: { + type: "hyperdrive", + id, + dev: { remote: true }, + }, + }, + }, + dev: { + remoteProxyConnectionString: + session.remoteProxyConnectionString, + }, + }, + ], + }; + const queryMf = new Miniflare(queryOptions); + try { + for (const reload of [false, true]) { + if (reload) { + await queryMf.setOptions(queryOptions); + } + const response = await queryMf.dispatchFetch("http://localhost/"); + expect(response.status).toBe(200); + expect(await response.json()).toEqual([ + { remote_hyperdrive_probe: 1 }, + ]); + } + } finally { + await queryMf.dispose(); + } + } finally { + await session.dispose(); + } + }); }); // Separate describe block for mTLS because it needs a custom remote-binding diff --git a/packages/wrangler/e2e/remote-binding/workers/hyperdrive-postgres-query.js b/packages/wrangler/e2e/remote-binding/workers/hyperdrive-postgres-query.js new file mode 100644 index 00000000000..ae6bb740272 --- /dev/null +++ b/packages/wrangler/e2e/remote-binding/workers/hyperdrive-postgres-query.js @@ -0,0 +1,23 @@ +import pg from "pg"; + +export default { + async fetch(request, env) { + const binding = env.HYPERDRIVE_BINDING; + const client = new pg.Client({ + host: binding.host, + port: binding.port, + user: binding.user, + password: binding.password, + database: binding.database, + }); + try { + await client.connect(); + const { rows } = await client.query( + "SELECT 1 AS remote_hyperdrive_probe" + ); + return Response.json(rows); + } finally { + await client.end(); + } + }, +}; diff --git a/packages/wrangler/e2e/remote-binding/workers/hyperdrive-query.js b/packages/wrangler/e2e/remote-binding/workers/hyperdrive-query.js new file mode 100644 index 00000000000..de9d1f40cd2 --- /dev/null +++ b/packages/wrangler/e2e/remote-binding/workers/hyperdrive-query.js @@ -0,0 +1,23 @@ +import { createConnection } from "mysql2/promise"; + +export default { + async fetch(request, env) { + const binding = env.HYPERDRIVE_BINDING; + const connection = await createConnection({ + host: binding.host, + port: binding.port, + user: binding.user, + password: binding.password, + database: binding.database, + disableEval: true, + }); + try { + const [rows] = await connection.query( + "SELECT 1 AS remote_hyperdrive_probe" + ); + return Response.json(rows); + } finally { + await connection.end(); + } + }, +}; diff --git a/packages/wrangler/package.json b/packages/wrangler/package.json index 37000b97ba5..c1ee8120ef1 100644 --- a/packages/wrangler/package.json +++ b/packages/wrangler/package.json @@ -126,6 +126,7 @@ "@types/minimatch": "^5.1.2", "@types/node": "catalog:default", "@types/node-forge": "^1.3.11", + "@types/pg": "8.15.4", "@types/prompts": "^2.0.14", "@types/resolve": "^1.20.6", "@types/shell-quote": "^1.7.2", @@ -161,10 +162,12 @@ "minimatch": "^5.1.0", "mock-socket": "^9.3.1", "msw": "catalog:default", + "mysql2": "^3.15.3", "node-forge": "^1.3.2", "open": "catalog:default", "p-queue": "^9.0.0", "patch-console": "^1.0.0", + "pg": "8.16.3", "pretty-bytes": "^6.0.0", "prompts": "^2.4.2", "qr": "^0.6.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2535fa8feb1..2915c8451bd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -4670,6 +4670,9 @@ importers: '@types/node-forge': specifier: ^1.3.11 version: 1.3.11 + '@types/pg': + specifier: 8.15.4 + version: 8.15.4 '@types/prompts': specifier: ^2.0.14 version: 2.0.14 @@ -4775,6 +4778,9 @@ importers: msw: specifier: catalog:default version: 2.14.6(@types/node@22.15.17)(typescript@5.8.3) + mysql2: + specifier: ^3.15.3 + version: 3.15.3 node-forge: specifier: ^1.3.2 version: 1.3.3 @@ -4787,6 +4793,9 @@ importers: patch-console: specifier: ^1.0.0 version: 1.0.0 + pg: + specifier: 8.16.3 + version: 8.16.3 pretty-bytes: specifier: ^6.0.0 version: 6.1.1