fix(VER-2346): stop reading stout/err if subprocess is killed

Author: dsd-at-dfinity

rs/tests/src/driver/process.rs

@@ -7,6 +7,8 @@ use std::process::{Command, ExitStatus, Stdio};
 use tokio::{
     io::{AsyncBufReadExt, AsyncRead, BufReader},
     process::{Child, Command as AsyncCommand},
+    select,
+    sync::watch::{channel, Receiver},
     task::{self, JoinHandle},
 };
 
@@ -31,19 +33,21 @@ impl Process {
         // signal.
         cmd.kill_on_drop(true);
 
-        // println!("AsyncCommand: {:?}", cmd);
         let mut child = cmd.spawn().expect("Spawning subprocess should succeed");
-        // println!("Child: {:?}", child);
 
         let stdout = child.stdout.take().unwrap();
+        let (kill_tx, kill_watch) = channel::<bool>(false);
+
         let stdout_jh = task::spawn(Self::listen_on_channel(
+            kill_watch.clone(),
             task_id.clone(),
             log.clone(),
             ChannelName::StdOut,
             stdout,
         ));
         let stderr = child.stderr.take().unwrap();
         let stderr_jh = task::spawn(Self::listen_on_channel(
+            kill_watch,
             task_id,
             log.clone(),
             ChannelName::StdErr,
@@ -60,6 +64,7 @@ impl Process {
             let pid = Pid::from_raw(self_.child.id().unwrap() as i32);
             move || {
                 let _ = kill(pid, Signal::SIGKILL);
+                let _ = kill_tx.send(true);
             }
         };
 
@@ -77,21 +82,34 @@ impl Process {
         child.wait().await
     }
 
-    async fn listen_on_channel<R>(task_id: TaskId, log: Logger, channel_tag: ChannelName, src: R)
-    where
+    async fn listen_on_channel<R>(
+        mut kill_watch: Receiver<bool>,
+        task_id: TaskId,
+        log: Logger,
+        channel_tag: ChannelName,
+        src: R,
+    ) where
         R: AsyncRead + Unpin,
     {
         let buffered_reader = BufReader::new(src);
         let mut lines = buffered_reader.lines();
         loop {
-            match lines.next_line().await {
-                Ok(Some(line)) => {
-                    let task_id: String = format!("{}", task_id);
-                    let output_channel: String = format!("{:?}", channel_tag);
-                    info!(log, "{}", line; "task_id" => task_id, "output_channel" => output_channel)
+            select! {
+                line_res = lines.next_line() => {
+                    match line_res {
+                        Ok(Some(line)) => {
+                            let task_id: String = format!("{}", task_id);
+                            let output_channel: String = format!("{:?}", channel_tag);
+                            info!(log, "{}", line; "task_id" => task_id, "output_channel" => output_channel)
+                        }
+                        Ok(None) => break,
+                        Err(e) => eprintln!("listen_on_channel(): {:?}", e),
+                    }
+                }
+                _ = kill_watch.changed() => {
+                    info!(log, "({}|{:?}): Kill received.", task_id, channel_tag);
+                    return;
                 }
-                Ok(None) => break,
-                Err(e) => eprintln!("listen_on_channel(): {:?}", e),
             }
         }
         info!(log, "({}|{:?}): Channel has closed.", task_id, channel_tag);

rs/tests/testing_verification/test_driver_e2e_scenarios.rs

@@ -167,6 +167,14 @@ fn get_all_e2e_test_scenarios() -> HashMap<String, SystemTestGroup> {
                 .add_test(systest!(test_to_succeed))
                 .without_farm(),
         ),
+        (
+            "test_child_process".to_string(),
+            SystemTestGroup::new()
+                .with_timeout_per_test(Duration::from_secs(5))
+                .with_setup(setup_to_succeed)
+                .add_test(systest!(spawning_process))
+                .without_farm(),
+        ),
     ])
 }
 
@@ -228,3 +236,13 @@ fn never_ending_task(env: TestEnv) {
         std::thread::sleep(Duration::from_secs(1));
     }
 }
+
+// Simulate a scenario where a test that starts a subprocess gets stopped by a timeout in the test
+// driver.
+fn spawning_process(_env: TestEnv) {
+    let _ = std::process::Command::new("sh")
+        .arg("-c")
+        .arg("for i in `seq 100`; do echo magicchild$i; sleep 1; done")
+        .spawn();
+    std::thread::sleep(std::time::Duration::from_secs(100));
+}

rs/tests/testing_verification/test_driver_tests.rs

@@ -496,6 +496,32 @@ fn test_overall_group_timeout_in_setup() {
     );
 }
 
+#[test]
+fn test_test_spawning_proc_gets_stopped() {
+    let result = execute_test_scenario_with_default_cmd("test_child_process");
+    assert!(
+        !result.status.success(),
+        "{}",
+        String::from_utf8_lossy(&result.stderr)
+    );
+
+    let err_str = String::from_utf8_lossy(&result.stderr);
+    // The process started by the test print a string magicchild1 at second 1, magicchild2 at
+    // second 2, etc. The per test timeout is 5 seconds, so, 'magicchild1' should be visible, but
+    // not 'magicchild10'.
+    // Further, we test that the test returns even though the child process of the test keeps
+    // stdout/err open.
+    assert!(err_str.contains("magicchild1"));
+    assert!(!err_str.contains("magicchild10"));
+    let summary = extract_report(result.stderr).expect("Failed to extract report from logs.");
+    assert_test_summary_size(&summary, 1, 1, 0);
+    assert_name_and_message_eq(
+        &summary.failure[0],
+        "spawning_process",
+        Some("Timeout after 5s"),
+    );
+}
+
 #[test]
 fn test_setup_failure_file_written() {
     let working_dir = create_unique_working_dir();