[debug] Trace the removal test on the new macOS runner image — do not merge

engine/mast-docker/src/command.rs

@@ -295,16 +295,36 @@ async fn wait_for_killed_child(child: &mut tokio::process::Child) -> std::io::Re
         .map(|_| ())
 }
 
+/// Whether any process is still in the group — `kill` with signal 0 checks
+/// without signalling.
+#[cfg(unix)]
+fn group_alive(pid: i32) -> bool {
+    unsafe { libc::kill(-pid, 0) == 0 }
+}
+
 #[cfg(unix)]
 async fn terminate_streaming_child(
     child: &mut tokio::process::Child,
     pid: u32,
     grace: Option<Duration>,
 ) -> std::io::Result<()> {
     if let Some(grace) = grace {
+        let deadline = tokio::time::Instant::now() + grace;
         kill_group(pid as i32, SIG_TERM);
         if let Ok(status) = tokio::time::timeout(grace, child.wait()).await {
-            return status.map(|_| ());
+            status?;
+            // The direct child is gone; whatever else it started in the group
+            // gets the rest of the grace to follow, then SIGKILL. A descendant
+            // that outlives the TERM — one that ignores it, or one caught
+            // mid-fork when it arrived — keeps the output pipes open, and a
+            // stop would otherwise wait on it for as long as it runs.
+            while group_alive(pid as i32) && tokio::time::Instant::now() < deadline {
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+            if group_alive(pid as i32) {
+                kill_group(pid as i32, SIG_KILL);
+            }
+            return Ok(());
         }
     }
     kill_group(pid as i32, SIG_KILL);
@@ -703,6 +723,50 @@ mod tests {
         assert!(started.elapsed() < Duration::from_secs(5), "cancel was not prompt");
     }
 
+    /// A descendant can outlive the group's SIGTERM — by ignoring it, as
+    /// here, or by being mid-fork when it lands, as a loaded macOS runner
+    /// showed — and it holds the output pipes. Once the direct child is gone,
+    /// the rest of the group must still be stopped within the grace.
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread")]
+    async fn cancellation_stops_a_descendant_that_outlives_the_term() {
+        let (tx, mut rx) = mpsc::channel(64);
+        // An ignored signal stays ignored across exec.
+        let argv: Vec<String> = ["sh", "-c", "(trap '' TERM; exec sleep 300) & echo $!; wait"]
+            .map(String::from)
+            .to_vec();
+        let cancel = CancellationToken::new();
+        let handle = {
+            let cancel = cancel.clone();
+            tokio::spawn(async move {
+                run_streaming(
+                    &argv,
+                    None,
+                    &[],
+                    tx,
+                    cancel,
+                    Duration::from_secs(60),
+                    Duration::from_secs(1),
+                )
+                .await
+            })
+        };
+        let sleeper: i32 = rx.recv().await.unwrap().line.trim().parse().unwrap();
+        let started = std::time::Instant::now();
+        cancel.cancel();
+        assert_eq!(handle.await.unwrap().unwrap(), CommandOutcome::Cancelled);
+        assert!(started.elapsed() < Duration::from_secs(5), "cancel was not prompt");
+        // Gone, not merely signalled: give init a moment to reap it.
+        let deadline = std::time::Instant::now() + Duration::from_secs(3);
+        while unsafe { libc::kill(sleeper, 0) } == 0 {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "the descendant that ignored SIGTERM outlived the cancel"
+            );
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+    }
+
     #[cfg(windows)]
     struct WindowsTestTree(Vec<u32>);
 

engine/mast-engine/src/ops.rs

@@ -328,7 +328,16 @@ impl Engine {
             Duration::from_secs(8),
         )
         .await;
-        let _ = forwarder.await;
+        // The process has exited: forward what it wrote, but only for a
+        // moment. The forwarder ends when every holder of the output pipes
+        // has closed them, and a background process the command left behind
+        // inherits them — waiting on it would hold the operation open for as
+        // long as that process runs. Its later output belongs to no
+        // operation, so it is dropped.
+        let mut forwarder = forwarder;
+        if tokio::time::timeout(Duration::from_secs(2), &mut forwarder).await.is_err() {
+            forwarder.abort();
+        }
         result.map_err(|e| ErrorInfo::Internal { message: redactor.redact(&e.to_string()) })
     }
 }

engine/mast-engine/tests/engine_tests.rs

@@ -3485,10 +3485,59 @@ async fn removing_project_cancels_its_running_command_before_forgetting_it() {
     assert!(matches!(terminal.kind, OperationEventKind::Cancelled));
 }
 
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread")]
+async fn a_command_that_leaves_a_background_process_still_finishes() {
+    use std::os::unix::fs::PermissionsExt;
+
+    let tmp = tempfile::tempdir().unwrap();
+    let project = make_project(tmp.path(), "leaves-a-daemon");
+    // The background sleeper inherits stdout and stderr, so the output pipes
+    // stay open long after the script itself has exited.
+    let script = project.join("start.sh");
+    std::fs::write(&script, "#!/bin/sh\nsleep 300 &\necho $! > daemon.pid\necho started\n").unwrap();
+    std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
+    let engine = test_engine(tmp.path(), Arc::new(FakeConnector(FakeAdapter::new())));
+    run_action(&engine, Action::ImportProject { path: project.to_string_lossy().into() }).await;
+    let id = engine.snapshot().projects[0].id.clone();
+    run_action(&engine, Action::SetProjectCommands { id: id.clone(), commands: vec![
+        mast_contract::ProjectCommand {
+            name: "start".into(), command: "./start.sh".into(),
+            auto_start: false, cwd: None, after: None, ready_when: None,
+            auto_restart: false, restart_when_changed: Vec::new(), from_manifest: false,
+        },
+    ] }).await;
+    let op = engine.dispatch(Action::RunProjectCommand { id, name: "start".into() }).unwrap();
+    let mut events = engine.operation_events(op).unwrap();
+    let ended = tokio::time::timeout(Duration::from_secs(15), async {
+        while let Some(event) = events.next().await {
+            if event.kind.is_terminal() {
+                return event.kind;
+            }
+        }
+        panic!("operation ended without a terminal event");
+    })
+    .await;
+    if let Ok(pid) = std::fs::read_to_string(project.join("daemon.pid")) {
+        let _ = std::process::Command::new("kill").arg(pid.trim()).status();
+    }
+    let ended = ended.expect("the operation must end with the command, not with what it left running");
+    assert!(matches!(ended, OperationEventKind::Completed), "{ended:?}");
+}
+
 #[cfg(unix)]
 #[tokio::test(flavor = "multi_thread")]
 async fn project_removal_blocks_new_commands_until_shutdown_finishes() {
+    // DEBUG (ci-debug branch only): repeat to catch the macOS-runner failure.
+    for round in 0..15 {
+        removal_round(round).await;
+    }
+}
+
+#[cfg(unix)]
+async fn removal_round(round: u32) {
     use std::os::unix::fs::PermissionsExt;
+    eprintln!("DEBUG round {round}");
 
     let tmp = tempfile::tempdir().unwrap();
     let project = make_project(tmp.path(), "slow-shutdown");
@@ -3538,16 +3587,41 @@ async fn project_removal_blocks_new_commands_until_shutdown_finishes() {
         assert!(matches!(engine.dispatch(action), Err(ErrorInfo::Conflict { .. })));
     }
     std::fs::write(project.join("release-stop"), "").unwrap();
+    // DEBUG (ci-debug branch only): record what both operations do after the
+    // release, with timestamps, and dump them plus the process table on timeout.
+    let t0 = std::time::Instant::now();
+    let trace = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
     let mut removed = engine.operation_events(removing).unwrap();
-    tokio::time::timeout(Duration::from_secs(30), async {
-        while let Some(event) = removed.next().await {
-            if event.kind.is_terminal() {
-                assert!(matches!(event.kind, OperationEventKind::Completed));
-                return;
+    let record = trace.clone();
+    let outcome = tokio::time::timeout(Duration::from_secs(30), async move {
+        let mut output_open = true;
+        loop {
+            tokio::select! {
+                event = output.next(), if output_open => match event {
+                    Some(e) => record.lock().unwrap().push(format!("{:?} serve: {:?}", t0.elapsed(), e.kind)),
+                    None => { output_open = false; record.lock().unwrap().push(format!("{:?} serve: stream closed", t0.elapsed())); }
+                },
+                event = removed.next() => match event {
+                    Some(e) => {
+                        record.lock().unwrap().push(format!("{:?} remove: {:?}", t0.elapsed(), e.kind));
+                        if e.kind.is_terminal() { return e.kind; }
+                    }
+                    None => panic!("removal ended without a terminal event"),
+                },
             }
         }
-        panic!("removal ended without a terminal event");
-    }).await.unwrap();
+    }).await;
+    let kind = match outcome {
+        Ok(kind) => kind,
+        Err(_) => {
+            let ps = std::process::Command::new("ps").args(["-ax", "-o", "pid,ppid,pgid,stat,etime,command"]).output().unwrap();
+            let ps = String::from_utf8_lossy(&ps.stdout);
+            let relevant: Vec<&str> = ps.lines().filter(|l| l.contains("serve.sh") || l.contains("sleep") || l.contains("PID")).collect();
+            panic!("removal did not finish in 30s\ntrace:\n{}\nprocesses:\n{}\nrelease-stop exists: {}", trace.lock().unwrap().join("\n"), relevant.join("\n"), project.join("release-stop").exists());
+        }
+    };
+    eprintln!("DEBUG trace:\n{}", trace.lock().unwrap().join("\n"));
+    assert!(matches!(kind, OperationEventKind::Completed), "{kind:?}");
     assert!(engine.snapshot().projects.is_empty());
 
     // The reservation ends with the operation; importing the same path must