fix(execute): stop waiting for processes a task left running

CHANGELOG.md

@@ -1,5 +1,6 @@
 # Changelog
 
+- **Fixed** `vp run` no longer waits for a process that a task left running when that process still holds the task's output or its connection to the runner, such as a preview server started by a test setup and stopped by killing only its `vp` wrapper. This affected cached tasks and tasks run with `--log=grouped` or `--log=labeled`, which could wait until that process exited, or forever. `vp run` now gives that process a few seconds after the task exits, then moves on, and doesn't cache the run, because that process's output or reports to the runner may be missing ([#875](https://github.com/voidzero-dev/vite-task/pull/875)).
 - **Perf** When tasks miss the cache, `vp run` no longer waits for the disk after saving each task's result, so the run finishes sooner. If the computer loses power or the operating system crashes during `vp run`, the cache entries that the run saved can be lost, and those tasks run again next time instead of hitting the cache ([#871](https://github.com/voidzero-dev/vite-task/pull/871)).
 - **Fixed** Running `vp run` in a package that sits inside a workspace folder but isn't listed in its `pnpm-workspace.yaml` `packages` or `package.json` `workspaces` now runs that package's own tasks, instead of treating the enclosing workspace as its root. When both configurations exist, `pnpm-workspace.yaml` takes precedence ([#859](https://github.com/voidzero-dev/vite-task/pull/859)).
 - **Fixed** A mistyped or misplaced setting in a task now gets an error that names it. For example, `cache: { outputs: [...] }` now fails with an `unknown field` error that names `outputs` and lists the accepted settings, instead of `data did not match any variant of untagged enum UserTaskDefinition` ([vite-plus#2635](https://github.com/voidzero-dev/vite-plus/issues/2635), [#815](https://github.com/voidzero-dev/vite-task/pull/815)).

crates/vt/Cargo.toml

@@ -40,6 +40,7 @@ tokio = { workspace = true, features = [
   "macros",
   "process",
   "sync",
+  "time",
 ] }
 tokio-util = { workspace = true, features = ["rt"] }
 tracing = { workspace = true }

crates/vt/src/session/event.rs

@@ -125,6 +125,11 @@ pub enum CacheNotUpdatedReason {
     /// inputs and outputs that are not all of them.
     #[cfg_attr(not(fspy), expect(dead_code, reason = "only fspy tracking can be incomplete"))]
     TrackingIncomplete,
+    /// The task exited, but a process it started still held its stdout,
+    /// stderr or runner connection open after the grace period, so the
+    /// runner stopped waiting for it. Its output or its reports to the
+    /// runner may be incomplete.
+    ProcessLeftRunning,
 }
 
 #[derive(Debug)]

crates/vt/src/session/execute/cache_update.rs

@@ -46,18 +46,24 @@ type TrackedEnvQueryValues = BTreeMap<TrackedEnvQuery, BTreeMap<Str, EnvValueHas
 /// `finish()` call; this function never reports by itself. The guard clauses
 /// run in priority order — each names the reason the run is *not* cached, and
 /// only a run that passes them all is stored. A run cancelled by Ctrl-C or
-/// fast-fail (`cancel_token`) isn't cached.
+/// fast-fail (`cancel_token`) isn't cached, and neither is one that left a
+/// process holding its pipes or runner connection (`left_running`).
 #[expect(
     clippy::too_many_arguments,
     reason = "the run's full context is genuinely needed to decide and store the cache entry"
 )]
+#[expect(
+    clippy::too_many_lines,
+    reason = "the guard clauses are listed in priority order in one place"
+)]
 pub(super) async fn update_cache(
     cache: &ExecutionCache,
     workspace_root: &Arc<AbsolutePath>,
     cache_dir: &AbsolutePath,
     state: CacheState<'_>,
     outcome: &ChildOutcome,
     reports: Option<&Reports>,
+    left_running: bool,
     duration: Duration,
     cancel_token: &CancellationToken,
 ) -> (CacheUpdateStatus, Option<ExecutionError>) {
@@ -91,6 +97,13 @@ pub(super) async fn update_cache(
         return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::NonZeroExitStatus), None);
     }
 
+    if left_running {
+        // The runner stopped waiting for a process the task left running,
+        // which still held its stdout, stderr or runner connection. Output
+        // and reports from that process may be missing.
+        return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::ProcessLeftRunning), None);
+    }
+
     // The accesses, or `None` when the task was not tracked at all. An
     // `Err` means it made more file accesses than the tracking channel had
     // room for, so what arrived is a subset of what it touched, and an entry

crates/vt/src/session/execute/mod.rs

@@ -15,7 +15,7 @@ use std::{
     collections::BTreeMap,
     ffi::{OsStr, OsString},
     sync::Arc,
-    time::Instant,
+    time::{Duration, Instant},
 };
 
 use futures_util::future::LocalBoxFuture;
@@ -468,20 +468,13 @@ async fn run(
     //    pipe_stdio alone makes the combined future large enough to trip
     //    clippy::large_futures in every caller otherwise.
     let RunHandles { sinks, ipc } = mode.run_handles();
-    let (wait_result, ipc_server_result) = if let Some(IpcHandles { stop_accepting, driver }) = ipc
-    {
-        let child_work =
-            Box::pin(run_child(child, sinks, Some(stop_accepting), fast_fail_token.clone()));
-        let (wait_result, join_result) = tokio::join!(child_work, driver);
-        if let Err(e) = &join_result {
-            tracing::warn!(?e, "IPC server failed; cache will not be updated");
-        }
-        (wait_result, Some(join_result.map(Recorder::into_reports)))
+    let (wait_result, ipc_server_result) = if let Some(ipc) = ipc {
+        run_child_with_ipc(child, sinks, ipc, fast_fail_token.clone()).await
     } else {
         let child_work = Box::pin(run_child(child, sinks, None, fast_fail_token.clone()));
         (child_work.await, None)
     };
-    let outcome = wait_result.map_err(Report::failed)?;
+    let ChildRun { outcome, left_running, .. } = wait_result.map_err(Report::failed)?;
     let duration = start.elapsed();
 
     // Extract reports, or short-circuit when the IPC server failed. An Err
@@ -516,6 +509,7 @@ async fn run(
                 state,
                 &outcome,
                 reports.as_ref(),
+                left_running,
                 duration,
                 &cancel_token,
             )
@@ -623,44 +617,107 @@ async fn replay_cache_hit(
     Report::CacheHit
 }
 
+/// How long the runner keeps reading a task's stdout and stderr after the
+/// task's own process has exited, and then how long it keeps serving the
+/// task's runner connections. A pipe or connection still open after that is
+/// held by a process the task left running, so the runner stops waiting
+/// instead of waiting for that process to exit.
+const LEFTOVER_GRACE: Duration = Duration::from_secs(2);
+
+/// What [`run_child`] observed.
+struct ChildRun {
+    outcome: ChildOutcome,
+    /// `true` when a process the child started still held its stdout, stderr
+    /// or runner connection after [`LEFTOVER_GRACE`], so the runner stopped
+    /// waiting. Output or reports may be incomplete.
+    left_running: bool,
+}
+
+/// [`run_child`] with the IPC server's driver polled alongside it, which must
+/// make progress during the pipe drain (see step 6 in [`run`]).
+///
+/// Once the child has exited, the driver drains for at most
+/// [`LEFTOVER_GRACE`]. A client still connected then belongs to a process the
+/// task left running: the driver isn't polled again (dropping it with the
+/// run's state closes the connection), the reports are `None`, and
+/// [`ChildRun::left_running`] is set.
+async fn run_child_with_ipc(
+    child: ChildHandle,
+    sinks: Option<PipeSinks<'_>>,
+    IpcHandles { stop_accepting, driver }: IpcHandles<'_>,
+    fast_fail_token: CancellationToken,
+) -> (Result<ChildRun, ExecutionError>, Option<Result<Reports, vt_server::Error>>) {
+    let mut child_work = Box::pin(run_child(child, sinks, Some(stop_accepting), fast_fail_token));
+    let mut join_result = None;
+    let mut wait_result = loop {
+        tokio::select! {
+            wait_result = &mut child_work => break wait_result,
+            result = &mut *driver, if join_result.is_none() => join_result = Some(result),
+        }
+    };
+    if join_result.is_none() {
+        join_result = tokio::time::timeout(LEFTOVER_GRACE, driver).await.ok();
+        if join_result.is_none()
+            && let Ok(run) = &mut wait_result
+        {
+            run.left_running = true;
+        }
+    }
+    if let Some(Err(e)) = &join_result {
+        tracing::warn!(?e, "IPC server failed; cache will not be updated");
+    }
+    (wait_result, join_result.map(|result| result.map(Recorder::into_reports)))
+}
+
 /// Phase 6: drain the child's pipes (if piped) and wait for exit. A pipe
 /// failure cancels the child so the wait future kills it instead of orphaning
 /// it. After the child exits (on every path), `stop_accepting` is signalled so
 /// the IPC server stops accepting and starts draining.
+///
+/// The pipes usually close when the child exits, but a process the child
+/// started can inherit them and keep them open. Once the child has exited,
+/// they're read for [`LEFTOVER_GRACE`] at most.
 async fn run_child(
     mut child: ChildHandle,
     sinks: Option<PipeSinks<'_>>,
     stop_accepting: Option<&StopAccepting>,
     fast_fail_token: CancellationToken,
-) -> Result<ChildOutcome, ExecutionError> {
-    let pipe_result: Result<(), ExecutionError> = if let Some(sinks) = sinks {
+) -> Result<ChildRun, ExecutionError> {
+    let mut wait = child.wait;
+    let mut left_running = false;
+    let (pipe_result, wait_result) = if let Some(sinks) = sinks {
         let stdout = child.stdout.take().expect("SpawnStdio::Piped yields a stdout pipe");
         let stderr = child.stderr.take().expect("SpawnStdio::Piped yields a stderr pipe");
-        #[expect(
-            clippy::large_futures,
-            reason = "pipe_stdio streams child I/O and creates a large future"
-        )]
-        let r = pipe_stdio(stdout, stderr, sinks, fast_fail_token.clone()).await;
-        r.map_err(|err| ExecutionError::ForwardTaskProcessOutput(err.into()))
-    } else {
-        Ok(())
-    };
-
-    let wait_result = match pipe_result {
-        Ok(()) => {
-            child.wait.await.map_err(|err| ExecutionError::WaitForTaskProcessExit(err.into()))
-        }
-        Err(err) => {
-            // Pipe failed — cancel so `child.wait` kills the child instead of
-            // orphaning it. Still signal the server below so it can drain.
-            fast_fail_token.cancel();
-            let _ = child.wait.await;
-            Err(err)
+        let mut pipes = Box::pin(pipe_stdio(stdout, stderr, sinks, fast_fail_token.clone()));
+        tokio::select! {
+            pipe_result = &mut pipes => {
+                if pipe_result.is_err() {
+                    // Pipe failed — cancel so `child.wait` kills the child
+                    // instead of orphaning it.
+                    fast_fail_token.cancel();
+                }
+                (pipe_result, wait.await)
+            }
+            wait_result = &mut wait => {
+                let pipe_result = match tokio::time::timeout(LEFTOVER_GRACE, &mut pipes).await {
+                    Ok(pipe_result) => pipe_result,
+                    Err(_elapsed) => {
+                        left_running = true;
+                        Ok(())
+                    }
+                };
+                (pipe_result, wait_result)
+            }
         }
+    } else {
+        (Ok(()), wait.await)
     };
 
+    // Signal on every path, so the server can drain.
     if let Some(stop_accepting) = stop_accepting {
         stop_accepting.signal();
     }
-    wait_result
+    pipe_result.map_err(|err| ExecutionError::ForwardTaskProcessOutput(err.into()))?;
+    let outcome = wait_result.map_err(|err| ExecutionError::WaitForTaskProcessExit(err.into()))?;
+    Ok(ChildRun { outcome, left_running })
 }

crates/vt/src/session/reporter/summary.rs

@@ -134,6 +134,11 @@ pub enum SpawnOutcome {
         /// Set when a runner-aware tool called `disableCache()`, skipping
         /// cache update.
         tool_disabled_cache: bool,
+        /// `true` when the task exited but a process it started still held
+        /// its stdout, stderr or runner connection, so the runner stopped
+        /// waiting for it and didn't update the cache.
+        #[serde(default)]
+        process_left_running: bool,
         /// Why uploading the entry to the remote cache failed, if it did.
         /// The local cache was still updated.
         upload_error: Option<SavedError>,
@@ -359,6 +364,10 @@ impl TaskResult {
             cache_update_status,
             CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::TrackingIncomplete)
         );
+        let process_left_running = matches!(
+            cache_update_status,
+            CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::ProcessLeftRunning)
+        );
 
         match cache_status {
             // The only error a cache hit can have is a failed restore.
@@ -380,6 +389,7 @@ impl TaskResult {
                     ipc_server_error,
                     tool_disabled_cache,
                     tracking_incomplete,
+                    process_left_running,
                 ),
             },
             CacheStatus::Miss(cache_miss) => Self::Spawned {
@@ -394,6 +404,7 @@ impl TaskResult {
                     ipc_server_error,
                     tool_disabled_cache,
                     tracking_incomplete,
+                    process_left_running,
                 ),
             },
         }
@@ -412,6 +423,11 @@ impl TaskResult {
 
 /// Build a [`SpawnOutcome`] from process exit status and optional pre-converted error.
 /// A failed upload is set later, with [`TaskResult::set_upload_error`].
+#[expect(
+    clippy::too_many_arguments,
+    clippy::fn_params_excessive_bools,
+    reason = "one parameter per reason the cache wasn't updated, copied into `SpawnOutcome::Success`"
+)]
 fn spawn_outcome_from_execution(
     exit_status: Option<std::process::ExitStatus>,
     saved_error: Option<&SavedError>,
@@ -420,6 +436,7 @@ fn spawn_outcome_from_execution(
     ipc_server_error: Option<SavedError>,
     tool_disabled_cache: bool,
     tracking_incomplete: bool,
+    process_left_running: bool,
 ) -> SpawnOutcome {
     match (exit_status, saved_error) {
         // Spawn error — process never ran
@@ -432,6 +449,7 @@ fn spawn_outcome_from_execution(
             ipc_server_error,
             tool_disabled_cache,
             tracking_incomplete,
+            process_left_running,
             upload_error: None,
         },
         // Process exited with non-zero code
@@ -454,6 +472,7 @@ fn spawn_outcome_from_execution(
             ipc_server_error: None,
             tool_disabled_cache: false,
             tracking_incomplete: false,
+            process_left_running: false,
             upload_error: None,
         },
     }
@@ -599,6 +618,19 @@ impl TaskResult {
                 &[],
             );
         }
+        // The runner stopped waiting for a process the task left running.
+        if let Self::Spawned {
+            outcome: SpawnOutcome::Success { process_left_running: true, .. },
+            ..
+        } = self
+        {
+            return (
+                Str::from(
+                    "→ Not cached: the task exited, but a process it started kept its output or runner connection open",
+                ),
+                &[],
+            );
+        }
         // fspy-unsupported-on-this-OS message — same overrides precedence as above
         if let Self::Spawned {
             outcome: SpawnOutcome::Success { fspy_unsupported: true, .. }, ..
@@ -1210,6 +1242,7 @@ mod tests {
                     ipc_server_error: None,
                     tracking_incomplete: false,
                     tool_disabled_cache: false,
+                    process_left_running: false,
                     upload_error: Some(error),
                 },
             },
@@ -1245,6 +1278,7 @@ mod tests {
                     ipc_server_error: None,
                     tracking_incomplete: false,
                     tool_disabled_cache: false,
+                    process_left_running: false,
                     upload_error: None,
                 },
             },

crates/vt_bin/Cargo.toml

@@ -29,6 +29,7 @@ tokio = { workspace = true, features = ["full"] }
 vt_path = { workspace = true }
 vt_str = { workspace = true }
 vt = { workspace = true }
+vt_client = { workspace = true }
 which = { workspace = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]

crates/vt_bin/src/main.rs

@@ -3,9 +3,12 @@ use vt::{Command, ExitStatus, Session};
 use vt_bin::OwnedSessionConfig;
 
 fn main() -> ! {
-    let status: ExitStatus =
-        tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap().block_on(run());
+    let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
+    let status: ExitStatus = runtime.block_on(run());
 
+    // Exit without dropping the runtime: dropping it waits for its blocking
+    // threads, and one can be stuck reading an IPC connection that a process
+    // a task left running still holds open.
     std::process::exit(i32::from(status.0));
 }