Services: Fix memory leaks in watch mode

.github/workflows/tests.yml

@@ -64,3 +64,17 @@ jobs:
       - run: npm ci
       - run: npm run lint
       - run: npm run format:check
+
+  test-garbage-collection:
+    timeout-minutes: 5
+    runs-on: ubuntu-20.04
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-node@v3
+        with:
+          node-version: 16
+          cache: npm
+      - uses: google/wireit@setup-github-actions-caching/v1
+
+      - run: npm ci
+      - run: npm run test:gc

package.json

@@ -44,6 +44,7 @@
     "test:failures": "wireit",
     "test:freshness": "wireit",
     "test:ide": "wireit",
+    "test:gc": "wireit",
     "test:glob": "wireit",
     "test:json-schema": "wireit",
     "test:optimize-mkdirs": "wireit",
@@ -221,6 +222,14 @@
       "files": [],
       "output": []
     },
+    "test:gc": {
+      "command": "cross-env NODE_OPTIONS=--enable-source-maps node --expose-gc node_modules/uvu/bin.js lib/test \"^gc\\.test\\.js$\"",
+      "dependencies": [
+        "build"
+      ],
+      "files": [],
+      "output": []
+    },
     "test:glob": {
       "command": "cross-env NODE_OPTIONS=--enable-source-maps uvu lib/test \"^glob\\.test\\.js$\"",
       "dependencies": [

src/cli.ts

@@ -9,7 +9,6 @@ import {Analyzer} from './analyzer.js';
 import {Executor} from './executor.js';
 import {WorkerPool} from './util/worker-pool.js';
 import {unreachable} from './util/unreachable.js';
-import {Deferred} from './util/deferred.js';
 import {Failure} from './event.js';
 import {logger, getOptions} from './cli-options.js';
 
@@ -71,22 +70,20 @@ const run = async (): Promise<Result<void, Failure[]>> => {
     }
   }
 
-  const abort = new Deferred<void>();
-  process.on('SIGINT', () => {
-    abort.resolve();
-  });
-
   if (options.watch) {
     const {Watcher} = await import('./watcher.js');
-    await Watcher.watch(
+    const watcher = new Watcher(
       options.script,
       options.extraArgs,
       logger,
       workerPool,
       cache,
-      options.failureMode,
-      abort
+      options.failureMode
     );
+    process.on('SIGINT', () => {
+      watcher.abort();
+    });
+    await watcher.watch();
   } else {
     const analyzer = new Analyzer();
     const {config} = await analyzer.analyze(options.script, options.extraArgs);
@@ -99,9 +96,11 @@ const run = async (): Promise<Result<void, Failure[]>> => {
       workerPool,
       cache,
       options.failureMode,
-      abort,
       undefined
     );
+    process.on('SIGINT', () => {
+      executor.abort();
+    });
     const result = await executor.execute();
     if (!result.ok) {
       return result;

src/event.ts

@@ -93,7 +93,8 @@ export type Failure =
   | DependencyOnMissingPackageJson
   | DependencyOnMissingScript
   | DependencyInvalid
-  | ServiceExitedUnexpectedly;
+  | ServiceExitedUnexpectedly
+  | Aborted;
 
 interface ErrorBase<T extends PackageReference = ScriptReference>
   extends EventBase<T> {
@@ -249,6 +250,14 @@ export interface ServiceExitedUnexpectedly extends ErrorBase {
   reason: 'service-exited-unexpectedly';
 }
 
+/**
+ * A script was killed or is refusing to run because it was intentionally
+ * aborted. Usually due to an error occuring in another script somewhere.
+ */
+export interface Aborted extends ErrorBase {
+  reason: 'aborted';
+}
+
 /**
  * We reached the point of doing cyclic dependency checking, and one of our
  * transitive dependencies had not transitioned to being locally validated.

src/execution/base.ts

@@ -26,6 +26,20 @@ export type ExecutionResult = Result<Fingerprint, Failure[]>;
  */
 export type FailureMode = 'no-new' | 'continue' | 'kill';
 
+let executionConstructorHook:
+  | ((executor: BaseExecution<ScriptConfig>) => void)
+  | undefined;
+
+/**
+ * For GC testing only. A function that is called whenever an Execution is
+ * constructed.
+ */
+export function registerExecutionConstructorHook(
+  fn: typeof executionConstructorHook
+) {
+  executionConstructorHook = fn;
+}
+
 /**
  * A single execution of a specific script.
  */
@@ -36,6 +50,7 @@ export abstract class BaseExecution<T extends ScriptConfig> {
   private _fingerprint?: Promise<ExecutionResult>;
 
   constructor(config: T, executor: Executor, logger: Logger) {
+    executionConstructorHook?.(this);
     this._config = config;
     this._executor = executor;
     this._logger = logger;

src/execution/service.ts

@@ -525,6 +525,7 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand<ServiceScri
         });
         return this._state.started.promise;
       }
+      case 'depsStarting':
       case 'starting': {
         return this._state.started.promise;
       }
@@ -535,13 +536,23 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand<ServiceScri
       case 'failed': {
         return Promise.resolve({ok: false, error: [this._state.failure]});
       }
+      case 'stopping':
+      case 'stopped': {
+        return Promise.resolve({
+          ok: false,
+          error: [
+            {
+              type: 'failure',
+              script: this._config,
+              reason: 'aborted',
+            },
+          ],
+        });
+      }
       case 'initial':
       case 'executingDeps':
-      case 'fingerprinting':
       case 'stoppingAdoptee':
-      case 'depsStarting':
-      case 'stopping':
-      case 'stopped':
+      case 'fingerprinting':
       case 'detached': {
         throw unexpectedState(this._state);
       }

src/executor.ts

@@ -47,6 +47,18 @@ export type ServiceMap = Map<ScriptReferenceString, ServiceScriptExecution>;
  */
 export type FailureMode = 'no-new' | 'continue' | 'kill';
 
+let executorConstructorHook: ((executor: Executor) => void) | undefined;
+
+/**
+ * For GC testing only. A function that is called whenever an Executor is
+ * constructed.
+ */
+export function registerExecutorConstructorHook(
+  fn: typeof executorConstructorHook
+) {
+  executorConstructorHook = fn;
+}
+
 /**
  * Executes a script that has been analyzed and validated by the Analyzer.
  */
@@ -75,24 +87,15 @@ export class Executor {
     workerPool: WorkerPool,
     cache: Cache | undefined,
     failureMode: FailureMode,
-    abort: Deferred<void>,
     previousIterationServices: ServiceMap | undefined
   ) {
+    executorConstructorHook?.(this);
     this._rootConfig = rootConfig;
     this._logger = logger;
     this._workerPool = workerPool;
     this._cache = cache;
     this._previousIterationServices = previousIterationServices;
 
-    // If this entire execution is aborted because e.g. the user sent a SIGINT
-    // to the Wireit process, then dont start new scripts, and kill running
-    // ones.
-    void abort.promise.then(() => {
-      this._stopStartingNewScripts.resolve();
-      this._killRunningScripts.resolve();
-      this._stopServices.resolve();
-    });
-
     // If a failure occurs, then whether we stop starting new scripts or kill
     // running ones depends on the failure mode setting.
     void this._failureOccured.promise.then(() => {
@@ -121,6 +124,16 @@ export class Executor {
     });
   }
 
+  /**
+   * If this entire execution is aborted because e.g. the user sent a SIGINT to
+   * the Wireit process, then dont start new scripts, and kill running ones.
+   */
+  abort() {
+    this._stopStartingNewScripts.resolve();
+    this._killRunningScripts.resolve();
+    this._stopServices.resolve();
+  }
+
   /**
    * Execute the root script.
    */
@@ -154,6 +167,17 @@ export class Executor {
     if (!rootExecutionResult.ok) {
       errors.push(...rootExecutionResult.error);
     }
+    // Wait for all persistent services to start.
+    for (const service of this._persistentServices.values()) {
+      // Persistent services start automatically, so calling start() here should
+      // be a no-op, but it lets us get the started promise.
+      const result = await service.start();
+      if (!result.ok) {
+        errors.push(...result.error);
+      }
+    }
+    // Wait for all ephemeral services to have terminated (either started and
+    // stopped, or never needed to start).
     const ephemeralServiceResults = await Promise.all(
       this._ephemeralServices.map((service) => service.terminated)
     );
@@ -204,12 +228,34 @@ export class Executor {
       if (config.command === undefined) {
         execution = new NoCommandScriptExecution(config, this, this._logger);
       } else if (config.service) {
+        const adoptee = this._previousIterationServices?.get(key);
+        if (adoptee !== undefined) {
+          // Remove the adoptee from the map so that this executor doesn't hold
+          // a reference to it. Otherwise, we'll maintain a chain of references
+          // going all the way back through all previous executions, which will
+          // leak memory in watch mode.
+          //
+          //               executor N
+          //  break this -----> | [previousIterationServices]
+          //   reference        v
+          //               service N-1
+          //                    | [executor]
+          //                    v
+          //               executor N-1
+          //                    | [previousIterationServices]
+          //                    v
+          //               sevice N-2
+          //                    | [executor]
+          //                    v
+          //                   ...
+          this._previousIterationServices!.delete(key);
+        }
         execution = new ServiceScriptExecution(
           config,
           this,
           this._logger,
           this._stopServices.promise,
-          this._previousIterationServices?.get(key)
+          adoptee
         );
         if (config.isPersistent) {
           this._persistentServices.set(key, execution);

src/logging/default-logger.ts

@@ -204,6 +204,12 @@ export class DefaultLogger implements Logger {
             console.error(`❌${prefix} Service exited unexpectedly`);
             break;
           }
+          case 'aborted': {
+            // This event isn't very useful to log. Things get aborted only
+            // because of a failure somewhere else, which should already get
+            // reported.
+            break;
+          }
         }
         break;
       }