[rush-daemon-transport] Reap a dead daemon's orphaned operation processes on reclaim

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"
+}

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<boolean> {
+  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<DaemonOrphanReapOutcome> {
+  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<DaemonOrphanReapOutcome> {
+  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<void> {
+  if (owner) await reapDeadDaemonProcessGroupAsync(owner.pid);
+}

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: " <state> <ppid> <pgrp> ...".
+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<void>;
+  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
+};

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<void> {
-  if (isLockfilePidAlive(readDaemonLockfile(paths.lockfilePath))) {
+  const owner: ReturnType<typeof readDaemonLockfile> = 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);
 }
 

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([]);
+});

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');
+});

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<IOrphanedGroup> {
+  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<boolean> {
+  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();
+});