fix(test): improve how --fail-fast shuts down when hitting limit (#16956)

Closes #15650

cli/build.rs

@@ -130,6 +130,10 @@ mod ts {
         path_dts.join(format!("lib.{}.d.ts", name)).display()
       );
     }
+    println!(
+      "cargo:rerun-if-changed={}",
+      cwd.join("js").join("40_testing.js").display()
+    );
 
     // create a copy of the vector that includes any op crate libs to be passed
     // to the JavaScript compiler to build into the snapshot

cli/js/40_testing.js

@@ -1086,6 +1086,9 @@
     }
 
     for (const desc of filtered) {
+      if (ops.op_tests_should_stop()) {
+        break;
+      }
       ops.op_dispatch_test_event({ wait: desc.id });
       const earlier = DateNow();
       const result = await runTest(desc);

cli/lsp/testing/execution.rs

@@ -13,6 +13,7 @@ use crate::lsp::logging::lsp_log;
 use crate::ops;
 use crate::proc_state;
 use crate::tools::test;
+use crate::tools::test::FailFastTracker;
 use crate::tools::test::TestEventSender;
 use crate::util::checksum;
 use crate::worker::create_main_worker_for_test_or_bench;
@@ -144,25 +145,29 @@ impl LspTestFilter {
   }
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn test_specifier(
   ps: proc_state::ProcState,
   permissions: Permissions,
   specifier: ModuleSpecifier,
   mode: test::TestMode,
-  sender: &TestEventSender,
+  sender: TestEventSender,
+  fail_fast_tracker: FailFastTracker,
   token: CancellationToken,
   filter: test::TestFilter,
 ) -> Result<(), AnyError> {
   if !token.is_cancelled() {
+    let stdout = StdioPipe::File(sender.stdout());
+    let stderr = StdioPipe::File(sender.stderr());
     let mut worker = create_main_worker_for_test_or_bench(
       &ps,
       specifier.clone(),
       permissions,
-      vec![ops::testing::init(sender.clone(), filter)],
+      vec![ops::testing::init(sender, fail_fast_tracker, filter)],
       Stdio {
         stdin: StdioPipe::Inherit,
-        stdout: StdioPipe::File(sender.stdout()),
+        stdout,
-        stderr: StdioPipe::File(sender.stderr()),
+        stderr,
       },
     )
     .await?;
@@ -262,19 +267,17 @@ impl TestRun {
     )
     .await?;
 
-    let (sender, mut receiver) = mpsc::unbounded_channel::<test::TestEvent>();
-    let sender = TestEventSender::new(sender);
-
     let (concurrent_jobs, fail_fast) =
       if let DenoSubcommand::Test(test_flags) = ps.options.sub_command() {
-        (
+        (test_flags.concurrent_jobs.into(), test_flags.fail_fast)
-          test_flags.concurrent_jobs.into(),
-          test_flags.fail_fast.map(|count| count.into()),
-        )
       } else {
         unreachable!("Should always be Test subcommand.");
       };
 
+    let (sender, mut receiver) = mpsc::unbounded_channel::<test::TestEvent>();
+    let sender = TestEventSender::new(sender);
+    let fail_fast_tracker = FailFastTracker::new(fail_fast);
+
     let mut queue = self.queue.iter().collect::<Vec<&ModuleSpecifier>>();
     queue.sort();
 
@@ -288,6 +291,7 @@ impl TestRun {
       let ps = ps.clone();
       let permissions = permissions.clone();
       let mut sender = sender.clone();
+      let fail_fast_tracker = fail_fast_tracker.clone();
       let lsp_filter = self.filters.get(&specifier);
       let filter = test::TestFilter {
         substring: None,
@@ -305,13 +309,17 @@ impl TestRun {
       let tests = tests_.clone();
 
       tokio::task::spawn_blocking(move || {
+        if fail_fast_tracker.should_stop() {
+          return Ok(());
+        }
         let origin = specifier.to_string();
         let file_result = run_local(test_specifier(
           ps,
           permissions,
           specifier,
           test::TestMode::Executable,
-          &sender,
+          sender.clone(),
+          fail_fast_tracker,
           token,
           filter,
         ));
@@ -427,12 +435,6 @@ impl TestRun {
               );
             }
           }
-
-          if let Some(count) = fail_fast {
-            if summary.failed >= count {
-              break;
-            }
-          }
         }
 
         let elapsed = Instant::now().duration_since(earlier);

cli/ops/testing.rs

@@ -1,10 +1,12 @@
 // Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
 
+use crate::tools::test::FailFastTracker;
 use crate::tools::test::TestDescription;
 use crate::tools::test::TestEvent;
 use crate::tools::test::TestEventSender;
 use crate::tools::test::TestFilter;
 use crate::tools::test::TestLocation;
+use crate::tools::test::TestResult;
 use crate::tools::test::TestStepDescription;
 
 use deno_core::error::generic_error;
@@ -23,7 +25,11 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use uuid::Uuid;
 
-pub fn init(sender: TestEventSender, filter: TestFilter) -> Extension {
+pub fn init(
+  sender: TestEventSender,
+  fail_fast_tracker: FailFastTracker,
+  filter: TestFilter,
+) -> Extension {
   Extension::builder()
     .ops(vec![
       op_pledge_test_permissions::decl(),
@@ -32,9 +38,11 @@ pub fn init(sender: TestEventSender, filter: TestFilter) -> Extension {
       op_register_test::decl(),
       op_register_test_step::decl(),
       op_dispatch_test_event::decl(),
+      op_tests_should_stop::decl(),
     ])
     .state(move |state| {
       state.put(sender.clone());
+      state.put(fail_fast_tracker.clone());
       state.put(filter.clone());
       Ok(())
     })
@@ -178,7 +186,18 @@ fn op_dispatch_test_event(
   state: &mut OpState,
   event: TestEvent,
 ) -> Result<(), AnyError> {
+  if matches!(
+    event,
+    TestEvent::Result(_, TestResult::Cancelled | TestResult::Failed(_), _)
+  ) {
+    state.borrow::<FailFastTracker>().add_failure();
+  }
   let mut sender = state.borrow::<TestEventSender>().clone();
   sender.send(event).ok();
   Ok(())
 }
+
+#[op]
+fn op_tests_should_stop(state: &mut OpState) -> bool {
+  state.borrow::<FailFastTracker>().should_stop()
+}

cli/tools/test.rs

@@ -53,6 +53,7 @@ use std::io::Write;
 use std::num::NonZeroUsize;
 use std::path::Path;
 use std::path::PathBuf;
+use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
@@ -713,18 +714,25 @@ async fn test_specifier(
   permissions: Permissions,
   specifier: ModuleSpecifier,
   mode: TestMode,
-  sender: &TestEventSender,
+  sender: TestEventSender,
+  fail_fast_tracker: FailFastTracker,
   options: TestSpecifierOptions,
 ) -> Result<(), AnyError> {
+  let stdout = StdioPipe::File(sender.stdout());
+  let stderr = StdioPipe::File(sender.stderr());
   let mut worker = create_main_worker_for_test_or_bench(
     &ps,
     specifier.clone(),
     permissions,
-    vec![ops::testing::init(sender.clone(), options.filter.clone())],
+    vec![ops::testing::init(
+      sender,
+      fail_fast_tracker,
+      options.filter.clone(),
+    )],
     Stdio {
       stdin: StdioPipe::Inherit,
-      stdout: StdioPipe::File(sender.stdout()),
+      stdout,
-      stderr: StdioPipe::File(sender.stderr()),
+      stderr,
     },
   )
   .await?;
@@ -999,9 +1007,9 @@ async fn test_specifiers(
   };
 
   let (sender, mut receiver) = unbounded_channel::<TestEvent>();
+  let fail_fast_tracker = FailFastTracker::new(options.fail_fast);
   let sender = TestEventSender::new(sender);
   let concurrent_jobs = options.concurrent_jobs;
-  let fail_fast = options.fail_fast;
 
   let join_handles =
     specifiers_with_mode.iter().map(move |(specifier, mode)| {
@@ -1011,15 +1019,21 @@ async fn test_specifiers(
       let mode = mode.clone();
       let mut sender = sender.clone();
       let options = options.clone();
+      let fail_fast_tracker = fail_fast_tracker.clone();
 
       tokio::task::spawn_blocking(move || {
+        if fail_fast_tracker.should_stop() {
+          return Ok(());
+        }
+
         let origin = specifier.to_string();
         let file_result = run_local(test_specifier(
           ps,
           permissions,
           specifier,
           mode,
-          &sender,
+          sender.clone(),
+          fail_fast_tracker,
           options,
         ));
         if let Err(error) = file_result {
@@ -1148,12 +1162,6 @@ async fn test_specifiers(
             );
           }
         }
-
-        if let Some(x) = fail_fast {
-          if summary.failed >= x.get() {
-            break;
-          }
-        }
       }
 
       let elapsed = Instant::now().duration_since(earlier);
@@ -1564,6 +1572,42 @@ pub async fn run_tests_with_watch(
   Ok(())
 }
 
+/// Tracks failures for the `--fail-fast` argument in
+/// order to tell when to stop running tests.
+#[derive(Clone)]
+pub struct FailFastTracker {
+  max_count: Option<usize>,
+  failure_count: Arc<AtomicUsize>,
+}
+
+impl FailFastTracker {
+  pub fn new(fail_fast: Option<NonZeroUsize>) -> Self {
+    Self {
+      max_count: fail_fast.map(|v| v.into()),
+      failure_count: Default::default(),
+    }
+  }
+
+  pub fn add_failure(&self) -> bool {
+    if let Some(max_count) = &self.max_count {
+      self
+        .failure_count
+        .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
+        >= *max_count
+    } else {
+      false
+    }
+  }
+
+  pub fn should_stop(&self) -> bool {
+    if let Some(max_count) = &self.max_count {
+      self.failure_count.load(std::sync::atomic::Ordering::SeqCst) >= *max_count
+    } else {
+      false
+    }
+  }
+}
+
 #[derive(Clone)]
 pub struct TestEventSender {
   sender: UnboundedSender<TestEvent>,
@@ -1596,6 +1640,7 @@ impl TestEventSender {
       TestEvent::Result(_, _, _)
         | TestEvent::StepWait(_)
         | TestEvent::StepResult(_, _, _)
+        | TestEvent::UncaughtError(_, _)
     ) {
       self.flush_stdout_and_stderr();
     }