From b694e093e7bc59a4058afeaf91607f75c32d277c Mon Sep 17 00:00:00 2001 From: Ryan Chou <88779759+ryanchou1994@users.noreply.github.com> Date: Tue, 8 Sep 2026 06:13:09 +0800 Subject: [PATCH] fix(core): listen on unix domain socket when SOCKET is set (#23238) The deploy docs still describe SOCKET, but the hono migration dropped it from the config parser and lib/index.ts always calls serve() with a TCP port, so the variable had no effect and PORT=null on top crashed the process with ERR_SOCKET_BAD_PORT. Parse SOCKET into config.connect.socket again and, when it is set, create the server with createAdaptorServer and listen on the socket path instead of a port. A stale socket file left by an unclean shutdown is removed before binding; only the binding process does that (cluster primary before forking), workers share the primary's handle and never touch the file. Fixes #18738 --- lib/config.test.ts | 18 ++++++++++ lib/config.ts | 3 ++ lib/index.test.ts | 58 ++++++++++++++++++++++++++++++++ lib/index.ts | 82 +++++++++++++++++++++++++++------------------- 4 files changed, 128 insertions(+), 33 deletions(-) diff --git a/lib/config.test.ts b/lib/config.test.ts index 1f107bb116..4d89fc0564 100644 --- a/lib/config.test.ts +++ b/lib/config.test.ts @@ -22,6 +22,24 @@ describe('config', () => { delete process.env.BILIBILI_COOKIE_34; }); + it('unix socket', async () => { + process.env.SOCKET = '/tmp/rsshub.sock'; + + const { config } = await import('./config'); + expect(config.connect.socket).toBe('/tmp/rsshub.sock'); + + delete process.env.SOCKET; + }); + + it('unix socket defaults to undefined', async () => { + process.env.SOCKET = ''; + + const { config } = await import('./config'); + expect(config.connect.socket).toBeUndefined(); + + delete process.env.SOCKET; + }); + it('email config', async () => { process.env['EMAIL_CONFIG_xx.qq.com'] = 'token1'; process.env['EMAIL_CONFIG_oo.qq.com'] = 'token2'; diff --git a/lib/config.ts b/lib/config.ts index 73d88db5f2..345a2800fd 100644 --- a/lib/config.ts +++ b/lib/config.ts @@ -14,6 +14,7 @@ type ConfigEnvKeys = | 'CHROMIUM_EXECUTABLE_PATH' // Network | 'PORT' + | 'SOCKET' | 'LISTEN_INADDR_ANY' | 'DISABLE_IPV6' | 'REQUEST_RETRY' @@ -270,6 +271,7 @@ export type Config = { // network connect: { port: number; + socket?: string; }; listenInaddrAny: boolean; disableIPv6: boolean; @@ -776,6 +778,7 @@ const calculateValue = () => { // network connect: { port: toInt(envs.PORT, 1200), // 监听端口 + socket: envs.SOCKET || undefined, // listen on a unix socket instead of a TCP port }, listenInaddrAny: toBoolean(envs.LISTEN_INADDR_ANY, true), // 是否允许公网连接，取值 0 1 disableIPv6: toBoolean(envs.DISABLE_IPV6, false), diff --git a/lib/index.test.ts b/lib/index.test.ts index 930df7feee..2aebc41069 100644 --- a/lib/index.test.ts +++ b/lib/index.test.ts @@ -1,6 +1,9 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; const serve = vi.fn<(...args: any[]) => any>(() => ({ close: vi.fn() })); +const listen = vi.fn(); +const createAdaptorServer = vi.fn<(...args: any[]) => any>(() => ({ listen, close: vi.fn() })); +const rmSync = vi.fn(); const logger = { info: vi.fn(), warn: vi.fn(), @@ -20,6 +23,13 @@ const availableParallelism = vi.fn(() => 2); vi.mock('@hono/node-server', () => ({ serve, + createAdaptorServer, +})); +vi.mock('node:fs', () => ({ + __esModule: true, + default: { + rmSync, + }, })); vi.mock('@/utils/logger', () => ({ default: logger, @@ -46,6 +56,9 @@ describe('index', () => { vi.resetModules(); vi.unstubAllEnvs(); serve.mockClear(); + createAdaptorServer.mockClear(); + listen.mockClear(); + rmSync.mockClear(); fork.mockClear(); availableParallelism.mockClear(); logger.info.mockClear(); @@ -64,6 +77,7 @@ describe('index', () => { hostname: '127.0.0.1', port: 12345, }); + expect(createAdaptorServer).not.toHaveBeenCalled(); }); it('forks workers when cluster is enabled and primary', async () => { @@ -93,4 +107,48 @@ describe('index', () => { port: 12347, }); }); + + it('listens on a unix socket instead of a port when SOCKET is set', async () => { + vi.stubEnv('ENABLE_CLUSTER', ''); + vi.stubEnv('LISTEN_INADDR_ANY', ''); + vi.stubEnv('SOCKET', '/tmp/rsshub-test.sock'); + vi.stubEnv('PORT', 'null'); + + const module = await import('@/index'); + expect(module.default).toBeDefined(); + expect(rmSync).toHaveBeenCalledWith('/tmp/rsshub-test.sock', { force: true }); + expect(createAdaptorServer).toHaveBeenCalledTimes(1); + expect(createAdaptorServer.mock.calls[0][0]).toMatchObject({ + hostname: '127.0.0.1', + }); + expect(listen).toHaveBeenCalledWith('/tmp/rsshub-test.sock'); + expect(serve).not.toHaveBeenCalled(); + }); + + it('cluster primary removes a stale unix socket before forking', async () => { + clusterState.isPrimary = true; + vi.stubEnv('ENABLE_CLUSTER', 'true'); + vi.stubEnv('SOCKET', '/tmp/rsshub-test.sock'); + availableParallelism.mockReturnValue(2); + + await import('@/index'); + + expect(rmSync).toHaveBeenCalledWith('/tmp/rsshub-test.sock', { force: true }); + expect(fork).toHaveBeenCalledTimes(2); + expect(createAdaptorServer).not.toHaveBeenCalled(); + expect(serve).not.toHaveBeenCalled(); + }); + + it('cluster worker listens on the unix socket without removing it', async () => { + clusterState.isPrimary = false; + vi.stubEnv('ENABLE_CLUSTER', 'true'); + vi.stubEnv('SOCKET', '/tmp/rsshub-test.sock'); + + await import('@/index'); + + expect(rmSync).not.toHaveBeenCalled(); + expect(createAdaptorServer).toHaveBeenCalledTimes(1); + expect(listen).toHaveBeenCalledWith('/tmp/rsshub-test.sock'); + expect(serve).not.toHaveBeenCalled(); + }); }); diff --git a/lib/index.ts b/lib/index.ts index d21bac888c..efce512296 100644 --- a/lib/index.ts +++ b/lib/index.ts @@ -1,8 +1,9 @@ import cluster from 'node:cluster'; +import fs from 'node:fs'; import os from 'node:os'; import process from 'node:process'; -import { serve } from '@hono/node-server'; +import { createAdaptorServer, serve } from '@hono/node-server'; import app from '@/app'; import { config } from '@/config'; @@ -10,20 +11,55 @@ import { getLocalhostAddress } from '@/utils/common-utils'; import logger from '@/utils/logger'; const port = config.connect.port; +const socketPath = config.connect.socket; const hostIPList = getLocalhostAddress(); +const serverOptions = { + fetch: app.fetch, + hostname: config.listenInaddrAny ? (config.disableIPv6 ? '0.0.0.0' : '::') : '127.0.0.1', + serverOptions: { + maxHeaderSize: 1024 * 32, + }, +}; + +const logListening = () => { + if (socketPath) { + logger.info(`🎉 RSSHub is running on unix socket ${socketPath}! Cheers!`); + return; + } + logger.info(`🎉 RSSHub is running on port ${port}! Cheers!`); + logger.info(`🔗 Local: 👉 http://localhost:${port}`); + if (config.listenInaddrAny) { + for (const ip of hostIPList) { + logger.info(`🔗 Network: 👉 http://${ip}:${port}`); + } + } +}; + +// A socket file left behind by an unclean shutdown makes listen() fail with EADDRINUSE. +// Only the process that binds the socket may remove it: a cluster worker must not touch +// the file because the primary already holds the listening handle. +const removeStaleSocket = () => { + if (socketPath) { + fs.rmSync(socketPath, { force: true }); + } +}; + +const listen = () => { + if (socketPath) { + const socketServer = createAdaptorServer(serverOptions); + socketServer.listen(socketPath); + return socketServer; + } + return serve({ ...serverOptions, port }); +}; + let server; if (config.enableCluster) { if (cluster.isPrimary) { - logger.info(`🎉 RSSHub is running on port ${port}! Cheers!`); - logger.info(`🔗 Local: 👉 http://localhost:${port}`); - if (config.listenInaddrAny) { - for (const ip of hostIPList) { - logger.info(`🔗 Network: 👉 http://${ip}:${port}`); - } - } - + logListening(); logger.info(`Primary ${process.pid} is running`); + removeStaleSocket(); const numCPUs = os.availableParallelism(); @@ -32,32 +68,12 @@ if (config.enableCluster) { } } else { logger.info(`Worker ${process.pid} is running`); - serve({ - fetch: app.fetch, - hostname: config.listenInaddrAny ? (config.disableIPv6 ? '0.0.0.0' : '::') : '127.0.0.1', - port, - serverOptions: { - maxHeaderSize: 1024 * 32, - }, - }); + listen(); } } else { - logger.info(`🎉 RSSHub is running on port ${port}! Cheers!`); - logger.info(`🔗 Local: 👉 http://localhost:${port}`); - if (config.listenInaddrAny) { - for (const ip of hostIPList) { - logger.info(`🔗 Network: 👉 http://${ip}:${port}`); - } - } - - server = serve({ - fetch: app.fetch, - hostname: config.listenInaddrAny ? (config.disableIPv6 ? '0.0.0.0' : '::') : '127.0.0.1', - port, - serverOptions: { - maxHeaderSize: 1024 * 32, - }, - }); + logListening(); + removeStaleSocket(); + server = listen(); } export default server;