From 744f6d1cec3be6b52c84fc2fa36e1fd8534970d6 Mon Sep 17 00:00:00 2001 From: Brad Heller Date: Wed, 13 May 2026 10:53:13 +0100 Subject: [PATCH] refactor(tower-runtime): structured AppCompletion / Status::Failed plumbing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `LocalApp::start` previously spawned `execute_local_app` with a `oneshot::Sender`; every `?` early-return in that task dropped the sender silently, and `LocalApp::status()` reported the resulting closed channel as `Error::WaiterClosed` — losing the actual cause (env join errors, `tower_uv::Error` variants, panics, etc.). Replace the channel payload with an internal `AppCompletion` enum and a new `Status::Failed(AppFailure)` shape that preserves the structured error. Highlights: - `AppCompletion` (internal) has `Exit(i32)`, `Cancelled`, `Failed(AppFailure)` variants. The spawn wrapper in `LocalApp::start` runs `inner_execute_local_app` inside `AssertUnwindSafe(...).catch_unwind()` and always sends an `AppCompletion` before exiting, so a closed waiter channel can no longer mask a real failure. - `Status::Failed(AppFailure)` replaces the old `Status::Failed { error_code, error_message }` strings. `AppFailure` has: * `Runtime(Error)` — structured runtime error * `Panic(String)` — `catch_unwind` payload * `Platform { error_code, error_message }` — opaque strings for callers that construct `Status::Failed` from outside `tower_runtime` (e.g. Kubernetes pod-status reconciliation). Stringification (`error_code` / `error_message`) is the consumer's responsibility, not the runtime's. - `Status::Cancelled` is a new terminal variant for explicit cancellation via the `CancellationToken`. Previously cancellation was represented as `Crashed { code: -1 }`, indistinguishable from an app that legitimately exits with code 255 / -1 or an IO error in `wait_for_process`. `inner_execute_local_app` now returns `Err(Error::Cancelled)` on every cancellation path (4 prologue checks + 3 post-`wait_for_process` checks that disambiguate cancellation from a child's genuine non-zero exit), and the spawn wrapper special-cases that into `AppCompletion::Cancelled`. - `Error` gains `Clone + PartialEq + Eq` (all variants are unit, so this is free) so `Status` can keep deriving `Clone + PartialEq`, which is required by status caching in `LocalApp::status()` and by existing integration tests. - `tower-cmd` (the only other in-crate consumer of `Status::Failed`) is updated for the new variant shape and prints a dedicated "cancelled" message for `Status::Cancelled`. - The integration test `test_abort_on_dependency_installation_failure` keeps asserting that `uv sync` non-zero exits surface as `Status::Crashed { code }` — `Status::Failed` is reserved for platform failures where the child never ran. `Error::WaiterClosed` now only fires when the spawn was aborted or the runtime was dropped — both real bugs worth surfacing as-is. --- Cargo.lock | 23 +-- Cargo.toml | 2 +- crates/tower-cmd/src/run.rs | 21 ++- crates/tower-runtime/Cargo.toml | 1 + crates/tower-runtime/src/errors.rs | 2 +- crates/tower-runtime/src/lib.rs | 32 ++++- crates/tower-runtime/src/local.rs | 170 +++++++++++++++++------ crates/tower-runtime/tests/local_test.rs | 3 + 8 files changed, 193 insertions(+), 61 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4238e116..1b60b0f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -491,7 +491,7 @@ dependencies = [ [[package]] name = "config" -version = "0.3.61" +version = "0.3.62" dependencies = [ "base64", "chrono", @@ -598,7 +598,7 @@ checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" [[package]] name = "crypto" -version = "0.3.61" +version = "0.3.62" dependencies = [ "aes-gcm", "base64", @@ -3348,7 +3348,7 @@ dependencies = [ [[package]] name = "testutils" -version = "0.3.61" +version = "0.3.62" dependencies = [ "pem", "rsa", @@ -3618,7 +3618,7 @@ checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" [[package]] name = "tower" -version = "0.3.61" +version = "0.3.62" dependencies = [ "config", "pyo3", @@ -3646,7 +3646,7 @@ dependencies = [ [[package]] name = "tower-api" -version = "0.3.61" +version = "0.3.62" dependencies = [ "reqwest", "serde", @@ -3658,7 +3658,7 @@ dependencies = [ [[package]] name = "tower-cmd" -version = "0.3.61" +version = "0.3.62" dependencies = [ "axum", "bytes", @@ -3728,7 +3728,7 @@ checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" [[package]] name = "tower-package" -version = "0.3.61" +version = "0.3.62" dependencies = [ "async-compression", "flate2", @@ -3752,11 +3752,12 @@ dependencies = [ [[package]] name = "tower-runtime" -version = "0.3.61" +version = "0.3.62" dependencies = [ "async-trait", "chrono", "config", + "futures", "nix 0.30.1", "snafu", "tmpdir", @@ -3775,7 +3776,7 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" [[package]] name = "tower-telemetry" -version = "0.3.61" +version = "0.3.62" dependencies = [ "tracing", "tracing-appender", @@ -3784,7 +3785,7 @@ dependencies = [ [[package]] name = "tower-uv" -version = "0.3.61" +version = "0.3.62" dependencies = [ "async-compression", "async_zip", @@ -3803,7 +3804,7 @@ dependencies = [ [[package]] name = "tower-version" -version = "0.3.61" +version = "0.3.62" dependencies = [ "anyhow", "chrono", diff --git a/Cargo.toml b/Cargo.toml index a59a228d..8045433c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ resolver = "2" [workspace.package] edition = "2021" -version = "0.3.61" +version = "0.3.62" description = "Tower is the best way to host Python data apps in production" rust-version = "1.81" authors = ["Brad Heller ", "Ben Lovell "] diff --git a/crates/tower-cmd/src/run.rs b/crates/tower-cmd/src/run.rs index 9ad5a189..be8050e6 100644 --- a/crates/tower-cmd/src/run.rs +++ b/crates/tower-cmd/src/run.rs @@ -251,13 +251,22 @@ where output::error(&format!("Your local run crashed with exit code: {}", code)); return Err(Error::AppCrashed); } - Status::Failed { - error_code, - error_message, - } => { + Status::Cancelled => { + output::error("Your local run was cancelled."); + return Err(Error::AppCrashed); + } + Status::Failed(failure) => { + let detail = match failure { + tower_runtime::AppFailure::Runtime(e) => format!("{:?}", e), + tower_runtime::AppFailure::Panic(msg) => format!("panic: {}", msg), + tower_runtime::AppFailure::Platform { + error_code, + error_message, + } => format!("{} ({})", error_message, error_code), + }; output::error(&format!( - "Your local run failed due to a platform error (code: {}, message: {})", - error_code, error_message + "Your local run failed due to a platform error: {}", + detail )); return Err(Error::AppCrashed); } diff --git a/crates/tower-runtime/Cargo.toml b/crates/tower-runtime/Cargo.toml index 37b47f62..e28fab94 100644 --- a/crates/tower-runtime/Cargo.toml +++ b/crates/tower-runtime/Cargo.toml @@ -9,6 +9,7 @@ license = { workspace = true } [dependencies] async-trait = { workspace = true } chrono = { workspace = true } +futures = { workspace = true } nix = { workspace = true } snafu = { workspace = true } tmpdir = { workspace = true } diff --git a/crates/tower-runtime/src/errors.rs b/crates/tower-runtime/src/errors.rs index 0326cd27..75a6b09c 100644 --- a/crates/tower-runtime/src/errors.rs +++ b/crates/tower-runtime/src/errors.rs @@ -1,6 +1,6 @@ use snafu::prelude::*; -#[derive(Debug, Snafu)] +#[derive(Clone, Debug, PartialEq, Eq, Snafu)] pub enum Error { #[snafu(display("failed to RPC server"))] RuntimeStartFailed, diff --git a/crates/tower-runtime/src/lib.rs b/crates/tower-runtime/src/lib.rs index 1be53ce8..446264fe 100644 --- a/crates/tower-runtime/src/lib.rs +++ b/crates/tower-runtime/src/lib.rs @@ -45,9 +45,34 @@ pub enum Status { Crashed { code: i32, }, + /// The app was explicitly terminated via the `CancellationToken` — a + /// deliberate stop, not a failure or a crash. Distinct from `Crashed` + /// because there's no meaningful exit code to report. + Cancelled, /// A platform-level failure (not the user's app). For example, pod scheduling /// failures, image pull errors, or other infrastructure issues. - Failed { + Failed(AppFailure), +} + +/// A platform-level failure — distinct from a child-process crash +/// (`Status::Crashed`) because the user's app never ran to completion. +/// Consumers (currently `tower-runtime-entrypoint` and the runner-side k8s +/// backend) translate this into the `error_code` / `error_message` strings +/// that flow through the termination-log → control-plane log pipeline. +#[derive(Clone, Debug, PartialEq)] +pub enum AppFailure { + /// A structured runtime error from `tower_runtime` (e.g. + /// `UnsupportedPlatform`, `SpawnFailed`, `DependencyInstallationFailed`). + Runtime(Error), + /// A panic inside the spawned task. `catch_unwind` only returns + /// `Box`, so we surface the best-effort textual payload. + Panic(String), + /// A platform failure surfaced from outside `tower_runtime` (e.g. the + /// kubelet refusing to pull an image, an init container failing, or a + /// parsed entrypoint termination message). The `error_code` is opaque + /// to `tower_runtime` — the consumer that constructed this variant owns + /// the vocabulary. + Platform { error_code: String, error_message: String, }, @@ -58,7 +83,10 @@ impl Status { pub fn is_terminal(&self) -> bool { matches!( self, - Status::Exited | Status::Crashed { .. } | Status::Failed { .. } + Status::Exited + | Status::Crashed { .. } + | Status::Cancelled + | Status::Failed(_) ) } } diff --git a/crates/tower-runtime/src/local.rs b/crates/tower-runtime/src/local.rs index e388fb04..07b368cf 100644 --- a/crates/tower-runtime/src/local.rs +++ b/crates/tower-runtime/src/local.rs @@ -6,7 +6,7 @@ use std::process::Stdio; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; -use crate::{errors::Error, OutputSender, StartOptions, Status}; +use crate::{errors::Error, AppFailure, OutputSender, StartOptions, Status}; use tokio::{ fs, @@ -35,18 +35,51 @@ use tower_uv::Uv; use crate::{App, Channel, Output, FD}; +/// Result of running an app to completion. The spawned execution task always +/// sends one of these on its waiter channel before exiting, so `status()` can +/// distinguish a real platform failure from a closed channel. +/// +/// `Failed(AppFailure)` preserves the structured `Error` (or panic payload) +/// so consumers can act on the variant directly — `error_code` / `error_message` +/// stringification is the consumer's responsibility. +#[derive(Clone, Debug)] +enum AppCompletion { + /// Child process exited (0 = clean, non-zero = crashed). + Exit(i32), + /// The cancellation token fired — explicit termination, not a failure. + /// Distinct from `Exit(non_zero)` so consumers can tell a deliberate stop + /// apart from a real crash. + Cancelled, + /// Platform-level failure inside the spawned task before/around the child + /// process (e.g. `tower_uv::Error`, env join failure, panic). + Failed(AppFailure), +} + +/// Best-effort extraction of a panic message from a `catch_unwind` payload. +fn panic_payload_message(payload: &Box) -> String { + if let Some(s) = payload.downcast_ref::<&'static str>() { + (*s).to_string() + } else if let Some(s) = payload.downcast_ref::() { + s.clone() + } else { + "execute_local_app panicked with non-string payload".to_string() + } +} + pub struct LocalApp { status: Mutex>, // waiter is what we use to communicate that the overall process is finished by the execution - // handle. - waiter: Mutex>, + // handle. The spawned task always sends an `AppCompletion` before it ends; if the channel + // ever closes without a send, that's a real bug (task aborted, runtime dropped) and surfaces + // as `Error::WaiterClosed`. + waiter: Mutex>, // terminator is what we use to flag that we want to terminate the child process. terminator: CancellationToken, // execute_handle keeps track of the current state of the execution lifecycle. - execute_handle: Option>>, + execute_handle: Option>, } // Helper function to check if a file is executable @@ -99,11 +132,18 @@ async fn find_bash() -> Result { } } -async fn execute_local_app( +/// Run the app to completion, returning the child's exit code on success or an +/// `Error` for any platform failure. Cancellation is reported as `Ok(-1)` to +/// preserve the historical `Status::Crashed { code: -1 }` semantic. +/// +/// The caller is responsible for delivering the result on the waiter channel — +/// see `LocalApp::start`, which wraps this in `catch_unwind` and an unconditional +/// `sx.send(AppCompletion::...)` so that no failure path can silently close the +/// channel. +async fn inner_execute_local_app( opts: StartOptions, - sx: oneshot::Sender, cancel_token: CancellationToken, -) -> Result<(), Error> { +) -> Result { let ctx = opts.ctx.clone(); let package = opts.package; let environment = opts.environment; @@ -157,9 +197,6 @@ async fn execute_local_app( // We do this before instantiating `Uv` because that can be somewhat time consuming. Likewise // this stops us from instantiating a bash process. if cancel_token.is_cancelled() { - // if there's a waiter, we want them to know that the process was cancelled so we have - // to return something on the relevant channel. - let _ = sx.send(-1); return Err(Error::Cancelled); } @@ -176,7 +213,14 @@ async fn execute_local_app( ) .await?; - let _ = sx.send(wait_for_process(ctx.clone(), &cancel_token, child).await); + let code = wait_for_process(ctx.clone(), &cancel_token, child).await; + // `wait_for_process` returns -1 on both cancellation and IO error; + // disambiguate so a cancellation surfaces as `Status::Cancelled` + // rather than `Status::Crashed { code: -1 }`. + if cancel_token.is_cancelled() { + return Err(Error::Cancelled); + } + Ok(code) } else { // we put Uv in to protected mode when there's no caching configured/enabled. let protected_mode = opts.cache_dir.is_none(); @@ -197,8 +241,6 @@ async fn execute_local_app( // Quickly do a check to see if there was a cancellation before we do a subprocess spawn to // ensure everything is in place. if cancel_token.is_cancelled() { - // again tell any waiters that we cancelled. - let _ = sx.send(-1); return Err(Error::Cancelled); } @@ -224,17 +266,21 @@ async fn execute_local_app( // Wait for venv to finish up. let res = wait_for_process(ctx.clone(), &cancel_token, child).await; + // Distinguish cancellation from a real uv-venv non-zero exit. + if cancel_token.is_cancelled() { + return Err(Error::Cancelled); + } + if res != 0 { - // If the venv process failed, we want to return an error. - let _ = sx.send(res); - return Err(Error::VirtualEnvCreationFailed); + // `uv venv` exited non-zero — surface as a crash with the child's + // exit code. `Status::Failed` is reserved for platform errors + // where the child never ran at all. + return Ok(res); } // Check once more if the process was cancelled before we do a uv sync. The sync itself, // once started, will take a while and we have logic for checking for cancellation. if cancel_token.is_cancelled() { - // again tell any waiters that we cancelled. - let _ = sx.send(-1); return Err(Error::Cancelled); } @@ -259,6 +305,13 @@ async fn execute_local_app( Ok(child) => { let mut res = run_setup_child(&ctx, &cancel_token, &opts.output_sender, child).await; + // If sync was cancelled, don't bother retrying — bail out + // cleanly so the receiver sees `Status::Cancelled` instead of + // `Status::Crashed { code: -1 }`. + if cancel_token.is_cancelled() { + return Err(Error::Cancelled); + } + // If the install failed, retry with the legacy setuptools<82 // pin. Some apps (those whose transitive deps rely on // pkg_resources) need that pin to install successfully; we don't @@ -276,21 +329,22 @@ async fn execute_local_app( .sync_with_legacy_setuptools_pin(&working_dir, &env_vars) .await?; res = run_setup_child(&ctx, &cancel_token, &opts.output_sender, retry_child).await; + if cancel_token.is_cancelled() { + return Err(Error::Cancelled); + } } if res != 0 { - // If the sync process failed, we want to return an error. - let _ = sx.send(res); - return Err(Error::DependencyInstallationFailed); + // `uv sync` exited non-zero — surface as a crash with the + // child's exit code. `Status::Failed` is reserved for + // platform errors where the child never ran at all. + return Ok(res); } } } // Check once more to see if the process was cancelled, this will bail us out early. if cancel_token.is_cancelled() { - // if there's a waiter, we want them to know that the process was cancelled so we have - // to return something on the relevant channel. - let _ = sx.send(-1); return Err(Error::Cancelled); } @@ -313,11 +367,12 @@ async fn execute_local_app( BufReader::new(stderr), )); - let _ = sx.send(wait_for_process(ctx.clone(), &cancel_token, child).await); + let code = wait_for_process(ctx.clone(), &cancel_token, child).await; + if cancel_token.is_cancelled() { + return Err(Error::Cancelled); + } + Ok(code) } - - // Everything was properly executed I suppose. - return Ok(()); } impl Drop for LocalApp { @@ -338,12 +393,36 @@ impl Drop for LocalApp { impl App for LocalApp { async fn start(opts: StartOptions) -> Result { + use futures::FutureExt; + use std::panic::AssertUnwindSafe; + let terminator = CancellationToken::new(); - let (sx, rx) = oneshot::channel::(); + let (sx, rx) = oneshot::channel::(); let waiter = Mutex::new(rx); - let handle = tokio::spawn(execute_local_app(opts, sx, terminator.clone())); + let task_terminator = terminator.clone(); + let handle = tokio::spawn(async move { + // `catch_unwind` turns a panic inside the spawned task into a normal + // `Err(payload)` so we can still report it on the waiter channel + // instead of leaving the receiver to observe a silently-closed + // channel (the historical `WaiterClosed` bug). + // + // `Err(Error::Cancelled)` is special-cased: cancellation is a + // deliberate stop, not a failure, so it surfaces as + // `AppCompletion::Cancelled` → `Status::Cancelled` rather than + // `AppCompletion::Failed(AppFailure::Runtime(Error::Cancelled))`. + let completion = match AssertUnwindSafe(inner_execute_local_app(opts, task_terminator)) + .catch_unwind() + .await + { + Ok(Ok(code)) => AppCompletion::Exit(code), + Ok(Err(Error::Cancelled)) => AppCompletion::Cancelled, + Ok(Err(e)) => AppCompletion::Failed(AppFailure::Runtime(e)), + Err(panic) => AppCompletion::Failed(AppFailure::Panic(panic_payload_message(&panic))), + }; + let _ = sx.send(completion); + }); let execute_handle = Some(handle); Ok(Self { @@ -377,17 +456,28 @@ impl App for LocalApp { match res { Err(TryRecvError::Empty) => Ok(Status::Running), + // The spawned task always sends an `AppCompletion` before + // exiting, so a closed channel here means the task was + // aborted or the runtime dropped — a real bug worth + // surfacing as-is. Err(TryRecvError::Closed) => Err(Error::WaiterClosed), - Ok(t) => { - // We save this for the next time this gets called. - if t == 0 { - *status = Some(Status::Exited); - Ok(Status::Exited) - } else { - let next_status = Status::Crashed { code: t }; - *status = Some(next_status.clone()); - Ok(next_status) - } + Ok(AppCompletion::Exit(0)) => { + *status = Some(Status::Exited); + Ok(Status::Exited) + } + Ok(AppCompletion::Exit(code)) => { + let next_status = Status::Crashed { code }; + *status = Some(next_status.clone()); + Ok(next_status) + } + Ok(AppCompletion::Cancelled) => { + *status = Some(Status::Cancelled); + Ok(Status::Cancelled) + } + Ok(AppCompletion::Failed(failure)) => { + let next_status = Status::Failed(failure); + *status = Some(next_status.clone()); + Ok(next_status) } } } diff --git a/crates/tower-runtime/tests/local_test.rs b/crates/tower-runtime/tests/local_test.rs index 5dda4d9d..b0ab679a 100644 --- a/crates/tower-runtime/tests/local_test.rs +++ b/crates/tower-runtime/tests/local_test.rs @@ -373,6 +373,9 @@ async fn test_abort_on_dependency_installation_failure() { Status::None => { panic!("App should have a status"); } + Status::Cancelled => { + panic!("App should have crashed, not been cancelled"); + } Status::Failed { .. } => { panic!("App should have crashed, not failed with a platform error"); }