From 40b6e91053c540cf86b1291d094452f96d6e17f4 Mon Sep 17 00:00:00 2001 From: arlo Date: Thu, 6 Aug 2026 11:23:18 +0800 Subject: [PATCH] fix: atomically allocate standalone ws ports --- .../devframe/src/rpc/transports/ws-server.ts | 51 ++++++++++++++++--- .../devframe/src/rpc/transports/ws.test.ts | 24 +++++++++ 2 files changed, 68 insertions(+), 7 deletions(-) diff --git a/packages/devframe/src/rpc/transports/ws-server.ts b/packages/devframe/src/rpc/transports/ws-server.ts index 3319d91c..9d75cf87 100644 --- a/packages/devframe/src/rpc/transports/ws-server.ts +++ b/packages/devframe/src/rpc/transports/ws-server.ts @@ -4,6 +4,7 @@ import type { NodeAdapter } from 'crossws/adapters/node' import type { Buffer } from 'node:buffer' import type { Server as HttpServer, IncomingMessage } from 'node:http' import type { Server as HttpsServer, ServerOptions as HttpsServerOptions } from 'node:https' +import type { AddressInfo } from 'node:net' import type { Duplex } from 'node:stream' import type { RpcFunctionDefinitionAny } from '../types' import { createServer as createHttpServer } from 'node:http' @@ -44,9 +45,12 @@ export interface WsRpcTransportOptions { * this transport detaches the upgrade listener without closing the server. */ server?: HttpServer | HttpsServer - /** Port for a newly-created standalone WS server. */ + /** + * Port for the standalone WebSocket server. Defaults to `0`, which lets the + * operating system assign an available port. + */ port?: number - /** Host for a newly-created standalone WS server. Defaults to `localhost`. */ + /** Host for the standalone WebSocket server. Defaults to `localhost`. */ host?: string /** * Restrict the WS endpoint to a single upgrade route (e.g. `/__devframe_ws`). When @@ -168,6 +172,10 @@ export interface WsRpcTransport { * `peers` and pub/sub. See https://crossws.h3.dev. */ ws: NodeAdapter + /** Resolves when the transport-owned server is listening. */ + ready: Promise + /** Returns the bound address, or `null` when the server is not listening. */ + address: () => AddressInfo | string | null /** Remove the upgrade listener from a shared `server` (a no-op otherwise). */ detach: () => void /** @@ -184,6 +192,27 @@ const EMPTY_DEFS: ReadonlyMap { + return new Promise((resolve, reject) => { + const onError = (error: Error) => reject(error) + server.once('error', onError) + try { + server.listen(port, host, () => { + server.off('error', onError) + resolve() + }) + } + catch (error) { + server.off('error', onError) + reject(error) + } + }) +} + /** Compare two URL paths ignoring a trailing slash. */ function pathMatches(a: string, b: string): boolean { const strip = (p: string) => (p.length > 1 && p.endsWith('/') ? p.slice(0, -1) : p) @@ -276,9 +305,9 @@ function routeUpgrades( * `server` (sharing its port, optionally scoped to a `path`), or let this * helper create a standalone server from `port` / `host` / `https`. * - * Returns the crossws node adapter plus `detach` (remove the upgrade - * listener from a shared `server`) and `close` (full deterministic - * teardown). + * Returns the crossws node adapter, standalone-server readiness/address + * accessors, `detach` (remove the upgrade listener from a shared `server`), + * and `close` (full deterministic teardown). */ export function attachWsRpcTransport< ClientFunctions extends object, @@ -388,6 +417,7 @@ export function attachWsRpcTransport< }) let detach = NOOP + let ready = Promise.resolve() // A server created (and thus owned) by this transport. Nothing else // handles its upgrades, so off-route clients are rejected promptly. let ownedServer: HttpServer | HttpsServer | undefined @@ -399,7 +429,7 @@ export function attachWsRpcTransport< else if (https) { ownedServer = createHttpsServer(https) detach = routeUpgrades(ownedServer, ws, path, true, allowedOrigins) - ownedServer.listen(port, host) + ready = listen(ownedServer, port ?? 0, host) } else { // Standalone server on its own port. Plain HTTP requests get the @@ -409,11 +439,15 @@ export function attachWsRpcTransport< res.end('Upgrade Required') }) detach = routeUpgrades(ownedServer, ws, path, true, allowedOrigins) - ownedServer.listen(port, host) + ready = listen(ownedServer, port ?? 0, host) } + const activeServer = server ?? ownedServer + return { ws, + ready, + address: () => activeServer?.address() ?? null, detach, async close() { // Detach our upgrade listener first so a shared host server stops @@ -425,6 +459,9 @@ export function attachWsRpcTransport< ws.closeAll(undefined, undefined, true) if (ownedServer) { const srv = ownedServer + await ready.catch(() => {}) + if (!srv.listening) + return await new Promise(r => srv.close(() => r())) } }, diff --git a/packages/devframe/src/rpc/transports/ws.test.ts b/packages/devframe/src/rpc/transports/ws.test.ts index b4606146..d35bab83 100644 --- a/packages/devframe/src/rpc/transports/ws.test.ts +++ b/packages/devframe/src/rpc/transports/ws.test.ts @@ -183,6 +183,30 @@ describe('ws client post on a non-open socket', () => { }) describe('devframe rpc', () => { + it('atomically allocates unique ports for concurrent standalone transports', async () => { + const HOST = '127.0.0.1' + const transports = Array.from({ length: 3 }, () => { + const server = createRpcServer, Record>({}) + return attachWsRpcTransport(server, { host: HOST }) + }) + + try { + await Promise.all(transports.map(transport => transport.ready)) + const ports = transports.map((transport) => { + const address = transport.address() + if (!address || typeof address === 'string') + throw new TypeError('Expected an IP socket address') + return address.port + }) + + expect(ports.every(port => port > 0)).toBe(true) + expect(new Set(ports).size).toBe(transports.length) + } + finally { + await Promise.all(transports.map(transport => transport.close())) + } + }) + it('should work w/ ws transport', async () => { // Use 127.0.0.1 on both client and server so they agree on the // address family — `localhost` resolution is ambiguous (IPv4 vs IPv6)