diff --git a/engine/mast-docker/src/command.rs b/engine/mast-docker/src/command.rs index 6e165aa..a6a1312 100644 --- a/engine/mast-docker/src/command.rs +++ b/engine/mast-docker/src/command.rs @@ -295,6 +295,13 @@ 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, @@ -302,9 +309,22 @@ async fn terminate_streaming_child( grace: Option, ) -> 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 = ["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); diff --git a/engine/mast-engine/src/ops.rs b/engine/mast-engine/src/ops.rs index f454ebb..2c4cb1c 100644 --- a/engine/mast-engine/src/ops.rs +++ b/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()) }) } } diff --git a/engine/mast-engine/tests/engine_tests.rs b/engine/mast-engine/tests/engine_tests.rs index 0720e8b..c68dcdd 100644 --- a/engine/mast-engine/tests/engine_tests.rs +++ b/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::::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