diff --git a/apps/rush-cli-client/README.md b/apps/rush-cli-client/README.md index 1d4cf98eb5..ef74ee076e 100644 --- a/apps/rush-cli-client/README.md +++ b/apps/rush-cli-client/README.md @@ -247,19 +247,39 @@ attests a restart request, not completion of successor startup or success of a c `rush-client daemon stop` requires protocol >= 0.6 and waits for `shutdownAck` followed by EOF. It reports `state: "shutdownAccepted"` with exit code 0; this -does not assert successful workspace disposal. An absent/unreachable daemon, -unsupported protocol, missing acknowledgement, or timeout returns exit code 1. -It does not auto-start anything. +does not assert successful workspace disposal. Stop is idempotent: when nothing +listens at the endpoint it reports `state: "notRunning"` with exit code 0. An +unsupported protocol, missing acknowledgement, handshake failure, or timeout +returns exit code 1. It does not auto-start anything. + +`rush-client daemon stop --force` stops a running daemon the same way, then waits +(up to 15 seconds) for it to release its listener and ownership record and removes +any remaining artifacts, such as an abandoned startup reservation, reporting them in +`removedPaths`. When none is listening, it removes this workspace's leftover ownership record +(`.pid.json`), socket, and startup reservation (`.starting`), then reports +`state: "reset"` and the `removedPaths` (or `state: "notRunning"` if nothing was +left behind). It holds the start mutex, proves that no listener is bound, and +refuses (exit 1) while the recorded owner PID still exists and cannot be shown to +be a reused PID. It never kills a process. Automatic startup already reclaims +the common leftovers on its own (see below); this is the documented escape hatch +that every fail-closed startup message points to. `rush-client daemon restart` first verifies that the selected Rush version has a launcher and captures the original lock's PID/start timestamp, checking that it matches pong's positive PID and the selected endpoint, then performs acknowledged shutdown. It waits for original ownership release or a demonstrably dead owner -before calling the existing locked starter. A live/reused owner fails closed at +before calling the existing locked starter. A live owner fails closed at the startup deadline; no PID is killed and no live ownership record is deleted. A newly started/reused successor must pass hello/ping before reporting `state: "ready"`. -An absent daemon must be started explicitly with `daemon start`. +When nothing listens at the endpoint, restart starts a daemon exactly like `daemon start`. + +Automatic and explicit startup reclaim stale artifacts only when that is provably +safe: while holding the start mutex with no `.starting` reservation, a socket +without an ownership record, or an unreadable/corrupt record, is removed only after +a connection attempt is refused (so no listener exists). On Linux, a record whose +PID now belongs to a process that started after the record's `startedAt` (PID reuse) +is treated as dead; other platforms fail closed and point to `daemon stop --force`. Restart is explicit even when automatic startup or CI execution routing is disabled, but conflicts with `--no-daemon`. The two-phase host retains ownership diff --git a/apps/rush-cli-client/src/daemonCommands.ts b/apps/rush-cli-client/src/daemonCommands.ts index d2c3e592b1..b6ad07750d 100644 --- a/apps/rush-cli-client/src/daemonCommands.ts +++ b/apps/rush-cli-client/src/daemonCommands.ts @@ -7,9 +7,14 @@ import { DaemonClient, connectOrStartDaemonAsync, requestDaemonShutdownAsync, + resetDaemonArtifactsAsync, type IConnectOrStartDaemonOptions } from '@rushstack/rush-client-core'; -import type { IDaemonLockfile } from '@rushstack/rush-daemon-transport'; +import { + DaemonTransportError, + DaemonTransportErrorCode, + type IDaemonLockfile +} from '@rushstack/rush-daemon-transport'; import type { IDaemonRequestAdmissionOptions } from '@rushstack/rush-daemon-protocol'; import { getDaemonConnectionOptionsAsync } from './daemonConnectionOptions'; @@ -17,6 +22,8 @@ import { printDaemonLogAsync } from './daemonLogs'; import { executeDaemonGraphCommandAsync } from './daemonGraph'; import { writeStreamAsync } from './writeStreamAsync'; +const FORCE_STOP_WAIT_MS: number = 15000; + export interface IDaemonCommandOptions { readonly argv: ReadonlyArray; readonly environment: Readonly; @@ -38,14 +45,18 @@ export async function executeDaemonCommandAsync(options: IDaemonCommandOptions): } if ( (options.argv.length !== 1 && - !(command === 'logs' && options.argv.length === 2 && options.argv[1] === '--follow')) || + !( + options.argv.length === 2 && + ((command === 'logs' && options.argv[1] === '--follow') || + (command === 'stop' && options.argv[1] === '--force')) + )) || (command !== 'start' && command !== 'status' && command !== 'stop' && command !== 'restart' && command !== 'logs') ) { - throw new Error('Usage: rush-client daemon start|status|stop|restart|logs [--follow]'); + throw new Error('Usage: rush-client daemon start|status|stop [--force]|restart|logs [--follow]'); } if (!options.rushJsonPath) throw new Error('Daemon management requires a repository containing rush.json.'); const mayStart: boolean = command === 'start' || command === 'restart'; @@ -76,13 +87,52 @@ export async function executeDaemonCommandAsync(options: IDaemonCommandOptions): } // Status observes the selected endpoint, including a compatible daemon from a different client version. // It never starts a process or trusts a PID file as evidence of readiness. - const client: DaemonClient = + const client: DaemonClient | undefined = command === 'start' ? await connectOrStartDaemonAsync(connectionOptions) - : await DaemonClient.connectAsync({ socketPath: connectionOptions.paths.socketPath }); + : await connectExistingAsync(connectionOptions, command !== 'status'); + if (!client) { + if (command === 'restart') { + // Nothing to shut down: restart behaves like start. + const started: DaemonClient = await connectOrStartDaemonAsync(connectionOptions); + try { + await writeStatusAsync({ + state: 'ready', + socketPath: connectionOptions.paths.socketPath, + ...(await started.status) + }); + } finally { + await started.closeAsync(); + } + return; + } + const { removedPaths } = + options.argv[1] === '--force' + ? await resetDaemonArtifactsAsync(connectionOptions.paths) + : { removedPaths: [] }; + await writeStatusAsync({ + state: removedPaths.length > 0 ? 'reset' : 'notRunning', + socketPath: connectionOptions.paths.socketPath, + ...(options.argv[1] === '--force' ? { removedPaths } : {}) + }); + return; + } try { if (command === 'stop') { await client.shutdownAsync(); + if (options.argv[1] === '--force') { + // Wait for the acknowledged daemon to release its listener and record, then clear leftovers + // such as an abandoned startup reservation in the same invocation. + const { removedPaths } = await resetDaemonArtifactsAsync(connectionOptions.paths, { + waitTimeoutMs: FORCE_STOP_WAIT_MS + }); + await writeStatusAsync({ + state: 'shutdownAccepted', + socketPath: connectionOptions.paths.socketPath, + removedPaths + }); + return; + } await writeStatusAsync({ state: 'shutdownAccepted', socketPath: connectionOptions.paths.socketPath @@ -105,6 +155,25 @@ export async function executeDaemonCommandAsync(options: IDaemonCommandOptions): } } +/** Returns undefined when nothing listens at the endpoint and `allowAbsent` is set; other failures propagate. */ +async function connectExistingAsync( + options: IConnectOrStartDaemonOptions, + allowAbsent: boolean +): Promise { + try { + return await DaemonClient.connectAsync({ socketPath: options.paths.socketPath }); + } catch (error) { + if ( + allowAbsent && + error instanceof DaemonTransportError && + error.code === DaemonTransportErrorCode.connectionRefused + ) { + return undefined; + } + throw error; + } +} + async function restartDaemonAsync( client: DaemonClient, options: IConnectOrStartDaemonOptions diff --git a/apps/rush-cli-client/src/test/launchClient.test.ts b/apps/rush-cli-client/src/test/launchClient.test.ts index 14c8d888d9..b3ec694fca 100644 --- a/apps/rush-cli-client/src/test/launchClient.test.ts +++ b/apps/rush-cli-client/src/test/launchClient.test.ts @@ -248,13 +248,82 @@ describe('standalone rushx fallback', () => { expect((await invokeAsync(true, false, false, ['daemon', 'logs', '--follow', 'extra'])).code).toBe(1); }); - it('does not start an absent daemon when stop or restart cannot be acknowledged', async () => { - for (const verb of ['stop', 'restart']) { - const result: IInvocationResult = await invokeAsync(true, false, false, ['daemon', verb]); - expect(result.code).toBe(1); - expect(result.stderr).toContain('Could not connect to daemon'); + it('treats stop as idempotent and restart as start when no daemon is running', async () => { + const { paths } = getDaemonConnectionOptions(folder, Rush.version, {}, false); + for (const args of [ + ['daemon', 'stop'], + ['daemon', 'stop', '--force'] + ]) { + const result: IInvocationResult = await invokeAsync(true, false, false, args); + expect(result).toMatchObject({ code: 0, stderr: '' }); + expect(JSON.parse(result.stdout)).toEqual({ + state: 'notRunning', + socketPath: paths.socketPath, + ...(args[2] ? { removedPaths: [] } : {}) + }); } - }); + try { + const restarted: IInvocationResult = await invokeAsync(true, false, false, ['daemon', 'restart']); + expect(restarted.stderr).toBe(''); + expect(restarted.code).toBe(0); + expect(JSON.parse(restarted.stdout)).toMatchObject({ state: 'ready', socketPath: paths.socketPath }); + const stopped: IInvocationResult = await invokeAsync(true, false, false, ['daemon', 'stop']); + expect(stopped.code).toBe(0); + expect(JSON.parse(stopped.stdout)).toMatchObject({ state: 'shutdownAccepted' }); + } finally { + const deadline: number = Date.now() + 7000; + while (fs.existsSync(paths.lockfilePath) && Date.now() < deadline) await delayAsync(50); + expect(fs.existsSync(paths.lockfilePath)).toBe(false); + } + }, 30000); + + it('stop --force clears an abandoned startup reservation next to a running daemon', async () => { + const { paths } = getDaemonConnectionOptions(folder, Rush.version, {}, false); + const reservation: string = `${paths.lockfilePath}.starting`; + try { + expect((await invokeAsync(true, false, false, ['daemon', 'start'])).code).toBe(0); + fs.writeFileSync(reservation, 'abandoned'); + const result: IInvocationResult = await invokeAsync(true, false, false, ['daemon', 'stop', '--force']); + expect(result).toMatchObject({ code: 0, stderr: '' }); + expect(JSON.parse(result.stdout)).toEqual({ + state: 'shutdownAccepted', + socketPath: paths.socketPath, + removedPaths: [reservation] + }); + expect(fs.existsSync(reservation)).toBe(false); + expect(fs.existsSync(paths.lockfilePath)).toBe(false); + } finally { + const deadline: number = Date.now() + 7000; + while (fs.existsSync(paths.lockfilePath) && Date.now() < deadline) await delayAsync(50); + } + }, 30000); + + (process.platform === 'win32' ? it.skip : it)( + 'stop --force removes stale artifacts left by a killed daemon', + async () => { + const { paths } = getDaemonConnectionOptions(folder, Rush.version, {}, false); + fs.mkdirSync(path.dirname(paths.lockfilePath), { recursive: true, mode: 0o700 }); + const listener: ChildProcess = spawn( + process.execPath, + [ + '-e', + `require('net').createServer().listen(${JSON.stringify(paths.socketPath)}, () => process.kill(process.pid, 'SIGKILL'))` + ], + { stdio: 'ignore' } + ); + await once(listener, 'close'); + fs.writeFileSync(paths.lockfilePath, 'garbage{'); + const result: IInvocationResult = await invokeAsync(true, false, false, ['daemon', 'stop', '--force']); + expect(result).toMatchObject({ code: 0, stderr: '' }); + expect(JSON.parse(result.stdout)).toEqual({ + state: 'reset', + socketPath: paths.socketPath, + removedPaths: [paths.lockfilePath, paths.socketPath] + }); + expect(fs.existsSync(paths.lockfilePath)).toBe(false); + expect(fs.existsSync(paths.socketPath)).toBe(false); + } + ); it.each([false, true])( 'restarts after ownership release and stops the successor (embedded: %s)', diff --git a/common/changes/@rushstack/rush-cli-client/daemon-recovery_2026-09-23.json b/common/changes/@rushstack/rush-cli-client/daemon-recovery_2026-09-23.json new file mode 100644 index 0000000000..ddc56a8ade --- /dev/null +++ b/common/changes/@rushstack/rush-cli-client/daemon-recovery_2026-09-23.json @@ -0,0 +1,10 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-cli-client", + "comment": "Make `rush-client daemon stop` idempotent (state notRunning, exit 0), make `daemon restart` start a daemon when none is running, and add `daemon stop --force` to remove stale workspace daemon artifacts.", + "type": "patch" + } + ], + "packageName": "@rushstack/rush-cli-client" +} diff --git a/common/changes/@rushstack/rush-client-core/daemon-recovery_2026-09-23.json b/common/changes/@rushstack/rush-client-core/daemon-recovery_2026-09-23.json new file mode 100644 index 0000000000..4dbf5f28e0 --- /dev/null +++ b/common/changes/@rushstack/rush-client-core/daemon-recovery_2026-09-23.json @@ -0,0 +1,10 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-client-core", + "comment": "Reclaim provably stale daemon artifacts (a socket without an ownership record, a corrupt record, or a Linux PID reused since the record was written) instead of permanently disabling the daemon, and add resetDaemonArtifactsAsync() as the explicit recovery path referenced by fail-closed messages.", + "type": "patch" + } + ], + "packageName": "@rushstack/rush-client-core" +} diff --git a/common/reviews/api/rush-client-core.api.md b/common/reviews/api/rush-client-core.api.md index 2305d7bad2..74c883aba2 100644 --- a/common/reviews/api/rush-client-core.api.md +++ b/common/reviews/api/rush-client-core.api.md @@ -80,6 +80,16 @@ export interface IConnectOrStartDaemonOptions extends Omit; +} + // @beta export interface IDaemonClientConnectOptions { // (undocumented) @@ -130,4 +140,7 @@ export interface IDaemonStartCommand { // @beta export function requestDaemonShutdownAsync(client: DaemonClient, paths: IDaemonPaths, timeoutMs?: number): Promise>; +// @beta +export function resetDaemonArtifactsAsync(paths: IDaemonPaths, options?: IDaemonArtifactResetOptions): Promise; + ``` diff --git a/libraries/rush-client-core/README.md b/libraries/rush-client-core/README.md index 6c1afe75e6..86142f10ee 100644 --- a/libraries/rush-client-core/README.md +++ b/libraries/rush-client-core/README.md @@ -48,7 +48,13 @@ handing the explicit command to a detached startup helper. The helper spawns wit a shell and retains that reservation until the daemon completes hello/ping readiness, independently of whether the requesting client survives. Clients still await hello/pong under bounded backoff. Stdout/stderr go to `.log`. No PID -is killed; a live (possibly reused) PID with an unreachable socket fails closed. +is killed. While holding the mutex with no startup reservation, stale leftovers are +reclaimed only when provably safe: a socket without an ownership record, or a corrupt +record, once a connection attempt is refused (no listener exists); and, on Linux, a +record whose PID now belongs to a process that started after the record's `startedAt` +(PID reuse, detected from `/proc`). Any other live PID with an unreachable socket fails +closed, pointing to `resetDaemonArtifactsAsync()` (`rush-client daemon stop --force`), +which removes the record, socket and reservation after the same no-listener/no-live-owner checks. The helper uses a stable tool cwd, and the starting client awaits its exit after readiness. The explicit launcher's cwd is unchanged. @@ -90,7 +96,8 @@ identifies the original ownership record by its `pid` and `startedAt`, captured before sending shutdown. Startup waits until that record disappears, another owner replaces it, or its owner is demonstrably dead. A new owner is checked by hello/ping; it is never blindly reclaimed. Signal 0 is only a liveness probe; no -process is killed. A live/reused owner times out conservatively, while corrupt or +process is killed. A live owner times out conservatively (a Linux PID provably reused +since `startedAt` counts as dead), while corrupt or unreadable metadata fails closed. During a captured predecessor handoff, transient Windows sharing-denied reads stay unknown and are retried only within the existing startup deadline. They never diff --git a/libraries/rush-client-core/src/DaemonOwnership.ts b/libraries/rush-client-core/src/DaemonOwnership.ts new file mode 100644 index 0000000000..a7226d0ee9 --- /dev/null +++ b/libraries/rush-client-core/src/DaemonOwnership.ts @@ -0,0 +1,258 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as fs from 'node:fs'; +import * as net from 'node:net'; +import * as path from 'node:path'; +import { setTimeout as delayAsync } from 'node:timers/promises'; + +import type { IDaemonLockfile, IDaemonPaths } from '@rushstack/rush-daemon-transport'; + +import { DaemonClientError } from './DaemonClientError'; +import { getDaemonStartupFilePath } from './DaemonStartup'; +import { isProcessStartedAfter } from './ProcessStartTime'; +import { tryAcquireStartupLockAsync, type IStartupLock } from './StartupLock'; + +const PROBE_TIMEOUT_MS: number = 1000; +const RESET_RETRY_MS: number = 100; + +/** Printed wherever automatic recovery fails closed. */ +export const DAEMON_RESET_HINT: string = + 'If no daemon is running for this workspace, run "rush-client daemon stop --force" to remove its stale files.'; + +export type DaemonOwnership = Pick; + +type OwnershipState = + | { readonly kind: 'absent' } + | { readonly kind: 'corrupt'; readonly raw: string } + | { readonly kind: 'owned'; readonly raw: string; readonly owner: DaemonOwnership }; + +/** Options for {@link resetDaemonArtifactsAsync}. @beta */ +export interface IDaemonArtifactResetOptions { + /** How long to keep re-checking a bound listener, live owner, or held start mutex. Defaults to 0. */ + readonly waitTimeoutMs?: number; +} + +/** The result of {@link resetDaemonArtifactsAsync}. @beta */ +export interface IDaemonArtifactResetResult { + /** The stale files that were removed; empty when nothing was left behind. */ + readonly removedPaths: ReadonlyArray; +} + +export function isDaemonOwnership(record: unknown): record is DaemonOwnership { + return ( + typeof record === 'object' && + record !== null && + 'pid' in record && + typeof record.pid === 'number' && + Number.isSafeInteger(record.pid) && + record.pid > 0 && + 'startedAt' in record && + typeof record.startedAt === 'string' && + Number.isFinite(Date.parse(record.startedAt)) + ); +} + +export function readDaemonOwnership(lockfilePath: string): DaemonOwnership | undefined { + let record: unknown; + try { + record = JSON.parse(fs.readFileSync(lockfilePath, 'utf8')); + } catch (error) { + if (hasErrorCode(error, 'ENOENT')) return undefined; + throw new DaemonClientError( + 'startupFailed', + `Cannot safely read ${lockfilePath}; refusing automatic reclaim. ${DAEMON_RESET_HINT}`, + { cause: error } + ); + } + if (!isDaemonOwnership(record)) { + throw new DaemonClientError( + 'startupFailed', + `Invalid daemon ownership record in ${lockfilePath}; refusing automatic reclaim. ${DAEMON_RESET_HINT}` + ); + } + return { pid: record.pid, startedAt: record.startedAt }; +} + +export function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + if (hasErrorCode(error, 'ESRCH')) return false; + throw error; + } +} + +/** + * True unless the recorded owner is demonstrably gone: its PID does not exist, or the process now using + * that PID started after the record was written (PID reuse). + */ +export function isOwnerProcessAlive(owner: { readonly pid: number; readonly startedAt?: unknown }): boolean { + if (!isProcessAlive(owner.pid)) return false; + return !(typeof owner.startedAt === 'string' && isProcessStartedAfter(owner.pid, owner.startedAt)); +} + +/** + * Makes stale ownership reclaimable when that is provably safe. The caller must hold the start mutex and + * have observed no startup reservation, so no legitimate daemon can be between bind and record publication; + * a refused connection then proves no listener exists. + */ +export async function reclaimAbandonedOwnershipAsync(paths: IDaemonPaths): Promise { + const state: OwnershipState = inspectOwnership(paths.lockfilePath); + if (state.kind === 'owned' && !isProcessAlive(state.owner.pid)) return; + if (state.kind === 'owned' && !isProcessStartedAfter(state.owner.pid, state.owner.startedAt)) { + throw new DaemonClientError( + 'startupFailed', + `PID ${state.owner.pid} still exists but the daemon is not ready. It may be a daemon that is still shutting down, or a reused PID; refusing to kill it or remove ${paths.lockfilePath}. ${DAEMON_RESET_HINT}` + ); + } + if (state.kind === 'absent' && (process.platform === 'win32' || !fs.existsSync(paths.socketPath))) return; + if (!(await isEndpointUnboundAsync(paths.socketPath))) { + throw new DaemonClientError( + 'startupFailed', + `${describeOwnership(state, paths)}, but ${paths.socketPath} did not refuse a connection; refusing automatic reclaim. ${DAEMON_RESET_HINT}` + ); + } + // The transport reclaim then removes the unbound socket under its own two-factor checks. + if (state.kind !== 'absent') removeIfUnchanged(paths.lockfilePath, state.raw); +} + +/** + * Removes this workspace's leftover daemon files (ownership record, socket, and startup reservation) + * after verifying that no listener is bound and that the recorded owner, if any, is gone. + * @remarks Never kills a process. Fails when another client holds the start mutex, a listener is bound, + * or the recorded owner is alive; with `waitTimeoutMs`, those conditions are re-checked until the deadline + * (for example, while a daemon that just acknowledged shutdown finishes its cleanup). + * @beta + */ +export async function resetDaemonArtifactsAsync( + paths: IDaemonPaths, + options?: IDaemonArtifactResetOptions +): Promise { + const deadline: number = Date.now() + (options?.waitTimeoutMs ?? 0); + while (true) { + const outcome: IDaemonArtifactResetResult | DaemonClientError = await tryResetDaemonArtifactsAsync(paths); + if (!(outcome instanceof DaemonClientError)) return outcome; + if (Date.now() >= deadline) throw outcome; + await delayAsync(Math.min(RESET_RETRY_MS, Math.max(1, deadline - Date.now()))); + } +} + +/** Returns a (not thrown) error for conditions that may clear on their own. */ +async function tryResetDaemonArtifactsAsync( + paths: IDaemonPaths +): Promise { + if (!fs.existsSync(path.dirname(paths.lockfilePath))) return { removedPaths: [] }; + const lock: IStartupLock | undefined = await tryAcquireStartupLockAsync(paths); + if (!lock) { + return new DaemonClientError( + 'startupFailed', + `Another client is starting the daemon for ${paths.lockfilePath}; retry after it finishes.` + ); + } + try { + if (!(await isEndpointUnboundAsync(paths.socketPath))) { + return new DaemonClientError( + 'startupFailed', + `A process is still listening at ${paths.socketPath}; use "rush-client daemon stop" to stop it.` + ); + } + const state: OwnershipState = inspectOwnership(paths.lockfilePath); + if (state.kind === 'owned' && isOwnerProcessAlive(state.owner)) { + return new DaemonClientError( + 'startupFailed', + `PID ${state.owner.pid} still owns ${paths.lockfilePath}; it may be a daemon that is shutting down. Wait for it to exit (or stop that process yourself), then retry. No process was killed.` + ); + } + const removedPaths: string[] = []; + if (state.kind !== 'absent' && removeIfUnchanged(paths.lockfilePath, state.raw)) { + removedPaths.push(paths.lockfilePath); + } + const others: string[] = [getDaemonStartupFilePath(paths)]; + if (process.platform !== 'win32') others.push(paths.socketPath); + for (const filePath of others) { + if (tryUnlink(filePath)) removedPaths.push(filePath); + } + return { removedPaths }; + } finally { + await lock.releaseAsync(); + } +} + +/** Resolves true only when a connection attempt proves that nothing listens at the endpoint. */ +export function isEndpointUnboundAsync(socketPath: string): Promise { + return new Promise((resolve) => { + const socket: net.Socket = net.createConnection(socketPath); + const settle = (unbound: boolean): void => { + socket.destroy(); + resolve(unbound); + }; + socket.setTimeout(PROBE_TIMEOUT_MS); + socket.once('connect', () => settle(false)); + socket.once('timeout', () => settle(false)); + socket.once('error', (error) => settle(hasErrorCode(error, 'ECONNREFUSED') || hasErrorCode(error, 'ENOENT'))); + }); +} + +export function hasErrorCode(error: unknown, code: string): boolean { + return typeof error === 'object' && error !== null && 'code' in error && error.code === code; +} + +function inspectOwnership(lockfilePath: string): OwnershipState { + let raw: string; + try { + raw = fs.readFileSync(lockfilePath, 'utf8'); + } catch (error) { + if (hasErrorCode(error, 'ENOENT')) return { kind: 'absent' }; + throw new DaemonClientError( + 'startupFailed', + `Cannot safely read ${lockfilePath}; refusing automatic reclaim. ${DAEMON_RESET_HINT}`, + { cause: error } + ); + } + let record: unknown; + try { + record = JSON.parse(raw); + } catch { + return { kind: 'corrupt', raw }; + } + return isDaemonOwnership(record) + ? { kind: 'owned', raw, owner: { pid: record.pid, startedAt: record.startedAt } } + : { kind: 'corrupt', raw }; +} + +function describeOwnership(state: OwnershipState, paths: IDaemonPaths): string { + switch (state.kind) { + case 'absent': + return `Socket ${paths.socketPath} has no ownership record`; + case 'corrupt': + return `Invalid daemon ownership record in ${paths.lockfilePath}`; + default: + return `PID ${state.owner.pid} was reused by a process that started after ${paths.lockfilePath} was written`; + } +} + +function removeIfUnchanged(filePath: string, expected: string): boolean { + let current: string; + try { + current = fs.readFileSync(filePath, 'utf8'); + } catch (error) { + if (hasErrorCode(error, 'ENOENT')) return false; + throw error; + } + if (current !== expected) { + throw new DaemonClientError('startupFailed', `${filePath} changed during reclaim; refusing to remove it.`); + } + return tryUnlink(filePath); +} + +function tryUnlink(filePath: string): boolean { + try { + fs.unlinkSync(filePath); + return true; + } catch (error) { + if (hasErrorCode(error, 'ENOENT')) return false; + throw error; + } +} diff --git a/libraries/rush-client-core/src/ProcessStartTime.ts b/libraries/rush-client-core/src/ProcessStartTime.ts new file mode 100644 index 0000000000..a258752e5d --- /dev/null +++ b/libraries/rush-client-core/src/ProcessStartTime.ts @@ -0,0 +1,67 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as fs from 'node:fs'; +import { performance } from 'node:perf_hooks'; + +// USER_HZ is fixed at 100 on mainstream Linux ABIs; it is verified against this process before use. +const USER_HZ: number = 100; +const PROC_STAT_START_TIME_FIELD: number = 22; +const MAX_CALIBRATION_ERROR_MS: number = 1000; +/** A process that began this long after a record was written cannot be the record's writer. */ +const MIN_REUSE_MARGIN_MS: number = 2000; + +/** + * Returns the wall-clock start time of `pid`, or `undefined` when it cannot be determined reliably. + * Only Linux `/proc` is supported; other platforms always return `undefined` (unknown). + */ +export function tryGetProcessStartTimeMs(pid: number): number | undefined { + if (process.platform !== 'linux') return undefined; + const uptimeSeconds: number | undefined = readUptimeSeconds(); + const selfStartSeconds: number | undefined = readStartSeconds('self'); + if (uptimeSeconds === undefined || selfStartSeconds === undefined) return undefined; + const now: number = Date.now(); + const selfStartMs: number = now - (uptimeSeconds - selfStartSeconds) * 1000; + // Guards against an unexpected USER_HZ, a non-standard /proc, or a wall-clock jump since this process began. + if (Math.abs(selfStartMs - performance.timeOrigin) > MAX_CALIBRATION_ERROR_MS) return undefined; + const startSeconds: number | undefined = readStartSeconds(pid); + return startSeconds === undefined ? undefined : now - (uptimeSeconds - startSeconds) * 1000; +} + +/** + * True only when `pid` provably started after `recordedAt`, so it cannot be the process that wrote a record + * at that time (the PID was reused). Unknown start times return false. + */ +export function isProcessStartedAfter(pid: number, recordedAt: string): boolean { + const recordedMs: number = Date.parse(recordedAt); + if (!Number.isFinite(recordedMs)) return false; + const startMs: number | undefined = tryGetProcessStartTimeMs(pid); + return startMs !== undefined && startMs > recordedMs + MIN_REUSE_MARGIN_MS; +} + +function readUptimeSeconds(): number | undefined { + try { + const uptime: number = Number.parseFloat(fs.readFileSync('/proc/uptime', 'utf8').split(' ')[0]); + return Number.isFinite(uptime) ? uptime : undefined; + } catch { + return undefined; + } +} + +function readStartSeconds(pid: number | 'self'): number | undefined { + let stat: string; + try { + stat = fs.readFileSync(`/proc/${pid}/stat`, 'utf8'); + } catch { + return undefined; + } + // The command name (field 2) may contain spaces and parentheses; fields after it never do. + const commandEnd: number = stat.lastIndexOf(')'); + if (commandEnd < 0) return undefined; + const fields: string[] = stat + .slice(commandEnd + 1) + .trim() + .split(' '); + const jiffies: number = Number(fields[PROC_STAT_START_TIME_FIELD - 3]); + return Number.isSafeInteger(jiffies) && jiffies >= 0 ? jiffies / USER_HZ : undefined; +} diff --git a/libraries/rush-client-core/src/connectOrStartDaemon.ts b/libraries/rush-client-core/src/connectOrStartDaemon.ts index 0889608be2..c70284d8e6 100644 --- a/libraries/rush-client-core/src/connectOrStartDaemon.ts +++ b/libraries/rush-client-core/src/connectOrStartDaemon.ts @@ -21,6 +21,15 @@ import { import { DaemonClient, type IDaemonClientConnectOptions } from './DaemonClient'; import { DaemonClientError } from './DaemonClientError'; import { getDaemonLogFilePath } from './DaemonLogFile'; +import { + DAEMON_RESET_HINT, + hasErrorCode, + isDaemonOwnership, + isOwnerProcessAlive, + readDaemonOwnership, + reclaimAbandonedOwnershipAsync, + type DaemonOwnership +} from './DaemonOwnership'; import { getDaemonStartupFilePath, reserveDaemonStartup, @@ -113,7 +122,7 @@ export async function connectOrStartDaemonAsync( if (Date.now() >= deadline) { throw startupError( options, - `has an unresolved startup handoff at ${getDaemonStartupFilePath(options.paths)}; refusing another launch` + `has an unresolved startup handoff at ${getDaemonStartupFilePath(options.paths)}; refusing another launch. ${DAEMON_RESET_HINT}` ); } await delayAsync(Math.min(100, Math.max(1, deadline - Date.now())), undefined, { @@ -127,7 +136,7 @@ export async function connectOrStartDaemonAsync( const handoff: DaemonClient | undefined = await waitForHandoffAsync(options, deadline); if (handoff) return handoff; if (Date.now() >= deadline) throw startupError(options, 'exceeded its deadline before reclaim'); - assertNoLiveOwner(options.paths); + await reclaimAbandonedOwnershipAsync(options.paths); await reclaimStaleDaemonAsync(options.paths); if (Date.now() >= deadline) throw startupError(options, 'exceeded its deadline before spawn'); options.abortSignal?.throwIfAborted(); @@ -276,8 +285,13 @@ async function waitForHandoffAsync( let backoffMs: number = 50; while (Date.now() < deadline) { const owner: IDaemonLockfile | undefined = readDaemonLockfile(options.paths.lockfilePath); - // Only wait on a fully published endpoint; malformed or ambiguous ownership still fails closed. - if (!owner || owner.socketPath !== options.paths.socketPath || !isProcessAlive(owner.pid)) + // Only wait on a fully published endpoint; malformed or ambiguous ownership is resolved by reclaim. + if ( + !owner || + !isDaemonOwnership(owner) || + owner.socketPath !== options.paths.socketPath || + !isOwnerProcessAlive(owner) + ) return undefined; await delayAsync(Math.min(backoffMs, Math.max(1, deadline - Date.now())), undefined, { signal: options.abortSignal @@ -291,57 +305,11 @@ async function waitForHandoffAsync( } function assertNoLiveOwner(paths: IDaemonPaths): void { - const owner: Pick | undefined = readDaemonOwnership( - paths.lockfilePath - ); - if (!owner) { - if (process.platform !== 'win32' && fs.existsSync(paths.socketPath)) { - throw new DaemonClientError( - 'startupFailed', - `Socket ${paths.socketPath} has no ownership record; refusing automatic reclaim.` - ); - } - return; - } - if (!isProcessAlive(owner.pid)) return; + const owner: DaemonOwnership | undefined = readDaemonOwnership(paths.lockfilePath); + if (!owner || !isOwnerProcessAlive(owner)) return; throw new DaemonClientError( 'startupFailed', - `PID ${owner.pid} still exists but the daemon is not ready. It may be a reused PID; refusing to kill it or remove ${paths.lockfilePath}.` - ); -} - -function readDaemonOwnership(lockfilePath: string): Pick | undefined { - let record: unknown; - try { - record = JSON.parse(fs.readFileSync(lockfilePath, 'utf8')); - } catch (error) { - if (hasErrorCode(error, 'ENOENT')) return undefined; - throw new DaemonClientError( - 'startupFailed', - `Cannot safely read ${lockfilePath}; refusing automatic reclaim.`, - { cause: error } - ); - } - if (!isDaemonOwnership(record)) { - throw new DaemonClientError( - 'startupFailed', - `Invalid daemon ownership record in ${lockfilePath}; refusing automatic reclaim.` - ); - } - return { pid: record.pid, startedAt: record.startedAt }; -} - -function isDaemonOwnership(record: unknown): record is Pick { - return ( - typeof record === 'object' && - record !== null && - 'pid' in record && - typeof record.pid === 'number' && - Number.isSafeInteger(record.pid) && - record.pid > 0 && - 'startedAt' in record && - typeof record.startedAt === 'string' && - Number.isFinite(Date.parse(record.startedAt)) + `PID ${owner.pid} still exists but the daemon is not ready. It may be a daemon that is still shutting down, or a reused PID; refusing to kill it or remove ${paths.lockfilePath}. ${DAEMON_RESET_HINT}` ); } @@ -394,7 +362,7 @@ async function waitForPreviousDaemonAsync( deadline, abortSignal ); - if (!owner || owner.pid !== pid || owner.startedAt !== startedAt || !isProcessAlive(owner.pid)) return; + if (!owner || owner.pid !== pid || owner.startedAt !== startedAt || !isOwnerProcessAlive(owner)) return; if (Date.now() >= deadline) { throw new DaemonClientError( 'timeout', @@ -408,16 +376,6 @@ async function waitForPreviousDaemonAsync( } } -function isProcessAlive(pid: number): boolean { - try { - process.kill(pid, 0); - return true; - } catch (error) { - if (hasErrorCode(error, 'ESRCH')) return false; - throw error; - } -} - async function spawnDetachedAsync( options: IConnectOrStartDaemonOptions, deadline: number @@ -526,7 +484,3 @@ function startupError(options: IConnectOrStartDaemonOptions, reason: string): Da `Daemon startup ${reason}. Inspect ${getDaemonLogFilePath(options.paths)} and retry, or use --no-daemon.` ); } - -function hasErrorCode(error: unknown, code: string): boolean { - return typeof error === 'object' && error !== null && 'code' in error && error.code === code; -} diff --git a/libraries/rush-client-core/src/index.ts b/libraries/rush-client-core/src/index.ts index b2962188b7..7597de72ab 100644 --- a/libraries/rush-client-core/src/index.ts +++ b/libraries/rush-client-core/src/index.ts @@ -15,6 +15,11 @@ export { } from './DaemonClient'; export { DaemonClientError, type DaemonClientErrorCode } from './DaemonClientError'; export { getDaemonLogFilePath } from './DaemonLogFile'; +export { + resetDaemonArtifactsAsync, + type IDaemonArtifactResetOptions, + type IDaemonArtifactResetResult +} from './DaemonOwnership'; export { executeWithDaemonRestartAsync } from './executeWithDaemonRestart'; export { connectOrStartDaemonAsync, diff --git a/libraries/rush-client-core/src/test/ProcessStartTime.test.ts b/libraries/rush-client-core/src/test/ProcessStartTime.test.ts new file mode 100644 index 0000000000..a7a9ff4418 --- /dev/null +++ b/libraries/rush-client-core/src/test/ProcessStartTime.test.ts @@ -0,0 +1,31 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { spawn, type ChildProcess } from 'node:child_process'; +import { once } from 'node:events'; +import { performance } from 'node:perf_hooks'; + +import { isProcessStartedAfter, tryGetProcessStartTimeMs } from '../ProcessStartTime'; + +const linuxIt: typeof it = process.platform === 'linux' ? it : it.skip; + +describe('process start time', () => { + linuxIt('estimates this process start time from /proc', () => { + const startMs: number | undefined = tryGetProcessStartTimeMs(process.pid); + expect(startMs).toEqual(expect.any(Number)); + expect(Math.abs(startMs! - performance.timeOrigin)).toBeLessThan(1000); + }); + + linuxIt('detects a PID whose process started after a record was written', () => { + expect(isProcessStartedAfter(process.pid, new Date(Date.now() - 3600000).toISOString())).toBe(true); + expect(isProcessStartedAfter(process.pid, new Date().toISOString())).toBe(false); + }); + + it('treats unknown start times and invalid timestamps as not provably reused', async () => { + const exited: ChildProcess = spawn(process.execPath, ['-e', ''], { stdio: 'ignore' }); + await once(exited, 'close'); + expect(tryGetProcessStartTimeMs(exited.pid!)).toBeUndefined(); + expect(isProcessStartedAfter(exited.pid!, new Date(0).toISOString())).toBe(false); + expect(isProcessStartedAfter(process.pid, 'not a timestamp')).toBe(false); + }); +}); diff --git a/libraries/rush-client-core/src/test/connectOrStartDaemon.test.ts b/libraries/rush-client-core/src/test/connectOrStartDaemon.test.ts index df7fc390cd..0f0957b1d2 100644 --- a/libraries/rush-client-core/src/test/connectOrStartDaemon.test.ts +++ b/libraries/rush-client-core/src/test/connectOrStartDaemon.test.ts @@ -4,16 +4,22 @@ import { spawn, type ChildProcess } from 'node:child_process'; import { once } from 'node:events'; import * as fs from 'node:fs'; +import * as net from 'node:net'; import * as os from 'node:os'; import * as path from 'node:path'; import { setTimeout as delayAsync } from 'node:timers/promises'; import { FileSystem } from '@rushstack/node-core-library'; -import type { IDaemonLockfile, IDaemonPaths } from '@rushstack/rush-daemon-transport'; +import { + readDaemonLockfile, + type IDaemonLockfile, + type IDaemonPaths +} from '@rushstack/rush-daemon-transport'; import { DaemonClient } from '../DaemonClient'; import { captureDaemonRequest } from '../captureDaemonRequest'; import { getDaemonLogFilePath } from '../DaemonLogFile'; +import { resetDaemonArtifactsAsync } from '../DaemonOwnership'; import { connectOrStartDaemonAsync, type IConnectOrStartDaemonOptions } from '../connectOrStartDaemon'; import { executeWithDaemonRestartAsync } from '../executeWithDaemonRestart'; import { getDaemonStartupFilePath } from '../DaemonStartup'; @@ -98,6 +104,19 @@ describe('detached daemon startup', () => { }; } + async function leaveStaleSocketAsync(): Promise { + // A listener killed without cleanup leaves a bound-nowhere socket file behind. + const child: ChildProcess = spawn( + process.execPath, + [ + '-e', + `require('net').createServer().listen(${JSON.stringify(paths.socketPath)}, () => process.kill(process.pid, 'SIGKILL'))` + ], + { stdio: 'ignore' } + ); + await once(child, 'close'); + } + async function killStarterBeforeBindAsync(): Promise { fs.writeFileSync(path.join(folder, 'hold-prebind'), ''); const starter = startClient(); @@ -156,6 +175,7 @@ describe('detached daemon startup', () => { code: 1, stderr: expect.stringContaining('unresolved startup handoff') }); + expect((await result).stderr).toContain('daemon stop --force'); expect(fs.readFileSync(startupPath, 'utf8')).toBe(contents); expect(fs.existsSync(path.join(folder, 'starts'))).toBe(false); }); @@ -581,19 +601,134 @@ describe('detached daemon startup', () => { } ); - it('never reclaims a live or reused PID', async () => { + it('reclaims a parseable record with an invalid timestamp without waiting out the deadline', async () => { + const record: string = JSON.stringify({ + pid: process.pid, + protocolVersion: { major: 0, minor: 6 }, + startedAt: 'invalid', + socketPath: paths.socketPath + }); + fs.writeFileSync(paths.lockfilePath, record); + const started: number = Date.now(); + const client = await connectOrStartDaemonAsync(options); + await client.closeAsync(); + expect(Date.now() - started).toBeLessThan(options.startupTimeoutMs! - 2000); + expect(readDaemonLockfile(paths.lockfilePath)?.pid).not.toBe(process.pid); + }); + + (process.platform === 'win32' ? it.skip : it)( + 'force reset waits for a listener to release the endpoint', + async () => { + fs.writeFileSync(getDaemonStartupFilePath(paths), 'abandoned'); + const listener: net.Server = net.createServer((socket) => socket.destroy()); + await new Promise((resolve) => listener.listen(paths.socketPath, resolve)); + await expect(resetDaemonArtifactsAsync(paths)).rejects.toThrow('still listening'); + const closing: NodeJS.Timeout = setTimeout(() => listener.close(), 300); + try { + expect(await resetDaemonArtifactsAsync(paths, { waitTimeoutMs: 5000 })).toEqual({ + removedPaths: [getDaemonStartupFilePath(paths)] + }); + } finally { + clearTimeout(closing); + listener.close(); + } + } + ); + + it('never reclaims a live PID that may still own the record', async () => { const record: string = JSON.stringify({ pid: process.pid, startedAt: new Date().toISOString() }); fs.writeFileSync(paths.lockfilePath, record); - await expect(connectOrStartDaemonAsync(options)).rejects.toThrow('may be a reused PID'); + await expect(connectOrStartDaemonAsync(options)).rejects.toThrow('or a reused PID'); + await expect(connectOrStartDaemonAsync(options)).rejects.toThrow('daemon stop --force'); + expect(fs.readFileSync(paths.lockfilePath, 'utf8')).toBe(record); + await expect(resetDaemonArtifactsAsync(paths)).rejects.toThrow(`PID ${process.pid} still owns`); expect(fs.readFileSync(paths.lockfilePath, 'utf8')).toBe(record); }); - it('fails closed on corrupt ownership records', async () => { + it('reclaims a corrupt ownership record once the endpoint refuses connections', async () => { fs.writeFileSync(paths.lockfilePath, 'not json'); - await expect(connectOrStartDaemonAsync(options)).rejects.toThrow('refusing automatic reclaim'); - expect(fs.readFileSync(paths.lockfilePath, 'utf8')).toBe('not json'); + const client = await connectOrStartDaemonAsync(options); + await client.closeAsync(); + expect(fs.readFileSync(path.join(folder, 'starts'), 'utf8').trim().split('\n')).toHaveLength(1); + expect(readDaemonLockfile(paths.lockfilePath)?.pid).toBe( + Number(fs.readFileSync(path.join(folder, 'starts'), 'utf8').trim()) + ); }); + it('fails closed on a corrupt ownership record while something still listens', async () => { + fs.writeFileSync(paths.lockfilePath, 'not json'); + const listener: net.Server = net.createServer((socket) => socket.destroy()); + await new Promise((resolve) => listener.listen(paths.socketPath, resolve)); + try { + await expect(connectOrStartDaemonAsync({ ...options, startupTimeoutMs: 2000 })).rejects.toThrow( + 'did not refuse a connection' + ); + await expect(resetDaemonArtifactsAsync(paths)).rejects.toThrow('still listening'); + expect(fs.readFileSync(paths.lockfilePath, 'utf8')).toBe('not json'); + expect(fs.existsSync(path.join(folder, 'starts'))).toBe(false); + } finally { + await new Promise((resolve) => listener.close(() => resolve())); + } + }); + + (process.platform === 'win32' ? it.skip : it)( + 'reclaims a socket without an ownership record once it refuses connections', + async () => { + await leaveStaleSocketAsync(); + expect(fs.existsSync(paths.socketPath)).toBe(true); + const client = await connectOrStartDaemonAsync(options); + await client.closeAsync(); + expect(fs.readFileSync(path.join(folder, 'starts'), 'utf8').trim().split('\n')).toHaveLength(1); + } + ); + + (process.platform === 'linux' ? it : it.skip)( + 'reclaims a record whose live PID started after the record was written', + async () => { + const unrelated: ChildProcess = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { + stdio: 'ignore' + }); + await once(unrelated, 'spawn'); + try { + await leaveStaleSocketAsync(); + fs.writeFileSync( + paths.lockfilePath, + JSON.stringify({ + pid: unrelated.pid, + protocolVersion: { major: 0, minor: 6 }, + startedAt: new Date(Date.now() - 3600000).toISOString(), + socketPath: paths.socketPath + }) + ); + const started: number = Date.now(); + const client = await connectOrStartDaemonAsync(options); + await client.closeAsync(); + // Not the full startup deadline spent polling the unrelated process. + expect(Date.now() - started).toBeLessThan(options.startupTimeoutMs! - 2000); + expect(unrelated.exitCode).toBeNull(); + expect(readDaemonLockfile(paths.lockfilePath)?.pid).not.toBe(unrelated.pid); + } finally { + const closed: Promise = once(unrelated, 'close'); + unrelated.kill('SIGKILL'); + await closed; + } + } + ); + + (process.platform === 'win32' ? it.skip : it)( + 'force reset removes stale artifacts without starting a daemon', + async () => { + await leaveStaleSocketAsync(); + fs.writeFileSync(paths.lockfilePath, 'garbage{'); + fs.writeFileSync(getDaemonStartupFilePath(paths), 'abandoned'); + expect(await resetDaemonArtifactsAsync(paths)).toEqual({ + removedPaths: [paths.lockfilePath, getDaemonStartupFilePath(paths), paths.socketPath] + }); + expect(await resetDaemonArtifactsAsync(paths)).toEqual({ removedPaths: [] }); + expect(fs.existsSync(path.join(folder, 'starts'))).toBe(false); + } + ); + it('starts after ownership release even while the original process remains alive', async () => { const client = await connectOrStartDaemonAsync({ ...options,