From a94cf3046991d684562ae74ec3cb06984ae0a0eb Mon Sep 17 00:00:00 2001 From: Rocky shi Date: Fri, 9 Oct 2026 15:46:30 +0000 Subject: [PATCH 1/4] feat: implement source IP affinity for cluster workers --- .github/workflows/ci.yml | 4 + src/adapters/express/adapter.ts | 12 +- src/adapters/express/index.ts | 12 +- src/adapters/fastify/adapter.ts | 31 +- src/adapters/fastify/index.ts | 12 +- src/adapters/hono/adapter.ts | 16 +- src/adapters/hono/index.ts | 11 +- src/adapters/koa/adapter.ts | 12 +- src/adapters/koa/index.ts | 14 +- src/adapters/native/adapter.ts | 12 +- src/adapters/native/index.ts | 12 +- src/index.ts | 2 + src/lib/adapter-resolver.ts | 22 +- src/lib/app.ts | 1 + src/lib/bootstrap.ts | 83 ++++- src/lib/cluster/affinity-router.ts | 41 +++ src/lib/cluster/connection-policy.ts | 56 +++ src/lib/cluster/ipc-types.ts | 32 ++ src/lib/cluster/master.ts | 229 ++++++++++-- src/lib/cluster/socket-dispatcher.ts | 238 ++++++++++++ src/lib/cluster/socket-receiver.ts | 103 ++++++ src/lib/cluster/worker-slots.ts | 16 + src/lib/cluster/worker.ts | 54 ++- src/lib/socket-http-server.ts | 142 ++++++++ src/types/adapter.ts | 16 + src/types/app.ts | 6 +- src/types/index.ts | 2 + test/fixtures/public-surface/coverage.json | 7 +- test/integration/cluster/affinity-fixture.mjs | 118 ++++++ .../integration/cluster/capacity-loss.test.ts | 4 +- test/integration/cluster/ip-affinity.test.ts | 342 ++++++++++++++++++ test/unit/adapter-resolver.test.ts | 23 +- test/unit/cluster/affinity-router.test.ts | 40 ++ test/unit/cluster/socket-http-server.test.ts | 155 ++++++++ test/unit/cluster/socket-receiver.test.ts | 65 ++++ website/docs/en/api/config.md | 2 +- website/docs/en/guide/adapters.md | 10 + website/docs/en/guide/cluster.md | 26 +- website/docs/en/guide/deployment.md | 2 +- website/docs/zh/api/config.md | 2 +- website/docs/zh/guide/adapters.md | 10 + website/docs/zh/guide/cluster.md | 26 +- website/docs/zh/guide/deployment.md | 2 +- 43 files changed, 1901 insertions(+), 124 deletions(-) create mode 100644 src/lib/cluster/affinity-router.ts create mode 100644 src/lib/cluster/connection-policy.ts create mode 100644 src/lib/cluster/socket-dispatcher.ts create mode 100644 src/lib/cluster/socket-receiver.ts create mode 100644 src/lib/cluster/worker-slots.ts create mode 100644 src/lib/socket-http-server.ts create mode 100644 test/integration/cluster/affinity-fixture.mjs create mode 100644 test/integration/cluster/ip-affinity.test.ts create mode 100644 test/unit/cluster/affinity-router.test.ts create mode 100644 test/unit/cluster/socket-http-server.test.ts create mode 100644 test/unit/cluster/socket-receiver.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 27265eac..cf2a3f58 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -473,6 +473,10 @@ jobs: test/unit/frontend-deploy-validation.test.ts test/unit/frontend-runtime.test.ts test/unit/frontend-repairs.test.ts + test/unit/cluster/affinity-router.test.ts + test/unit/cluster/socket-receiver.test.ts + test/unit/cluster/socket-http-server.test.ts + test/integration/cluster/ip-affinity.test.ts --reporter=verbose - name: Verify Windows package exports diff --git a/src/adapters/express/adapter.ts b/src/adapters/express/adapter.ts index 60d2d848..e7ea5e97 100644 --- a/src/adapters/express/adapter.ts +++ b/src/adapters/express/adapter.ts @@ -5,7 +5,11 @@ import type { Response as ExpressResponse, } from "express"; import crypto from "node:crypto"; -import { createServer } from "node:http"; +import { + createAdapterHTTPServer, + socketHandoffControls, +} from "../../lib/socket-http-server.js"; +import type { VextAdapterRuntimeContext } from "../../types/adapter.js"; import type { IncomingMessage, ServerResponse } from "node:http"; import { createVextRequest } from "./request.js"; import { createVextResponse } from "./response.js"; @@ -217,6 +221,7 @@ function collectRawBody( export function createExpressAdapter( options: ExpressAdapterOptions, app: VextApp, + context?: VextAdapterRuntimeContext, ): VextAdapter { // ── 创建 Express 实例 ──────────────────────────────────── // @@ -423,6 +428,7 @@ export function createExpressAdapter( return { name: "express", + supportsSocketHandoff: Boolean(context?.socketHandoff), // ── registerMiddleware ─────────────────────────────────── // @@ -642,7 +648,8 @@ export function createExpressAdapter( // 注册兜底中间件（确保在所有路由之后） registerFallbacks(); - const server = createServer( + const server = createAdapterHTTPServer( + context, createNodeServerOptions(options?.server), expressApp, ); @@ -663,6 +670,7 @@ export function createExpressAdapter( : host; resolve({ + ...socketHandoffControls(server), port: actualPort, host: actualHost, diff --git a/src/adapters/express/index.ts b/src/adapters/express/index.ts index 6a60ce25..27670502 100644 --- a/src/adapters/express/index.ts +++ b/src/adapters/express/index.ts @@ -1,6 +1,10 @@ import { createExpressAdapter } from "./adapter.js"; import type { ExpressAdapterOptions } from "./adapter.js"; -import type { VextAdapter } from "../../types/adapter.js"; +import type { + VextAdapter, + VextAdapterFactory, + VextAdapterRuntimeContext, +} from "../../types/adapter.js"; import type { VextApp } from "../../types/app.js"; /** @@ -65,11 +69,11 @@ export type { ExpressAdapterOptions } from "./adapter.js"; */ export function expressAdapter( options?: ExpressAdapterOptions, -): (app: VextApp) => VextAdapter { +): VextAdapterFactory { const opts = options ?? {}; - return (app: VextApp): VextAdapter => { - return createExpressAdapter(opts, app); + return (app: VextApp, context?: VextAdapterRuntimeContext): VextAdapter => { + return createExpressAdapter(opts, app, context); }; } diff --git a/src/adapters/fastify/adapter.ts b/src/adapters/fastify/adapter.ts index 469987c7..715631c1 100644 --- a/src/adapters/fastify/adapter.ts +++ b/src/adapters/fastify/adapter.ts @@ -1,6 +1,10 @@ import Fastify from "fastify"; import type { FastifyInstance, FastifyRequest, FastifyReply } from "fastify"; -import { createServer } from "node:http"; +import { + createAdapterHTTPServer, + socketHandoffControls, +} from "../../lib/socket-http-server.js"; +import type { VextAdapterRuntimeContext } from "../../types/adapter.js"; import type { IncomingMessage, ServerResponse, Server } from "node:http"; import crypto from "node:crypto"; import { createVextRequest } from "./request.js"; @@ -181,6 +185,7 @@ const _noop = async (): Promise => {}; export function createFastifyAdapter( options: FastifyAdapterOptions, app: VextApp, + context?: VextAdapterRuntimeContext, ): VextAdapter { // ── 创建 Fastify 实例 ──────────────────────────────────── // @@ -197,16 +202,18 @@ export function createFastifyAdapter( adapterBodyLimit: options.bodyLimit, }); const serverConfig = app.config.server; - const serverFactory = hasServerConfig(serverConfig) - ? (handler: NodeRequestHandler): Server => { - const server = createServer( - createNodeServerOptions(serverConfig), - handler, - ); - applyServerConfig(server, serverConfig); - return server; - } - : undefined; + const serverFactory = + context?.socketHandoff || hasServerConfig(serverConfig) + ? (handler: NodeRequestHandler): Server => { + const server = createAdapterHTTPServer( + context, + createNodeServerOptions(serverConfig), + handler, + ); + applyServerConfig(server, serverConfig); + return server; + } + : undefined; const fastify: FastifyInstance = Fastify({ logger: options.logger ?? false, @@ -263,6 +270,7 @@ export function createFastifyAdapter( return { name: "fastify", + supportsSocketHandoff: Boolean(context?.socketHandoff), // ── registerMiddleware ─────────────────────────────────── // @@ -574,6 +582,7 @@ export function createFastifyAdapter( : host; return { + ...socketHandoffControls(fastify.server), port: actualPort, host: actualHost, diff --git a/src/adapters/fastify/index.ts b/src/adapters/fastify/index.ts index 07dcdb27..86ac1d8b 100644 --- a/src/adapters/fastify/index.ts +++ b/src/adapters/fastify/index.ts @@ -1,6 +1,10 @@ import { createFastifyAdapter } from "./adapter.js"; import type { FastifyAdapterOptions } from "./adapter.js"; -import type { VextAdapter } from "../../types/adapter.js"; +import type { + VextAdapter, + VextAdapterFactory, + VextAdapterRuntimeContext, +} from "../../types/adapter.js"; import type { VextApp } from "../../types/app.js"; /** @@ -91,11 +95,11 @@ export type { FastifyAdapterOptions } from "./adapter.js"; */ export function fastifyAdapter( options?: FastifyAdapterOptions, -): (app: VextApp) => VextAdapter { +): VextAdapterFactory { const opts = options ?? {}; - return (app: VextApp): VextAdapter => { - return createFastifyAdapter(opts, app); + return (app: VextApp, context?: VextAdapterRuntimeContext): VextAdapter => { + return createFastifyAdapter(opts, app, context); }; } diff --git a/src/adapters/hono/adapter.ts b/src/adapters/hono/adapter.ts index a94a6c20..2eba7b13 100644 --- a/src/adapters/hono/adapter.ts +++ b/src/adapters/hono/adapter.ts @@ -1,6 +1,10 @@ import { Hono, type Context } from "hono"; import { Buffer } from "node:buffer"; -import { createServer } from "node:http"; +import { + createAdapterHTTPServer, + socketHandoffControls, +} from "../../lib/socket-http-server.js"; +import type { VextAdapterRuntimeContext } from "../../types/adapter.js"; import type { IncomingMessage, ServerResponse } from "node:http"; import crypto from "node:crypto"; import { createRouter } from "route-core"; @@ -292,7 +296,10 @@ async function writeWebResponse( * @param app VextApp 实例（用于传递给 createVextRequest 的 app 引用） * @returns VextAdapter 实例 */ -export function createHonoAdapter(app: VextApp): VextAdapter { +export function createHonoAdapter( + app: VextApp, + context?: VextAdapterRuntimeContext, +): VextAdapter { const hono = new Hono(); const explicitHeadRouter = createRouter({ ignoreTrailingSlash: true, @@ -310,6 +317,7 @@ export function createHonoAdapter(app: VextApp): VextAdapter { return { name: "hono", + supportsSocketHandoff: Boolean(context?.socketHandoff), registerMiddleware(middleware: VextMiddleware): void { globalMiddlewares.push(middleware); @@ -505,7 +513,8 @@ export function createHonoAdapter(app: VextApp): VextAdapter { const requestHandler = this.buildHandler(); return new Promise((resolve, reject) => { - const server = createServer( + const server = createAdapterHTTPServer( + context, createNodeServerOptions(options?.server), requestHandler, ); @@ -525,6 +534,7 @@ export function createHonoAdapter(app: VextApp): VextAdapter { : host; resolve({ + ...socketHandoffControls(server), port: actualPort, host: actualHost, diff --git a/src/adapters/hono/index.ts b/src/adapters/hono/index.ts index 4f0dbe6a..120fdb4a 100644 --- a/src/adapters/hono/index.ts +++ b/src/adapters/hono/index.ts @@ -1,5 +1,9 @@ import { createHonoAdapter } from "./adapter.js"; -import type { VextAdapter } from "../../types/adapter.js"; +import type { + VextAdapter, + VextAdapterFactory, + VextAdapterRuntimeContext, +} from "../../types/adapter.js"; import type { VextApp } from "../../types/app.js"; /** @@ -17,8 +21,9 @@ import type { VextApp } from "../../types/app.js"; * * 内置字符串方式仍然可用：`adapter: "hono"`。 */ -export function honoAdapter(): (app: VextApp) => VextAdapter { - return (app: VextApp): VextAdapter => createHonoAdapter(app); +export function honoAdapter(): VextAdapterFactory { + return (app: VextApp, context?: VextAdapterRuntimeContext): VextAdapter => + createHonoAdapter(app, context); } export { createHonoAdapter } from "./adapter.js"; diff --git a/src/adapters/koa/adapter.ts b/src/adapters/koa/adapter.ts index 2f044203..1095a0b7 100644 --- a/src/adapters/koa/adapter.ts +++ b/src/adapters/koa/adapter.ts @@ -1,7 +1,11 @@ import Koa from "koa"; import Router from "@koa/router"; import type { Context as KoaContext } from "koa"; -import { createServer } from "node:http"; +import { + createAdapterHTTPServer, + socketHandoffControls, +} from "../../lib/socket-http-server.js"; +import type { VextAdapterRuntimeContext } from "../../types/adapter.js"; import type { IncomingMessage, ServerResponse } from "node:http"; import crypto from "node:crypto"; import { createVextRequest } from "./request.js"; @@ -242,6 +246,7 @@ function getKoaParams(ctx: KoaContext): Record { export function createKoaAdapter( options: KoaAdapterOptions, vextApp: VextApp, + context?: VextAdapterRuntimeContext, ): VextAdapter { // ── 创建 Koa 实例 ───────────────────────────────────────── // @@ -497,6 +502,7 @@ export function createKoaAdapter( return { name: "koa", + supportsSocketHandoff: Boolean(context?.socketHandoff), // ── registerMiddleware ─────────────────────────────────── // @@ -591,7 +597,8 @@ export function createKoaAdapter( registerKoaMiddleware(); const handler = koaApp.callback(); - const server = createServer( + const server = createAdapterHTTPServer( + context, createNodeServerOptions(options?.server), handler, ); @@ -612,6 +619,7 @@ export function createKoaAdapter( : host; resolve({ + ...socketHandoffControls(server), port: actualPort, host: actualHost, diff --git a/src/adapters/koa/index.ts b/src/adapters/koa/index.ts index 04b2679f..857a19ff 100644 --- a/src/adapters/koa/index.ts +++ b/src/adapters/koa/index.ts @@ -1,6 +1,10 @@ import { createKoaAdapter } from "./adapter.js"; import type { KoaAdapterOptions } from "./adapter.js"; -import type { VextAdapter } from "../../types/adapter.js"; +import type { + VextAdapter, + VextAdapterFactory, + VextAdapterRuntimeContext, +} from "../../types/adapter.js"; import type { VextApp } from "../../types/app.js"; /** @@ -64,13 +68,11 @@ export type { KoaAdapterOptions } from "./adapter.js"; * import { koaAdapter } from 'vextjs/adapters/koa' * export default { adapter: koaAdapter({ bodyLimit: '5mb' }) } */ -export function koaAdapter( - options?: KoaAdapterOptions, -): (app: VextApp) => VextAdapter { +export function koaAdapter(options?: KoaAdapterOptions): VextAdapterFactory { const opts = options ?? {}; - return (app: VextApp): VextAdapter => { - return createKoaAdapter(opts, app); + return (app: VextApp, context?: VextAdapterRuntimeContext): VextAdapter => { + return createKoaAdapter(opts, app, context); }; } diff --git a/src/adapters/native/adapter.ts b/src/adapters/native/adapter.ts index b41663c4..cf225464 100644 --- a/src/adapters/native/adapter.ts +++ b/src/adapters/native/adapter.ts @@ -1,4 +1,8 @@ -import { createServer } from "node:http"; +import { + createAdapterHTTPServer, + socketHandoffControls, +} from "../../lib/socket-http-server.js"; +import type { VextAdapterRuntimeContext } from "../../types/adapter.js"; import type { IncomingMessage, ServerResponse, Server } from "node:http"; import crypto from "node:crypto"; import { createRouter } from "route-core"; @@ -218,6 +222,7 @@ const _noop = async (): Promise => {}; export function createNativeAdapter( options: NativeAdapterOptions, app: VextApp, + context?: VextAdapterRuntimeContext, ): VextAdapter { // ── 创建 route-core 路由器 ──────────────────────────────── // @@ -611,6 +616,7 @@ export function createNativeAdapter( return { name: "native", + supportsSocketHandoff: Boolean(context?.socketHandoff), // ── registerMiddleware ─────────────────────────────────── // @@ -716,7 +722,8 @@ export function createNativeAdapter( host: string = "0.0.0.0", options?: VextAdapterListenOptions, ): Promise { - const server: Server = createServer( + const server: Server = createAdapterHTTPServer( + context, createNodeServerOptions(options?.server), handleRequest, ); @@ -737,6 +744,7 @@ export function createNativeAdapter( : host; resolve({ + ...socketHandoffControls(server), port: actualPort, host: actualHost, diff --git a/src/adapters/native/index.ts b/src/adapters/native/index.ts index b9072a0a..294c0970 100644 --- a/src/adapters/native/index.ts +++ b/src/adapters/native/index.ts @@ -1,6 +1,10 @@ import { createNativeAdapter } from "./adapter.js"; import type { NativeAdapterOptions } from "./adapter.js"; -import type { VextAdapter } from "../../types/adapter.js"; +import type { + VextAdapter, + VextAdapterFactory, + VextAdapterRuntimeContext, +} from "../../types/adapter.js"; import type { VextApp } from "../../types/app.js"; /** @@ -81,11 +85,11 @@ export type { NativeAdapterOptions } from "./adapter.js"; */ export function nativeAdapter( options?: NativeAdapterOptions, -): (app: VextApp) => VextAdapter { +): VextAdapterFactory { const opts = options ?? {}; - return (app: VextApp): VextAdapter => { - return createNativeAdapter(opts, app); + return (app: VextApp, context?: VextAdapterRuntimeContext): VextAdapter => { + return createNativeAdapter(opts, app, context); }; } diff --git a/src/index.ts b/src/index.ts index 8ec666f1..096b163d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -76,6 +76,8 @@ export type { // Adapter 类型 VextAdapter, + VextAdapterRuntimeContext, + VextAdapterFactory, VextServerHandle, // App / Config / Services 类型 diff --git a/src/lib/adapter-resolver.ts b/src/lib/adapter-resolver.ts index 7b2eab6a..60796216 100644 --- a/src/lib/adapter-resolver.ts +++ b/src/lib/adapter-resolver.ts @@ -1,4 +1,8 @@ -import type { VextAdapter } from "../types/adapter.js"; +import type { + VextAdapter, + VextAdapterFactory, + VextAdapterRuntimeContext, +} from "../types/adapter.js"; import type { VextApp, VextConfig } from "../types/app.js"; /** @@ -167,12 +171,13 @@ async function loadBuiltInAdapterWithDiagnostics( async function loadBuiltInAdapter( name: string, app: VextApp, + context?: VextAdapterRuntimeContext, ): Promise { switch (name) { case "native": { const { createNativeAdapter } = await import("../adapters/native/adapter.js"); - return createNativeAdapter({}, app); + return createNativeAdapter({}, app, context); } case "hono": { @@ -186,7 +191,7 @@ async function loadBuiltInAdapter( async () => { const { createHonoAdapter } = await import("../adapters/hono/index.js"); - return createHonoAdapter(app); + return createHonoAdapter(app, context); }, ); } @@ -202,7 +207,7 @@ async function loadBuiltInAdapter( async () => { const { createFastifyAdapter } = await import("../adapters/fastify/adapter.js"); - return createFastifyAdapter({}, app); + return createFastifyAdapter({}, app, context); }, ); } @@ -218,7 +223,7 @@ async function loadBuiltInAdapter( async () => { const { createExpressAdapter } = await import("../adapters/express/adapter.js"); - return createExpressAdapter({}, app); + return createExpressAdapter({}, app, context); }, ); } @@ -234,7 +239,7 @@ async function loadBuiltInAdapter( async () => { const { createKoaAdapter } = await import("../adapters/koa/adapter.js"); - return createKoaAdapter({}, app); + return createKoaAdapter({}, app, context); }, ); } @@ -282,18 +287,19 @@ async function loadBuiltInAdapter( export async function resolveAdapter( config: VextConfig, app: VextApp, + context?: VextAdapterRuntimeContext, ): Promise { const adapterConfig = config.adapter ?? "native"; // 字符串 → 内置 adapter（动态 import 按需加载） if (typeof adapterConfig === "string") { - return loadBuiltInAdapter(adapterConfig, app); + return loadBuiltInAdapter(adapterConfig, app, context); } // 函数 → adapter 工厂函数（如 fastifyAdapter({ bodyLimit: 5MB }) 返回的 (app) => VextAdapter） // 用户通过 import { fastifyAdapter } from 'vextjs/adapters/fastify' 使用 if (typeof adapterConfig === "function") { - const adapter = (adapterConfig as (app: VextApp) => VextAdapter)(app); + const adapter = (adapterConfig as VextAdapterFactory)(app, context); validateAdapterInterface(adapter); return adapter; } diff --git a/src/lib/app.ts b/src/lib/app.ts index 14f18b90..a814589a 100644 --- a/src/lib/app.ts +++ b/src/lib/app.ts @@ -593,6 +593,7 @@ export function createApp(config: VextConfig): { "[vextjs] server close failed during shutdown", ); } + if (serverResult.status === "timed-out") serverHandle.forceClose?.(); } // ── 步骤 2：按 LIFO 顺序执行 onClose 钩子 ── diff --git a/src/lib/bootstrap.ts b/src/lib/bootstrap.ts index babebfed..22510a2f 100644 --- a/src/lib/bootstrap.ts +++ b/src/lib/bootstrap.ts @@ -1,3 +1,7 @@ +import type { + SocketWorkerContext, + ClusterWorkerPolicy, +} from "./cluster/connection-policy.js"; import { join } from "node:path"; import cluster from "node:cluster"; import { @@ -208,6 +212,8 @@ async function resolveStartupConfig( */ export async function bootstrap( rootDir = process.cwd(), + socketContext?: SocketWorkerContext, + expectedPolicy?: ClusterWorkerPolicy, ): Promise { // 资源引用（用于错误边界清理） let internals: AppInternals | null = null; @@ -245,6 +251,16 @@ export async function bootstrap( () => finalizeConfig(rawConfig), { phase: "config" }, ); + if ( + (socketContext && !cluster.isWorker) || + (expectedPolicy && + ((config.cluster?.sticky ?? "none") !== expectedPolicy.sticky || + (config.cluster?.workers ?? "auto") !== expectedPolicy.workers)) + ) { + throw new Error( + "[cluster] Worker connection policy differs from Master; restart Master to change sticky or workers", + ); + } const lifecycleLevel = getLifecycleLevel(rawConfig); sendLifecycleLevelToParent(lifecycleLevel); @@ -282,10 +298,27 @@ export async function bootstrap( // 默认 native adapter（零外部依赖），其他 adapter 需用户额外安装对应框架。 app.adapter = await startupProfiler.time( "start.adapter", - () => resolveAdapter(config, app), + () => + resolveAdapter( + config, + app, + socketContext + ? { + socketHandoff: { + host: socketContext.host, + port: socketContext.port, + }, + } + : undefined, + ), { phase: "config" }, ); + if (socketContext && !app.adapter.supportsSocketHandoff) { + throw new Error( + `[cluster] Adapter "${app.adapter.name}" does not support sticky="ip" socket handoff`, + ); + } app.logger.info("[vextjs] initializing..."); // ── 步骤 ①+: i18n 语言包自动加载 ───────────────────── @@ -1265,6 +1298,7 @@ async function startClusterMaster(rootDir: string): Promise { workers: [], }), workers, + listen: { host: config.host ?? "0.0.0.0", port: config.port }, autoRestart: (clusterConfig.autoRestart as boolean | undefined) ?? true, maxRestarts: (clusterConfig.maxRestarts as number | undefined) ?? 5, restartWindow: @@ -1287,6 +1321,8 @@ async function startClusterMaster(rootDir: string): Promise { const snapshotWorkers = () => [...master.getWorkerMetas().values()].map((worker) => ({ id: worker.id, + slotId: worker.slotId, + generation: worker.generation, state: worker.state, startTime: worker.startTime, lastHeartbeat: worker.lastHeartbeat, @@ -1298,11 +1334,12 @@ async function startClusterMaster(rootDir: string): Promise { runtimeIdentity, summary: { state: "ready", - host: config.host ?? "0.0.0.0", - port: config.port, + host: master.getListenAddress()?.host ?? config.host ?? "0.0.0.0", + port: master.getListenAddress()?.port ?? config.port, cluster: true, workers: master.getReadyWorkerCount(), totalWorkers: master.getTargetWorkerCount(), + connections: master.getConnectionSnapshot(), }, workers: snapshotWorkers(), event: { type: "worker-ready", ...event }, @@ -1317,6 +1354,16 @@ async function startClusterMaster(rootDir: string): Promise { process.exit(1); }); }); + master.on("listener-error", (error: Error) => { + console.error("[cluster] TCP listener failed:", error); + void master.gracefulShutdown("TCP listener failure", 1); + }); + master.on("worker-metrics", () => { + void patchRuntimeSnapshotSafe(rootDir, { + runtimeIdentity, + summary: { connections: master.getConnectionSnapshot() }, + }); + }); master.on("worker-exit", (event) => { void patchRuntimeSnapshotSafe(rootDir, { runtimeIdentity, @@ -1324,6 +1371,7 @@ async function startClusterMaster(rootDir: string): Promise { state: master.getReadyWorkerCount() > 0 ? "ready" : "degraded", workers: master.getReadyWorkerCount(), totalWorkers: master.getTargetWorkerCount(), + connections: master.getConnectionSnapshot(), }, workers: snapshotWorkers(), event: { type: "worker-exit", ...event }, @@ -1336,6 +1384,7 @@ async function startClusterMaster(rootDir: string): Promise { state: "ready", workers: master.getReadyWorkerCount(), totalWorkers: master.getTargetWorkerCount(), + connections: master.getConnectionSnapshot(), }, workers: snapshotWorkers(), reload: { status: "success", ...event }, @@ -1357,20 +1406,22 @@ async function startClusterMaster(rootDir: string): Promise { runtimeIdentity, summary: { state: "ready", - host: config.host ?? "0.0.0.0", - port: config.port, + host: master.getListenAddress()?.host ?? config.host ?? "0.0.0.0", + port: master.getListenAddress()?.port ?? config.port, cluster: true, workers: readyWorkers, totalWorkers: master.getTargetWorkerCount(), + connections: master.getConnectionSnapshot(), built: isBuilt, }, workers: snapshotWorkers(), event: { type: "ready", - host: config.host ?? "0.0.0.0", - port: config.port, + host: master.getListenAddress()?.host ?? config.host ?? "0.0.0.0", + port: master.getListenAddress()?.port ?? config.port, workers: readyWorkers, totalWorkers: master.getTargetWorkerCount(), + connections: master.getConnectionSnapshot(), }, }); const parentReadyLog = isEnvFlagEnabled( @@ -1380,8 +1431,8 @@ async function startClusterMaster(rootDir: string): Promise { process.send({ type: "ready", server: { - host: config.host ?? "0.0.0.0", - port: config.port, + host: master.getListenAddress()?.host ?? config.host ?? "0.0.0.0", + port: master.getListenAddress()?.port ?? config.port, }, startupProfile: startupProfiler.toJSON(), detail: { @@ -1389,15 +1440,21 @@ async function startClusterMaster(rootDir: string): Promise { cluster: true, workers: readyWorkers, totalWorkers: master.getTargetWorkerCount(), + connections: master.getConnectionSnapshot(), }, }); } if (!parentReadyLog) { - printReadyLog(console, config.host ?? "0.0.0.0", config.port, { - prefix: "[vextjs]", - suffix: `(workers=${readyWorkers}/${master.getTargetWorkerCount()})`, - }); + printReadyLog( + console, + master.getListenAddress()?.host ?? config.host ?? "0.0.0.0", + master.getListenAddress()?.port ?? config.port, + { + prefix: "[vextjs]", + suffix: `(workers=${readyWorkers}/${master.getTargetWorkerCount()})`, + }, + ); } } diff --git a/src/lib/cluster/affinity-router.ts b/src/lib/cluster/affinity-router.ts new file mode 100644 index 00000000..a7e01849 --- /dev/null +++ b/src/lib/cluster/affinity-router.ts @@ -0,0 +1,41 @@ +import { createHash } from "node:crypto"; +import { isIP } from "node:net"; + +/** URL canonicalizes IPv6 compression; mapped IPv4 shares the IPv4 key. */ +export function normalizeSourceIP(address: string): string { + const version = isIP(address); + if (version === 4) return address; + if (version !== 6) throw new Error("Invalid TCP source IP"); + const zone = address.indexOf("%"); + const plain = zone < 0 ? address : address.slice(0, zone); + const canonical = new URL(`http://[${plain}]/`).hostname.slice(1, -1); + const mapped = /^::ffff:([a-f\d]+):([a-f\d]+)$/i.exec(canonical); + if (mapped) { + const high = parseInt(mapped[1]!, 16), + low = parseInt(mapped[2]!, 16); + return `${high >>> 8}.${high & 255}.${low >>> 8}.${low & 255}`; + } + return zone < 0 ? canonical : `${canonical}%${address.slice(zone + 1)}`; +} + +/** Rendezvous scoring over logical slots, never over PIDs or transient IDs. */ +export function selectAffinitySlot( + address: string, + slots: readonly number[], +): number | undefined { + const key = normalizeSourceIP(address); + let winner: number | undefined, score: Buffer | undefined; + for (const slot of slots) { + const next = createHash("sha256") + .update(key) + .update("\0") + .update(String(slot)) + .digest(); + const comparison = score ? Buffer.compare(next, score) : 1; + if (comparison > 0 || (comparison === 0 && slot < winner!)) { + winner = slot; + score = next; + } + } + return winner; +} diff --git a/src/lib/cluster/connection-policy.ts b/src/lib/cluster/connection-policy.ts new file mode 100644 index 00000000..8391b376 --- /dev/null +++ b/src/lib/cluster/connection-policy.ts @@ -0,0 +1,56 @@ +/** Shared bounds for the two-phase, paused socket handoff. */ +export const SOCKET_HANDOFF_POLICY = Object.freeze({ + version: 1, + timeout: 5_000, + maxPending: 1_024, + maxPendingPerWorker: 64, +}); + +export interface SocketWorkerContext { + version: number; + generation: string; + host: string; + port: number; +} + +export const SOCKET_WORKER_CONTEXT_ENV = "VEXT_SOCKET_WORKER_CONTEXT"; + +export const CLUSTER_WORKER_POLICY_ENV = "VEXT_CLUSTER_WORKER_POLICY"; +export interface ClusterWorkerPolicy { + sticky: "none" | "ip"; + workers: "auto" | "auto-1" | number; +} + +export function parseClusterWorkerPolicy(value: string): ClusterWorkerPolicy { + const policy = JSON.parse(value) as ClusterWorkerPolicy; + if ( + !policy || + !["none", "ip"].includes(policy.sticky) || + !( + policy.workers === "auto" || + policy.workers === "auto-1" || + (Number.isInteger(policy.workers) && + Number(policy.workers) >= 1 && + Number(policy.workers) <= 64) + ) + ) { + throw new Error("[cluster] invalid worker policy"); + } + return policy; +} + +export function parseSocketWorkerContext(value: string): SocketWorkerContext { + const context = JSON.parse(value) as SocketWorkerContext; + if ( + context.version !== SOCKET_HANDOFF_POLICY.version || + typeof context.generation !== "string" || + !context.generation || + typeof context.host !== "string" || + !context.host || + !Number.isInteger(context.port) || + context.port < 1 || + context.port > 65535 + ) + throw new Error("[cluster] invalid socket worker context"); + return context; +} diff --git a/src/lib/cluster/ipc-types.ts b/src/lib/cluster/ipc-types.ts index aad80267..1ee8f846 100644 --- a/src/lib/cluster/ipc-types.ts +++ b/src/lib/cluster/ipc-types.ts @@ -19,6 +19,31 @@ // ── Worker → Master 消息 ──────────────────────────────────── +export interface SocketPreparedMessage { + type: "socket-prepared"; + transferId: string; + generation: string; + accepted: boolean; +} + +export interface SocketOfferMessage { + type: "socket-offer"; + transferId: string; + generation: string; +} + +export interface SocketCommitMessage { + type: "socket-commit"; + transferId: string; + generation: string; +} + +export interface SocketCancelMessage { + type: "socket-cancel"; + transferId: string; + generation: string; +} + /** * Worker 就绪通知 * @@ -85,6 +110,7 @@ export interface WorkerRequestRestartMessage { * Worker → Master 消息联合类型 */ export type WorkerToMasterMessage = + | SocketPreparedMessage | WorkerReadyMessage | WorkerHeartbeatMessage | WorkerMetricsMessage @@ -150,6 +176,9 @@ export interface MasterBroadcastMessage { * Master → Worker 消息联合类型 */ export type MasterToWorkerMessage = + | SocketOfferMessage + | SocketCommitMessage + | SocketCancelMessage | MasterSetTitleMessage | MasterShutdownMessage | MasterHealthCheckMessage @@ -185,6 +214,9 @@ export interface WorkerMetrics { * 存储在 ClusterMaster.workers Map 中。 */ export interface WorkerMeta { + /** Stable affinity slot; process generations can change within this slot. */ + slotId?: number; + generation?: string; /** cluster.Worker.id */ id: number; /** Worker 启动时间戳（Date.now()） */ diff --git a/src/lib/cluster/master.ts b/src/lib/cluster/master.ts index 6d6734e3..35d0281c 100644 --- a/src/lib/cluster/master.ts +++ b/src/lib/cluster/master.ts @@ -1,3 +1,10 @@ +import { SocketDispatcher } from "./socket-dispatcher.js"; +import { WorkerSlots } from "./worker-slots.js"; +import { + SOCKET_WORKER_CONTEXT_ENV, + SOCKET_HANDOFF_POLICY, + CLUSTER_WORKER_POLICY_ENV, +} from "./connection-policy.js"; import cluster, { type Worker as ClusterWorker } from "node:cluster"; import { EventEmitter } from "node:events"; import { resolveWorkerCount } from "./worker-count.js"; @@ -6,6 +13,7 @@ import type { WorkerMeta, WorkerToMasterMessage, WorkerMetrics, + MasterToWorkerMessage, } from "./ipc-types.js"; function isEnvFlagEnabled(value: string | undefined): boolean { @@ -57,6 +65,8 @@ function shouldLogStartupLifecycle(): boolean { export interface ClusterMasterConfig { /** 内部宿主在所有 worker 停止后完成诊断落盘；异常不阻断关闭。 */ onStopped?: (exitCode: number) => Promise; + /** Actual public listener, owned by Master when sticky is ip. */ + listen?: { host: string; port: number }; /** Worker 数量配置 */ workers: "auto" | "auto-1" | number; /** Worker 崩溃后是否自动重启 */ @@ -90,7 +100,7 @@ export interface ClusterMasterConfig { pidFile: string; /** 进程标题前缀 */ titlePrefix: string; - /** 粘性会话模式 */ + /** TCP 源 IP 连接亲和性模式 */ sticky: "none" | "ip"; } @@ -165,6 +175,8 @@ export class ClusterMaster extends EventEmitter { /** Worker 元数据表（key = cluster.Worker.id） */ private workers = new Map(); private nextWorkerId = 1; + private readonly slots = new WorkerSlots(); + private dispatcher?: SocketDispatcher; /** 窗口内的重启时间戳（用于频率保护） */ private restartTimestamps: number[] = []; @@ -287,23 +299,50 @@ export class ClusterMaster extends EventEmitter { throw new Error(`[cluster] ${pidResult.error}`); } - // ── 配置 cluster 调度策略 ───────────────────────────── - // - // Linux + 非 sticky: SCHED_NONE - // → 内核 SO_REUSEPORT，由内核分配连接，性能最优 - // - // 其他情况: SCHED_RR (Round-Robin) - // → Node.js 内置负载均衡 - // - if (process.platform === "linux" && this.config.sticky === "none") { - cluster.schedulingPolicy = cluster.SCHED_NONE; + // Ordinary Cluster uses Node's shared-listener policy. IP mode owns TCP ingress. + if (this.config.sticky === "none") { + cluster.schedulingPolicy = + process.platform === "linux" ? cluster.SCHED_NONE : cluster.SCHED_RR; } else { - cluster.schedulingPolicy = cluster.SCHED_RR; - } - if (this.config.sticky === "ip") { - console.warn( - '[cluster] sticky="ip" is deprecated and does not provide IP affinity; using round-robin. Configure affinity in an external load balancer.', - ); + try { + if (!this.config.listen) + throw new Error( + '[cluster] sticky="ip" requires a public listen endpoint', + ); + this.dispatcher = new SocketDispatcher( + () => + [...this.slots.entries()].flatMap(([slot, id]) => { + const meta = this.workers.get(id), + worker = cluster.workers?.[id]; + return meta?.state === "ready" && + worker?.isConnected() && + !worker.isDead() && + !worker.process.killed + ? [{ slot, generation: meta.generation!, worker }] + : []; + }), + (worker) => { + // Preserve unexpected-exit/auto-restart semantics; never mark as a normal drain. + if (!worker.isDead() && !worker.process.killed) { + try { + worker.process.kill("SIGKILL"); + } catch { + if (!worker.isDead()) worker.kill("SIGKILL"); + } + } + }, + (error) => this.emit("listener-error", error), + ); + await this.dispatcher.listen( + this.config.listen.port, + this.config.listen.host, + ); + } catch (error) { + this.startupFailed = true; + this.isStarting = false; + this.cleanup(); + throw error; + } } if (shouldLogStartupLifecycle()) { @@ -329,7 +368,7 @@ export class ClusterMaster extends EventEmitter { // for (let i = 0; i < this.workerCount; i++) { try { - await this.forkWorker(); + await this.forkWorker({ slotId: i }); } catch (err) { if (i === 0) { // 首个 Worker 失败 → Fail Fast @@ -358,6 +397,7 @@ export class ClusterMaster extends EventEmitter { const readyCount = this.getReadyWorkerCount(); this.isStarting = false; + if (readyCount > 0) this.dispatcher?.open(); this.checkAllDead(); if (shouldLogStartupLifecycle()) { console.log( @@ -382,6 +422,7 @@ export class ClusterMaster extends EventEmitter { async gracefulShutdown(trigger: string, exitCode = 0): Promise { if (this.isShuttingDown) return; this.isShuttingDown = true; + this.dispatcher?.beginShutdown(); for (const timer of this.pendingRestarts) clearTimeout(timer); this.pendingRestarts.clear(); @@ -400,7 +441,7 @@ export class ClusterMaster extends EventEmitter { if (worker && !worker.isDead()) { const meta = this.workers.get(id); if (meta) meta.state = "draining"; - worker.send({ + this.sendControl(worker, { type: "shutdown", timeout: this.config.reload.shutdownTimeout, }); @@ -452,7 +493,15 @@ export class ClusterMaster extends EventEmitter { this.isReloading = true; - const oldWorkerIds = [...this.workers.keys()]; + // Snapshot only current ready owners, excluding recovery/replacement candidates. + const oldWorkerIds = [...this.workers.values()] + .filter( + (meta) => + meta.state === "ready" && + (meta.slotId === undefined || + this.slots.owner(meta.slotId) === meta.id), + ) + .map((meta) => meta.id); console.log( `[cluster] rolling restart triggered by ${trigger}, replacing ${oldWorkerIds.length} workers`, ); @@ -472,9 +521,21 @@ export class ClusterMaster extends EventEmitter { break; } + const oldMeta = this.workers.get(oldId); + // Another slot may already have crashed/recovered while the prior slot drained. + if ( + !oldMeta || + oldMeta.state !== "ready" || + (oldMeta.slotId !== undefined && + this.slots.owner(oldMeta.slotId) !== oldId) + ) + continue; try { // Fork 新 Worker - const _newWorker = await this.forkWorker(); + const _newWorker = await this.forkWorker({ + slotId: oldMeta.slotId, + replaceWorkerId: oldId, + }); // 新 Worker ready → 通知旧 Worker shutdown const oldWorker = cluster.workers?.[oldId]; @@ -482,7 +543,7 @@ export class ClusterMaster extends EventEmitter { const oldMeta = this.workers.get(oldId); if (oldMeta) oldMeta.state = "draining"; - oldWorker.send({ + this.sendControl(oldWorker, { type: "shutdown", timeout: this.config.reload.shutdownTimeout, }); @@ -528,7 +589,7 @@ export class ClusterMaster extends EventEmitter { for (const [id] of this.workers) { const worker = cluster.workers?.[id]; if (worker && !worker.isDead()) { - worker.send({ type: "broadcast", payload }); + this.sendControl(worker, { type: "broadcast", payload }); } } } @@ -572,9 +633,17 @@ export class ClusterMaster extends EventEmitter { return !this.isShuttingDown; } - /** - * getTargetWorkerCount — 获取配置计算出的目标 Worker 数量 - */ + /** Bounded handoff counters, sampled by the existing runtime metrics path. */ + getConnectionSnapshot() { + return this.dispatcher?.snapshot() ?? { mode: "none", pending: 0 }; + } + + /** Master's bound public endpoint in IP mode. */ + getListenAddress() { + return this.dispatcher?.address() ?? this.config.listen; + } + + /** Actual target Worker count for this Master lifecycle. */ getTargetWorkerCount(): number { return this.workerCount; } @@ -603,19 +672,39 @@ export class ClusterMaster extends EventEmitter { * @throws Worker 在 readyTimeout 内未就绪或在就绪前退出 */ private async forkWorker( - options: { retryOnEarlyExit?: boolean } = {}, + options: { + retryOnEarlyExit?: boolean; + slotId?: number; + replaceWorkerId?: number; + } = {}, ): Promise { this.activeForks++; try { const nextId = this.nextWorkerId++; + const slotId = options.slotId ?? nextId - 1; + const generation = String(nextId); + const endpoint = this.dispatcher?.address(); const worker = cluster.fork({ VEXT_WORKER_ID: String(nextId), VEXT_MODE: "start", + [CLUSTER_WORKER_POLICY_ENV]: JSON.stringify({ + sticky: this.config.sticky, + workers: this.config.workers, + }), + [SOCKET_WORKER_CONTEXT_ENV]: endpoint + ? JSON.stringify({ + version: SOCKET_HANDOFF_POLICY.version, + generation, + ...endpoint, + }) + : "", }); const meta: WorkerMeta = { id: worker.id, + slotId, + generation, startTime: Date.now(), restartCount: 0, lastHeartbeat: Date.now(), @@ -627,14 +716,10 @@ export class ClusterMaster extends EventEmitter { } // 设置 Worker 进程标题 - try { - worker.send({ - type: "set-title", - title: `${this.config.titlePrefix}:worker:${nextId}`, - }); - } catch { - // Worker 可能在 send 之前就退出了 - } + this.sendControl(worker, { + type: "set-title", + title: `${this.config.titlePrefix}:worker:${nextId}`, + }); // 注册 IPC 消息监听 worker.on("message", (msg: unknown) => @@ -643,6 +728,22 @@ export class ClusterMaster extends EventEmitter { // 等待 Worker 就绪 await this.waitForWorkerReady(worker); + const owner = this.slots.owner(slotId); + if ( + this.isShuttingDown || + (owner !== undefined && owner !== options.replaceWorkerId) + ) { + this.terminateUnreadyWorker(worker); + throw new Error( + `worker ${worker.id} affinity slot is already owned or shutting down`, + ); + } + // One synchronous commit; candidates never add an extra routable slot. + this.slots.activate(slotId, worker.id); + if (owner !== undefined) { + const old = this.workers.get(owner); + if (old) old.state = "draining"; + } this.ownedStartupWorkerIds.delete(worker.id); this.capacityLossEmitted = false; @@ -694,6 +795,23 @@ export class ClusterMaster extends EventEmitter { ) { return; } + const endpoint = this.dispatcher?.address(); + const reported = (msg as { server?: { host: string; port: number } }) + .server; + if ( + endpoint && + (reported?.host !== endpoint.host || reported.port !== endpoint.port) + ) { + settled = true; + this.terminateUnreadyWorker(worker); + cleanup(); + reject( + new Error( + `worker ${worker.id} did not initialize the Master socket endpoint`, + ), + ); + return; + } settled = true; cleanup(); const meta = this.workers.get(worker.id); @@ -796,6 +914,10 @@ export class ClusterMaster extends EventEmitter { const message = msg as WorkerToMasterMessage; switch (message.type) { + case "socket-prepared": { + this.dispatcher?.prepared(worker, message); + break; + } case "heartbeat": { const meta = this.workers.get(worker.id); if (meta) { @@ -806,6 +928,7 @@ export class ClusterMaster extends EventEmitter { case "metrics": { this.workerMetrics.set(worker.id, message.data); + this.emit("worker-metrics", { workerId: worker.id }); break; } @@ -828,6 +951,21 @@ export class ClusterMaster extends EventEmitter { } } + private sendControl( + worker: ClusterWorker, + message: MasterToWorkerMessage, + ): void { + if (worker.isDead() || !worker.isConnected() || worker.process.killed) + return; + try { + // A callback also consumes asynchronous channel errors. Existing readiness, + // heartbeat and shutdown deadlines still own recovery and process cleanup. + worker.send(message, () => {}); + } catch { + // The process can exit between the connection check and send. + } + } + /** * replaceWorker — 替换单个 Worker（用于 request-restart 场景） * @@ -839,14 +977,19 @@ export class ClusterMaster extends EventEmitter { */ private async replaceWorker(oldWorker: ClusterWorker): Promise { if (this.isShuttingDown || this.isReloading) return; + const oldMeta = this.workers.get(oldWorker.id); + if (!oldMeta || oldMeta.state !== "ready") return; try { - await this.forkWorker(); + await this.forkWorker({ + slotId: oldMeta.slotId, + replaceWorkerId: oldWorker.id, + }); if (!oldWorker.isDead()) { const meta = this.workers.get(oldWorker.id); if (meta) meta.state = "draining"; - oldWorker.send({ + this.sendControl(oldWorker, { type: "shutdown", timeout: this.config.reload.shutdownTimeout, }); @@ -882,6 +1025,8 @@ export class ClusterMaster extends EventEmitter { const wasOwnedStartupCandidate = this.ownedStartupWorkerIds.delete( worker.id, ); + if (meta?.slotId !== undefined) this.slots.remove(meta.slotId, worker.id); + this.dispatcher?.workerExited(worker.id); this.workers.delete(worker.id); this.workerMetrics.delete(worker.id); @@ -945,8 +1090,15 @@ export class ClusterMaster extends EventEmitter { const timer = setTimeout(async () => { this.pendingRestarts.delete(timer); if (this.isShuttingDown) return; + if ( + meta?.slotId !== undefined && + this.slots.owner(meta.slotId) !== undefined + ) { + this.checkAllDead(); + return; + } try { - await this.forkWorker({ retryOnEarlyExit: true }); + await this.forkWorker({ retryOnEarlyExit: true, slotId: meta?.slotId }); } catch (err) { console.error( `[cluster] failed to spawn replacement: ${(err as Error).message}`, @@ -1189,6 +1341,7 @@ export class ClusterMaster extends EventEmitter { * 关闭定时器 + 移除信号处理器 + 删除 PID 文件。 */ private cleanup(): void { + this.dispatcher?.beginShutdown(); this.stopHealthCheck(); this.removeSignalHandlers(); this.removeWorkerExitListenerIfTerminated(); diff --git a/src/lib/cluster/socket-dispatcher.ts b/src/lib/cluster/socket-dispatcher.ts new file mode 100644 index 00000000..3b5c284e --- /dev/null +++ b/src/lib/cluster/socket-dispatcher.ts @@ -0,0 +1,238 @@ +import { createServer, type Server, type Socket } from "node:net"; +import type { Worker } from "node:cluster"; +import { randomUUID } from "node:crypto"; +import { selectAffinitySlot } from "./affinity-router.js"; +import { SOCKET_HANDOFF_POLICY } from "./connection-policy.js"; +import type { SocketPreparedMessage } from "./ipc-types.js"; + +export interface SocketTarget { + slot: number; + generation: string; + worker: Worker; +} +interface Transfer { + socket: Socket; + target: SocketTarget; + timer: ReturnType; +} + +/** Accepts paused sockets; never reads HTTP or replays a handed-off connection. */ +export class SocketDispatcher { + private readonly server: Server; + private readonly transfers = new Map(); + private readonly perWorker = new Map(); + private readonly prefix = randomUUID(); + private sequence = 0; + private accepting = false; + private stopped = false; + private endpoint?: { host: string; port: number }; + private readonly counters = { + committed: 0, + rejected: 0, + timedOut: 0, + sendFailed: 0, + backpressure: 0, + }; + + constructor( + private readonly targets: () => SocketTarget[], + private readonly unresponsive: (worker: Worker) => void, + onError?: (error: Error) => void, + ) { + this.server = createServer({ pauseOnConnect: true }, (socket) => + this.dispatch(socket), + ); + // After successful binding, listener errors are observed by the host. + this.server.on("error", (error) => { + this.accepting = false; + if (this.endpoint) onError?.(error); + }); + } + + async listen( + port: number, + host: string, + ): Promise<{ host: string; port: number }> { + await new Promise((resolve, reject) => { + const error = (err: Error) => { + this.server.off("listening", ready); + reject(err); + }; + const ready = () => { + this.server.off("error", error); + resolve(); + }; + this.server.once("error", error); + this.server.once("listening", ready); + this.server.listen(port, host); + }); + const address = this.server.address(); + if (!address || typeof address === "string") + throw new Error("Missing TCP listener address"); + this.endpoint = { host: address.address, port: address.port }; + return this.endpoint; + } + + open(): void { + if (!this.stopped) this.accepting = true; + } + address(): { host: string; port: number } | undefined { + return this.endpoint; + } + snapshot() { + return { mode: "ip", pending: this.transfers.size, ...this.counters }; + } + + prepared(worker: Worker, message: SocketPreparedMessage): void { + const transfer = this.transfers.get(message.transferId); + if ( + !transfer || + transfer.target.worker.id !== worker.id || + transfer.target.generation !== message.generation + ) + return; + if ( + !message.accepted || + this.stopped || + !worker.isConnected() || + worker.process.killed + ) { + this.counters.rejected++; + this.cancel(message.transferId); + return; + } + try { + worker.send( + { + type: "socket-commit", + transferId: message.transferId, + generation: message.generation, + }, + (error: Error | null) => { + if (!this.transfers.has(message.transferId)) return; + if (error) { + this.counters.sendFailed++; + this.unresponsive(worker); + this.cancel(message.transferId); + } else { + this.counters.committed++; + this.finish(message.transferId); + } + }, + ); + } catch { + this.counters.sendFailed++; + this.unresponsive(worker); + this.cancel(message.transferId); + } + } + + workerExited(workerId: number): void { + for (const [id, transfer] of this.transfers) { + if (transfer.target.worker.id === workerId) this.finish(id); + } + } + + beginShutdown(): void { + if (this.stopped) return; + this.stopped = true; + this.accepting = false; + // Start close before Worker IPC disconnect; never await before draining Workers. + this.server.close(() => {}); + for (const id of this.transfers.keys()) this.cancel(id); + } + + private dispatch(socket: Socket): void { + socket.on("error", () => {}); + const targets = this.targets(); + let selected: number | undefined; + try { + selected = socket.remoteAddress + ? selectAffinitySlot( + socket.remoteAddress, + targets.map((t) => t.slot), + ) + : undefined; + } catch { + /* Invalid/missing TCP address: reject rather than invent a shared key. */ + } + const target = targets.find((t) => t.slot === selected); + if ( + !this.accepting || + !target || + this.transfers.size >= SOCKET_HANDOFF_POLICY.maxPending || + (this.perWorker.get(target.worker.id) ?? 0) >= + SOCKET_HANDOFF_POLICY.maxPendingPerWorker + ) { + this.counters.rejected++; + socket.destroy(); + return; + } + const id = `${this.prefix}:${++this.sequence}`; + const timer = setTimeout(() => { + this.counters.timedOut++; + // A stopped/event-loop-blocked Worker cannot run its own deadline. Exclude + // and terminate it before releasing credits, so its IPC queue stays bounded. + this.unresponsive(target.worker); + this.cancel(id); + }, SOCKET_HANDOFF_POLICY.timeout); + timer.unref(); + this.transfers.set(id, { socket, target, timer }); + this.perWorker.set( + target.worker.id, + (this.perWorker.get(target.worker.id) ?? 0) + 1, + ); + try { + const available = target.worker.send( + { type: "socket-offer", transferId: id, generation: target.generation }, + socket, + { keepOpen: false }, + (error: Error | null) => { + if (error && this.transfers.has(id)) { + this.counters.sendFailed++; + this.unresponsive(target.worker); + this.cancel(id); + } + }, + ); + if (!available) this.counters.backpressure++; + } catch { + this.counters.sendFailed++; + this.unresponsive(target.worker); + this.cancel(id); + } + } + + private cancel(id: string): void { + const transfer = this.transfers.get(id); + if (!transfer) return; + try { + if ( + transfer.target.worker.isConnected() && + !transfer.target.worker.process.killed + ) + transfer.target.worker.send( + { + type: "socket-cancel", + transferId: id, + generation: transfer.target.generation, + }, + () => {}, + ); + } catch { + /* Receiver's independent prepare deadline still destroys the socket. */ + } + this.finish(id); + } + + private finish(id: string): void { + const transfer = this.transfers.get(id); + if (!transfer) return; + this.transfers.delete(id); + clearTimeout(transfer.timer); + transfer.socket.destroy(); + const count = (this.perWorker.get(transfer.target.worker.id) ?? 1) - 1; + if (count) this.perWorker.set(transfer.target.worker.id, count); + else this.perWorker.delete(transfer.target.worker.id); + } +} diff --git a/src/lib/cluster/socket-receiver.ts b/src/lib/cluster/socket-receiver.ts new file mode 100644 index 00000000..1eb546fc --- /dev/null +++ b/src/lib/cluster/socket-receiver.ts @@ -0,0 +1,103 @@ +import { Socket } from "node:net"; +import type { VextServerHandle } from "../../types/adapter.js"; +import { SOCKET_HANDOFF_POLICY } from "./connection-policy.js"; +import type { + MasterToWorkerMessage, + SocketPreparedMessage, +} from "./ipc-types.js"; + +/** Worker owns paused offers until commit, with an independent bounded deadline. */ +export class SocketReceiver { + private readonly offers = new Map< + string, + { socket: Socket; deadline: number; timer: ReturnType } + >(); + private closed = false; + constructor( + private readonly generation: string, + private readonly handle: VextServerHandle, + private readonly reply: (message: SocketPreparedMessage) => void, + ) {} + + message(message: MasterToWorkerMessage, socket?: unknown): void { + if ( + message.type !== "socket-offer" && + message.type !== "socket-commit" && + message.type !== "socket-cancel" + ) + return; + if ( + message.generation !== this.generation || + typeof message.transferId !== "string" + ) { + if (socket instanceof Socket) socket.destroy(); + return; + } + const id = message.transferId; + if (message.type === "socket-offer") { + const accepted = + !this.closed && + socket instanceof Socket && + !socket.destroyed && + !!this.handle.receiveSocket && + !this.offers.has(id) && + this.offers.size < SOCKET_HANDOFF_POLICY.maxPendingPerWorker; + if (accepted) { + socket.pause(); + const timer = setTimeout( + () => this.cancel(id), + SOCKET_HANDOFF_POLICY.timeout, + ); + timer.unref(); + this.offers.set(id, { + socket, + deadline: Date.now() + SOCKET_HANDOFF_POLICY.timeout, + timer, + }); + socket.once("close", () => { + if (this.offers.get(id)?.socket === socket) this.remove(id); + }); + socket.on("error", () => {}); + } else if (socket instanceof Socket) socket.destroy(); + try { + this.reply({ + type: "socket-prepared", + transferId: id, + generation: this.generation, + accepted, + }); + } catch { + this.cancel(id); + } + } else if (message.type === "socket-cancel") this.cancel(id); + else { + const offer = this.remove(id); + if (!offer) return; // Expired/duplicate commit cannot resurrect an offer. + if (this.closed || Date.now() >= offer.deadline) { + offer.socket.destroy(); + return; + } + try { + this.handle.receiveSocket!(offer.socket); + } catch { + offer.socket.destroy(); + } + } + } + + close(): void { + this.closed = true; + for (const id of this.offers.keys()) this.cancel(id); + } + private cancel(id: string): void { + this.remove(id)?.socket.destroy(); + } + private remove(id: string) { + const offer = this.offers.get(id); + if (offer) { + clearTimeout(offer.timer); + this.offers.delete(id); + } + return offer; + } +} diff --git a/src/lib/cluster/worker-slots.ts b/src/lib/cluster/worker-slots.ts new file mode 100644 index 00000000..688bcb53 --- /dev/null +++ b/src/lib/cluster/worker-slots.ts @@ -0,0 +1,16 @@ +/** Identity only: readiness and health remain in ClusterMaster.WorkerMeta. */ +export class WorkerSlots { + private readonly current = new Map(); + activate(slot: number, workerId: number): void { + this.current.set(slot, workerId); + } + remove(slot: number, workerId: number): void { + if (this.current.get(slot) === workerId) this.current.delete(slot); + } + entries(): IterableIterator<[number, number]> { + return this.current.entries(); + } + owner(slot: number): number | undefined { + return this.current.get(slot); + } +} diff --git a/src/lib/cluster/worker.ts b/src/lib/cluster/worker.ts index dcefd460..6555a846 100644 --- a/src/lib/cluster/worker.ts +++ b/src/lib/cluster/worker.ts @@ -1,3 +1,13 @@ +import cluster from "node:cluster"; +import { SocketReceiver } from "./socket-receiver.js"; +import { + parseSocketWorkerContext, + SOCKET_WORKER_CONTEXT_ENV, + parseClusterWorkerPolicy, + CLUSTER_WORKER_POLICY_ENV, + type SocketWorkerContext, + type ClusterWorkerPolicy, +} from "./connection-policy.js"; import type { VextApp } from "../../types/app.js"; import type { AppInternals } from "../app.js"; import type { VextServerHandle } from "../../types/adapter.js"; @@ -107,6 +117,7 @@ interface WorkerContext { app: VextApp | null; internals: AppInternals | null; serverHandle: VextServerHandle | null; + socketReceiver?: SocketReceiver; /** 是否已触发关闭 */ isShuttingDown: boolean; } @@ -146,7 +157,11 @@ interface WorkerContext { */ export async function workerMain( rootDir: string, - bootstrapFn: (rootDir: string) => Promise<{ + bootstrapFn: ( + rootDir: string, + socketContext?: SocketWorkerContext, + expectedPolicy?: ClusterWorkerPolicy, + ) => Promise<{ app: VextApp; serverHandle: VextServerHandle; internals: AppInternals; @@ -180,10 +195,39 @@ export async function workerMain( try { // ── 1. 执行标准 bootstrap ──────────────────────────── - const result = await bootstrapFn(rootDir); + const serializedContext = cluster.isWorker + ? process.env[SOCKET_WORKER_CONTEXT_ENV] + : undefined; + const socketContext = serializedContext + ? parseSocketWorkerContext(serializedContext) + : undefined; + const serializedPolicy = cluster.isWorker + ? process.env[CLUSTER_WORKER_POLICY_ENV] + : undefined; + const expectedPolicy = serializedPolicy + ? parseClusterWorkerPolicy(serializedPolicy) + : undefined; + const result = expectedPolicy + ? await bootstrapFn(rootDir, socketContext, expectedPolicy) + : await bootstrapFn(rootDir); ctx.app = result.app; ctx.internals = result.internals; ctx.serverHandle = result.serverHandle; + if (socketContext) { + if (!result.serverHandle.receiveSocket || !result.serverHandle.forceClose) + throw new Error( + "[cluster] adapter did not return socket handoff controls", + ); + ctx.socketReceiver = new SocketReceiver( + socketContext.generation, + result.serverHandle, + (message) => { + process.send?.(message, (error: Error | null) => { + if (error) shutdownWorker(ctx, config); + }); + }, + ); + } // ── 2. 注册 IPC 消息处理器 ────────────────────────── registerIPCHandlers(ctx, config); @@ -245,11 +289,12 @@ export async function workerMain( * - broadcast: 转发给 app 事件系统（后续扩展用） */ function registerIPCHandlers(ctx: WorkerContext, config: WorkerConfig): void { - process.on("message", (msg: unknown) => { + process.on("message", (msg: unknown, handle?: unknown) => { if (typeof msg !== "object" || msg === null) return; const message = msg as MasterToWorkerMessage; + ctx.socketReceiver?.message(message, handle); switch (message.type) { case "set-title": { process.title = message.title; @@ -301,6 +346,7 @@ function registerIPCHandlers(ctx: WorkerContext, config: WorkerConfig): void { function shutdownWorker(ctx: WorkerContext, config: WorkerConfig): void { if (ctx.isShuttingDown) return; ctx.isShuttingDown = true; + ctx.socketReceiver?.close(); console.log(`[worker:${config.workerId}] shutting down...`); @@ -466,7 +512,7 @@ function sendToMaster( | WorkerRequestRestartMessage, ): void { try { - process.send?.(msg); + process.send?.(msg, () => {}); } catch { // IPC channel 已关闭或其他通信错误 // 不抛出 — 避免影响 Worker 正常运行 diff --git a/src/lib/socket-http-server.ts b/src/lib/socket-http-server.ts new file mode 100644 index 00000000..14d4f36f --- /dev/null +++ b/src/lib/socket-http-server.ts @@ -0,0 +1,142 @@ +import { + Server, + createServer, + type RequestListener, + type ServerOptions, +} from "node:http"; +import { isIP, type AddressInfo, type Socket } from "node:net"; +import type { + VextAdapterRuntimeContext, + VextServerHandle, +} from "../types/adapter.js"; + +/** Node HTTP parser/timeouts with a logical listener receiving IPC sockets. */ +export class SocketHTTPServer extends Server { + private opened = false; + private draining = false; + private closed = false; + private readonly sockets = new Set(); + private readonly callbacks: Array<(error?: Error) => void> = []; + private finishing = false; + private readonly endpoint: AddressInfo; + + constructor( + options: ServerOptions, + handler: RequestListener, + endpoint: { host: string; port: number }, + ) { + super(options, handler); + this.endpoint = { + address: endpoint.host, + port: endpoint.port, + family: isIP(endpoint.host) === 6 ? "IPv6" : "IPv4", + }; + this.prependListener("request", (_request, response) => { + response.once("finish", () => { + if (this.draining) this.closeIdleConnections(); + }); + }); + } + + override listening = false; + override address(): AddressInfo | null { + return this.listening ? this.endpoint : null; + } + override listen(...args: unknown[]): this { + if (this.opened || this.draining) + throw new Error("Socket HTTP server cannot listen twice"); + this.opened = true; + this.listening = true; + const callback = args.at(-1); + process.nextTick(() => { + if (!this.listening) return; // An immediate close must not start HTTP timers. + // Public lifecycle event initializes Node's HTTP connection tracker. + this.emit("listening"); + if (typeof callback === "function") callback.call(this); + }); + return this; + } + + receiveSocket(socket: Socket): void { + if (!this.listening || socket.destroyed) { + socket.destroy(); + throw new Error("Socket HTTP receiver is closing"); + } + this.sockets.add(socket); + socket.once("close", () => { + this.sockets.delete(socket); + this.finishClose(); + }); + try { + this.emit("connection", socket); + socket.resume(); + } catch (error) { + socket.destroy(); + throw error; + } + } + + override getConnections( + callback: (error: Error | null, count: number) => void, + ): this { + process.nextTick(callback, null, this.sockets.size); + return this; + } + + override close(callback?: (error?: Error) => void): this { + if (this.closed) { + if (callback) process.nextTick(callback); + return this; + } + if (callback) this.callbacks.push(callback); + this.draining = true; + this.listening = false; + this.closeIdleConnections(); + this.finishClose(); + return this; + } + + forceClose(): void { + this.draining = true; + this.listening = false; + for (const socket of this.sockets) socket.destroy(); + this.finishClose(); + } + + private finishClose(): void { + if (!this.draining || this.sockets.size || this.finishing || this.closed) + return; + this.finishing = true; + // Also disposes the HTTP connection checking interval. No OS listener exists. + super.close((error?: Error) => { + this.closed = true; + const result = + (error as NodeJS.ErrnoException | undefined)?.code === + "ERR_SERVER_NOT_RUNNING" + ? undefined + : error; + for (const callback of this.callbacks.splice(0)) callback(result); + }); + } +} + +export function createAdapterHTTPServer( + context: VextAdapterRuntimeContext | undefined, + options: ServerOptions, + handler: RequestListener, +): Server { + return context?.socketHandoff + ? new SocketHTTPServer(options, handler, context.socketHandoff) + : createServer(options, handler); +} + +export function socketHandoffControls( + server: Server, +): Pick { + return server instanceof SocketHTTPServer + ? { + receiveSocket: (socket) => server.receiveSocket(socket), + forceClose: () => server.forceClose(), + } + : {}; +} diff --git a/src/types/adapter.ts b/src/types/adapter.ts index 05701922..a767b7c8 100644 --- a/src/types/adapter.ts +++ b/src/types/adapter.ts @@ -6,6 +6,16 @@ export interface VextAdapterListenOptions { server?: VextServerConfig; } +/** Per-instance context supplied by the actual Cluster Worker startup path. */ +export interface VextAdapterRuntimeContext { + socketHandoff?: { host: string; port: number }; +} + +export type VextAdapterFactory = ( + app: import("./app.js").VextApp, + context?: VextAdapterRuntimeContext, +) => VextAdapter; + /** * VextAdapter — 框架底层适配器接口 * @@ -34,6 +44,8 @@ export interface VextAdapterListenOptions { * - 优雅关闭编排（bootstrap/shutdown 职责） */ export interface VextAdapter { + /** Explicit opt-in: listen returns receiveSocket/forceClose in handoff mode. */ + readonly supportsSocketHandoff?: boolean; /** adapter 名称标识（用于日志和错误信息） */ readonly name: string; @@ -131,6 +143,10 @@ export interface VextAdapter { * shutdown 流程通过 close() 停止接受新连接并等待飞行中请求完成。 */ export interface VextServerHandle { + /** Accept a committed, paused IPC socket (Cluster sticky:ip only). */ + receiveSocket?(socket: import("node:net").Socket): void; + /** Synchronously release owned sockets when the shared shutdown budget expires. */ + forceClose?(): void; /** * 停止接受新连接，等待飞行中请求完成后关闭 * diff --git a/src/types/app.ts b/src/types/app.ts index 5885534d..452c6895 100644 --- a/src/types/app.ts +++ b/src/types/app.ts @@ -1,4 +1,4 @@ -import type { VextAdapter } from "./adapter.js"; +import type { VextAdapter, VextAdapterFactory } from "./adapter.js"; import type { VextMiddleware, VextHandler } from "./middleware.js"; import type { VextHooks } from "./hooks.js"; import type { VextFetch, VextFetchConfig } from "../lib/fetch.js"; @@ -1005,7 +1005,7 @@ export interface VextClusterConfig { /** 进程标题前缀 @default 'vext' */ titlePrefix: string; - /** 默认 none；ip 是兼容保留值，使用 RR 并诊断，未实现 IP 粘性。@default 'none' */ + /** 默认 none；ip 按 TCP 源 IP 将独立连接分配给稳定 Worker 槽位。@default 'none' */ sticky: "none" | "ip"; } @@ -1142,7 +1142,7 @@ export interface VextConfig { * 函数：adapter 工厂函数（如 fastifyAdapter({ bodyLimit: 5MB })） * 对象：第三方 adapter 实例（必须实现 VextAdapter 接口） */ - adapter: string | ((app: VextApp) => VextAdapter) | VextAdapter; + adapter: string | VextAdapterFactory | VextAdapter; /** 是否信任代理（影响 req.ip / req.protocol 从 X-Forwarded-* 读取） */ trustProxy: boolean; diff --git a/src/types/index.ts b/src/types/index.ts index 9751f9d6..6b2389a9 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -98,6 +98,8 @@ export type { // ── Adapter 类型 ──────────────────────────────────────────── export type { VextAdapter, + VextAdapterRuntimeContext, + VextAdapterFactory, VextAdapterListenOptions, VextServerHandle, } from "./adapter.js"; diff --git a/test/fixtures/public-surface/coverage.json b/test/fixtures/public-surface/coverage.json index 56ac1232..be69ca0f 100644 --- a/test/fixtures/public-surface/coverage.json +++ b/test/fixtures/public-surface/coverage.json @@ -292,7 +292,8 @@ "config:shutdown.onFatalError", "config:shutdown.timeout", "config:trustProxy", - "symbol:vextjs#VextPluginConfig" + "symbol:vextjs#VextPluginConfig", + "config:adapter.supportsSocketHandoff" ] }, { @@ -863,7 +864,9 @@ "symbol:vextjs/adapters/koa#koaAdapter", "symbol:vextjs/adapters/native#NativeAdapterOptions", "symbol:vextjs/adapters/native#createNativeAdapter", - "symbol:vextjs/adapters/native#nativeAdapter" + "symbol:vextjs/adapters/native#nativeAdapter", + "symbol:vextjs#VextAdapterFactory", + "symbol:vextjs#VextAdapterRuntimeContext" ] }, { diff --git a/test/integration/cluster/affinity-fixture.mjs b/test/integration/cluster/affinity-fixture.mjs new file mode 100644 index 00000000..5c4fd1fb --- /dev/null +++ b/test/integration/cluster/affinity-fixture.mjs @@ -0,0 +1,118 @@ +// Real ClusterMaster/workerMain and HTTP adapters, controlled through portable IPC. +import cluster from "node:cluster"; +import { fileURLToPath } from "node:url"; +import { ClusterMaster } from "../../../dist/lib/cluster/master.js"; +import { workerMain } from "../../../dist/lib/cluster/worker.js"; +import { createApp, DEFAULT_CONFIG } from "../../../dist/lib/app.js"; +import { resolveAdapter } from "../../../dist/lib/adapter-resolver.js"; + +const adapterName = process.env.AFFINITY_ADAPTER ?? "native"; +const factory = process.env.AFFINITY_FACTORY === "1"; +const entry = fileURLToPath(import.meta.url); + +if (cluster.isWorker) { + await workerMain(process.cwd(), async (_root, context) => { + if ( + process.env.AFFINITY_FAIL_CANDIDATE === "1" && + Number(process.env.VEXT_WORKER_ID) > 2 + ) + throw new Error("candidate failure"); + const config = { + ...DEFAULT_CONFIG, + _testMode: false, + logger: { ...DEFAULT_CONFIG.logger, level: "silent" }, + adapter: factory + ? (await import(`../../../dist/adapters/${adapterName}/index.js`))[ + `${adapterName}Adapter` + ]() + : adapterName, + cluster: { ...DEFAULT_CONFIG.cluster, sticky: "ip", workers: 2 }, + shutdown: { ...DEFAULT_CONFIG.shutdown, timeout: 500 }, + }; + const { app, internals } = createApp(config); + if ( + process.env.AFFINITY_SLOW_CLOSE === "1" && + process.env.VEXT_WORKER_ID === "1" + ) { + app.onClose(() => new Promise((resolve) => setTimeout(resolve, 350))); + } + app.adapter = await resolveAdapter(config, app, { socketHandoff: context }); + app.adapter.registerRoute("GET", "/", [ + async (request, response) => { + const delay = Number(request.query.delay ?? 0); + if (delay) { + process.send?.({ type: "test-in-flight" }); + await new Promise((resolve) => setTimeout(resolve, delay)); + } + response.rawJson({ + pid: process.pid, + generation: process.env.VEXT_WORKER_ID, + remote: request.ip, + }); + }, + ]); + const serverHandle = await app.adapter.listen(context.port, context.host); + return { app, internals, serverHandle }; + }); +} else { + cluster.setupPrimary({ exec: entry, execArgv: [] }); + const master = new ClusterMaster({ + workers: 2, + sticky: "ip", + listen: { host: "127.0.0.1", port: 0 }, + pidFile: process.env.AFFINITY_PID_FILE, + healthCheck: { enabled: false }, + autoRestart: process.env.AFFINITY_AUTO_RESTART !== "0", + restartBaseDelay: 20, + restartMaxDelay: 50, + reload: { workerDelay: 0, readyTimeout: 3000, shutdownTimeout: 700 }, + }); + master.on( + "all-workers-dead", + () => void master.gracefulShutdown("capacity loss", 1), + ); + master.on("worker-ready", ({ workerId }) => { + cluster.workers[workerId]?.on("message", (message) => { + if (message.type === "test-in-flight") + process.send?.({ type: "in-flight" }); + }); + }); + for (const event of ["worker-ready", "worker-exit", "reload-complete"]) { + master.on(event, (data) => process.send?.({ type: event, data })); + } + process.on("message", async (message) => { + try { + if (message.type === "reload") { + await master.rollingRestart("test IPC"); + process.send?.({ type: "reloaded" }); + } else if (message.type === "crash") { + cluster.workers[message.id]?.process.kill("SIGKILL"); + } else if (message.type === "disconnect-and-stop") { + const worker = cluster.workers[message.id]; + worker.once("disconnect", () => { + void master.gracefulShutdown("disconnected Worker IPC"); + }); + worker.process.disconnect(); + } else if (message.type === "snapshot") { + process.send?.({ + type: "snapshot", + workers: [...master.getWorkerMetas().values()], + connections: master.getConnectionSnapshot(), + }); + } + } catch (error) { + process.send?.({ type: "error", error: error.message }); + } + }); + try { + await master.start(); + process.send?.({ + type: "listening", + endpoint: master.getListenAddress(), + workers: [...master.getWorkerMetas().values()], + }); + } catch (error) { + process.send?.({ type: "error", error: error.message }); + process.exit(1); + } +} diff --git a/test/integration/cluster/capacity-loss.test.ts b/test/integration/cluster/capacity-loss.test.ts index c6d4fa07..68c566f8 100644 --- a/test/integration/cluster/capacity-loss.test.ts +++ b/test/integration/cluster/capacity-loss.test.ts @@ -196,7 +196,7 @@ it("does not announce readiness after terminal capacity loss during startup", as } }); -it("keeps remaining workers alive and diagnoses sticky=ip before a normal shutdown", async () => { +it("keeps remaining sticky=ip workers alive before a normal shutdown", async () => { const root = await fixture(false, 2, "ip"); const proc = run(root, "ipc"); try { @@ -213,7 +213,7 @@ it("keeps remaining workers alive and diagnoses sticky=ip before a normal shutdo }, 20); }); expect(proc.child.exitCode).toBeNull(); - expect(proc.output()).toContain("does not provide IP affinity"); + expect(proc.output()).not.toContain("does not provide IP affinity"); expect(proc.output()).not.toContain("fatal capacity loss"); const exit = once(proc.child, "exit"); proc.child.send!({ type: "shutdown" }); diff --git a/test/integration/cluster/ip-affinity.test.ts b/test/integration/cluster/ip-affinity.test.ts new file mode 100644 index 00000000..2f6d39f9 --- /dev/null +++ b/test/integration/cluster/ip-affinity.test.ts @@ -0,0 +1,342 @@ +import { fork, type ChildProcess } from "node:child_process"; +import http from "node:http"; +import net from "node:net"; +import { once } from "node:events"; +import { mkdtempSync, rmSync, existsSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { selectAffinitySlot } from "../../../src/lib/cluster/affinity-router.js"; + +const processes: ChildProcess[] = []; +const directories: string[] = []; +afterEach(async () => { + for (const child of processes.splice(0)) { + if (child.exitCode === null && child.signalCode === null) { + if (child.connected) child.send({ type: "shutdown" }); + await Promise.race([ + once(child, "exit"), + new Promise((resolve) => setTimeout(resolve, 3000)), + ]); + if (child.exitCode === null && child.signalCode === null) + child.kill("SIGKILL"); + } + } + for (const directory of directories.splice(0)) + rmSync(directory, { recursive: true, force: true }); +}); + +function waitMessage(child: ChildProcess, type: string): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + cleanup(); + reject(new Error(`Timeout waiting for ${type}`)); + }, 10000); + const message = (value: any) => { + if (value.type === type) { + cleanup(); + resolve(value); + } else if (value.type === "error") { + cleanup(); + reject(new Error(value.error)); + } + }; + const exit = () => { + cleanup(); + reject(new Error(`Master exited before ${type}`)); + }; + const cleanup = () => { + clearTimeout(timer); + child.off("message", message); + child.off("exit", exit); + }; + child.on("message", message); + child.once("exit", exit); + }); +} + +async function start( + adapter = "native", + factory = false, + extra: Record = {}, +) { + const directory = mkdtempSync(join(tmpdir(), "vext-affinity-")); + directories.push(directory); + const pidFile = join(directory, "master.pid"); + const child = fork( + join(process.cwd(), "test/integration/cluster/affinity-fixture.mjs"), + [], + { + stdio: ["ignore", "pipe", "pipe", "ipc"], + execArgv: [], + env: { + ...process.env, + AFFINITY_ADAPTER: adapter, + AFFINITY_FACTORY: factory ? "1" : "0", + AFFINITY_PID_FILE: pidFile, + ...extra, + }, + }, + ); + processes.push(child); + let logs = ""; + child.stdout?.on("data", (chunk) => { + logs += chunk; + }); + child.stderr?.on("data", (chunk) => { + logs += chunk; + }); + try { + return { child, pidFile, ...(await waitMessage(child, "listening")) }; + } catch (error) { + throw new Error(`${error}\n${logs}`); + } +} + +function request( + port: number, + address = "127.0.0.1", + path = "/", + agent: http.Agent | false = false, +): Promise { + return new Promise((resolve, reject) => { + const req = http.get( + { host: "127.0.0.1", port, path, localAddress: address, agent }, + (response) => { + let body = ""; + response.on("data", (chunk) => { + body += chunk; + }); + response.on("end", () => { + try { + resolve(JSON.parse(body)); + } catch { + reject(new Error(body)); + } + }); + }, + ); + req.setTimeout(5000, () => req.destroy(new Error("HTTP request timeout"))); + req.on("error", reject); + }); +} + +describe("real IP sticky Cluster", () => { + for (const adapter of ["native", "hono", "fastify", "express", "koa"]) { + for (const factory of [false, true]) { + it(`${adapter} ${factory ? "factory" : "string"}: independent connections keep affinity and source IP`, async () => { + const { endpoint } = await start(adapter, factory); + const replies = await Promise.all( + Array.from({ length: 24 }, () => request(endpoint.port)), + ); + expect(new Set(replies.map((reply) => reply.pid)).size).toBe(1); + expect(replies.every((reply) => reply.remote === "127.0.0.1")).toBe( + true, + ); + const agent = new http.Agent({ keepAlive: true }); + try { + expect( + (await request(endpoint.port, "127.0.0.1", "/", agent)).pid, + ).toBe(replies[0].pid); + expect( + (await request(endpoint.port, "127.0.0.1", "/", agent)).pid, + ).toBe(replies[0].pid); + } finally { + agent.destroy(); + } + }); + } + } + + it("replaces generations without adding routable slots and cleans PID on shutdown", async () => { + const { child, endpoint, workers, pidFile } = await start(); + const before = await request(endpoint.port); + const ready = waitMessage(child, "reloaded"); + child.send({ type: "reload" }); + await ready; + const after = await request(endpoint.port); + expect(after.pid).not.toBe(before.pid); + const snapshot = waitMessage(child, "snapshot"); + child.send({ type: "snapshot" }); + const result = await snapshot; + expect(result.workers.map((worker: any) => worker.slotId).sort()).toEqual( + workers.map((worker: any) => worker.slotId).sort(), + ); + expect(result.workers).toHaveLength(2); + expect(result.connections.pending).toBe(0); + const exit = once(child, "exit"); + child.send({ type: "shutdown" }); + expect((await exit)[0]).toBe(0); + expect(existsSync(pidFile)).toBe(false); + }); + + it("keeps old Workers when candidates fail", async () => { + const { child, endpoint } = await start("native", false, { + AFFINITY_FAIL_CANDIDATE: "1", + }); + const before = await request(endpoint.port); + const reloaded = waitMessage(child, "reloaded"); + child.send({ type: "reload" }); + await reloaded; + expect((await request(endpoint.port)).pid).toBe(before.pid); + }); + + it("shuts down cleanly when a live Worker has already disconnected IPC", async () => { + const { child, workers, pidFile } = await start("native", false, { + AFFINITY_SLOW_CLOSE: "1", + }); + const exit = once(child, "exit"); + child.send({ + type: "disconnect-and-stop", + id: workers.find((worker: any) => worker.slotId === 0).id, + }); + expect((await exit)[0]).toBe(0); + expect(existsSync(pidFile)).toBe(false); + }); + + it("does not add a new slot when another Worker crashes during rolling drain", async () => { + const { child, workers } = await start("native", false, { + AFFINITY_SLOW_CLOSE: "1", + }); + const firstCandidate = waitMessage(child, "worker-ready"); + const reloaded = waitMessage(child, "reloaded"); + child.send({ type: "reload" }); + await firstCandidate; + const recovered = waitMessage(child, "worker-ready"); + child.send({ + type: "crash", + id: workers.find((worker: any) => worker.slotId === 1).id, + }); + await recovered; + await reloaded; + const snapshot = waitMessage(child, "snapshot"); + child.send({ type: "snapshot" }); + const result = await snapshot; + expect(result.workers).toHaveLength(2); + expect(result.workers.map((worker: any) => worker.slotId).sort()).toEqual([ + 0, 1, + ]); + expect( + result.workers.every((worker: any) => worker.state === "ready"), + ).toBe(true); + }); + + it("recovers the same slot after a real Worker crash", async () => { + const { child, endpoint, workers } = await start(); + const before = await request(endpoint.port); + const slot = selectAffinitySlot( + "127.0.0.1", + workers.map((worker: any) => worker.slotId), + ); + const victim = workers.find((worker: any) => worker.slotId === slot); + const ready = waitMessage(child, "worker-ready"); + child.send({ type: "crash", id: victim.id }); + await ready; + const after = await request(endpoint.port); + expect(after.pid).not.toBe(before.pid); + const snapshot = waitMessage(child, "snapshot"); + child.send({ type: "snapshot" }); + expect( + (await snapshot).workers.some( + (worker: any) => + worker.slotId === slot && worker.generation === after.generation, + ), + ).toBe(true); + }); + + it.skipIf(process.platform === "win32")( + "distributes actual source IPs and only remaps the unavailable slot", + async () => { + const { child, endpoint, workers } = await start("native", false, { + AFFINITY_AUTO_RESTART: "0", + }); + const addresses = Array.from( + { length: 12 }, + (_, i) => `127.0.0.${i + 1}`, + ); + const before = await Promise.all( + addresses.map((ip) => request(endpoint.port, ip)), + ); + expect(new Set(before.map((reply) => reply.pid)).size).toBe(2); + for (let i = 0; i < addresses.length; i++) { + const slot = selectAffinitySlot(addresses[i]!, [0, 1]); + expect(before[i].generation).toBe( + workers.find((worker: any) => worker.slotId === slot).generation, + ); + } + const victim = workers[0]; + const exited = waitMessage(child, "worker-exit"); + child.send({ type: "crash", id: victim.id }); + await exited; + const after = await Promise.all( + addresses.map((ip) => request(endpoint.port, ip)), + ); + expect(new Set(after.map((reply) => reply.pid)).size).toBe(1); + for (let i = 0; i < addresses.length; i++) { + if (before[i].generation !== victim.generation) + expect(after[i].pid).toBe(before[i].pid); + } + }, + ); + + it.skipIf(process.platform === "win32")( + "bounds stalled handoffs and recovers an unresponsive Worker", + async () => { + const { child, endpoint, workers } = await start(); + const slot = selectAffinitySlot("127.0.0.1", [0, 1]); + const victim = workers.find((worker: any) => worker.slotId === slot); + const initial = await request(endpoint.port); + process.kill(initial.pid, "SIGSTOP"); + const sockets = Array.from({ length: 120 }, () => + net.connect(endpoint.port, "127.0.0.1"), + ); + for (const socket of sockets) socket.on("error", () => {}); + try { + await new Promise((resolve) => setTimeout(resolve, 150)); + const snapshot = waitMessage(child, "snapshot"); + child.send({ type: "snapshot" }); + const status = await snapshot; + expect(status.connections.pending).toBeGreaterThan(0); + expect(status.connections.pending).toBeLessThanOrEqual(64); + expect(status.connections.rejected).toBeGreaterThan(0); + const recovered = await waitMessage(child, "worker-ready"); + expect(recovered.data.workerId).not.toBe(victim.id); + expect((await request(endpoint.port)).pid).not.toBe(initial.pid); + } finally { + for (const socket of sockets) socket.destroy(); + try { + process.kill(initial.pid, "SIGCONT"); + } catch { + /* already killed */ + } + } + }, + ); + + it("closes partial-header connections within the shutdown budget", async () => { + const { child, endpoint } = await start("fastify"); + const socket = net.connect(endpoint.port, "127.0.0.1"); + await once(socket, "connect"); + socket.write("GET / HTTP/1.1\r\nHost: localhost\r\n"); + socket.on("error", () => {}); + await new Promise((resolve) => setTimeout(resolve, 50)); + const closed = once(socket, "close"), + exit = once(child, "exit"); + child.send({ type: "shutdown" }); + await closed; + expect((await exit)[0]).toBe(0); + }); + + for (const adapter of ["native", "fastify"]) { + it(`${adapter}: graceful Master shutdown waits for an in-flight response`, async () => { + const { child, endpoint } = await start(adapter); + const inFlight = waitMessage(child, "in-flight"); + const response = request(endpoint.port, "127.0.0.1", "/?delay=120"); + await inFlight; + const exit = once(child, "exit"); + child.send({ type: "shutdown" }); + expect((await response).pid).toBeGreaterThan(0); + expect((await exit)[0]).toBe(0); + }); + } +}); diff --git a/test/unit/adapter-resolver.test.ts b/test/unit/adapter-resolver.test.ts index 6ef605e7..618912af 100644 --- a/test/unit/adapter-resolver.test.ts +++ b/test/unit/adapter-resolver.test.ts @@ -9,7 +9,10 @@ import { resolveAdapter, } from "../../src/lib/adapter-resolver.js"; import { _validateConfig } from "../../src/lib/config-loader.js"; -import type { VextAdapter } from "../../src/types/adapter.js"; +import type { + VextAdapter, + VextAdapterRuntimeContext, +} from "../../src/types/adapter.js"; import type { VextApp, VextConfig } from "../../src/types/app.js"; function createCustomAdapter(): VextAdapter { @@ -44,6 +47,24 @@ function createStaleCustomAdapter() { } describe("adapter resolver configuration contract", () => { + it("passes per-instance runtime context to custom factories", async () => { + const context: VextAdapterRuntimeContext = { + socketHandoff: { host: "127.0.0.1", port: 3001 }, + }; + const adapter = createCustomAdapter(); + const app = {} as VextApp; + const factory = ( + receivedApp: VextApp, + receivedContext?: VextAdapterRuntimeContext, + ) => { + expect(receivedApp).toBe(app); + expect(receivedContext).toBe(context); + return adapter; + }; + expect( + await resolveAdapter({ adapter: factory } as VextConfig, app, context), + ).toBe(adapter); + }); it("accepts and returns a documented adapter object instance", async () => { const adapter = createCustomAdapter(); expect(() => _validateConfig({ adapter })).not.toThrow(); diff --git a/test/unit/cluster/affinity-router.test.ts b/test/unit/cluster/affinity-router.test.ts new file mode 100644 index 00000000..48420988 --- /dev/null +++ b/test/unit/cluster/affinity-router.test.ts @@ -0,0 +1,40 @@ +import { describe, it, expect } from "vitest"; +import { + normalizeSourceIP, + selectAffinitySlot, +} from "../../../src/lib/cluster/affinity-router.js"; +import { WorkerSlots } from "../../../src/lib/cluster/worker-slots.js"; + +describe("source IP affinity", () => { + it("canonicalizes equivalent addresses and rejects non-IP keys", () => { + expect(normalizeSourceIP("::ffff:127.0.0.1")).toBe("127.0.0.1"); + expect(normalizeSourceIP("0:0:0:0:0:ffff:7f00:1")).toBe("127.0.0.1"); + expect(normalizeSourceIP("2001:0db8:0:0:0:0:0:1")).toBe("2001:db8::1"); + expect(normalizeSourceIP("fe80::1%eth0")).toBe("fe80::1%eth0"); + expect(() => normalizeSourceIP("proxy.example")).toThrow(); + }); + + it("keeps healthy-slot mappings when one slot disappears", () => { + const used = new Set(); + for (let ip = 1; ip < 255; ip++) { + const address = `127.0.0.${ip}`; + const initial = selectAffinitySlot(address, [0, 1, 2, 3])!; + used.add(initial); + expect(selectAffinitySlot(address, [3, 2, 1, 0])).toBe(initial); + if (initial !== 2) + expect(selectAffinitySlot(address, [0, 1, 3])).toBe(initial); + } + expect(used.size).toBe(4); + expect(selectAffinitySlot("127.0.0.1", [])).toBeUndefined(); + }); + + it("does not remove a new owner on a late old-generation exit", () => { + const slots = new WorkerSlots(); + slots.activate(0, 1); + slots.activate(0, 2); + slots.remove(0, 1); + expect(slots.owner(0)).toBe(2); + slots.remove(0, 2); + expect(slots.owner(0)).toBeUndefined(); + }); +}); diff --git a/test/unit/cluster/socket-http-server.test.ts b/test/unit/cluster/socket-http-server.test.ts new file mode 100644 index 00000000..53c09194 --- /dev/null +++ b/test/unit/cluster/socket-http-server.test.ts @@ -0,0 +1,155 @@ +import http from "node:http"; +import net from "node:net"; +import { once } from "node:events"; +import { describe, expect, it } from "vitest"; +import { SocketHTTPServer } from "../../../src/lib/socket-http-server.js"; + +async function ingress( + handler: http.RequestListener, + options: http.ServerOptions = {}, +) { + let receiver: SocketHTTPServer; + const server = net.createServer({ pauseOnConnect: true }, (socket) => + receiver.receiveSocket(socket), + ); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const endpoint = server.address() as net.AddressInfo; + receiver = new SocketHTTPServer(options, handler, { + host: endpoint.address, + port: endpoint.port, + }); + await new Promise((resolve) => + receiver.listen(endpoint.port, endpoint.address, resolve), + ); + return { + receiver, + port: endpoint.port, + async cleanup() { + const closing = new Promise((resolve) => + server.close(() => resolve()), + ); + receiver.forceClose(); + await new Promise((resolve) => receiver.close(() => resolve())); + await closing; + }, + }; +} + +describe("managed Node HTTP lifecycle", () => { + it("cancels the listening event when closed before the next tick", async () => { + const receiver = new SocketHTTPServer({}, (_req, res) => res.end(), { + host: "127.0.0.1", + port: 80, + }); + let listened = false; + receiver.once("listening", () => { + listened = true; + }); + receiver.listen(80, "127.0.0.1"); + await new Promise((resolve) => receiver.close(() => resolve())); + expect(listened).toBe(false); + expect(receiver.address()).toBeNull(); + }); + it("reports handed-off connections and waits for in-flight responses on repeated close", async () => { + let started!: () => void; + const began = new Promise((resolve) => { + started = resolve; + }); + const test = await ingress((_request, response) => { + started(); + setTimeout(() => response.end("done"), 80); + }); + try { + const response = new Promise((resolve, reject) => { + http + .get( + { port: test.port, host: "127.0.0.1", agent: false }, + (result) => { + let body = ""; + result.on("data", (chunk) => { + body += chunk; + }); + result.on("end", () => resolve(body)); + }, + ) + .on("error", reject); + }); + await began; + expect(test.receiver.address()?.port).toBe(test.port); + expect( + await new Promise((resolve, reject) => + test.receiver.getConnections((error, count) => + error ? reject(error) : resolve(count), + ), + ), + ).toBe(1); + let finished = false; + const closed = new Promise((resolve, reject) => + test.receiver.close((error) => { + finished = true; + error ? reject(error) : resolve(); + }), + ); + const again = new Promise((resolve, reject) => + test.receiver.close((error) => (error ? reject(error) : resolve())), + ); + expect(finished).toBe(false); + expect(test.receiver.listening).toBe(false); + expect(await response).toBe("done"); + await Promise.all([closed, again]); + expect(test.receiver.address()).toBeNull(); + } finally { + await test.cleanup(); + } + }); + + it("enforces Node headersTimeout on externally accepted sockets", async () => { + const test = await ingress((_req, res) => res.end("unexpected"), { + headersTimeout: 40, + requestTimeout: 500, + connectionsCheckingInterval: 10, + }); + const socket = net.connect(test.port, "127.0.0.1"); + try { + await once(socket, "connect"); + socket.write("GET / HTTP/1.1\r\nHost: localhost\r\n"); + let result = ""; + socket.on("data", (chunk) => { + result += chunk; + }); + await once(socket, "close"); + expect(result).toContain("408 Request Timeout"); + } finally { + socket.destroy(); + await test.cleanup(); + } + }); + + it("force closes upgraded sockets as well as ordinary HTTP sockets", async () => { + const test = await ingress((_req, res) => res.end("ok")); + test.receiver.on("upgrade", (_req, socket) => + socket.write( + "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: test\r\n\r\n", + ), + ); + const socket = net.connect(test.port, "127.0.0.1"); + try { + await once(socket, "connect"); + socket.write( + "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: test\r\n\r\n", + ); + expect(String((await once(socket, "data"))[0])).toContain( + "101 Switching", + ); + const closed = once(socket, "close"); + test.receiver.forceClose(); + await closed; + await new Promise((resolve) => + test.receiver.close(() => resolve()), + ); + } finally { + socket.destroy(); + await test.cleanup(); + } + }); +}); diff --git a/test/unit/cluster/socket-receiver.test.ts b/test/unit/cluster/socket-receiver.test.ts new file mode 100644 index 00000000..e2a3fe2c --- /dev/null +++ b/test/unit/cluster/socket-receiver.test.ts @@ -0,0 +1,65 @@ +import { Socket } from "node:net"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { SocketReceiver } from "../../../src/lib/cluster/socket-receiver.js"; +import { SOCKET_HANDOFF_POLICY } from "../../../src/lib/cluster/connection-policy.js"; + +afterEach(() => vi.useRealTimers()); +describe("paused socket ownership", () => { + function setup() { + const receiveSocket = vi.fn(); + const reply = vi.fn(); + const receiver = new SocketReceiver( + "one", + { host: "127.0.0.1", port: 80, close: async () => {}, receiveSocket }, + reply, + ); + return { receiver, receiveSocket, reply }; + } + const offer = { + type: "socket-offer", + generation: "one", + transferId: "a", + } as const; + + it("starts application processing only on commit, once", () => { + const { receiver, receiveSocket, reply } = setup(); + const socket = new Socket(); + receiver.message(offer, socket); + expect(reply).toHaveBeenCalledWith({ + ...offer, + type: "socket-prepared", + accepted: true, + }); + expect(receiveSocket).not.toHaveBeenCalled(); + receiver.message({ ...offer, type: "socket-commit" }); + receiver.message({ ...offer, type: "socket-commit" }); + expect(receiveSocket).toHaveBeenCalledExactlyOnceWith(socket); + socket.destroy(); + receiver.close(); + }); + + it("expires offers independently of Master and ignores late commits", () => { + vi.useFakeTimers(); + const { receiver, receiveSocket } = setup(); + const socket = new Socket(); + receiver.message(offer, socket); + vi.advanceTimersByTime(SOCKET_HANDOFF_POLICY.timeout); + expect(socket.destroyed).toBe(true); + receiver.message({ ...offer, type: "socket-commit" }); + expect(receiveSocket).not.toHaveBeenCalled(); + receiver.close(); + }); + + it("rejects old-generation offers and cancels pending offers on close", () => { + const { receiver, receiveSocket } = setup(); + const invalid = new Socket(); + receiver.message({ ...offer, generation: "old" }, invalid); + expect(invalid.destroyed).toBe(true); + const socket = new Socket(); + receiver.message(offer, socket); + receiver.close(); + expect(socket.destroyed).toBe(true); + receiver.message({ ...offer, type: "socket-commit" }); + expect(receiveSocket).not.toHaveBeenCalled(); + }); +}); diff --git a/website/docs/en/api/config.md b/website/docs/en/api/config.md index ae4299b9..9d2214da 100644 --- a/website/docs/en/api/config.md +++ b/website/docs/en/api/config.md @@ -1128,7 +1128,7 @@ Cluster multi-process configuration. For the complete interface definition, see | `memoryThreshold` | `number` | `1073741824` | Worker heapUsed threshold in bytes; a periodic breach asks Master for replacement rather than exiting immediately | | `pidFile` | `string` | `'.vext.pid'` | PID file path (for `vext stop` / `vext reload` to locate the process) | | `titlePrefix` | `string` | `'vext'` | Worker process title prefix | -| `sticky` | `'none' \| 'ip'` | `'none'` | ip is deprecated and diagnosed, still round-robin; use external load-balancer affinity. Planned removal in the next breaking release | +| `sticky` | `'none' \| 'ip'` | `'none'` | none uses ordinary Cluster dispatch; ip routes TCP source IPs to stable Worker slots. Proxies/NAT can create hotspots; see the Cluster guide | ### `healthCheck` — heartbeat detection diff --git a/website/docs/en/guide/adapters.md b/website/docs/en/guide/adapters.md index 46db02a4..55f8af3b 100644 --- a/website/docs/en/guide/adapters.md +++ b/website/docs/en/guide/adapters.md @@ -541,6 +541,16 @@ The current Hono Adapter is a Node.js runtime capability: it receives requests t | Response still shows old name | Check environment override, working directory, and old process; rebuild/start | Recheck four requests and actual port | | EADDRINUSE | Stop the instance you started earlier or change port and request URL | Confirm the new instance listens before retrying | +## Cluster source IP affinity and custom adapters + +All five built-in adapters share a Node HTTP receiver for `cluster.sticky: "ip"`, retaining their routing, request, and shutdown lifecycles. Fastify uses its official `serverFactory` and continues through ready/listen/close. + +The public `VextAdapterFactory` type is `(app, context?: VextAdapterRuntimeContext) => VextAdapter`. The second argument is per-instance runtime context. Only an actual Cluster Worker with affinity enabled receives `context.socketHandoff`; its host/port describe Master's bound public endpoint. A custom factory must not change runtime mode solely because the project config contains sticky. + +Custom adapters keep their original required interface by default. To support this mode, declare `supportsSocketHandoff: true`, consume the factory context, and return a `VextServerHandle` with `receiveSocket(socket)` and synchronous `forceClose()` from `listen()`. The Worker binds neither a public nor a private TCP port in this mode. receiveSocket accepts a committed, paused socket, registers ownership, and then resumes reading. close waits for in-flight connections; forceClose releases every owned socket, including upgrades, within the framework's absolute shutdown deadline. Missing capability fails before listening; claiming support without returning the controls fails startup. + +Existing one-argument factories remain valid in ordinary mode. Wrappers around built-in factories must forward context, for example `(app, context) => nativeAdapter()(app, context)`. Injecting `buildHandler()` into an unlistened server does not by itself provide Node timeout, connection-counting, and shutdown contracts. + ## Next step - Understand the Adapter-related configuration items in [Configuration](/guide/configuration) diff --git a/website/docs/en/guide/cluster.md b/website/docs/en/guide/cluster.md index ad9616ab..6cc18905 100644 --- a/website/docs/en/guide/cluster.md +++ b/website/docs/en/guide/cluster.md @@ -154,9 +154,9 @@ export default { // Worker process title prefix titlePrefix: "vext", - // sticky session mode ('none' | 'ip') + // Connection affinity mode ('none' | 'ip') // 'none' — not enabled (default) - // 'ip' — currently selects a scheduling-policy branch, not IP affinity + // 'ip' — assign TCP source IPs to stable Worker slots sticky: "none", }, } satisfies VextConfigOverride; @@ -180,7 +180,25 @@ Each Worker adds an application instance, database pool, cache, and heap. Choose Workers do not share ordinary variables, Service instances, or in-memory Stores. Use shared storage for data that must be globally consistent. For example, an in-memory rate limit counts separately in each Worker and is not a global quota; see [Rate Limiting](/guide/rate-limit). Sessions and caches likewise depend on their configured Stores. -`sticky: "ip"` is deprecated and emits a startup diagnostic. It still uses round-robin and does not assign Workers by client IP. Use external load-balancer affinity for sessions or reconnection state. The ip option is planned for removal in the next breaking release. +### Source IP affinity + +Set `cluster.sticky: "ip"` when independent connections from the same TCP source IP must reach the same Worker. Master accepts and pauses each connection, chooses a stable logical slot using the source address, then hands the socket to its Worker over IPC. The Worker handles application traffic directly. All five built-in adapters support this through string configuration and official factories. + +```ts +export default { + cluster: { enabled: true, workers: 2, sticky: "ip" }, +}; +``` + +The default `"none"` keeps ordinary Cluster listening: Node SCHED_NONE on Linux, SCHED_RR on other platforms. `"ip"` directly implements affinity, replacing the previous round-robin placeholder; no new configuration or migration switch is required. It activates only during Cluster startup; dev/testing and ordinary single-process startup keep their existing paths. + +Routing uses the TCP peer address, without inspecting `X-Forwarded-For`, `Forwarded`, cookies, or SIDs. Application `trustProxy` and `req.ip` parsing cannot change connection routing that has already happened. Proxies and NAT can merge many clients into one source IP and concentrate traffic on one Worker. External load balancers generally choose instances or separate upstreams, rather than Workers behind a shared port. + +With stable slots and availability, new connections from one IP select the same slot. When a slot fails, only its traffic falls back; restoration can return it to its preferred slot. Rolling replacement preserves slot IDs, but does not preserve process memory. Old connections remain with the old Worker until completion or the shutdown budget expires. Changing sticky or workers requires a full Master restart; rolling reload rejects candidates whose policy differs from Master. + +Handoffs have bounded credits and deadlines: currently at most 1,024 pending globally and 64 per Worker, with a 5-second handoff deadline. These internal safeguards are separate from HTTP request execution timeouts. A saturated target slot rejects new connections rather than spilling them into another Worker. An unconfirmed, timed-out Worker is terminated and handled by the existing auto-restart policy. The framework does not replay requests; clients must consider idempotency when retrying. Runtime snapshot `summary.connections` records pending, committed, rejected, timedOut, sendFailed, and backpressure at Worker metrics intervals and lifecycle events. Committed means that the commit instruction was sent, rather than that an HTTP request succeeded. + +IP affinity can serve applications with process-local state across multiple connections, but does not provide highly available sessions or automatically add Socket.IO, TLS, or HTTP/2. Shared storage remains preferable for consistency and recovery. Custom adapters must explicitly support socket handoff; see [Adapters](/guide/adapters). ## CLI commands @@ -515,7 +533,7 @@ HTTP Cluster Workers do not execute Jobs by default, so multiple HTTP Workers do ### What should we pay attention to when using WebSocket/SSE in Cluster mode? -An established long connection stays with the Worker holding it and can disconnect when that Worker exits. Design reconnection, cross-request state, broadcast, and shutdown timeouts explicitly. Current `sticky: "ip"` does not implement IP affinity and cannot guarantee that reconnection returns to the old Worker. Protocol support also depends on the adapter and application. +An established long connection stays with the Worker holding it and can disconnect when that Worker exits. Design reconnection, cross-request state, broadcast, and shutdown timeouts explicitly. `sticky: "ip"` preserves source IP reconnection affinity while slots and availability are stable; replacement, failure, or address changes cannot preserve the original process or its memory. Protocol support also depends on the adapter and application. ### What is the appropriate number of Workers? diff --git a/website/docs/en/guide/deployment.md b/website/docs/en/guide/deployment.md index 58bfdea7..2d80876a 100644 --- a/website/docs/en/guide/deployment.md +++ b/website/docs/en/guide/deployment.md @@ -852,7 +852,7 @@ upstream vext_backend { } ``` -Vext Cluster reload does not update Master configuration, publish images, change database schemas, or preserve every long connection. `sticky: "ip"` is deprecated and diagnoses its round-robin behavior; use external load-balancer affinity. With zero Workers and no pending replacement, the standard host records failed state, cleans its PID, and exits 1 for supervisor recovery. Remaining healthy Workers keep running, so monitor ready capacity and business health. +Vext Cluster reload does not update Master configuration, publish images, change database schemas, or preserve every long connection. `sticky: "ip"` routes TCP peer addresses to stable Worker slots inside one instance. Proxies/NAT can create hotspots, and failures or rolling replacement do not migrate in-memory sessions. External instance affinity does not provide Worker affinity behind a shared port. With zero Workers and no pending replacement, the standard host records failed state, cleans its PID, and exits 1 for supervisor recovery. Remaining healthy Workers keep running, so monitor ready capacity and business health. ## Monitor alarms diff --git a/website/docs/zh/api/config.md b/website/docs/zh/api/config.md index 57d5b76b..f9131da8 100644 --- a/website/docs/zh/api/config.md +++ b/website/docs/zh/api/config.md @@ -1243,7 +1243,7 @@ Cluster 多进程配置。完整接口定义见 `src/types/app.ts` `VextClusterC | `memoryThreshold` | `number` | `1073741824` | Worker heapUsed 阈值（bytes）；周期超限请求 Master 替换，不立即退出 | | `pidFile` | `string` | `'.vext.pid'` | PID 文件路径（供 `vext stop` / `vext reload` 定位进程） | | `titlePrefix` | `string` | `'vext'` | Worker 进程标题前缀 | -| `sticky` | `'none' \| 'ip'` | `'none'` | ip 已弃用并诊断，仍为 round-robin；IP 会话亲和性请使用外部负载均衡，计划在下一个破坏性版本移除 ip | +| `sticky` | `'none' \| 'ip'` | `'none'` | none 使用普通 Cluster 分发；ip 按 TCP 源 IP 选择稳定 Worker 槽位，代理/NAT 可能产生热点，详见 Cluster 指南 | ### `healthCheck` — 心跳检测 diff --git a/website/docs/zh/guide/adapters.md b/website/docs/zh/guide/adapters.md index 4a037c66..4570eb46 100644 --- a/website/docs/zh/guide/adapters.md +++ b/website/docs/zh/guide/adapters.md @@ -545,6 +545,16 @@ export default { | 修改后响应仍显示旧名称 | 核对环境配置覆盖、运行目录和旧进程，重新 build/start | 检查四个请求及实际端口 | | EADDRINUSE | 停止自己先前启动的实例，或调整端口和请求 URL | 确认新实例成功监听后重试 | +## Cluster 源 IP 亲和性与自定义 Adapter + +五种内置 Adapter 在 `cluster.sticky: "ip"` 下共用 Node HTTP 接收层，保留各自的路由、请求处理与关闭生命周期。Fastify 通过其正式 `serverFactory` 接入，仍执行 ready/listen/close。 + +公开类型 `VextAdapterFactory` 的签名是 `(app, context?: VextAdapterRuntimeContext) => VextAdapter`。第二参数是每个实例的运行上下文；只有实际启用亲和性的 Cluster Worker 才传入 `context.socketHandoff`，其中 host/port 是 Master 已绑定的公共地址。自定义工厂不能仅因项目配置含 sticky 就自行切换运行模式。 + +自定义 Adapter 默认仍只需实现原有接口。支持该模式时，应设置 `supportsSocketHandoff: true`，消费工厂上下文，并让 `listen()` 返回带 `receiveSocket(socket)` 和同步 `forceClose()` 的 `VextServerHandle`。此时 Worker 不绑定公共或私有 TCP 端口；receiveSocket 接收已提交的暂停 socket，登记所有权后才恢复读取。close 等待在途连接，forceClose 释放全部持有的连接（包括 upgrade），并与框架 shutdown 的绝对期限配合。缺少能力时在监听前报错；宣称支持却缺少返回控制方法时启动失败。 + +旧的一参数工厂在普通模式下继续可用。封装内置工厂时需要把上下文一起传递，例如 `(app, context) => nativeAdapter()(app, context)`。仅将 `buildHandler()` 注入未监听 server 无法保证 Node 超时、连接统计和关闭合同，不能作为完整的 handoff 实现。 + ## 下一步 - 了解 [配置](/zh/guide/configuration) 中 Adapter 相关的配置项 diff --git a/website/docs/zh/guide/cluster.md b/website/docs/zh/guide/cluster.md index cbc22ffb..594e6f9a 100644 --- a/website/docs/zh/guide/cluster.md +++ b/website/docs/zh/guide/cluster.md @@ -154,9 +154,9 @@ export default { // Worker 进程标题前缀 titlePrefix: "vext", - // sticky session 模式（'none' | 'ip'） + // 连接亲和性模式（'none' | 'ip'） // 'none' — 不启用（默认） - // 'ip' — 当前仅影响调度策略分支，不提供 IP 粘性分配保证 + // 'ip' — 按 TCP 源 IP 分配到稳定 Worker 槽位 sticky: "none", }, } satisfies VextConfigOverride; @@ -180,7 +180,25 @@ export default { Worker 之间不共享普通变量、Service 实例或内存 Store。需要全局一致的数据，应选择具有共享语义的存储；例如内存限流是各 Worker 独立计数，不能视为全局配额，见[限流](/zh/guide/rate-limit)。Session 与缓存的共享方式同样需按对应 Store 设计。 -`sticky: "ip"` 是弃用的兼容保留值：启动时给出诊断并沿用 Node SCHED_RR，没有实现按客户端 IP 映射 Worker。需要 IP affinity 时在外部负载均衡配置，并用共享 Session/状态存储处理跨 Worker 状态；该值计划在下一破坏性版本移除。 +### 源 IP 亲和性 + +需要让同一 TCP 源 IP 的多条独立连接进入同一 Worker 时，设置 `cluster.sticky: "ip"`。Master 接收并暂停新连接，按源地址和稳定逻辑槽位计算路由，再通过 IPC 把 socket 交给 Worker；业务数据由 Worker 直接处理。五种内置 Adapter 的字符串配置与官方工厂均支持此模式。 + +```ts +export default { + cluster: { enabled: true, workers: 2, sticky: "ip" }, +}; +``` + +默认 `"none"` 继续使用普通 Cluster 监听路径：Linux 使用 Node SCHED_NONE，其他平台使用 SCHED_RR。`"ip"` 直接实现真实亲和性，替换旧的 RR 占位行为；无需新增配置或迁移开关。仅在 Cluster 启动时生效，dev/testing 和普通单进程启动仍沿用原路径。 + +路由依据是 TCP 对端地址，不读取 `X-Forwarded-For`、`Forwarded`、cookie 或 SID；应用的 `trustProxy` 与 `req.ip` 解析不能改变已完成的连接分发。反向代理或 NAT 会把多个客户端合并为同一个源 IP，可能集中到一个 Worker。外部负载均衡通常只能选择实例或独立上游，不能自动选择共享端口中的 Worker。 + +槽位和可用集合稳定时，同一 IP 的新连接选择同一槽位。崩溃时仅不可用槽位的流量回退，恢复后可能回到原槽位。滚动替换保留槽位编号，但替换进程后内存状态不会保留；旧连接仍由旧 Worker 完成或在关闭预算耗尽时断开。改变 sticky 或 workers 配置需要完整重启 Master，rolling reload 会拒绝与 Master 策略不一致的候选。 + +交接设有有界额度和超时：当前全局最多 1,024 条待交接连接、每个 Worker 最多 64 条，交接期限为 5 秒；这些内部防护值不是 HTTP 请求执行时限。目标槽位额度耗尽时关闭新连接，不把它改投到其他繁忙程度较低的 Worker；超时且无法确认交接的进程会被终止并进入现有 auto-restart 规则。框架不会重放请求，客户端重试需考虑幂等性。指标通过 runtime snapshot 的 `summary.connections` 记录，按 Worker 指标周期及生命周期事件更新，包含 pending、committed、rejected、timedOut、sendFailed 与 backpressure；committed 表示提交指令发送完成，不表示 HTTP 请求成功。 + +IP 亲和性适合依赖进程内状态的多连接应用，但不提供 Session 高可用，也不自动启用 Socket.IO、TLS 或 HTTP/2。状态一致性及故障恢复仍建议使用共享存储。自定义 Adapter 必须显式支持 socket handoff，参见[适配器](/zh/guide/adapters)。 ## CLI 命令 @@ -523,7 +541,7 @@ HTTP cluster worker 默认不执行 Job，避免多个 HTTP worker 各自触发 ### Cluster 模式下 WebSocket / SSE 需要注意什么？ -已建立的长连接由持有它的 Worker 处理，Worker 退出时仍可能断开。重连、跨请求状态、消息广播和关闭超时需单独设计；当前 `sticky: "ip"` 未实现 IP 粘性分配，不能据此保证同一客户端重新连接到原 Worker。是否支持具体协议还取决于适配器与应用实现。 +已建立的长连接由持有它的 Worker 处理，Worker 退出时仍可能断开。重连、跨请求状态、消息广播和关闭超时需单独设计；`sticky: "ip"` 可在槽位与可用集合稳定时保持源 IP 的重连亲和性；Worker 替换、故障或源地址变化后不能保证原进程或内存状态仍在。是否支持具体协议还取决于适配器与应用实现。 ### Worker 数量设多少合适？ diff --git a/website/docs/zh/guide/deployment.md b/website/docs/zh/guide/deployment.md index 5af5e942..f1e1ee22 100644 --- a/website/docs/zh/guide/deployment.md +++ b/website/docs/zh/guide/deployment.md @@ -855,7 +855,7 @@ upstream vext_backend { } ``` -Vext Cluster reload 不更新 Master 自身配置，也不会替你发布镜像、切换数据结构或保证全部长连接无损。`sticky: "ip"` 是弃用的 RR 兼容值，使用时有诊断；IP affinity 交由外部负载均衡。全部 Worker 消失且没有恢复工作时，标准 Master 清理 PID 并以 1 退出，供 PM2/容器监督恢复；外部监督需要退避。继续使用业务健康检查和有效 Worker 容量告警。 +Vext Cluster reload 不更新 Master 自身配置，也不会替你发布镜像、切换数据结构或保证全部长连接无损。`sticky: "ip"` 按 TCP 源地址在实例内选择稳定 Worker 槽位；代理/NAT 可能造成热点，故障和滚动替换不会迁移内存 Session。外部负载均衡的实例亲和性不等于共享端口内的 Worker 亲和性。全部 Worker 消失且没有恢复工作时，标准 Master 清理 PID 并以 1 退出，供 PM2/容器监督恢复；外部监督需要退避。继续使用业务健康检查和有效 Worker 容量告警。 ## 监控告警 From 3b0845b9f36ce9f168b6fbe9a98aa842c129c8f9 Mon Sep 17 00:00:00 2001 From: Rocky shi Date: Fri, 9 Oct 2026 15:54:59 +0000 Subject: [PATCH 2/4] test: accept reset sockets during forced cluster shutdown --- test/integration/cluster/ip-affinity.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/test/integration/cluster/ip-affinity.test.ts b/test/integration/cluster/ip-affinity.test.ts index 2f6d39f9..84bfccfa 100644 --- a/test/integration/cluster/ip-affinity.test.ts +++ b/test/integration/cluster/ip-affinity.test.ts @@ -320,7 +320,11 @@ describe("real IP sticky Cluster", () => { socket.write("GET / HTTP/1.1\r\nHost: localhost\r\n"); socket.on("error", () => {}); await new Promise((resolve) => setTimeout(resolve, 50)); - const closed = once(socket, "close"), + // Forced teardown can produce ECONNRESET on Windows; the contract is that + // the connection closes and Master finishes normally, rather than a FIN. + const closed = new Promise((resolve) => + socket.once("close", () => resolve()), + ), exit = once(child, "exit"); child.send({ type: "shutdown" }); await closed; From e6b6cde5e02032dbe42500e55cd69a1e19cc7492 Mon Sep 17 00:00:00 2001 From: Rocky shi Date: Fri, 9 Oct 2026 16:11:02 +0000 Subject: [PATCH 3/4] ci: bound full-suite coverage concurrency --- .github/workflows/ci.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cf2a3f58..7dd5b54f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -306,7 +306,9 @@ jobs: run: npm run build - name: Run tests with coverage - run: npm run test:cov + # Full-suite coverage runs native TS compilers alongside real fork/HTTP + # fixtures; bound parallelism to keep their existing 30s deadlines useful. + run: npm run test:cov -- --maxWorkers=2 env: VEXT_TEST_MONGODB_URI: mongodb://127.0.0.1:27017/vextjs_ci_cov VEXT_TEST_REDIS_URL: redis://127.0.0.1:6379 From 818baeb148e0f87efbfeb4b94c2c2f30930f9601 Mon Sep 17 00:00:00 2001 From: Rocky shi Date: Fri, 9 Oct 2026 23:16:43 +0000 Subject: [PATCH 4/4] fix: withdraw stopping workers and harden shutdown cleanup --- .github/workflows/ci.yml | 1 + src/lib/app.ts | 12 +- src/lib/bootstrap.ts | 26 ++-- src/lib/cluster/ipc-types.ts | 18 ++- src/lib/cluster/master.ts | 27 +++- src/lib/cluster/socket-dispatcher.ts | 9 ++ src/lib/cluster/worker.ts | 24 +++- test/integration/cluster/affinity-fixture.mjs | 34 ++++- test/integration/cluster/ip-affinity.test.ts | 118 +++++++++++++++++- test/unit/app.test.ts | 59 +++++++++ test/unit/cluster/master.test.ts | 81 ++++++++++++ test/unit/cluster/socket-dispatcher.test.ts | 59 +++++++++ website/docs/en/guide/cluster.md | 3 + website/docs/zh/guide/cluster.md | 3 + 14 files changed, 446 insertions(+), 28 deletions(-) create mode 100644 test/unit/cluster/socket-dispatcher.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7dd5b54f..dca26be0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -477,6 +477,7 @@ jobs: test/unit/frontend-repairs.test.ts test/unit/cluster/affinity-router.test.ts test/unit/cluster/socket-receiver.test.ts + test/unit/cluster/socket-dispatcher.test.ts test/unit/cluster/socket-http-server.test.ts test/integration/cluster/ip-affinity.test.ts --reporter=verbose diff --git a/src/lib/app.ts b/src/lib/app.ts index a814589a..20bba43f 100644 --- a/src/lib/app.ts +++ b/src/lib/app.ts @@ -593,7 +593,17 @@ export function createApp(config: VextConfig): { "[vextjs] server close failed during shutdown", ); } - if (serverResult.status === "timed-out") serverHandle.forceClose?.(); + if (serverResult.status === "timed-out") { + try { + serverHandle.forceClose?.(); + } catch (error) { + shutdownError = error; + app.logger.error( + { error: shutdownErrorMessage(error) }, + "[vextjs] server force close failed during shutdown", + ); + } + } } // ── 步骤 2：按 LIFO 顺序执行 onClose 钩子 ── diff --git a/src/lib/bootstrap.ts b/src/lib/bootstrap.ts index 22510a2f..24d94895 100644 --- a/src/lib/bootstrap.ts +++ b/src/lib/bootstrap.ts @@ -1364,19 +1364,21 @@ async function startClusterMaster(rootDir: string): Promise { summary: { connections: master.getConnectionSnapshot() }, }); }); - master.on("worker-exit", (event) => { - void patchRuntimeSnapshotSafe(rootDir, { - runtimeIdentity, - summary: { - state: master.getReadyWorkerCount() > 0 ? "ready" : "degraded", - workers: master.getReadyWorkerCount(), - totalWorkers: master.getTargetWorkerCount(), - connections: master.getConnectionSnapshot(), - }, - workers: snapshotWorkers(), - event: { type: "worker-exit", ...event }, + for (const type of ["worker-exit", "worker-stopping"] as const) { + master.on(type, (event) => { + void patchRuntimeSnapshotSafe(rootDir, { + runtimeIdentity, + summary: { + state: master.getReadyWorkerCount() > 0 ? "ready" : "degraded", + workers: master.getReadyWorkerCount(), + totalWorkers: master.getTargetWorkerCount(), + connections: master.getConnectionSnapshot(), + }, + workers: snapshotWorkers(), + event: { type, ...event }, + }); }); - }); + } master.on("reload-complete", (event) => { void patchRuntimeSnapshotSafe(rootDir, { runtimeIdentity, diff --git a/src/lib/cluster/ipc-types.ts b/src/lib/cluster/ipc-types.ts index 1ee8f846..af181268 100644 --- a/src/lib/cluster/ipc-types.ts +++ b/src/lib/cluster/ipc-types.ts @@ -5,7 +5,7 @@ * 此文件定义双向消息的类型联合，确保 IPC 消息的类型安全。 * * 消息方向： - * - WorkerToMasterMessage: Worker → Master（ready / heartbeat / metrics / request-restart） + * - WorkerToMasterMessage: Worker → Master（ready / heartbeat / stopping / metrics / request-restart） * - MasterToWorkerMessage: Master → Worker（set-title / shutdown / health-check / broadcast） * * 设计原则： @@ -83,6 +83,12 @@ export interface WorkerHeartbeatMessage { memory: number; } +/** Worker has begun local shutdown; keep automatic recovery after its exit. */ +export interface WorkerStoppingMessage { + type: "stopping"; + workerId: string; +} + /** * Worker 指标上报消息 * @@ -113,6 +119,7 @@ export type WorkerToMasterMessage = | SocketPreparedMessage | WorkerReadyMessage | WorkerHeartbeatMessage + | WorkerStoppingMessage | WorkerMetricsMessage | WorkerRequestRestartMessage; @@ -236,7 +243,14 @@ export interface WorkerMeta { * starting → ready → draining → dead * ↑ │ * └── auto-restart ──┘ + * Local shutdown uses ready → stopping → dead and retains auto-restart. + * draining denotes intentional Master replacement/shutdown without recovery. * * @see 12c-lifecycle.md §1（状态机） */ -export type WorkerState = "starting" | "ready" | "draining" | "dead"; +export type WorkerState = + | "starting" + | "ready" + | "stopping" + | "draining" + | "dead"; diff --git a/src/lib/cluster/master.ts b/src/lib/cluster/master.ts index 35d0281c..629cffd8 100644 --- a/src/lib/cluster/master.ts +++ b/src/lib/cluster/master.ts @@ -123,6 +123,8 @@ export type ClusterMasterInput = Partial< export interface ClusterMasterEvents { /** Worker 就绪 */ "worker-ready": { workerId: number; pid: number }; + /** Worker began local shutdown and no longer accepts new connections. */ + "worker-stopping": { workerId: number }; /** Worker 退出 */ "worker-exit": { workerId: number; @@ -731,11 +733,14 @@ export class ClusterMaster extends EventEmitter { const owner = this.slots.owner(slotId); if ( this.isShuttingDown || + this.workers.get(worker.id)?.state !== "ready" || + worker.isDead() || + !worker.isConnected() || (owner !== undefined && owner !== options.replaceWorkerId) ) { this.terminateUnreadyWorker(worker); throw new Error( - `worker ${worker.id} affinity slot is already owned or shutting down`, + `worker ${worker.id} cannot activate affinity slot: not ready, already owned or shutting down`, ); } // One synchronous commit; candidates never add an extra routable slot. @@ -795,6 +800,13 @@ export class ClusterMaster extends EventEmitter { ) { return; } + if (this.workers.get(worker.id)?.state === "stopping") { + settled = true; + this.terminateUnreadyWorker(worker); + cleanup(); + reject(new Error(`worker ${worker.id} stopped before ready`)); + return; + } const endpoint = this.dispatcher?.address(); const reported = (msg as { server?: { host: string; port: number } }) .server; @@ -918,6 +930,19 @@ export class ClusterMaster extends EventEmitter { this.dispatcher?.prepared(worker, message); break; } + case "stopping": { + const meta = this.workers.get(worker.id); + if ( + !meta || + meta.generation !== message.workerId || + (meta.state !== "ready" && meta.state !== "starting") + ) + break; + meta.state = "stopping"; + this.dispatcher?.workerStopping(worker.id); + this.emit("worker-stopping", { workerId: worker.id }); + break; + } case "heartbeat": { const meta = this.workers.get(worker.id); if (meta) { diff --git a/src/lib/cluster/socket-dispatcher.ts b/src/lib/cluster/socket-dispatcher.ts index 3b5c284e..85e13f3b 100644 --- a/src/lib/cluster/socket-dispatcher.ts +++ b/src/lib/cluster/socket-dispatcher.ts @@ -133,6 +133,15 @@ export class SocketDispatcher { } } + workerStopping(workerId: number): void { + for (const [id, transfer] of this.transfers) { + if (transfer.target.worker.id === workerId) { + this.counters.rejected++; + this.cancel(id); + } + } + } + beginShutdown(): void { if (this.stopped) return; this.stopped = true; diff --git a/src/lib/cluster/worker.ts b/src/lib/cluster/worker.ts index 6555a846..8ed03a54 100644 --- a/src/lib/cluster/worker.ts +++ b/src/lib/cluster/worker.ts @@ -16,6 +16,7 @@ import type { WorkerMetricsMessage, WorkerReadyMessage, WorkerRequestRestartMessage, + WorkerStoppingMessage, MasterToWorkerMessage, } from "./ipc-types.js"; import { checkClusterCompatibility } from "./cluster-checks.js"; @@ -229,6 +230,19 @@ export async function workerMain( ); } + // Signal, fatal-error and IPC shutdown all use this same internals object. + // Observe its entry once, before application close hooks or server.close(). + const shutdown = ctx.internals.shutdown.bind(ctx.internals); + ctx.internals.shutdown = (...args) => { + if (!ctx.isShuttingDown) { + ctx.isShuttingDown = true; + ctx.socketReceiver?.close(); + cleanupTimers(ctx); + sendToMaster({ type: "stopping", workerId: config.workerId }); + } + return shutdown(...args); + }; + // ── 2. 注册 IPC 消息处理器 ────────────────────────── registerIPCHandlers(ctx, config); @@ -345,14 +359,9 @@ function registerIPCHandlers(ctx: WorkerContext, config: WorkerConfig): void { */ function shutdownWorker(ctx: WorkerContext, config: WorkerConfig): void { if (ctx.isShuttingDown) return; - ctx.isShuttingDown = true; - ctx.socketReceiver?.close(); console.log(`[worker:${config.workerId}] shutting down...`); - // 先清理定时器 - cleanupTimers(ctx); - if (ctx.internals && ctx.serverHandle) { ctx.internals.shutdown(ctx.serverHandle).catch((err) => { console.error( @@ -363,6 +372,8 @@ function shutdownWorker(ctx: WorkerContext, config: WorkerConfig): void { }); } else { // internals 不可用（bootstrap 可能未完成），直接退出 + ctx.isShuttingDown = true; + cleanupTimers(ctx); process.exit(0); } } @@ -509,7 +520,8 @@ function sendToMaster( | WorkerReadyMessage | WorkerHeartbeatMessage | WorkerMetricsMessage - | WorkerRequestRestartMessage, + | WorkerRequestRestartMessage + | WorkerStoppingMessage, ): void { try { process.send?.(msg, () => {}); diff --git a/test/integration/cluster/affinity-fixture.mjs b/test/integration/cluster/affinity-fixture.mjs index 5c4fd1fb..29a0f005 100644 --- a/test/integration/cluster/affinity-fixture.mjs +++ b/test/integration/cluster/affinity-fixture.mjs @@ -5,10 +5,13 @@ import { ClusterMaster } from "../../../dist/lib/cluster/master.js"; import { workerMain } from "../../../dist/lib/cluster/worker.js"; import { createApp, DEFAULT_CONFIG } from "../../../dist/lib/app.js"; import { resolveAdapter } from "../../../dist/lib/adapter-resolver.js"; +import { setupShutdown } from "../../../dist/lib/shutdown.js"; const adapterName = process.env.AFFINITY_ADAPTER ?? "native"; const factory = process.env.AFFINITY_FACTORY === "1"; const entry = fileURLToPath(import.meta.url); +const workerShutdownSeconds = 1; +const masterShutdownMs = 2000; if (cluster.isWorker) { await workerMain(process.cwd(), async (_root, context) => { @@ -27,7 +30,7 @@ if (cluster.isWorker) { ]() : adapterName, cluster: { ...DEFAULT_CONFIG.cluster, sticky: "ip", workers: 2 }, - shutdown: { ...DEFAULT_CONFIG.shutdown, timeout: 500 }, + shutdown: { ...DEFAULT_CONFIG.shutdown, timeout: workerShutdownSeconds }, }; const { app, internals } = createApp(config); if ( @@ -52,6 +55,19 @@ if (cluster.isWorker) { }, ]); const serverHandle = await app.adapter.listen(context.port, context.host); + const forceClose = serverHandle.forceClose.bind(serverHandle); + serverHandle.forceClose = () => { + process.send?.({ + type: "test-force-close", + workerId: process.env.VEXT_WORKER_ID, + }); + forceClose(); + }; + app.onClose(setupShutdown({ internals, serverHandle, logger: app.logger })); + process.on("message", (message) => { + if (message.type === "test-local-stop") + void internals.shutdown(serverHandle); + }); return { app, internals, serverHandle }; }); } else { @@ -65,7 +81,11 @@ if (cluster.isWorker) { autoRestart: process.env.AFFINITY_AUTO_RESTART !== "0", restartBaseDelay: 20, restartMaxDelay: 50, - reload: { workerDelay: 0, readyTimeout: 3000, shutdownTimeout: 700 }, + reload: { + workerDelay: 0, + readyTimeout: 3000, + shutdownTimeout: masterShutdownMs, + }, }); master.on( "all-workers-dead", @@ -75,9 +95,15 @@ if (cluster.isWorker) { cluster.workers[workerId]?.on("message", (message) => { if (message.type === "test-in-flight") process.send?.({ type: "in-flight" }); + else if (message.type === "test-force-close") process.send?.(message); }); }); - for (const event of ["worker-ready", "worker-exit", "reload-complete"]) { + for (const event of [ + "worker-ready", + "worker-stopping", + "worker-exit", + "reload-complete", + ]) { master.on(event, (data) => process.send?.({ type: event, data })); } process.on("message", async (message) => { @@ -87,6 +113,8 @@ if (cluster.isWorker) { process.send?.({ type: "reloaded" }); } else if (message.type === "crash") { cluster.workers[message.id]?.process.kill("SIGKILL"); + } else if (message.type === "local-stop") { + cluster.workers[message.id]?.send({ type: "test-local-stop" }); } else if (message.type === "disconnect-and-stop") { const worker = cluster.workers[message.id]; worker.once("disconnect", () => { diff --git a/test/integration/cluster/ip-affinity.test.ts b/test/integration/cluster/ip-affinity.test.ts index 84bfccfa..96da5220 100644 --- a/test/integration/cluster/ip-affinity.test.ts +++ b/test/integration/cluster/ip-affinity.test.ts @@ -80,6 +80,8 @@ async function start( ); processes.push(child); let logs = ""; + const messages: any[] = []; + child.on("message", (message) => messages.push(message)); child.stdout?.on("data", (chunk) => { logs += chunk; }); @@ -87,7 +89,13 @@ async function start( logs += chunk; }); try { - return { child, pidFile, ...(await waitMessage(child, "listening")) }; + return { + child, + pidFile, + messages, + getLogs: () => logs, + ...(await waitMessage(child, "listening")), + }; } catch (error) { throw new Error(`${error}\n${logs}`); } @@ -313,8 +321,73 @@ describe("real IP sticky Cluster", () => { }, ); - it("closes partial-header connections within the shutdown budget", async () => { - const { child, endpoint } = await start("fastify"); + for (const adapter of ["native", "fastify"]) { + for (const autoRestart of [false, true]) { + it(`${adapter}: local shutdown excludes the slot and autoRestart=${autoRestart} keeps its recovery policy`, async () => { + const { child, endpoint, workers } = await start(adapter, false, { + AFFINITY_SLOW_CLOSE: "1", + AFFINITY_AUTO_RESTART: autoRestart ? "1" : "0", + }); + const before = await request(endpoint.port); + const slot = selectAffinitySlot("127.0.0.1", [0, 1]); + const victim = workers.find((worker: any) => worker.slotId === slot); + const inFlight = waitMessage(child, "in-flight"); + const response = request(endpoint.port, "127.0.0.1", "/?delay=120"); + await inFlight; + const stopping = waitMessage(child, "worker-stopping"); + const exited = waitMessage(child, "worker-exit"); + const recovered = autoRestart + ? waitMessage(child, "worker-ready") + : undefined; + child.send({ type: "local-stop", id: victim.id }); + expect((await stopping).data.workerId).toBe(victim.id); + expect((await request(endpoint.port)).pid).not.toBe(before.pid); + expect((await response).pid).toBe(before.pid); + expect((await exited).data).toMatchObject({ + workerId: victim.id, + code: 0, + signal: null, + }); + if (recovered) await recovered; + const snapshot = waitMessage(child, "snapshot"); + child.send({ type: "snapshot" }); + const result = await snapshot; + expect(result.workers).toHaveLength(autoRestart ? 2 : 1); + expect( + result.workers.every((worker: any) => worker.state === "ready"), + ).toBe(true); + if (autoRestart) { + const replacement = result.workers.find( + (worker: any) => worker.slotId === slot, + ); + expect(replacement.id).not.toBe(victim.id); + expect((await request(endpoint.port)).generation).toBe( + replacement.generation, + ); + } + }); + } + it.skipIf(process.platform === "win32")( + `${adapter}: SIGTERM uses the same local shutdown notification`, + async () => { + const { child, endpoint, workers } = await start(adapter, false, { + AFFINITY_SLOW_CLOSE: "1", + AFFINITY_AUTO_RESTART: "0", + }); + const before = await request(endpoint.port); + const stopping = waitMessage(child, "worker-stopping"); + process.kill(before.pid, "SIGTERM"); + expect((await stopping).data.workerId).toBe( + workers.find((worker: any) => worker.generation === before.generation) + .id, + ); + expect((await request(endpoint.port)).pid).not.toBe(before.pid); + }, + ); + } + + it("force closes partial headers in the Worker before Master fallback", async () => { + const { child, endpoint, messages, getLogs } = await start("fastify"); const socket = net.connect(endpoint.port, "127.0.0.1"); await once(socket, "connect"); socket.write("GET / HTTP/1.1\r\nHost: localhost\r\n"); @@ -329,8 +402,47 @@ describe("real IP sticky Cluster", () => { child.send({ type: "shutdown" }); await closed; expect((await exit)[0]).toBe(0); + expect( + messages.some((message: any) => message.type === "test-force-close"), + ).toBe(true); + expect( + messages.filter((message: any) => message.type === "worker-exit"), + ).toHaveLength(2); + expect( + messages + .filter((message: any) => message.type === "worker-exit") + .every( + (message: any) => + message.data.code === 0 && message.data.signal === null, + ), + ).toBe(true); + expect(getLogs()).not.toContain("SIGKILL"); }); + it.skipIf(process.platform === "win32")( + "Master force kills a stopped Worker when it cannot enforce its own budget", + async () => { + const { child, endpoint, messages, getLogs } = await start(); + const initial = await request(endpoint.port); + process.kill(initial.pid, "SIGSTOP"); + const exit = once(child, "exit"); + try { + child.send({ type: "shutdown" }); + expect((await exit)[0]).toBe(0); + expect( + messages.some((message: any) => message.type === "test-force-close"), + ).toBe(false); + expect(getLogs()).toContain("shutdown timeout (2000ms), SIGKILL"); + } finally { + try { + process.kill(initial.pid, "SIGCONT"); + } catch { + /* already exited */ + } + } + }, + ); + for (const adapter of ["native", "fastify"]) { it(`${adapter}: graceful Master shutdown waits for an in-flight response`, async () => { const { child, endpoint } = await start(adapter); diff --git a/test/unit/app.test.ts b/test/unit/app.test.ts index faec8914..d3b6e32e 100644 --- a/test/unit/app.test.ts +++ b/test/unit/app.test.ts @@ -352,6 +352,65 @@ describe("createApp", () => { ); }); + it("continues cleanup and preserves failure when forceClose throws after the deadline", async () => { + vi.useFakeTimers(); + try { + const { app, internals } = createApp({ + ...DEFAULT_CONFIG, + shutdown: { timeout: 0.02 }, + _testMode: true, + }); + const error = vi.fn(); + app.setLogger(() => ({ error })); + const order: string[] = []; + app.onClose(() => { + order.push("onClose"); + }); + const cacheClose = vi.fn(() => { + order.push("cache"); + }); + (app.cache._getResponseCache() as { close?: () => void }).close = + cacheClose; + app.hooks.on("app:close", ({ phase }) => { + order.push(phase); + }); + const failure = new Error("force-close-failed"); + const handle = { + host: "127.0.0.1", + port: 80, + close: vi.fn(() => new Promise(() => {})), + forceClose: vi.fn(() => { + order.push("forceClose"); + throw failure; + }), + }; + const stopping = internals.shutdown(handle, { skipExit: true }); + const rejection = expect(stopping).rejects.toBe(failure); + expect(internals.shutdown(handle, { skipExit: true })).toBe(stopping); + await vi.advanceTimersByTimeAsync(20); + await rejection; + expect(order).toEqual([ + "before", + "forceClose", + "onClose", + "cache", + "after", + ]); + expect(cacheClose).toHaveBeenCalledOnce(); + expect(handle.forceClose).toHaveBeenCalledOnce(); + expect(error).toHaveBeenCalledWith( + { error: "force-close-failed" }, + "[vextjs] server force close failed during shutdown", + ); + await expect( + internals.shutdown(handle, { skipExit: true }), + ).resolves.toBeUndefined(); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + it("applies one absolute shutdown deadline and still invokes remaining cleanup", async () => { const { app, internals } = createApp({ ...DEFAULT_CONFIG, diff --git a/test/unit/cluster/master.test.ts b/test/unit/cluster/master.test.ts index 1a8dd33d..ec8fd23e 100644 --- a/test/unit/cluster/master.test.ts +++ b/test/unit/cluster/master.test.ts @@ -28,6 +28,7 @@ import { } from "../../../src/lib/cluster/master.js"; import { applyClusterWorkerEnv } from "../../../src/lib/bootstrap.js"; import { EventEmitter } from "node:events"; +import cluster from "node:cluster"; const repoRoot = path.resolve( path.dirname(fileURLToPath(import.meta.url)), @@ -40,6 +41,86 @@ function readRepoFile(relativePath: string): string { // ── DEFAULT_CLUSTER_CONFIG ────────────────────────────────── +describe("Worker stopping state", () => { + function setup(state = "ready") { + const master = new ClusterMaster(); + const runtime = master as any; + const meta = { id: 1, slotId: 0, generation: "7", state }; + runtime.workers.set(1, meta); + runtime.slots.activate(0, 1); + const workerStopping = vi.fn(); + runtime.dispatcher = { workerStopping }; + return { master, runtime, meta, workerStopping }; + } + it("withdraws readiness once and cancels pending offers", () => { + const { master, runtime, meta, workerStopping } = setup(); + const observer = vi.fn(); + master.on("worker-stopping", observer); + runtime.handleWorkerMessage({ id: 1 }, { type: "stopping", workerId: "7" }); + runtime.handleWorkerMessage({ id: 1 }, { type: "stopping", workerId: "7" }); + expect(meta.state).toBe("stopping"); + expect(master.getReadyWorkerCount()).toBe(0); + expect(workerStopping).toHaveBeenCalledExactlyOnceWith(1); + expect(observer).toHaveBeenCalledExactlyOnceWith({ workerId: 1 }); + }); + it("ignores stale generations and preserves intentional draining", () => { + const { runtime, meta, workerStopping } = setup(); + runtime.handleWorkerMessage( + { id: 1 }, + { type: "stopping", workerId: "old" }, + ); + expect(meta.state).toBe("ready"); + meta.state = "draining"; + runtime.handleWorkerMessage({ id: 1 }, { type: "stopping", workerId: "7" }); + expect(meta.state).toBe("draining"); + expect(workerStopping).not.toHaveBeenCalled(); + }); + it("cannot make a stopped startup candidate ready again", async () => { + const { runtime, meta } = setup("starting"); + runtime.dispatcher.address = () => undefined; + const terminate = vi + .spyOn(runtime, "terminateUnreadyWorker") + .mockImplementation(() => {}); + const worker = Object.assign(new EventEmitter(), { id: 1 }); + const ready = runtime.waitForWorkerReady(worker); + runtime.handleWorkerMessage(worker, { type: "stopping", workerId: "7" }); + worker.emit("message", { type: "ready" }); + await expect(ready).rejects.toThrow("stopped before ready"); + expect(meta.state).toBe("stopping"); + expect(terminate).toHaveBeenCalledExactlyOnceWith(worker); + }); + it("rechecks readiness after the ready promise before activating a slot", async () => { + const master = new ClusterMaster(); + const runtime = master as any; + const worker = Object.assign(new EventEmitter(), { + id: 1, + process: { pid: 123, killed: false }, + isDead: () => false, + isConnected: () => true, + send: vi.fn(() => true), + }); + const fork = vi.spyOn(cluster, "fork").mockReturnValue(worker as never); + vi.spyOn(runtime, "waitForWorkerReady").mockImplementation(async () => { + runtime.workers.get(1).state = "stopping"; + }); + const terminate = vi + .spyOn(runtime, "terminateUnreadyWorker") + .mockImplementation(() => {}); + const observer = vi.fn(); + master.on("worker-ready", observer); + try { + await expect(runtime.forkWorker({ slotId: 0 })).rejects.toThrow( + "cannot activate affinity slot", + ); + expect(runtime.slots.owner(0)).toBeUndefined(); + expect(observer).not.toHaveBeenCalled(); + expect(terminate).toHaveBeenCalledExactlyOnceWith(worker); + } finally { + fork.mockRestore(); + } + }); +}); + describe("DEFAULT_CLUSTER_CONFIG", () => { it("should have workers set to 'auto'", () => { expect(DEFAULT_CLUSTER_CONFIG.workers).toBe("auto"); diff --git a/test/unit/cluster/socket-dispatcher.test.ts b/test/unit/cluster/socket-dispatcher.test.ts new file mode 100644 index 00000000..3f6b9168 --- /dev/null +++ b/test/unit/cluster/socket-dispatcher.test.ts @@ -0,0 +1,59 @@ +import net from "node:net"; +import type { Worker } from "node:cluster"; +import { expect, it, vi } from "vitest"; +import { SocketDispatcher } from "../../../src/lib/cluster/socket-dispatcher.js"; + +it("cancels uncommitted offers when a Worker stops and ignores its late prepare", async () => { + let offered!: (message: any) => void; + const offer = new Promise((resolve) => { + offered = resolve; + }); + const send = vi.fn((message, ...args) => { + if (message.type === "socket-offer") offered(message); + const callback = args.at(-1); + if (typeof callback === "function") callback(null); + return true; + }); + const worker = { + id: 1, + process: { killed: false }, + isConnected: () => true, + send, + } as unknown as Worker; + const unresponsive = vi.fn(); + const dispatcher = new SocketDispatcher( + () => [{ slot: 0, generation: "one", worker }], + unresponsive, + ); + const endpoint = await dispatcher.listen(0, "127.0.0.1"); + dispatcher.open(); + const socket = net.connect(endpoint.port, endpoint.host); + socket.on("error", () => {}); + try { + const message = await offer; + const closed = new Promise((resolve) => + socket.once("close", () => resolve()), + ); + dispatcher.workerStopping(worker.id); + await closed; + dispatcher.prepared(worker, { + ...message, + type: "socket-prepared", + accepted: true, + }); + expect(dispatcher.snapshot()).toMatchObject({ + pending: 0, + rejected: 1, + committed: 0, + timedOut: 0, + }); + expect(send.mock.calls.map(([sent]) => sent.type)).toEqual([ + "socket-offer", + "socket-cancel", + ]); + expect(unresponsive).not.toHaveBeenCalled(); + } finally { + socket.destroy(); + dispatcher.beginShutdown(); + } +}); diff --git a/website/docs/en/guide/cluster.md b/website/docs/en/guide/cluster.md index 6cc18905..2d332fad 100644 --- a/website/docs/en/guide/cluster.md +++ b/website/docs/en/guide/cluster.md @@ -196,6 +196,8 @@ Routing uses the TCP peer address, without inspecting `X-Forwarded-For`, `Forwar With stable slots and availability, new connections from one IP select the same slot. When a slot fails, only its traffic falls back; restoration can return it to its preferred slot. Rolling replacement preserves slot IDs, but does not preserve process memory. Old connections remain with the old Worker until completion or the shutdown budget expires. Changing sticky or workers requires a full Master restart; rolling reload rejects candidates whose policy differs from Master. +When a Worker enters application shutdown through SIGTERM/SIGINT or a fatal error, it notifies the Master to withdraw its routing eligibility. New connections can select remaining available slots, while in-flight connections retain the existing shutdown budget. Local exits still follow autoRestart and the restart budget to recover the same slot; Master-initiated replacements or shutdown do not create duplicate Workers. + Handoffs have bounded credits and deadlines: currently at most 1,024 pending globally and 64 per Worker, with a 5-second handoff deadline. These internal safeguards are separate from HTTP request execution timeouts. A saturated target slot rejects new connections rather than spilling them into another Worker. An unconfirmed, timed-out Worker is terminated and handled by the existing auto-restart policy. The framework does not replay requests; clients must consider idempotency when retrying. Runtime snapshot `summary.connections` records pending, committed, rejected, timedOut, sendFailed, and backpressure at Worker metrics intervals and lifecycle events. Committed means that the commit instruction was sent, rather than that an HTTP request succeeded. IP affinity can serve applications with process-local state across multiple connections, but does not provide highly available sessions or automatically add Socket.IO, TLS, or HTTP/2. Shared storage remains preferable for consistency and recovery. Custom adapters must explicitly support socket handoff; see [Adapters](/guide/adapters). @@ -464,6 +466,7 @@ The message types below are the exact string literals of the IPC payload `type` | ----------------- | -------------------------------------------------------------------------------------------------------------------------------------- | | `ready` | Worker initialization is completed and starts accepting requests | | `heartbeat` | Heartbeat response | +| `stopping` | Worker begins local shutdown; withdraw new connections and retain autoRestart recovery after exit | | `metrics` | Memory and other snapshots every 30 seconds; without a request metrics provider, counts are placeholder zero with `metricsUnavailable` | | `request-restart` | Worker requests itself to restart (if a memory leak is detected) | diff --git a/website/docs/zh/guide/cluster.md b/website/docs/zh/guide/cluster.md index 594e6f9a..4bb44613 100644 --- a/website/docs/zh/guide/cluster.md +++ b/website/docs/zh/guide/cluster.md @@ -196,6 +196,8 @@ export default { 槽位和可用集合稳定时，同一 IP 的新连接选择同一槽位。崩溃时仅不可用槽位的流量回退，恢复后可能回到原槽位。滚动替换保留槽位编号，但替换进程后内存状态不会保留；旧连接仍由旧 Worker 完成或在关闭预算耗尽时断开。改变 sticky 或 workers 配置需要完整重启 Master，rolling reload 会拒绝与 Master 策略不一致的候选。 +Worker 自行收到 SIGTERM/SIGINT 或因致命错误进入应用关闭流程时，会通知 Master 撤销新连接路由资格；新连接可选择其余可用槽位，在途连接继续使用原关闭预算。自行退出后仍按 autoRestart 与重启预算恢复原槽位，Master 主动替换或整体停止不会因此重复启动 Worker。 + 交接设有有界额度和超时：当前全局最多 1,024 条待交接连接、每个 Worker 最多 64 条，交接期限为 5 秒；这些内部防护值不是 HTTP 请求执行时限。目标槽位额度耗尽时关闭新连接，不把它改投到其他繁忙程度较低的 Worker；超时且无法确认交接的进程会被终止并进入现有 auto-restart 规则。框架不会重放请求，客户端重试需考虑幂等性。指标通过 runtime snapshot 的 `summary.connections` 记录，按 Worker 指标周期及生命周期事件更新，包含 pending、committed、rejected、timedOut、sendFailed 与 backpressure；committed 表示提交指令发送完成，不表示 HTTP 请求成功。 IP 亲和性适合依赖进程内状态的多连接应用，但不提供 Session 高可用，也不自动启用 Socket.IO、TLS 或 HTTP/2。状态一致性及故障恢复仍建议使用共享存储。自定义 Adapter 必须显式支持 socket handoff，参见[适配器](/zh/guide/adapters)。 @@ -472,6 +474,7 @@ Master 和 Worker 之间通过内部 IPC 协议通信。下表用于理解运行 | ----------------- | ------------------------------------------------------------------------------- | | `ready` | Worker 初始化完成，开始接受请求 | | `heartbeat` | 心跳响应 | +| `stopping` | Worker 开始自行关闭；停止分发新连接，退出后仍按 autoRestart 恢复 | | `metrics` | 每 30 秒上报内存等快照；无请求指标提供者时计数为占位 0，并带 metricsUnavailable | | `request-restart` | Worker 请求自身重启（如检测到内存泄漏） |