[rush] rushd: fix the bugs found by dogfooding the Rush daemon (stacked on #6046)

apps/heft/src/cli/HeftCommandLineParser.test.ts

@@ -0,0 +1,140 @@
+// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license.
+// See LICENSE in the project root for license information.
+
+import * as childProcess from 'node:child_process';
+import * as fs from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
+
+jest.setTimeout(30_000);
+
+const HEFT_START_PATH: string = path.resolve(__dirname, '../start.js');
+
+// A project with one task. The task sets process.exitCode to FIXTURE_EXIT_CODE in the Heft process, the
+// way a test that Jest runs in band can, and then fails if FIXTURE_FAIL is set.
+const FIXTURE_FILES: Record<string, string> = {
+  'package.json': JSON.stringify({ name: 'heft-exit-code-fixture', version: '1.0.0', private: true }),
+  'heft-plugin.json': JSON.stringify({
+    taskPlugins: [{ pluginName: 'exit-code-fixture-plugin', entryPoint: './plugin.js' }]
+  }),
+  'config/heft.json': JSON.stringify({
+    phasesByName: {
+      test: { tasksByName: { fixture: { taskPlugin: { pluginPackage: 'heft-exit-code-fixture' } } } }
+    }
+  }),
+  'plugin.js': `module.exports = class {
+  apply(taskSession) {
+    taskSession.hooks.run.tapPromise('exit-code-fixture-plugin', async () => {
+      process.exitCode = Number(process.env.FIXTURE_EXIT_CODE);
+      if (process.env.FIXTURE_FAIL) {
+        throw new Error('The fixture task failed on purpose');
+      }
+    });
+  }
+};
+`
+};
+
+// The same project, but its heft.json declares an alias with the name of the "test" phase's action, so Heft fails
+// before any task runs, with an error that nothing has reported yet.
+const ALIAS_CLASH_FIXTURE_FILES: Record<string, string> = {
+  ...FIXTURE_FILES,
+  'config/heft.json': JSON.stringify({
+    ...JSON.parse(FIXTURE_FILES['config/heft.json']),
+    aliasesByName: { test: { actionName: 'test' } }
+  })
+};
+
+function createFixtureFolder(files: Record<string, string>): string {
+  const fixtureFolderPath: string = fs.mkdtempSync(path.join(os.tmpdir(), 'heft-exit-code-'));
+  for (const [relativePath, contents] of Object.entries(files)) {
+    const filePath: string = path.join(fixtureFolderPath, relativePath);
+    fs.mkdirSync(path.dirname(filePath), { recursive: true });
+    fs.writeFileSync(filePath, contents);
+  }
+  return fixtureFolderPath;
+}
+
+interface IHeftResult {
+  exitCode: number | undefined;
+  output: string;
+}
+
+describe('HeftCommandLineParser', () => {
+  let fixtureFolderPath: string;
+  let aliasClashFixtureFolderPath: string;
+
+  beforeAll(() => {
+    fixtureFolderPath = createFixtureFolder(FIXTURE_FILES);
+    aliasClashFixtureFolderPath = createFixtureFolder(ALIAS_CLASH_FIXTURE_FILES);
+  });
+
+  afterAll(() => {
+    fs.rmSync(fixtureFolderPath, { recursive: true, force: true });
+    fs.rmSync(aliasClashFixtureFolderPath, { recursive: true, force: true });
+  });
+
+  async function runHeftTestAsync(
+    taskExitCode: number,
+    taskFails: boolean,
+    folderPath: string = fixtureFolderPath
+  ): Promise<IHeftResult> {
+    const env: NodeJS.ProcessEnv = { ...process.env, FIXTURE_EXIT_CODE: String(taskExitCode) };
+    delete env.FIXTURE_FAIL;
+    delete env._RUSH_REPORTER_CHILD_FD;
+    delete env._RUSH_REPORTER_CHILD_ACK_FD;
+    if (taskFails) {
+      env.FIXTURE_FAIL = '1';
+    }
+
+    const child: childProcess.ChildProcess = childProcess.spawn(process.execPath, [HEFT_START_PATH, 'test'], {
+      cwd: folderPath,
+      env,
+      stdio: ['ignore', 'pipe', 'pipe']
+    });
+    let output: string = '';
+    const appendOutput: (chunk: string) => void = (chunk: string) => {
+      output += chunk;
+    };
+    child.stdout?.setEncoding('utf8').on('data', appendOutput);
+    child.stderr?.setEncoding('utf8').on('data', appendOutput);
+    const exitCode: number | null = await new Promise((resolve, reject) => {
+      child.once('error', reject);
+      child.once('close', resolve);
+    });
+
+    return { exitCode: exitCode ?? undefined, output };
+  }
+
+  it('exits with 1 when a task fails after code in the Heft process set process.exitCode to 0', async () => {
+    const { exitCode, output }: IHeftResult = await runHeftTestAsync(0, true);
+    expect(output).toContain('The fixture task failed on purpose');
+    expect(exitCode).toBe(1);
+  });
+
+  it('exits with 1 when a task fails after code in the Heft process set a negative process.exitCode', async () => {
+    const { exitCode, output }: IHeftResult = await runHeftTestAsync(-1, true);
+    expect(output).toContain('The fixture task failed on purpose');
+    expect(exitCode).toBe(1);
+  });
+
+  it('keeps a positive process.exitCode when a task fails', async () => {
+    const { exitCode, output }: IHeftResult = await runHeftTestAsync(3, true);
+    expect(output).toContain('The fixture task failed on purpose');
+    expect(exitCode).toBe(3);
+  });
+
+  it('exits with 0 when the task succeeds after code in the Heft process set process.exitCode to 0', async () => {
+    const { exitCode, output }: IHeftResult = await runHeftTestAsync(0, false);
+    expect(output).not.toContain('The fixture task failed on purpose');
+    expect(exitCode).toBe(0);
+  });
+
+  it('writes an error that nothing reported before it exits, and exits with 1', async () => {
+    const { exitCode, output }: IHeftResult = await runHeftTestAsync(0, false, aliasClashFixtureFolderPath);
+    expect(output).toContain(
+      'The alias "test" specified in heft.json cannot be used because an action with that name already exists.'
+    );
+    expect(exitCode).toBe(1);
+  });
+});

apps/heft/src/cli/HeftCommandLineParser.ts

@@ -26,7 +26,7 @@ import { PhaseAction } from './actions/PhaseAction';
 import { RunAction } from './actions/RunAction';
 import type { IHeftActionOptions } from './actions/IHeftAction';
 import { AliasAction } from './actions/AliasAction';
-import { getToolParameterNamesFromArgs } from '../utilities/CliUtilities';
+import { getErrorExitCode, getToolParameterNamesFromArgs } from '../utilities/CliUtilities';
 import { Constants } from '../utilities/Constants';
 import { HeftChildReporter } from '../pluginFramework/logging/HeftChildReporter';
 
@@ -271,11 +271,6 @@ export class HeftCommandLineParser extends CommandLineParser {
       this.globalTerminal.writeErrorLine(error.stack!);
     }
 
-    const exitCode: string | number | undefined = process.exitCode;
-    if (!exitCode || typeof exitCode !== 'number' || exitCode > 0) {
-      process.exit(exitCode);
-    } else {
-      process.exit(1);
-    }
+    process.exit(getErrorExitCode(process.exitCode));
   }
 }

apps/heft/src/utilities/CliUtilities.ts

@@ -20,3 +20,15 @@ export function getToolParameterNamesFromArgs(argv: string[] = process.argv): Se
   }
   return toolParameters;
 }
+
+/**
+ * Get the exit code for a Heft process that is exiting because of an error. This is `exitCode` if it
+ * is a positive integer, and 1 otherwise, so it is never 0.
+ *
+ * @param exitCode - The current value of `process.exitCode`. Code that ran in the Heft process, such as
+ * a test that Jest ran in band, may have set it to 0.
+ */
+export function getErrorExitCode(exitCode: string | number | undefined): number {
+  const numericExitCode: number = Number(exitCode);
+  return Number.isInteger(numericExitCode) && numericExitCode > 0 ? numericExitCode : 1;
+}

apps/heft/src/utilities/WatchFileSystemAdapter.ts

@@ -129,6 +129,126 @@ interface ITimeEntry {
   safeTime: number;
 }
 
+/**
+ * What a run read. The run's watcher checks the changes that it reports against this.
+ */
+interface IRunInputs {
+  /**
+   * When the run started. The watcher reports files whose times are close to this time, or later.
+   */
+  baseline: number | undefined;
+  /**
+   * The time of each file that the run read, as the run recorded it
+   */
+  files: ReadonlyMap<string, number>;
+  /**
+   * The folders that the run read
+   */
+  contexts: ReadonlyMap<string, number>;
+  /**
+   * The times that the previous watcher had when the run started
+   */
+  times: ReadonlyMap<string, ITimeEntry> | undefined;
+}
+
+/**
+ * The fields of watchpack's internal `DirectoryWatcher` that {@link refreshPendingTimes} reads.
+ * They aren't part of watchpack's public API, so they are all optional.
+ */
+interface IDirectoryWatcherInternals {
+  path?: string;
+  /**
+   * The names of the files that have an OS event whose `fs.lstat()` hasn't finished yet. The change is
+   * recorded only when the `fs.lstat()` finishes.
+   */
+  _activeEvents?: Map<string, boolean>;
+}
+
+/**
+ * A watchpack instance with the internal state that {@link refreshPendingTimes} reads
+ */
+interface IWatchpackWithInternals extends Watchpack {
+  watcherManager?: {
+    directoryWatchers?: Map<string, IDirectoryWatcherInternals>;
+  };
+}
+
+const OUTDATED_ON_ATTACH_EXPLANATION: string = 'watch (outdated on attach)';
+// A new watcher's first scan reports each file whose mtime is later than the watcher's start time, less the file
+// system's accuracy. A watcher that attaches to the watcher of a folder that has already scanned reports the
+// folder in the same way. Watchpack's other explanations are for events from the file system.
+const SCAN_EXPLANATIONS: ReadonlySet<string> = new Set(['scan (file)', OUTDATED_ON_ATTACH_EXPLANATION]);
+
+function tryLstatSync(filePath: string): fs.Stats | undefined {
+  try {
+    return fs.lstatSync(filePath, { throwIfNoEntry: false });
+  } catch {
+    return undefined;
+  }
+}
+
+function getTimestamp(stats: fs.Stats): number {
+  return stats.mtime.getTime() || stats.ctime.getTime() || Date.now();
+}
+
+/**
+ * Watchpack records a file's new time only when its `fs.lstat()` after the file system's event finishes. If a
+ * run starts before then, the previous watcher's time for the file is from before the change. Read the time
+ * again, so that the run sees the change, and so that the run's watcher knows that the run saw it.
+ */
+function refreshPendingTimes(watcher: Watchpack, times: Map<string, ITimeEntry>): void {
+  const directoryWatchers: Map<string, IDirectoryWatcherInternals> | undefined = (
+    watcher as IWatchpackWithInternals
+  ).watcherManager?.directoryWatchers;
+  if (!(directoryWatchers instanceof Map)) {
+    return;
+  }
+
+  for (const { path: folderPath, _activeEvents: pendingNames } of directoryWatchers.values()) {
+    if (typeof folderPath !== 'string' || !(pendingNames instanceof Map)) {
+      continue;
+    }
+
+    for (const name of pendingNames.keys()) {
+      const filePath: string = path.join(folderPath, name);
+      const stats: fs.Stats | undefined = tryLstatSync(filePath);
+      if (!stats) {
+        times.delete(filePath);
+      } else if (!stats.isDirectory()) {
+        const timestamp: number = getTimestamp(stats);
+        times.set(filePath, { timestamp, safeTime: timestamp });
+      }
+    }
+  }
+}
+
+/**
+ * Returns false if the watcher reported the path only because the path's time is close to the run's start, and
+ * the run has already read the path as it is now.
+ */
+function isReportedChangeNew(filePath: string, explanation: string, inputs: IRunInputs): boolean {
+  if (!SCAN_EXPLANATIONS.has(explanation) || inputs.baseline === undefined) {
+    return true;
+  }
+
+  const stats: fs.Stats | undefined = tryLstatSync(filePath);
+  if (!stats) {
+    return true;
+  }
+
+  if (stats.isDirectory()) {
+    // The watcher of a folder that the run read reports the folder's files one by one
+    return explanation !== OUTDATED_ON_ATTACH_EXPLANATION || !inputs.contexts.has(filePath);
+  }
+
+  if (stats.ctimeMs >= inputs.baseline) {
+    return true;
+  }
+
+  const readTime: number | undefined = inputs.files.get(filePath) ?? inputs.times?.get(filePath)?.timestamp;
+  return readTime !== getTimestamp(stats);
+}
+
 /**
  * A filesystem adapter for use with the "fast-glob" package. This adapter tracks file system accesses
  * to initialize `watchpack`.
@@ -261,11 +381,21 @@ export class WatchFileSystemAdapter implements IWatchFileSystemAdapter {
   public setBaseline(): void {
     this.#lastQueryTime = Date.now();
 
-    if (this.#watcher) {
-      const times: Map<string, { timestamp: number; safeTime: number }> = new Map();
-      this.#watcher.pause();
-      this.#watcher.collectTimeInfoEntries(times, times);
+    const watcher: Watchpack | undefined = this.#watcher;
+    if (watcher) {
+      this.#watcher = undefined;
+      const times: Map<string, ITimeEntry> = new Map();
+      watcher.collectTimeInfoEntries(times, times);
+      refreshPendingTimes(watcher, times);
+      // Close the previous watcher instead of only pausing it. A paused watcher keeps its directory watchers,
+      // so every run would leak another set of them. A kept watcher also keeps its OS watch on a folder that
+      // was deleted and recreated, and later watchers on that folder would share the dead watch, so edits in
+      // the recreated folder would be reported one run late.
+      watcher.close();
       this.#times = times;
+    } else {
+      // Nothing is watching, so times collected for an earlier run may be out of date.
+      this.#times = undefined;
     }
   }
 
@@ -277,11 +407,39 @@ export class WatchFileSystemAdapter implements IWatchFileSystemAdapter {
       return;
     }
 
+    const inputs: IRunInputs = {
+      baseline: this.#lastQueryTime,
+      files: this.#files,
+      contexts: this.#contexts,
+      times: this.#times
+    };
+
     const watcher: Watchpack = new Watchpack({
       aggregateTimeout: 0,
       followSymlinks: false
     });
 
+    // A file that changed just before the run started is reported again when the watcher starts, although the
+    // run has read the change. Call onChange only for a change that the run may not have seen.
+    let hasNewChange: boolean = false;
+    const onReportedChange = (filePath: string, modifiedTime: number, explanation: string): void => {
+      hasNewChange ||= isReportedChangeNew(filePath, explanation, inputs);
+    };
+    const onReportedRemove = (): void => {
+      hasNewChange = true;
+    };
+    const onAggregated = (): void => {
+      if (hasNewChange) {
+        watcher.off('change', onReportedChange);
+        watcher.off('remove', onReportedRemove);
+        watcher.off('aggregated', onAggregated);
+        onChange();
+      }
+    };
+    watcher.on('change', onReportedChange);
+    watcher.on('remove', onReportedRemove);
+    watcher.on('aggregated', onAggregated);
+
     this.#watcher = watcher;
     watcher.watch({
       files: this.#files.keys(),
@@ -290,12 +448,11 @@ export class WatchFileSystemAdapter implements IWatchFileSystemAdapter {
       startTime: this.#lastQueryTime
     });
 
+    // The watcher checks its reports against `inputs`, so the next run records into new maps
     this.#lastFiles = this.#files;
     this.#files = new Map();
-    this.#contexts.clear();
+    this.#contexts = new Map();
     this.#missing.clear();
-
-    watcher.once('aggregated', onChange);
   }
 
   /**