diff --git a/common/changes/@rushstack/rush-daemon-transport/rushd-reap-orphans_2026-09-24-03-00.json b/common/changes/@rushstack/rush-daemon-transport/rushd-reap-orphans_2026-09-24-03-00.json new file mode 100644 index 0000000000..89a8bf208b --- /dev/null +++ b/common/changes/@rushstack/rush-daemon-transport/rushd-reap-orphans_2026-09-24-03-00.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon-transport", + "comment": "When reclaiming the socket of a daemon that died uncleanly (SIGKILL/OOM), terminate the operation processes still running in its process group (SIGTERM, then SIGKILL after a grace period) so a successor daemon does not re-run them concurrently.", + "type": "patch" + } + ], + "packageName": "@rushstack/rush-daemon-transport", + "email": "TheLarkInn@users.noreply.github.com" +} diff --git a/libraries/rush-daemon-transport/src/DaemonOrphanReaper.ts b/libraries/rush-daemon-transport/src/DaemonOrphanReaper.ts new file mode 100644 index 0000000000..bd4c61534b --- /dev/null +++ b/libraries/rush-daemon-transport/src/DaemonOrphanReaper.ts @@ -0,0 +1,100 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonLockfile } from './DaemonLockfile'; +import type { IDaemonProcessGroupOps } from './DaemonProcessGroup'; +import { POSIX_PROCESS_GROUP_OPS } from './DaemonProcessGroup'; + +const WINDOWS_PLATFORM: NodeJS.Platform = 'win32'; +// 0 and 1 are never daemons, and kill(-0)/kill(-1) would signal our own group or every process. +const FIRST_USER_PID: number = 2; +const POLL_INTERVAL_MS: number = 20; +const DEFAULT_GRACE_MS: number = 2000; + +/** Outcome of reaping a dead daemon's process group. */ +export type DaemonOrphanReapOutcome = 'none' | 'terminated' | 'killed'; + +/** Options for {@link reapDeadDaemonProcessGroupAsync}; every field defaults to the real process. */ +export interface IDaemonOrphanReaperOptions { + readonly ops?: IDaemonProcessGroupOps; + readonly platform?: NodeJS.Platform; + readonly selfPid?: number; + /** How long SIGTERM'd (and then SIGKILL'd) processes get to exit. */ + readonly graceMs?: number; +} + +interface IReapContext { + readonly ops: IDaemonProcessGroupOps; + readonly deadPid: number; + readonly graceMs: number; +} +type OrphanCheck = (pid: number) => boolean; + +function isNeitherSelfNorOwnGroup(pid: number, selfPid: number, ops: IDaemonProcessGroupOps): boolean { + // Fail closed: an unknown own group might be `pid` (e.g. rush-client run by an operation). + const ownGroupId: number | undefined = ops.ownGroupId(); + return pid !== selfPid && ownGroupId !== undefined && pid !== ownGroupId; +} + +function orphanChecks(options: IDaemonOrphanReaperOptions, ops: IDaemonProcessGroupOps): OrphanCheck[] { + return [ + () => (options.platform ?? process.platform) !== WINDOWS_PLATFORM, + (pid: number) => Number.isSafeInteger(pid) && pid >= FIRST_USER_PID, + (pid: number) => isNeitherSelfNorOwnGroup(pid, options.selfPid ?? process.pid, ops), + (pid: number) => !ops.isProcessAlive(pid), + (pid: number) => ops.groupExists(pid) + ]; +} + +async function waitForGroupExitAsync(context: IReapContext): Promise { + const deadline: number = context.ops.now() + context.graceMs; + while (context.ops.now() < deadline) { + if (!context.ops.groupExists(context.deadPid)) return true; + await context.ops.delayAsync(POLL_INTERVAL_MS); + } + return !context.ops.groupExists(context.deadPid); +} + +async function terminateGroupAsync(context: IReapContext): Promise { + context.ops.signalGroup(context.deadPid, 'SIGTERM'); + if (await waitForGroupExitAsync(context)) return 'terminated'; + context.ops.signalGroup(context.deadPid, 'SIGKILL'); + if (await waitForGroupExitAsync(context)) return 'killed'; + throw new Error(`Processes of dead daemon ${context.deadPid} survived SIGKILL; not reclaiming its socket.`); +} + +/** + * Terminates operation processes left behind by a daemon that died without joining them (SIGKILL, OOM). + * + * @remarks + * The daemon is spawned detached, so its pid is its process group id, and phased operation children inherit + * that group (children spawned with their own detached group are out of scope). Sends SIGTERM, then SIGKILL + * after `graceMs`, and throws if the group still has not exited after a further `graceMs`. + * PID-reuse guard: only group `deadPid` is signaled, and only once `deadPid` is proven dead while the group + * still exists; POSIX never reuses a pid still in use as a process group id, so every remaining member + * belongs to the dead daemon. Never signals the caller's pid or group, and does nothing when the caller's + * group is unknown (no `/proc`). Call only under the reclaim mutex. + */ +export async function reapDeadDaemonProcessGroupAsync( + deadPid: number, + options: IDaemonOrphanReaperOptions = {} +): Promise { + const context: IReapContext = createReapContext(deadPid, options); + if (!orphanChecks(options, context.ops).every((check: OrphanCheck) => check(deadPid))) return 'none'; + const outcome: DaemonOrphanReapOutcome = await terminateGroupAsync(context); + context.ops.log(`Reclaimed dead daemon ${deadPid}: its orphaned operation process group was ${outcome}.`); + return outcome; +} + +function createReapContext(deadPid: number, options: IDaemonOrphanReaperOptions): IReapContext { + return { + ops: options.ops ?? POSIX_PROCESS_GROUP_OPS, + deadPid, + graceMs: options.graceMs ?? DEFAULT_GRACE_MS + }; +} + +/** Reaps the orphaned operations of a reclaimed daemon's recorded owner, if there is one. */ +export async function reapOrphansOfDeadOwnerAsync(owner: IDaemonLockfile | undefined): Promise { + if (owner) await reapDeadDaemonProcessGroupAsync(owner.pid); +} diff --git a/libraries/rush-daemon-transport/src/DaemonProcessGroup.ts b/libraries/rush-daemon-transport/src/DaemonProcessGroup.ts new file mode 100644 index 0000000000..a8eb871383 --- /dev/null +++ b/libraries/rush-daemon-transport/src/DaemonProcessGroup.ts @@ -0,0 +1,78 @@ +// 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 { setTimeout as delayAsync } from 'node:timers/promises'; + +import { isDaemonProcessAlive } from './DaemonLockfile'; + +const NO_SIGNAL: number = 0; +const NO_SUCH_PROCESS: string = 'ESRCH'; +const PROC_SELF_STAT: string = '/proc/self/stat'; +const UTF8: BufferEncoding = 'utf8'; +const COMM_END: string = ')'; +const FIELD_SEPARATOR: string = ' '; +// After the ")" that ends the command name come: " ...". +const PGRP_FIELD_INDEX: number = 3; + +/** Process probing/signaling used to reap a dead daemon's process group; injectable for tests. */ +export interface IDaemonProcessGroupOps { + readonly isProcessAlive: (pid: number) => boolean; + readonly groupExists: (groupId: number) => boolean; + readonly signalGroup: (groupId: number, signal: NodeJS.Signals) => void; + /** The caller's own process group id, or `undefined` when the platform cannot report it. */ + readonly ownGroupId: () => number | undefined; + readonly delayAsync: (ms: number) => Promise; + readonly now: () => number; + readonly log: (message: string) => void; +} + +// Only ESRCH proves the group is gone; EPERM and other failures propagate so reclaim fails closed. +function rethrowUnlessNoSuchProcess(error: unknown): void { + if ((error as NodeJS.ErrnoException | undefined)?.code !== NO_SUCH_PROCESS) throw error; +} + +function groupExists(groupId: number): boolean { + try { + process.kill(-groupId, NO_SIGNAL); + return true; + } catch (error) { + rethrowUnlessNoSuchProcess(error); + return false; + } +} + +function signalGroup(groupId: number, signal: NodeJS.Signals): void { + try { + process.kill(-groupId, signal); + } catch (error) { + rethrowUnlessNoSuchProcess(error); + } +} + +function ownGroupId(): number | undefined { + try { + const stat: string = fs.readFileSync(PROC_SELF_STAT, UTF8); + const fields: string[] = stat.slice(stat.lastIndexOf(COMM_END)).split(FIELD_SEPARATOR); + return Number(fields[PGRP_FIELD_INDEX]); + } catch { + return undefined; + } +} + +function log(message: string): void { + process.emitWarning(message, { code: 'RUSH_DAEMON_ORPHANS_REAPED' }); +} + +/** The real POSIX implementation of {@link IDaemonProcessGroupOps}. */ +export const POSIX_PROCESS_GROUP_OPS: IDaemonProcessGroupOps = { + isProcessAlive: isDaemonProcessAlive, + groupExists, + signalGroup, + ownGroupId, + delayAsync: async (ms: number) => { + await delayAsync(ms); + }, + now: Date.now, + log +}; diff --git a/libraries/rush-daemon-transport/src/DaemonReclaim.ts b/libraries/rush-daemon-transport/src/DaemonReclaim.ts index e70d12f0f2..92f0ed9323 100644 --- a/libraries/rush-daemon-transport/src/DaemonReclaim.ts +++ b/libraries/rush-daemon-transport/src/DaemonReclaim.ts @@ -10,6 +10,7 @@ import { readDaemonLockfile, removeDaemonArtifacts } from './DaemonLockfile'; +import { reapOrphansOfDeadOwnerAsync } from './DaemonOrphanReaper'; import type { IDaemonPaths } from './DaemonPaths'; import { tryAcquireReclaimLock } from './DaemonReclaimLock'; import type { DaemonReclaimLockOutcome } from './DaemonReclaimLock'; @@ -24,7 +25,8 @@ import { DaemonTransportError, DaemonTransportErrorCode } from './DaemonTranspor * never reclaimed underneath itself. Reclaims are serialized through the * lockfile mutex ({@link tryAcquireReclaimLock}): only the mutex holder may * unlink the socket path, so a concurrent starter cannot delete a socket that - * another process just bound. + * another process just bound. Operation processes still running in the dead + * daemon's process group are terminated first (see `DaemonOrphanReaper`). * * @throws {@link DaemonTransportError} with code `daemonAlreadyRunning` when a * live (or plausibly live) daemon owns the path, or when another starter holds @@ -57,13 +59,16 @@ function reclaimLockPath(paths: IDaemonPaths): string { } async function reclaimUnderLockAsync(paths: IDaemonPaths): Promise { - if (isLockfilePidAlive(readDaemonLockfile(paths.lockfilePath))) { + const owner: ReturnType = readDaemonLockfile(paths.lockfilePath); + if (isLockfilePidAlive(owner)) { throwAlreadyRunning(paths, 'its lockfile PID is alive'); } const probeFailed: boolean = await probeConnectionFailsAsync(paths.socketPath); if (!probeFailed) { throwAlreadyRunning(paths, 'it answers a connect probe'); } + // A daemon that died uncleanly leaves its operations running; stop them before a successor re-runs them. + await reapOrphansOfDeadOwnerAsync(owner); removeDaemonArtifacts(paths.lockfilePath, paths.socketPath); } diff --git a/libraries/rush-daemon-transport/src/test/DaemonOrphanReaper.test.ts b/libraries/rush-daemon-transport/src/test/DaemonOrphanReaper.test.ts new file mode 100644 index 0000000000..a775495413 --- /dev/null +++ b/libraries/rush-daemon-transport/src/test/DaemonOrphanReaper.test.ts @@ -0,0 +1,64 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { reapDeadDaemonProcessGroupAsync } from '../DaemonOrphanReaper'; +import type { IDaemonOrphanReaperOptions } from '../DaemonOrphanReaper'; + +import { DEAD_PID, SELF_PID, createFakeGroup } from './OrphanReaperFixture'; +import type { IFakeGroup } from './OrphanReaperFixture'; + +const INIT_PID: number = 1; + +it('stops at SIGTERM when the orphaned group exits within the grace period', async () => { + const fake: IFakeGroup = createFakeGroup({ exitsOn: 'SIGTERM' }); + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID, fake.options)).resolves.toBe('terminated'); + expect(fake.signals).toEqual(['SIGTERM']); + expect(fake.logs).toEqual([expect.stringContaining(`dead daemon ${DEAD_PID}`)]); +}); + +it('escalates to SIGKILL when the group outlives the grace period', async () => { + const fake: IFakeGroup = createFakeGroup({ exitsOn: 'SIGKILL' }); + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID, fake.options)).resolves.toBe('killed'); + expect(fake.signals).toEqual(['SIGTERM', 'SIGKILL']); + expect(fake.logs).toEqual([expect.stringContaining('killed')]); +}); + +it('fails the reclaim when the group is still present after SIGKILL', async () => { + const fake: IFakeGroup = createFakeGroup({}); + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID, fake.options)).rejects.toThrow(/survived SIGKILL/); + expect(fake.signals).toEqual(['SIGTERM', 'SIGKILL']); +}); + +it('never signals a live daemon (its pid is not proven dead)', async () => { + const fake: IFakeGroup = createFakeGroup({ daemonAlive: true }); + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID, fake.options)).resolves.toBe('none'); + expect(fake.signals).toEqual([]); +}); + +it('never signals when no group with the dead pid remains', async () => { + const fake: IFakeGroup = createFakeGroup({}); + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID + INIT_PID, fake.options)).resolves.toBe('none'); + expect(fake.signals).toEqual([]); + expect(fake.logs).toEqual([]); +}); + +it.each([INIT_PID, SELF_PID, Number.NaN])('never signals the unsafe group id %p', async (pid: number) => { + const fake: IFakeGroup = createFakeGroup({ anyGroupExists: true, ownGroupId: DEAD_PID + INIT_PID }); + await expect(reapDeadDaemonProcessGroupAsync(pid, fake.options)).resolves.toBe('none'); + expect(fake.signals).toEqual([]); +}); + +it('never signals the caller own process group, or when that group is unknown', async () => { + const own: IFakeGroup = createFakeGroup({ ownGroupId: DEAD_PID }); + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID, own.options)).resolves.toBe('none'); + const unknown: IFakeGroup = createFakeGroup({ unknownOwnGroup: true }); + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID, unknown.options)).resolves.toBe('none'); + expect([...own.signals, ...unknown.signals]).toEqual([]); +}); + +it('never signals anything on Windows', async () => { + const win: IFakeGroup = createFakeGroup({}); + const winOptions: IDaemonOrphanReaperOptions = { ...win.options, platform: 'win32' }; + await expect(reapDeadDaemonProcessGroupAsync(DEAD_PID, winOptions)).resolves.toBe('none'); + expect(win.signals).toEqual([]); +}); diff --git a/libraries/rush-daemon-transport/src/test/DaemonProcessGroup.test.ts b/libraries/rush-daemon-transport/src/test/DaemonProcessGroup.test.ts new file mode 100644 index 0000000000..8ab89515a3 --- /dev/null +++ b/libraries/rush-daemon-transport/src/test/DaemonProcessGroup.test.ts @@ -0,0 +1,32 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { POSIX_PROCESS_GROUP_OPS } from '../DaemonProcessGroup'; + +const GROUP_ID: number = 4242; + +function throwErrno(code: string): never { + throw Object.assign(new Error(code), { code }); +} + +afterEach(() => { + jest.restoreAllMocks(); +}); + +it('treats only ESRCH as proof that a process group is gone', () => { + jest.spyOn(process, 'kill').mockImplementation(() => throwErrno('ESRCH')); + expect(POSIX_PROCESS_GROUP_OPS.groupExists(GROUP_ID)).toBe(false); + expect(() => POSIX_PROCESS_GROUP_OPS.signalGroup(GROUP_ID, 'SIGTERM')).not.toThrow(); +}); + +it('propagates EPERM instead of assuming the group is gone', () => { + jest.spyOn(process, 'kill').mockImplementation(() => throwErrno('EPERM')); + expect(() => POSIX_PROCESS_GROUP_OPS.groupExists(GROUP_ID)).toThrow('EPERM'); + expect(() => POSIX_PROCESS_GROUP_OPS.signalGroup(GROUP_ID, 'SIGKILL')).toThrow('EPERM'); +}); + +it('signals the negated group id', () => { + const kill: jest.SpyInstance = jest.spyOn(process, 'kill').mockImplementation(() => true); + POSIX_PROCESS_GROUP_OPS.signalGroup(GROUP_ID, 'SIGTERM'); + expect(kill).toHaveBeenCalledWith(-GROUP_ID, 'SIGTERM'); +}); diff --git a/libraries/rush-daemon-transport/src/test/OrphanReapOnReclaim.test.ts b/libraries/rush-daemon-transport/src/test/OrphanReapOnReclaim.test.ts new file mode 100644 index 0000000000..2e4a96a4cf --- /dev/null +++ b/libraries/rush-daemon-transport/src/test/OrphanReapOnReclaim.test.ts @@ -0,0 +1,70 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { spawn } from 'node:child_process'; +import type { ChildProcess } from 'node:child_process'; +import { once } from 'node:events'; +import type { Readable } from 'node:stream'; +import { setTimeout as delayAsync } from 'node:timers/promises'; + +import { DAEMON_PROTOCOL_VERSION } from '@rushstack/rush-daemon-protocol'; + +import { isDaemonProcessAlive, writeDaemonLockfile } from '../DaemonLockfile'; +import type { IDaemonPaths } from '../DaemonPaths'; +import { reclaimStaleDaemonAsync } from '../DaemonReclaim'; + +import { createTestDaemonPaths } from './TestDaemonFixture'; + +const posixIt: jest.It = process.platform === 'win32' ? it.skip : it; +const FIRST_ATTEMPT: number = 0; +const POLL_ATTEMPTS: number = 100; +const POLL_INTERVAL_MS: number = 20; +// A stand-in daemon: spawns one operation child and prints the child pid. Like a phased operation (spawned +// without `detached`), the child inherits the daemon's process group. +const FAKE_DAEMON_SCRIPT: string = + "const c=require('node:child_process').spawn(process.execPath,['-e','setInterval(()=>{},1000)'],{stdio:'ignore'});" + + "process.stdout.write(String(c.pid)+'\\n');setInterval(()=>{},1000);"; + +interface IOrphanedGroup { + readonly daemonPid: number; + readonly orphanPid: number; +} + +/** Spawns a detached fake daemon with one child in its group, then SIGKILLs only the daemon. */ +async function createOrphanedGroupAsync(): Promise { + const daemon: ChildProcess = spawn(process.execPath, ['-e', FAKE_DAEMON_SCRIPT], { + detached: true, + stdio: ['ignore', 'pipe', 'ignore'] + }); + const stdout: Readable = daemon.stdout as Readable; + const [chunk] = (await once(stdout, 'data')) as [Buffer]; + daemon.kill('SIGKILL'); + await once(daemon, 'exit'); + stdout.destroy(); + return { daemonPid: Number(daemon.pid), orphanPid: Number(chunk.toString().trim()) }; +} + +async function waitUntilDeadAsync(pid: number): Promise { + for (let attempt: number = FIRST_ATTEMPT; attempt < POLL_ATTEMPTS; attempt++) { + if (!isDaemonProcessAlive(pid)) return true; + await delayAsync(POLL_INTERVAL_MS); + } + return false; +} + +posixIt('reaps operation processes orphaned by a SIGKILLed daemon before reclaiming', async () => { + const warning: jest.SpyInstance = jest.spyOn(process, 'emitWarning').mockImplementation(() => undefined); + const { daemonPid, orphanPid } = await createOrphanedGroupAsync(); + expect(isDaemonProcessAlive(orphanPid)).toBe(true); + const paths: IDaemonPaths = createTestDaemonPaths(); + writeDaemonLockfile(paths.lockfilePath, { + pid: daemonPid, + protocolVersion: DAEMON_PROTOCOL_VERSION, + startedAt: new Date().toISOString(), + socketPath: paths.socketPath + }); + await reclaimStaleDaemonAsync(paths); + expect(await waitUntilDeadAsync(orphanPid)).toBe(true); + expect(warning).toHaveBeenCalledWith(expect.stringContaining(`dead daemon ${daemonPid}`), expect.anything()); + warning.mockRestore(); +}); diff --git a/libraries/rush-daemon-transport/src/test/OrphanReaperFixture.ts b/libraries/rush-daemon-transport/src/test/OrphanReaperFixture.ts new file mode 100644 index 0000000000..c0e1f8bede --- /dev/null +++ b/libraries/rush-daemon-transport/src/test/OrphanReaperFixture.ts @@ -0,0 +1,50 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonOrphanReaperOptions } from '../DaemonOrphanReaper'; +import type { IDaemonProcessGroupOps } from '../DaemonProcessGroup'; + +/** The pid of the fake dead daemon, which is also its process group id. */ +export const DEAD_PID: number = 4242; +/** The fake caller's own pid (and, by default, its process group id). */ +export const SELF_PID: number = 1000; +const GRACE_MS: number = 100; +const CLOCK_START: number = 0; + +/** A fake process table recording every signal sent and every message logged. */ +export interface IFakeGroup { + readonly signals: NodeJS.Signals[]; + readonly logs: string[]; + readonly options: IDaemonOrphanReaperOptions; +} + +/** Describes the fake process table. */ +export interface IFakeGroupSpec { + readonly daemonAlive?: boolean; + /** The signal after which the group is gone; omitted means it never exits. */ + readonly exitsOn?: NodeJS.Signals; + readonly ownGroupId?: number; + readonly unknownOwnGroup?: boolean; + readonly anyGroupExists?: boolean; +} + +/** Creates a fake process table with a virtual clock. */ +export function createFakeGroup(spec: IFakeGroupSpec): IFakeGroup { + const signals: NodeJS.Signals[] = []; + const logs: string[] = []; + let clock: number = CLOCK_START; + const ops: IDaemonProcessGroupOps = { + isProcessAlive: () => spec.daemonAlive === true, + groupExists: (groupId: number) => + (spec.anyGroupExists === true || groupId === DEAD_PID) && + !signals.some((signal: NodeJS.Signals) => signal === spec.exitsOn), + signalGroup: (groupId: number, signal: NodeJS.Signals) => signals.push(signal), + ownGroupId: () => (spec.unknownOwnGroup === true ? undefined : (spec.ownGroupId ?? SELF_PID)), + delayAsync: async (ms: number) => { + clock += ms; + }, + now: () => clock, + log: (message: string) => logs.push(message) + }; + return { signals, logs, options: { ops, platform: 'linux', selfPid: SELF_PID, graceMs: GRACE_MS } }; +}