fix: exec command with failFast should fail immediately (#665)

* better default value for exec concurrency

* reformat file

* change default value inside function

* add immediate fail fast

* refactor fail fast

- add tests

* remove unused import

* format files

* fix typo

* add delayed exit to make failure order predictable

* refactor

* remove useless code

packages/melos/lib/src/commands/exec.dart

@@ -11,8 +11,10 @@ mixin _ExecMixin on _Melos {
     Map<String, String> extraEnvironment = const {},
   }) async {
     concurrency ??= Platform.numberOfProcessors;
-    final workspace =
-        await createWorkspace(global: global, packageFilters: packageFilters);
+    final workspace = await createWorkspace(
+      global: global,
+      packageFilters: packageFilters,
+    );
     final packages = workspace.filteredPackages.values;
 
     await _execForAllPackages(
@@ -117,52 +119,66 @@ mixin _ExecMixin on _Melos {
       packages.map((package) => MapEntry(package.name, Completer<int?>())),
     );
 
-    await pool.forEach<Package, void>(sortedPackages, (package) async {
-      if (failFast && failures.isNotEmpty) {
-        return;
-      }
+    late final CancelableOperation<void> operation;
 
-      if (orderDependents) {
-        final dependenciesResults = await Future.wait(
-          package.allDependenciesInWorkspace.values
-              .map((package) => packageResults[package.name]?.future)
-              .whereNotNull(),
-        );
+    operation = CancelableOperation.fromFuture(
+      pool.forEach<Package, void>(sortedPackages, (package) async {
+        assert(!(failFast && failures.isNotEmpty));
+
+        if (orderDependents) {
+          final dependenciesResults = await Future.wait(
+            package.allDependenciesInWorkspace.values
+                .map((package) => packageResults[package.name]?.future)
+                .whereNotNull(),
+          );
 
-        final dependencyFailed = dependenciesResults
-            .any((exitCode) => exitCode == null || exitCode > 0);
-        if (dependencyFailed) {
-          packageResults[package.name]?.complete();
-          failures[package.name] = null;
-          return;
+          final dependencyFailed = dependenciesResults.any(
+            (exitCode) => exitCode == null || exitCode > 0,
+          );
+          if (dependencyFailed) {
+            packageResults[package.name]?.complete();
+            failures[package.name] = null;
+
+            return;
+          }
         }
-      }
 
-      if (!prefixLogs) {
-        logger
-          ..horizontalLine()
-          ..log(AnsiStyles.bgBlack.bold.italic('${package.name}:'));
-      }
+        if (!prefixLogs) {
+          logger
+            ..horizontalLine()
+            ..log(AnsiStyles.bgBlack.bold.italic('${package.name}:'));
+        }
 
-      final packageExitCode = await _execForPackage(
-        workspace,
-        package,
-        execArgs,
-        prefixLogs: prefixLogs,
-        extraEnvironment: additionalEnvironment,
-      );
-
-      packageResults[package.name]?.complete(packageExitCode);
-
-      if (packageExitCode > 0) {
-        failures[package.name] = packageExitCode;
-      } else if (!prefixLogs) {
-        logger.log(
-          AnsiStyles.bgBlack.bold.italic('${package.name}: ') +
-              AnsiStyles.bgBlack(successLabel),
+        final packageExitCode = await _execForPackage(
+          workspace,
+          package,
+          execArgs,
+          prefixLogs: prefixLogs,
+          extraEnvironment: additionalEnvironment,
         );
-      }
-    }).drain<void>();
+
+        packageResults[package.name]?.complete(packageExitCode);
+
+        if (packageExitCode > 0) {
+          failures[package.name] = packageExitCode;
+        } else if (!prefixLogs) {
+          logger.log(
+            AnsiStyles.bgBlack.bold.italic('${package.name}: ') +
+                AnsiStyles.bgBlack(successLabel),
+          );
+        }
+
+        if (packageExitCode > 0 && failFast) {
+          await operation.cancel();
+        }
+      }).drain<void>(),
+    );
+
+    await operation.valueOrCancellation();
+
+    if (failFast) {
+      runningPids.forEach(Process.killPid);
+    }
 
     logger
       ..horizontalLine()
@@ -181,6 +197,38 @@ mixin _ExecMixin on _Melos {
               'with exit code ${failures[packageName]})'}',
         );
       }
+
+      final canceled = <String>[];
+      for (final package in packages) {
+        if (failures.containsKey(package.name)) {
+          continue;
+        }
+
+        if (packageResults.containsKey(package.name)) {
+          final packageResult = packageResults[package.name]!;
+
+          if (packageResult.isCompleted) {
+            final exitCode = await packageResult.future;
+
+            if (exitCode == 0) {
+              continue;
+            }
+          }
+        }
+
+        canceled.add(package.name);
+      }
+
+      if (canceled.isNotEmpty) {
+        final canceledLogger = resultLogger
+            .child('$canceledLabel (in ${canceled.length} packages)');
+        for (final packageName in canceled) {
+          canceledLogger.child(
+            '${errorPackageNameStyle(packageName)} (due to failFast)',
+          );
+        }
+      }
+
       exitCode = 1;
     } else {
       resultLogger.child(successLabel);

packages/melos/lib/src/commands/runner.dart

@@ -4,6 +4,7 @@ import 'dart:io';
 import 'dart:math';
 
 import 'package:ansi_styles/ansi_styles.dart';
+import 'package:async/async.dart';
 import 'package:cli_util/cli_logging.dart';
 import 'package:collection/collection.dart';
 import 'package:file/local.dart';
@@ -31,8 +32,8 @@ import '../common/pending_package_update.dart';
 import '../common/platform.dart';
 import '../common/utils.dart' as utils;
 import '../common/utils.dart';
-import '../common/versioning.dart';
 import '../common/versioning.dart' as versioning;
+import '../common/versioning.dart';
 import '../global_options.dart';
 import '../lifecycle_hooks/lifecycle_hooks.dart';
 import '../logging.dart';

packages/melos/lib/src/common/utils.dart

@@ -457,6 +457,10 @@ Future<Process> startCommandRaw(
   );
 }
 
+final _runningPids = <int>[];
+
+List<int> get runningPids => UnmodifiableListView(_runningPids);
+
 Future<int> startCommand(
   List<String> command, {
   String? prefix,
@@ -479,6 +483,8 @@ Future<int> startCommand(
     includeParentEnvironment: includeParentEnvironment,
   );
 
+  _runningPids.add(process.pid);
+
   var stdoutStream = process.stdout;
   var stderrStream = process.stderr;
 
@@ -534,6 +540,8 @@ Future<int> startCommand(
   await processStderrCompleter.future;
   final exitCode = await process.exitCode;
 
+  _runningPids.remove(process.pid);
+
   if (onlyOutputOnError && exitCode > 0) {
     logger.stdout(utf8.decode(processStdout, allowMalformed: true));
     logger.stderr(utf8.decode(processStderr, allowMalformed: true));

packages/melos/lib/src/logging.dart

@@ -24,6 +24,7 @@ final successLabel = successLableColor(labelStyle('SUCCESS'));
 final warningLabel = warningLabelColor(labelStyle('WARNING'));
 final errorLabel = errorLabelColor(labelStyle('ERROR'));
 final failedLabel = errorLabelColor(labelStyle('FAILED'));
+final canceledLabel = errorLabelColor(labelStyle('CANCELED'));
 final hintLabel = hintLabelColor(labelStyle('HINT'));
 final runningLabel = commandLabelColor(labelStyle('RUNNING'));
 final checkLabel = AnsiStyles.greenBright('✓');

packages/melos/pubspec.yaml

@@ -17,6 +17,7 @@ executables:
 dependencies:
   ansi_styles: ^0.3.2+1
   args: ^2.4.2
+  async: ^2.11.0
   cli_launcher: ^0.3.1
   cli_util: ^0.4.1
   collection: ^1.18.0

packages/melos/test/commands/exec_test.dart

@@ -1,3 +1,5 @@
+import 'dart:io';
+
 import 'package:melos/melos.dart';
 import 'package:melos/src/common/io.dart';
 import 'package:melos/src/common/utils.dart';
@@ -72,6 +74,146 @@ ${'-' * terminalWidth}
       );
     });
 
+    group('concurrent processes', () {
+      /// Use this file instead of running "exit 1" so the failure
+      /// order is more predictable
+      void createDelayedExitFile(
+        Directory dir, {
+        int delay = 0,
+        int exitCode = 1,
+      }) {
+        File('${dir.path}/delayed_exit.dart').writeAsStringSync('''
+        import 'dart:io';
+        Future<void> main() async {
+          await Future.delayed(Duration(milliseconds: $delay));
+          exit($exitCode);
+        }
+        ''');
+      }
+
+      test('get cancel on first fail when fail fast is enabled', () async {
+        final workspaceDir = await createTemporaryWorkspace();
+
+        final a = await createProject(
+          workspaceDir,
+          const PubSpec(name: 'a'),
+        );
+
+        createDelayedExitFile(a, delay: 1000);
+
+        final b = await createProject(
+          workspaceDir,
+          const PubSpec(name: 'b'),
+        );
+
+        createDelayedExitFile(b, delay: 500);
+
+        final c = await createProject(
+          workspaceDir,
+          const PubSpec(name: 'c'),
+        );
+        createDelayedExitFile(c);
+
+        final logger = TestLogger();
+        final config = await MelosWorkspaceConfig.fromWorkspaceRoot(
+          workspaceDir,
+        );
+        final melos = Melos(
+          logger: logger,
+          config: config,
+        );
+
+        await melos.exec(
+          ['dart', 'delayed_exit.dart'],
+          concurrency: 3,
+          orderDependents: true,
+          failFast: true,
+        );
+
+        expect(
+          logger.output.normalizeNewLines(),
+          ignoringAnsii(
+            '''
+\$ melos exec
+  └> dart delayed_exit.dart
+     └> RUNNING (in 3 packages)
+
+${'-' * terminalWidth}
+${'-' * terminalWidth}
+
+\$ melos exec
+  └> dart delayed_exit.dart
+     └> FAILED (in 1 packages)
+        └> c (with exit code 1)
+     └> CANCELED (in 2 packages)
+        └> a (due to failFast)
+        └> b (due to failFast)
+''',
+          ),
+        );
+      });
+
+      test('keep running when fail fast is not enabled', () async {
+        final workspaceDir = await createTemporaryWorkspace();
+
+        final a = await createProject(
+          workspaceDir,
+          const PubSpec(name: 'a'),
+        );
+
+        createDelayedExitFile(a, delay: 1000);
+
+        final b = await createProject(
+          workspaceDir,
+          const PubSpec(name: 'b'),
+        );
+
+        createDelayedExitFile(b, delay: 500);
+
+        final c = await createProject(
+          workspaceDir,
+          const PubSpec(name: 'c'),
+        );
+        createDelayedExitFile(c);
+
+        final logger = TestLogger();
+        final config = await MelosWorkspaceConfig.fromWorkspaceRoot(
+          workspaceDir,
+        );
+        final melos = Melos(
+          logger: logger,
+          config: config,
+        );
+
+        await melos.exec(
+          ['dart', 'delayed_exit.dart'],
+          concurrency: 3,
+          orderDependents: true,
+        );
+
+        expect(
+          logger.output.normalizeNewLines(),
+          ignoringAnsii(
+            '''
+\$ melos exec
+  └> dart delayed_exit.dart
+     └> RUNNING (in 3 packages)
+
+${'-' * terminalWidth}
+${'-' * terminalWidth}
+
+\$ melos exec
+  └> dart delayed_exit.dart
+     └> FAILED (in 3 packages)
+        └> c (with exit code 1)
+        └> b (with exit code 1)
+        └> a (with exit code 1)
+''',
+          ),
+        );
+      });
+    });
+
     group('order dependents', () {
       test('sorts execution order topologically', () async {
         final workspaceDir = await createTemporaryWorkspace();
@@ -167,7 +309,7 @@ ${'-' * terminalWidth}
 
         await melos.exec(
           ['exit', '1'],
-          concurrency: 2,
+          concurrency: 3,
           orderDependents: true,
         );