[nextest-runner] wait for dispatcher to acknowledge retries

Gives the dispatcher a chance to control communication.

nextest-runner/src/runner/dispatcher.rs

@@ -471,10 +471,28 @@ where
             InternalEvent::Executor(ExecutorEvent::RetryStarted {
                 test_instance,
                 retry_data,
-            }) => self.callback_none_response(TestEventKind::TestRetryStarted {
-                test_instance,
-                retry_data,
-            }),
+                tx,
+            }) => {
+                if self.cancel_state.is_some() {
+                    // The run has been cancelled: don't send a message over the tx and don't start
+                    // any new units.
+                    return HandleEventResponse::None;
+                }
+
+                match tx.send(()) {
+                    Ok(_) => {}
+                    Err(_) => {
+                        // The test task died?
+                        debug!(test = ?test_instance.id(), "test task died, ignoring");
+                        return HandleEventResponse::None;
+                    }
+                }
+
+                self.callback_none_response(TestEventKind::TestRetryStarted {
+                    test_instance,
+                    retry_data,
+                })
+            }
             InternalEvent::Executor(ExecutorEvent::Finished {
                 test_instance,
                 success_output,

nextest-runner/src/runner/executor.rs

@@ -230,10 +230,23 @@ impl<'a> ExecutorContext<'a> {
             // is empty.
 
             if retry_data.attempt > 1 {
+                // Ensure that the dispatcher believes the run is still ongoing. If the run is
+                // cancelled, the dispatcher will let us know.
+                let (tx, rx) = oneshot::channel();
                 _ = resp_tx.send(ExecutorEvent::RetryStarted {
                     test_instance,
                     retry_data,
+                    tx,
                 });
+
+                match rx.await {
+                    Ok(()) => {}
+                    Err(_) => {
+                        // The receiver was dropped -- the dispatcher has signaled that this unit
+                        // should exit.
+                        return;
+                    }
+                }
             }
 
             // Some of this information is only useful for event reporting, but

nextest-runner/src/runner/internal_events.rs

@@ -89,6 +89,8 @@ pub(super) enum ExecutorEvent<'a> {
     RetryStarted {
         test_instance: TestInstance<'a>,
         retry_data: RetryData,
+        // This is used to indicate that the dispatcher still wants to run the test.
+        tx: oneshot::Sender<()>,
     },
     Finished {
         test_instance: TestInstance<'a>,