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..ba210a12799 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,41 @@ 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, + owner, 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 +330,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" * @@ -218,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"; @@ -244,7 +392,8 @@ export class HyperdriveProxyController { targetHost, targetPort, sslmodeRequire || sslmodeStrict, - tlsConfig + tlsConfig, + owner ); } else if (scheme === schemes.mysql) { return await handleMySQLTlsConnection( @@ -253,22 +402,228 @@ 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(); + } + dbSocket.destroy(); + return; + } + } + setupTLSConnection(clientSocket, dbSocket); + } + + /** + * 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); } - // Pipe plain tcp sockets - clientSocket.pipe(dbSocket); - dbSocket.pipe(clientSocket); } /** Begins tracking proxy servers created for the next runtime config. */ @@ -276,7 +631,10 @@ export class HyperdriveProxyController { 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 +644,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 +659,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(); + } + } } } @@ -330,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 */ @@ -378,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 @@ -437,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([ @@ -476,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", () => { @@ -529,6 +909,7 @@ function setupTLSConnection( tlsSocket.destroy(); }); + tlsSocket.once("close", () => clientSocket.destroy()); // Pipe sockets clientSocket.pipe(tlsSocket); tlsSocket.pipe(clientSocket); @@ -545,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/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/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); + }); +}); 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/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; + }, }; } 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/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/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, 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