feat(cli): auto-retry next port when serve port is in use (#6513)

* feat(cli): auto-retry next port when serve port is in use

When `qwen serve` or `npm run dev:daemon` encounters EADDRINUSE on the
default port (4170), automatically try the next available port (up to 10
attempts) instead of failing immediately. This allows running multiple
daemon instances side-by-side without manual port management.

- run-qwen-serve.ts: replace single listen() with recursive tryListen()
  that retries on EADDRINUSE; --port 0 (ephemeral) skips retry
- daemon-dev.js: pre-scan for an available port via net probe before
  spawning the daemon child, ensuring health-poll and web-shell target
  the correct URL
- Tests: retry-then-succeed, non-EADDRINUSE immediate-fail, and existing
  all-ports-exhausted test all pass (138 tests)

* test(cli): fix EADDRINUSE mock to create fresh server per listen attempt

The existing test reused a single fakeServer across all retry attempts,
accumulating 10+ once('listening') listeners and triggering a
MaxListenersExceededWarning. Create a new server per call to match
production behavior where each tryListen creates a fresh server.

* fix: address review feedback on port retry

- daemon-dev.js: strip IPv6 brackets before probe (ENOTFOUND fix),
  add port range/NaN validation
- run-qwen-serve.ts: remove duplicate runtime error listener
  (onListening already installs one via removeAllListeners + on)
- tests: add exhaustion (all 10 ports), port 0 EADDRINUSE no-retry,
  and stderr retry message assertion (140 tests pass)

* fix: update stale comment referencing removed server.once('error', reject)

* fix: address R2 review — port cap, TLS reuse, exhaustion summary

- daemon-dev.js: cap probe at port 65535, skip probe when user
  specifies --port, add --compacted-replay-max-bytes to whitelist
- run-qwen-serve.ts: move https.createServer before tryListen (avoid
  recreating TLS context per retry), cap retry at 65535, log summary
  "all ports X–Y are in use" on exhaustion
- tests: verify exhaustion summary stderr message

* fix: clear stale listening listeners on httpsServer before retry
This commit is contained in:
Shaojin Wen 2026-07-08 18:31:52 +08:00 committed by GitHub
parent c516ee8c2f
commit 271664b34b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 289 additions and 20 deletions

View file

@ -4495,11 +4495,12 @@ describe('runQwenServe channel worker supervisor', () => {
); );
const listenError = new Error('listen failed') as NodeJS.ErrnoException; const listenError = new Error('listen failed') as NodeJS.ErrnoException;
listenError.code = 'EADDRINUSE'; listenError.code = 'EADDRINUSE';
const fakeServer = createServer();
vi.spyOn(serverModule, 'createServeApp').mockReturnValue({ vi.spyOn(serverModule, 'createServeApp').mockReturnValue({
locals: {},
listen: vi.fn(() => { listen: vi.fn(() => {
setImmediate(() => fakeServer.emit('error', listenError)); const srv = createServer();
return fakeServer; setImmediate(() => srv.emit('error', listenError));
return srv;
}), }),
} as unknown as express.Application); } as unknown as express.Application);
const worker = makeWorker({ const worker = makeWorker({
@ -4535,6 +4536,180 @@ describe('runQwenServe channel worker supervisor', () => {
expect(pidfile.removeServeServiceInfo).toHaveBeenCalledWith(process.pid); expect(pidfile.removeServeServiceInfo).toHaveBeenCalledWith(process.pid);
}); });
it('retries the next port on EADDRINUSE and succeeds', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-port-retry-')),
);
const portsAttempted: number[] = [];
const stderrWrites: string[] = [];
const stderrSpy = vi
.spyOn(process.stderr, 'write')
.mockImplementation((chunk) => {
stderrWrites.push(String(chunk));
return true;
});
vi.spyOn(serverModule, 'createServeApp').mockReturnValue({
locals: {},
listen: vi.fn((port, _host, cb) => {
portsAttempted.push(port);
const srv = createServer();
if (portsAttempted.length === 1) {
const err = new Error('address in use') as NodeJS.ErrnoException;
err.code = 'EADDRINUSE';
setImmediate(() => srv.emit('error', err));
} else {
srv.listen(0, '127.0.0.1', () => {
setImmediate(() => {
srv.emit('listening');
if (typeof cb === 'function') cb();
});
});
}
return srv;
}),
} as unknown as express.Application);
const handle = await runQwenServe(
{
port: 4170,
hostname: '127.0.0.1',
mode: 'http-bridge',
workspace: tmpDir,
serveWebShell: false,
},
{
bridge: makeFakeBridge(),
resolveOnListen: true,
},
);
try {
stderrSpy.mockRestore();
expect(portsAttempted).toEqual([4170, 4171]);
expect(handle.url).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/);
expect(handle.url).not.toContain(':4170');
expect(
stderrWrites.some((w) =>
w.includes('port 4170 is in use, trying 4171'),
),
).toBe(true);
} finally {
await handle.close();
}
});
it('does not retry on non-EADDRINUSE listen errors', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-port-no-retry-')),
);
const portsAttempted: number[] = [];
const listenError = new Error('permission denied') as NodeJS.ErrnoException;
listenError.code = 'EACCES';
vi.spyOn(serverModule, 'createServeApp').mockReturnValue({
locals: {},
listen: vi.fn((port, _host, _cb) => {
portsAttempted.push(port);
const srv = createServer();
setImmediate(() => srv.emit('error', listenError));
return srv;
}),
} as unknown as express.Application);
await expect(
runQwenServe(
{
port: 4170,
hostname: '127.0.0.1',
mode: 'http-bridge',
workspace: tmpDir,
serveWebShell: false,
},
{ bridge: makeFakeBridge() },
),
).rejects.toBe(listenError);
expect(portsAttempted).toEqual([4170]);
});
it('rejects after exhausting all port retry attempts', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-port-exhaust-')),
);
const portsAttempted: number[] = [];
const stderrWrites: string[] = [];
const stderrSpy = vi
.spyOn(process.stderr, 'write')
.mockImplementation((chunk) => {
stderrWrites.push(String(chunk));
return true;
});
const listenError = new Error('address in use') as NodeJS.ErrnoException;
listenError.code = 'EADDRINUSE';
vi.spyOn(serverModule, 'createServeApp').mockReturnValue({
locals: {},
listen: vi.fn((port) => {
portsAttempted.push(port);
const srv = createServer();
setImmediate(() => srv.emit('error', listenError));
return srv;
}),
} as unknown as express.Application);
await expect(
runQwenServe(
{
port: 4170,
hostname: '127.0.0.1',
mode: 'http-bridge',
workspace: tmpDir,
serveWebShell: false,
},
{ bridge: makeFakeBridge() },
),
).rejects.toBe(listenError);
stderrSpy.mockRestore();
expect(portsAttempted).toEqual(
Array.from({ length: 10 }, (_, i) => 4170 + i),
);
expect(
stderrWrites.some((w) => w.includes('all ports 41704179 are in use')),
).toBe(true);
});
it('does not retry EADDRINUSE when port is 0 (ephemeral)', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-port0-no-retry-')),
);
const portsAttempted: number[] = [];
const listenError = new Error('address in use') as NodeJS.ErrnoException;
listenError.code = 'EADDRINUSE';
vi.spyOn(serverModule, 'createServeApp').mockReturnValue({
locals: {},
listen: vi.fn((port) => {
portsAttempted.push(port);
const srv = createServer();
setImmediate(() => srv.emit('error', listenError));
return srv;
}),
} as unknown as express.Application);
await expect(
runQwenServe(
{
port: 0,
hostname: '127.0.0.1',
mode: 'http-bridge',
workspace: tmpDir,
serveWebShell: false,
},
{ bridge: makeFakeBridge() },
),
).rejects.toBe(listenError);
expect(portsAttempted).toEqual([0]);
});
it('does not remove the channel pidfile reservation for handled uncaught exceptions', async () => { it('does not remove the channel pidfile reservation for handled uncaught exceptions', async () => {
tmpDir = fs.realpathSync( tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-channel-worker-crash-')), fs.mkdtempSync(path.join(os.tmpdir(), 'qws-channel-worker-crash-')),

View file

@ -199,6 +199,8 @@ function isNonNegativeIntegerMs(value: number): boolean {
const MAX_TIMEOUT_MS = 2_147_483_647; const MAX_TIMEOUT_MS = 2_147_483_647;
const MAX_PORT_ATTEMPTS = 10;
function assertTimerDelayInRange(name: string, value: number): void { function assertTimerDelayInRange(name: string, value: number): void {
if (value > MAX_TIMEOUT_MS) { if (value > MAX_TIMEOUT_MS) {
throw new TypeError( throw new TypeError(
@ -3897,11 +3899,11 @@ export async function runQwenServe(
process.on('uncaughtExceptionMonitor', onUncaughtExceptionMonitor); process.on('uncaughtExceptionMonitor', onUncaughtExceptionMonitor);
// Swap the boot-error listener for a runtime-error one // Swap the boot-error listener for a runtime-error one
// before resolving. `server.once('error', reject)` at the // before resolving. `tryListen`'s `server.once('error', ...)`
// bottom only catches errors BEFORE listening; post-listen // only catches errors BEFORE listening; post-listen errors
// errors (EMFILE after FD exhaustion, runtime errors on the // (EMFILE after FD exhaustion, runtime errors on the listener)
// listener) would be unhandled and crash the daemon. Use a // would be unhandled and crash the daemon. Use a persistent
// persistent listener that logs to stderr instead. // listener that logs to stderr instead.
server.removeAllListeners('error'); server.removeAllListeners('error');
server.on('error', (err) => { server.on('error', (err) => {
daemonLog.error('server error', err instanceof Error ? err : null); daemonLog.error('server error', err instanceof Error ? err : null);
@ -3948,8 +3950,8 @@ export async function runQwenServe(
} }
}; };
let server: Server; let server: Server;
let httpsServer: https.Server | undefined;
if (tlsOptions) { if (tlsOptions) {
let httpsServer: https.Server;
try { try {
httpsServer = https.createServer(tlsOptions, app); httpsServer = https.createServer(tlsOptions, app);
} catch (err) { } catch (err) {
@ -3966,13 +3968,53 @@ export async function runQwenServe(
); );
return; return;
} }
server = httpsServer.listen(opts.port, listenHostname, onListening);
} else {
server = app.listen(opts.port, listenHostname, onListening);
} }
server.once('error', (err) => {
removeCurrentServePidfile(); const tryListen = (attemptPort: number, attempt: number): void => {
reject(err); try {
}); if (httpsServer) {
// server.listen(port, host, cb) registers `cb` as a one-time
// `listening` listener. On failed attempts (EADDRINUSE),
// `listening` never fires so the listener accumulates. Clear
// stale listeners before each retry.
httpsServer.removeAllListeners('listening');
server = httpsServer.listen(attemptPort, listenHostname, onListening);
} else {
server = app.listen(attemptPort, listenHostname, onListening);
}
} catch (err) {
// Synchronous listen failure (e.g. invalid address) — not
// recoverable via port bump.
removeCurrentServePidfile();
reject(err instanceof Error ? err : new Error(String(err)));
return;
}
server.once('error', (err: NodeJS.ErrnoException) => {
server.close();
const nextPort = attemptPort + 1;
if (
err.code === 'EADDRINUSE' &&
opts.port !== 0 &&
nextPort <= 65535 &&
attempt < MAX_PORT_ATTEMPTS - 1
) {
writeStderrLine(
`qwen serve: port ${attemptPort} is in use, trying ${nextPort}...`,
);
tryListen(nextPort, attempt + 1);
} else {
if (err.code === 'EADDRINUSE' && attempt > 0) {
writeStderrLine(
`qwen serve: all ports ${opts.port}${attemptPort} are in use`,
);
}
removeCurrentServePidfile();
reject(err);
}
});
};
tryListen(opts.port, 0);
}); });
} }

View file

@ -7,6 +7,7 @@
import { spawn } from 'node:child_process'; import { spawn } from 'node:child_process';
import crypto from 'node:crypto'; import crypto from 'node:crypto';
import http from 'node:http'; import http from 'node:http';
import net from 'node:net';
import { dirname, join, resolve } from 'node:path'; import { dirname, join, resolve } from 'node:path';
import { fileURLToPath, pathToFileURL } from 'node:url'; import { fileURLToPath, pathToFileURL } from 'node:url';
import { platform } from 'node:os'; import { platform } from 'node:os';
@ -24,6 +25,7 @@ const serveOptionNames = new Set([
'--max-connections', '--max-connections',
'--require-auth', '--require-auth',
'--event-ring-size', '--event-ring-size',
'--compacted-replay-max-bytes',
]); ]);
function readOption(name) { function readOption(name) {
@ -86,6 +88,42 @@ function daemonUrl(hostname, port) {
return `http://${host}:${port}`; return `http://${host}:${port}`;
} }
const MAX_PORT_ATTEMPTS = 10;
function findAvailablePort(host, startPort) {
return new Promise((resolveFind, rejectFind) => {
let attempt = 0;
const tryNext = () => {
const port = startPort + attempt;
if (port > 65535 || attempt >= MAX_PORT_ATTEMPTS) {
rejectFind(
new Error(
`No available port found in range ${startPort}${Math.min(startPort + MAX_PORT_ATTEMPTS - 1, 65535)}`,
),
);
return;
}
const probe = net.createServer();
probe.once('error', (err) => {
probe.close();
if (err.code === 'EADDRINUSE') {
console.log(
`[daemon-dev] port ${port} is in use, trying ${port + 1}...`,
);
attempt++;
tryNext();
} else {
rejectFind(err);
}
});
probe.listen(port, host, () => {
probe.close(() => resolveFind(port));
});
};
tryNext();
});
}
function spawnDevProcess(label, command, commandArgs, options) { function spawnDevProcess(label, command, commandArgs, options) {
const child = spawn(command, commandArgs, { const child = spawn(command, commandArgs, {
stdio: 'inherit', stdio: 'inherit',
@ -199,15 +237,28 @@ try {
process.exit(1); process.exit(1);
} }
const port = readOption('--port') || '4170'; const hostname = readOption('--hostname') || '127.0.0.1';
if (port === '0') { const rawPort = readOption('--port') || '4170';
const startPort = parseInt(rawPort, 10);
if (!Number.isInteger(startPort) || startPort < 0 || startPort > 65535) {
console.error(
`daemon-dev: --port must be an integer 065535, got "${rawPort}".`,
);
process.exit(1);
}
if (startPort === 0) {
console.error( console.error(
'daemon-dev: --port 0 is not supported; the launcher needs a fixed port to poll for health.', 'daemon-dev: --port 0 is not supported; the launcher needs a fixed port to poll for health.',
); );
process.exit(1); process.exit(1);
} }
const probeHostname =
const hostname = readOption('--hostname') || '127.0.0.1'; hostname.startsWith('[') && hostname.endsWith(']')
? hostname.slice(1, -1)
: hostname;
const port = hasOption('--port')
? String(startPort)
: String(await findAvailablePort(probeHostname, startPort));
const token = const token =
readOption('--token') || readOption('--token') ||
process.env.QWEN_SERVER_TOKEN || process.env.QWEN_SERVER_TOKEN ||
@ -215,6 +266,7 @@ const token =
const workspace = resolve(readOption('--workspace') || process.cwd()); const workspace = resolve(readOption('--workspace') || process.cwd());
const serveArgs = serveArgsFromLauncherArgs(); const serveArgs = serveArgsFromLauncherArgs();
if (!hasOption('--port')) serveArgs.push('--port', port);
if (!hasOption('--workspace')) serveArgs.push('--workspace', workspace); if (!hasOption('--workspace')) serveArgs.push('--workspace', workspace);
const tsxLoaderUrl = pathToFileURL( const tsxLoaderUrl = pathToFileURL(