fix(testing): finish dev server async generator in cypress executor (#…

…29689)

<!-- Please make sure you have read the submission guidelines before
posting an PR -->
<!--
https://github.com/nrwl/nx/blob/master/CONTRIBUTING.md#-submitting-a-pr
-->

<!-- Please make sure that your commit message follows our format -->
<!-- Example: `fix(nx): must begin with lowercase` -->

<!-- If this is a particularly complex change or feature addition, you
can request a dedicated Nx release for this pull request branch. Mention
someone from the Nx team or the `@nrwl/nx-pipelines-reviewers` and they
will confirm if the PR warrants its own release for testing purposes,
and generate it for you if appropriate. -->

## Current Behavior
<!-- This is the behavior we have today -->

Running an e2e task using the `@nx/cypress:cypress` executor that starts
a dev server by running another task that uses the `@nx/js:node`
executor result in the process to hang after the e2e tests have finished
running.

## Expected Behavior
<!-- This is the behavior we should expect with the changes in this PR
-->

The async generator returned by starting the dev server and consumed by
the `@nx/cypress:cypress` executor should be finished, and the
`@nx/js:node` executor should properly clean up its child process once
the generator is finished.

## Related Issue(s)
<!-- Please link the issue being fixed so it gets closed when this is
merged. -->

Fixes #29571

packages/cypress/src/executors/cypress/cypress.impl.ts

@@ -58,13 +58,17 @@ export default async function cypressExecutor(
   process.env.NX_CYPRESS_TARGET_CONFIGURATION = context.configurationName;
   let success;
 
-  for await (const devServerValues of startDevServer(options, context)) {
+  const generatorInstance = startDevServer(options, context);
+  for await (const devServerValues of generatorInstance) {
     try {
       success = await runCypress(devServerValues.baseUrl, {
         ...options,
         portLockFilePath: devServerValues.portLockFilePath,
       });
-      if (!options.watch) break;
+      if (!options.watch) {
+        generatorInstance.return();
+        break;
+      }
     } catch (e) {
       logger.error(e.message);
       success = false;

packages/devkit/src/utils/async-iterable/create-async-iterable.ts

@@ -2,13 +2,15 @@ export interface AsyncPushCallbacks<T> {
   next: (value: T) => void;
   done: () => void;
   error: (err: unknown) => void;
+  registerCleanup?: (cb: () => void | Promise<void>) => void;
 }
 
 export function createAsyncIterable<T = unknown>(
   listener: (ls: AsyncPushCallbacks<T>) => void
 ): AsyncIterable<T> {
   let done = false;
   let error: unknown | null = null;
+  let cleanup: (() => void | Promise<void>) | undefined;
 
   const pushQueue: T[] = [];
   const pullQueue: Array<
@@ -42,6 +44,9 @@ export function createAsyncIterable<T = unknown>(
           }
           done = true;
         },
+        registerCleanup: (cb) => {
+          cleanup = cb;
+        },
       });
 
       return {
@@ -60,6 +65,12 @@ export function createAsyncIterable<T = unknown>(
             }
           );
         },
+        async return() {
+          if (cleanup) {
+            await cleanup();
+          }
+          return { value: undefined, done: true };
+        },
       };
     },
   } as AsyncIterable<T>;

packages/js/src/executors/node/node.impl.ts

@@ -32,11 +32,30 @@ interface ActiveTask {
   stop: (signal: NodeJS.Signals) => Promise<void>;
 }
 
-function debounce(fn: () => void, wait: number) {
+function debounce<T>(fn: () => Promise<T>, wait: number): () => Promise<T> {
   let timeoutId: NodeJS.Timeout;
+  let pendingPromise: Promise<T> | null = null;
+
   return () => {
     clearTimeout(timeoutId);
-    timeoutId = setTimeout(fn, wait);
+
+    if (!pendingPromise) {
+      pendingPromise = new Promise<T>((resolve, reject) => {
+        timeoutId = setTimeout(() => {
+          fn()
+            .then((result) => {
+              pendingPromise = null;
+              resolve(result);
+            })
+            .catch((error) => {
+              pendingPromise = null;
+              reject(error);
+            });
+        }, wait);
+      });
+    }
+
+    return pendingPromise;
   };
 }
 
@@ -97,7 +116,7 @@ export async function* nodeExecutor(
   yield* createAsyncIterable<{
     success: boolean;
     options?: Record<string, any>;
-  }>(async ({ done, next, error }) => {
+  }>(async ({ done, next, error, registerCleanup }) => {
     const processQueue = async () => {
       if (tasks.length === 0) return;
 
@@ -227,6 +246,10 @@ export async function* nodeExecutor(
       process.exit(128 + 1);
     });
 
+    registerCleanup(async () => {
+      await stopAllTasks('SIGTERM');
+    });
+
     if (options.runBuildTargetDependencies) {
       // If a all dependencies need to be rebuild on changes, then register with watcher
       // and run through CLI, otherwise only the current project will rebuild.