Worker: kill the whole process group when an action times out or is cancelled

nativelink-worker/src/namespace_utils.rs

@@ -24,6 +24,29 @@ pub struct MaybeNamespacedChild {
     child: tokio::process::Child,
 }
 
+/// SIGKILL to every process in the group `pgid` leads. ESRCH means the
+/// group is already gone, which is the outcome wanted; anything else, EPERM
+/// above all, means a survivor and is reported.
+#[cfg(target_os = "linux")]
+fn kill_process_group(pgid: u32) {
+    let Ok(pgid) = i32::try_from(pgid) else {
+        return;
+    };
+    // SAFETY: killpg only takes integers and has no memory safety
+    // considerations; a stale group id is reported as ESRCH, not acted on.
+    let rc = unsafe { libc::killpg(pgid, libc::SIGKILL) };
+    if rc != 0 {
+        let err = Error::last_os_error();
+        if err.raw_os_error() != Some(libc::ESRCH) {
+            error!(
+                pgid,
+                ?err,
+                "Could not kill the action's process group; a descendant may have survived"
+            );
+        }
+    }
+}
+
 impl MaybeNamespacedChild {
     pub const fn new(namespaced: bool, child: tokio::process::Child) -> Self {
         Self { namespaced, child }
@@ -34,7 +57,14 @@ impl MaybeNamespacedChild {
     }
 
     /// Send SIGTERM if namespaced which sends SIGKILL to the child, otherwise
-    /// send SIGKILL to the child.
+    /// SIGKILL to the child's whole process group and then the child.
+    ///
+    /// Not namespaced, the child is its own group leader, so the group is
+    /// the action and every descendant it forked: a timed-out or cancelled
+    /// action used to lose only its direct child while its descendants ran
+    /// on, holding memory and CPU against the next action (issue #225).
+    /// Namespaced, the leader is the stub that runs the sandbox and cleans
+    /// up after it, so the group is left alone and the stub is told.
     pub async fn kill(&mut self) -> Result<(), Error> {
         if self.namespaced {
             // It would be safer to call send_signal to use the pidfd to avoid
@@ -55,6 +85,10 @@ impl MaybeNamespacedChild {
                 return self.child.wait().await.map(|_| ());
             }
         }
+        #[cfg(target_os = "linux")]
+        if let Some(pgid) = self.child.id() {
+            kill_process_group(pgid);
+        }
         self.child.kill().await
     }
 

nativelink-worker/src/running_actions_manager.rs

@@ -2081,8 +2081,6 @@ impl RunningActionImpl {
                     let exit_status = maybe_exit_status.err_tip(|| "Failed to collect exit code of process")?;
                     // TODO(palfrey) We should implement stderr/stdout streaming to client here.
                     // If we get killed before the stream is started, then these will lock up.
-                    // TODO(palfrey) There is a significant bug here. If we kill the action and the action creates
-                    // child processes, it can create zombies. See: https://github.com/tracemachina/nativelink/issues/225
                     let (stdout, stderr) = if killed_action {
                         drop(timer);
                         (Bytes::new(), Bytes::new())

nativelink-worker/tests/running_actions_manager_test.rs

@@ -2926,6 +2926,121 @@ exit 0
         Ok(())
     }
 
+    #[cfg(target_os = "linux")]
+    #[nativelink_test]
+    async fn timed_out_action_takes_its_children_with_it() -> Result<(), Box<dyn core::error::Error>>
+    {
+        const WORKER_ID: &str = "foo_worker_id";
+        let (_, _, cas_store, _ac_store) = setup_stores().await?;
+        let root_action_directory = make_temp_path("root_action_directory");
+        fs::create_dir_all(&root_action_directory).await?;
+        // Under the (existing) action root, so its parent directory exists.
+        let pid_file = format!("{root_action_directory}/children.pids");
+
+        let running_actions_manager =
+            Arc::new(RunningActionsManagerImpl::new(RunningActionsManagerArgs {
+                root_action_directory: root_action_directory.clone(),
+                execution_configuration: ExecutionConfiguration::default(),
+                cas_store: cas_store.clone(),
+                ac_store: None,
+                historical_store: Store::new(cas_store.clone()),
+                upload_action_result_config: &UploadActionResultConfig {
+                    upload_ac_results_strategy: UploadCacheResultsStrategy::Never,
+                    ..Default::default()
+                },
+                max_action_timeout: Duration::MAX,
+                max_upload_timeout: Duration::from_secs(DEFAULT_MAX_UPLOAD_TIMEOUT),
+                max_cleanup_wait: Duration::from_secs(DEFAULT_MAX_CLEANUP_WAIT),
+                max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF),
+                timeout_handled_externally: false,
+                active_input_leases: false,
+                directory_cache: None,
+                // Host pids and a host-visible pid file: the point is what the kill
+                // reaches, not the namespaces.
+                use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::No,
+            })?);
+        // Two grandchildren that would outlive a kill aimed at the shell
+        // alone; their pids land in a file outside the action directory.
+        let command = Command {
+            arguments: vec![
+                "sh".to_string(),
+                "-c".to_string(),
+                format!("sleep 60 & echo $! > {pid_file}; sleep 60 & echo $! >> {pid_file}; wait"),
+            ],
+            working_directory: ".".to_string(),
+            environment_variables: vec![EnvironmentVariable {
+                name: "PATH".to_string(),
+                value: env::var("PATH").unwrap(),
+            }],
+            ..Default::default()
+        };
+        let command_digest = serialize_and_upload_message(
+            &command,
+            cas_store.as_pin(),
+            &mut DigestHasherFunc::Sha256.hasher(),
+        )
+        .await?;
+        let input_root_digest = serialize_and_upload_message(
+            &Directory::default(),
+            cas_store.as_pin(),
+            &mut DigestHasherFunc::Sha256.hasher(),
+        )
+        .await?;
+        let action = Action {
+            command_digest: Some(command_digest.into()),
+            input_root_digest: Some(input_root_digest.into()),
+            timeout: Some(prost_types::Duration {
+                seconds: 1,
+                nanos: 0,
+            }),
+            ..Default::default()
+        };
+        let action_digest = serialize_and_upload_message(
+            &action,
+            cas_store.as_pin(),
+            &mut DigestHasherFunc::Sha256.hasher(),
+        )
+        .await?;
+
+        let running_action_impl = running_actions_manager
+            .create_and_add_action(
+                WORKER_ID.to_string(),
+                StartExecute {
+                    request_metadata: None,
+                    execute_request: Some(ExecuteRequest {
+                        action_digest: Some(action_digest.into()),
+                        digest_function: ProtoDigestFunction::Sha256.into(),
+                        ..Default::default()
+                    }),
+                    operation_id: OperationId::default().to_string(),
+                    queued_timestamp: None,
+                    platform: None,
+                    worker_id: WORKER_ID.to_string(),
+                },
+            )
+            .await?;
+
+        let result = run_action(running_action_impl).await?;
+        let err = result
+            .error
+            .expect("a timed-out action must carry an error");
+        assert_eq!(err.code, Code::DeadlineExceeded, "{err}");
+
+        let pids: Vec<i32> = std::fs::read_to_string(&pid_file)?
+            .lines()
+            .filter_map(|line| line.trim().parse().ok())
+            .collect();
+        assert_eq!(pids.len(), 2, "both children should have written their pid");
+        // A moment for the kernel to reap; then neither pid may exist.
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        for pid in pids {
+            // SAFETY: signal 0 checks for existence and sends nothing.
+            let alive = unsafe { libc::kill(pid, 0) } == 0;
+            assert!(!alive, "child {pid} outlived the timed-out action");
+        }
+        Ok(())
+    }
+
     #[cfg_attr(feature = "nix", ignore)]
     #[nativelink_test]
     async fn entrypoint_sends_timeout_via_side_channel() -> Result<(), Box<dyn core::error::Error>>