diff --git a/README.md b/README.md index f7246873..86ee63ed 100644 --- a/README.md +++ b/README.md @@ -302,7 +302,11 @@ curl 'http://localhost:8080/dependency_edges?from=main&to=my-feature-branch' identity, liveness, and cache size usage without scraping logs. Unlike the query endpoints it is never gated on readiness, so it still responds on an un-ready or lame-ducked instance (the `ready` field reports the current state). The `cache` size fields are populated for the local-disk backend - and are `null` for a backend whose size is not cheaply knowable in-process. + and are `null` for a backend whose size is not cheaply knowable in-process. `activeQueries` counts + the queries accepted and not yet finished (queued, running, or abandoned by a client that stopped + waiting), and `workspaceBusy` is `true` while any operation holds `--workspacePath` (a query's + checkout and `bazel query`, or a startup warmup). A tool that shares the checkout with the server + should wait for `workspaceBusy` to be `false` before touching it. ```bash curl 'http://localhost:8080/metrics' @@ -313,6 +317,8 @@ curl 'http://localhost:8080/metrics' "version": "31.4.0", "uptimeSeconds": 3600, "ready": true, + "activeQueries": 1, + "workspaceBusy": true, "gitEngine": "subprocess", "trackDeps": false, "cache": {"directory": "/var/cache/bazel-diff", "remote": "s3://my-bucket/bazel-diff/", "entries": 128, "sizeBytes": 4823913, "sizeHuman": "4.6 MB"}, @@ -414,6 +420,16 @@ Notes and operational guidance: * The service checks out revisions inside `--workspacePath`, so point it at a dedicated clone, not a working tree you edit. All workspace-mutating work (git checkout + `bazel query`) is serialized, so a single instance answers one cold query at a time; the per-SHA cache absorbs the rest. +* `--requestTimeout ` (default `0`, no limit) bounds how long a query may run. A query that + exceeds it is cancelled, not merely abandoned: its `git` and `bazel` subprocesses are killed, the + hashes it had not finished generating are not written to the local or shared cache, and the `504` + is sent only after the query has released the workspace. The `504` body reports `cancelled` + (`true` once the query has stopped; `false` only if it was still winding down a minute later, in + which case `/metrics` `activeQueries` reports when it is done) and `workspaceBusy` (whether + another operation holds the workspace at that moment). A client that enforces its own deadline + instead cannot cancel the server-side work: it keeps running to completion, so poll `/metrics` for + `workspaceBusy` before reusing the checkout, or prefer `--requestTimeout`. A killed `git checkout` + can leave the clone mid-switch; the next query repairs it with `git checkout --force`. * Git operations (fetch and checkout) shell out to the `git` binary at `--gitPath` (default `git` on the `PATH`), so a `git` binary must be available on the host. The working tree is checked out on disk for `bazel query` to read. Because native git performs every fetch, all clone shapes are diff --git a/src/bazel.rs b/src/bazel.rs index b6ba6ba9..fb4e5faa 100644 --- a/src/bazel.rs +++ b/src/bazel.rs @@ -1,3 +1,4 @@ +use crate::cancel::{self, Cancellation}; use crate::proto::analysis::{ConfiguredTarget, CqueryResult}; use crate::proto::blaze_query::{ target, Attribute, ConfiguredRuleInput, PackageGroup, Repository, Rule, Target, @@ -8,9 +9,10 @@ use rayon::prelude::*; use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::fs::{self, File}; -use std::io::{BufRead, BufReader, Read}; +use std::io::{self, BufRead, BufReader, Read}; use std::path::{Path, PathBuf}; -use std::process::{Command, Stdio}; +use std::process::{Command, Output, Stdio}; +use std::sync::Arc; use std::time::Instant; use tempfile::NamedTempFile; @@ -32,6 +34,9 @@ pub struct BazelOptions { pub exclude_targets_query: Option, pub no_bazelrc: bool, pub verbose: bool, + /// Set by `bazel-diff serve` for the query being answered: every Bazel subprocess is run + /// through it so an abandoned query can kill them. `None` for the CLI commands. + pub cancellation: Option>, } impl BazelOptions { @@ -49,17 +54,36 @@ impl BazelOptions { command } - fn run_capture(&self, args: &[&str]) -> Result { + fn run_capture(&self, args: &[&str]) -> Result { let mut command = self.command(); command.args(args).stdin(Stdio::null()); if self.verbose { eprintln!("[Info] Command: {command:?}"); } - command - .output() + self.run(&mut command) .with_context(|| format!("failed to execute {}", self.bazel.display())) } + /// Runs a Bazel command like `Command::output`, through the request's cancellation token when + /// there is one. + fn run(&self, command: &mut Command) -> io::Result { + cancel::output(self.cancellation.as_deref(), command) + } + + /// [`run`](Self::run) for a command whose stdout is redirected to a file. + fn run_configured(&self, command: &mut Command) -> io::Result { + cancel::output_configured(self.cancellation.as_deref(), command) + } + + /// Fails once the query these options belong to has been cancelled, so a phase without a + /// subprocess to kill still stops promptly. + pub fn check_cancelled(&self) -> Result<()> { + if let Some(token) = &self.cancellation { + token.check()?; + } + Ok(()) + } + pub fn is_bzlmod_enabled(&self) -> bool { self.run_capture(&["mod", "graph"]) .map(|output| output.status.success()) @@ -278,7 +302,9 @@ impl BazelOptions { .stdout(File::create(output_file.path())?) .stdin(Stdio::null()) .stderr(Stdio::piped()); - let output = command.output().context("execute bazel mod show_repo")?; + let output = self + .run_configured(&mut command) + .context("execute bazel mod show_repo")?; if !output.stderr.is_empty() { eprint!("{}", String::from_utf8_lossy(&output.stderr)); } @@ -355,7 +381,9 @@ impl BazelOptions { eprintln!("[Info] Executing Query: {expression}"); eprintln!("[Info] Command: {command:?}"); } - let output = command.output().context("failed to execute Bazel query")?; + let output = self + .run_configured(&mut command) + .context("failed to execute Bazel query")?; let code = output.status.code().unwrap_or(-1); if code != 0 && !(self.keep_going && code == 3) { let stderr = String::from_utf8_lossy(&output.stderr); @@ -407,8 +435,8 @@ impl BazelOptions { .arg("--query_file") .arg(query_file.path()) .stdin(Stdio::null()); - let output = command - .output() + let output = self + .run(&mut command) .context("execute compatible-target cquery")?; if !output.status.success() { bail!( @@ -453,8 +481,8 @@ impl BazelOptions { if self.verbose { eprintln!("[Info] Command: {command:?}"); } - let output = command - .output() + let output = self + .run(&mut command) .context("execute bazel mod show_repo streamed_proto")?; if !output.status.success() { bail!( @@ -475,10 +503,10 @@ impl BazelOptions { eprintln!( "[BD-DBG][fingerprint-command] path=text_all_visible cmd={all_visible_command:?}" ); - let output = all_visible_command + all_visible_command .stdin(Stdio::null()) - .stderr(Stdio::piped()) - .output(); + .stderr(Stdio::piped()); + let output = self.run(&mut all_visible_command); if let Ok(output) = output { if output.status.success() { return Ok(String::from_utf8_lossy(&output.stdout).into_owned()); @@ -494,7 +522,9 @@ impl BazelOptions { } eprintln!("[BD-DBG][fingerprint-command] path=text_selected_repos cmd={command:?}"); command.stdin(Stdio::null()).stderr(Stdio::piped()); - let output = command.output().context("execute bazel mod show_repo")?; + let output = self + .run(&mut command) + .context("execute bazel mod show_repo")?; if !output.stderr.is_empty() && self.verbose { eprint!("{}", String::from_utf8_lossy(&output.stderr)); } @@ -1786,6 +1816,7 @@ mod tests { exclude_targets_query: None, no_bazelrc: false, verbose: false, + cancellation: None, }; options.exclude_targets_query = Some("attr(tags, manual, //...)".into()); assert_eq!( @@ -1812,6 +1843,7 @@ mod tests { exclude_targets_query: Some("attr(tags, manual, //...)".into()), no_bazelrc: false, verbose: false, + cancellation: None, }; let mut queries = Vec::new(); let targets = options @@ -1854,6 +1886,7 @@ mod tests { exclude_targets_query: None, no_bazelrc: false, verbose: false, + cancellation: None, }; let mut queries = Vec::new(); let targets = options @@ -2036,6 +2069,7 @@ exit 1 exclude_targets_query: None, no_bazelrc: false, verbose: false, + cancellation: None, }; let fingerprint = options.dependency_fingerprint().unwrap(); @@ -2106,6 +2140,7 @@ exit 1 exclude_targets_query: None, no_bazelrc: false, verbose: false, + cancellation: None, }; let show_repo_err = options.query_all_targets().unwrap_err(); diff --git a/src/cancel.rs b/src/cancel.rs new file mode 100644 index 00000000..b5b2ce0a --- /dev/null +++ b/src/cancel.rs @@ -0,0 +1,307 @@ +//! Cooperative cancellation for work that shells out to `git` and `bazel`. +//! +//! A [`Cancellation`] token is shared between the thread doing the work and the thread that may +//! abandon it. Every subprocess the work spawns goes through [`output`] or [`output_configured`], +//! which register the child so [`Cancellation::cancel`] can kill it, and which refuse to start a +//! child once the token is cancelled. Phases with no subprocess to kill (CPU-bound hashing, the +//! cache write) call [`Cancellation::check`] at their boundaries instead. +//! +//! Without a token (`None`), both helpers behave exactly like [`Command::output`], so the CLI +//! commands that never cancel anything are unaffected. + +use std::collections::HashMap; +use std::fmt; +use std::io::{self, Read}; +use std::process::{Child, Command, ExitStatus, Output, Stdio}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +/// The message carried by every error a cancelled token produces. +pub const CANCELLED: &str = "query cancelled"; + +/// Longest pause between exit polls of a running child. Polling starts at a millisecond and backs +/// off, so a quick `git rev-parse` is noticed almost at once and a long `bazel query` costs at most +/// this much extra latency. +const MAX_POLL_INTERVAL: Duration = Duration::from_millis(50); + +#[derive(Default)] +pub struct Cancellation { + cancelled: AtomicBool, + next_id: AtomicU64, + children: Mutex>>>, +} + +impl fmt::Debug for Cancellation { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("Cancellation") + .field("cancelled", &self.is_cancelled()) + .field("children", &self.running_children()) + .finish() + } +} + +impl Cancellation { + pub fn new() -> Self { + Self::default() + } + + pub fn is_cancelled(&self) -> bool { + self.cancelled.load(Ordering::SeqCst) + } + + /// Fails once the token is cancelled, so a phase boundary can stop work that has no + /// subprocess to kill. + pub fn check(&self) -> io::Result<()> { + if self.is_cancelled() { + return Err(cancelled_error()); + } + Ok(()) + } + + /// Number of subprocesses started through this token that have not exited yet. + pub fn running_children(&self) -> usize { + lock(&self.children).len() + } + + /// Marks the token cancelled and kills every registered subprocess. Returns how many kills + /// were delivered. Later calls to [`output`] with this token fail without spawning anything. + pub fn cancel(&self) -> usize { + self.cancelled.store(true, Ordering::SeqCst); + let children = lock(&self.children).values().cloned().collect::>(); + children + .iter() + .filter(|child| lock(child).kill().is_ok()) + .count() + } + + /// Runs `command` like [`Command::output`] (stdin closed, stdout and stderr captured), but + /// kills it on [`cancel`](Self::cancel) and fails with [`CANCELLED`] instead of returning + /// output the caller no longer wants. + pub fn output(&self, command: &mut Command) -> io::Result { + command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + self.output_configured(command) + } + + /// Like [`output`](Self::output), but keeps the stdio the caller configured, for commands + /// whose stdout is redirected to a file. Only piped streams are captured. + pub fn output_configured(&self, command: &mut Command) -> io::Result { + self.check()?; + let mut child = command.spawn()?; + // Both pipes are drained on their own threads so the child never blocks on a full pipe + // while this thread waits for it to exit. + let stdout = child.stdout.take().map(spawn_reader); + let stderr = child.stderr.take().map(spawn_reader); + let registered = self.register(child); + // A cancel that landed between `check` and `register` saw an empty registry. + if self.is_cancelled() { + let _ = lock(®istered.child).kill(); + } + let status = wait_for_exit(®istered.child)?; + drop(registered); + // After a cancel the readers are deliberately not joined: a grandchild the kill did not + // reach may still hold the pipes open, and nobody wants the output anyway. + self.check()?; + Ok(Output { + status, + stdout: join_reader(stdout)?, + stderr: join_reader(stderr)?, + }) + } + + fn register(&self, child: Child) -> Registered<'_> { + let id = self.next_id.fetch_add(1, Ordering::SeqCst); + let child = Arc::new(Mutex::new(child)); + lock(&self.children).insert(id, Arc::clone(&child)); + Registered { + token: self, + id, + child, + } + } +} + +/// Runs `command` through `token` when there is one, or exactly like [`Command::output`] when +/// there is not. +pub fn output(token: Option<&Cancellation>, command: &mut Command) -> io::Result { + match token { + Some(token) => token.output(command), + None => command.output(), + } +} + +/// [`output`] for a command whose stdio the caller configured; see +/// [`Cancellation::output_configured`]. +pub fn output_configured( + token: Option<&Cancellation>, + command: &mut Command, +) -> io::Result { + match token { + Some(token) => token.output_configured(command), + None => command.output(), + } +} + +/// A child that is registered with its token until it has been waited on, including when the +/// caller unwinds past the wait. +struct Registered<'a> { + token: &'a Cancellation, + id: u64, + child: Arc>, +} + +impl Drop for Registered<'_> { + fn drop(&mut self) { + lock(&self.token.children).remove(&self.id); + } +} + +fn spawn_reader(mut pipe: impl Read + Send + 'static) -> JoinHandle>> { + thread::spawn(move || { + let mut bytes = Vec::new(); + pipe.read_to_end(&mut bytes)?; + Ok(bytes) + }) +} + +fn join_reader(reader: Option>>>) -> io::Result> { + match reader { + Some(reader) => reader + .join() + .map_err(|_| io::Error::other("output reader panicked"))?, + None => Ok(Vec::new()), + } +} + +/// Waits for the child without holding its lock across the wait, so `cancel` can take the lock +/// to kill it. +fn wait_for_exit(child: &Mutex) -> io::Result { + let mut interval = Duration::from_millis(1); + loop { + if let Some(status) = lock(child).try_wait()? { + return Ok(status); + } + thread::sleep(interval); + interval = (interval * 2).min(MAX_POLL_INTERVAL); + } +} + +fn cancelled_error() -> io::Error { + io::Error::new(io::ErrorKind::Interrupted, CANCELLED) +} + +/// A poisoned lock here only means another thread panicked mid-bookkeeping; the registry and the +/// child handles are still usable. +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Instant; + + #[cfg(unix)] + fn shell(script: &str) -> Command { + let mut command = Command::new("sh"); + command.arg("-c").arg(script); + command + } + + #[test] + fn fresh_token_is_not_cancelled() { + let token = Cancellation::new(); + assert!(!token.is_cancelled()); + token.check().unwrap(); + assert_eq!(token.running_children(), 0); + assert_eq!(token.cancel(), 0); + assert!(token.is_cancelled()); + let error = token.check().unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::Interrupted); + assert_eq!(error.to_string(), CANCELLED); + assert!(format!("{token:?}").contains("cancelled: true")); + } + + #[cfg(unix)] + #[test] + fn output_captures_streams_and_status_like_command_output() { + let token = Cancellation::new(); + let captured = token + .output(&mut shell("echo out; echo err >&2; exit 3")) + .unwrap(); + assert_eq!(captured.status.code(), Some(3)); + assert_eq!(captured.stdout, b"out\n"); + assert_eq!(captured.stderr, b"err\n"); + assert_eq!(token.running_children(), 0); + + // Without a token the helpers defer to Command::output. + let plain = output_configured(None, &mut shell("echo plain")).unwrap(); + assert_eq!(plain.stdout, b"plain\n"); + let plain = output(None, &mut shell("exit 2")).unwrap(); + assert_eq!(plain.status.code(), Some(2)); + } + + #[cfg(unix)] + #[test] + fn output_configured_keeps_redirected_stdout() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("stdout.txt"); + let token = Cancellation::new(); + let mut command = shell("echo to-file; echo err >&2"); + command + .stdout(std::fs::File::create(&path).unwrap()) + .stderr(Stdio::piped()); + let result = output_configured(Some(&token), &mut command).unwrap(); + assert!(result.status.success()); + assert!(result.stdout.is_empty()); + assert_eq!(result.stderr, b"err\n"); + assert_eq!(std::fs::read_to_string(&path).unwrap(), "to-file\n"); + } + + #[cfg(unix)] + #[test] + fn cancel_kills_running_child_and_refuses_new_ones() { + let token = Arc::new(Cancellation::new()); + let worker_token = Arc::clone(&token); + let worker = thread::spawn(move || worker_token.output(&mut shell("exec sleep 30"))); + let started = Instant::now(); + while token.running_children() == 0 { + assert!( + started.elapsed() < Duration::from_secs(10), + "child never started" + ); + thread::sleep(Duration::from_millis(5)); + } + + assert_eq!(token.cancel(), 1); + let error = worker.join().unwrap().unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::Interrupted); + assert!( + started.elapsed() < Duration::from_secs(10), + "the sleeping child was not killed" + ); + assert_eq!(token.running_children(), 0); + + let refused = output(Some(&token), &mut shell("echo never")).unwrap_err(); + assert_eq!(refused.to_string(), CANCELLED); + // Cancelling again is harmless and kills nothing. + assert_eq!(token.cancel(), 0); + } + + #[test] + fn spawn_failure_surfaces_and_leaves_no_registration() { + let token = Cancellation::new(); + let error = token + .output(&mut Command::new("/definitely/missing/binary")) + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::NotFound); + assert_eq!(token.running_children(), 0); + } +} diff --git a/src/hash.rs b/src/hash.rs index 76440f1e..b61c0ff6 100644 --- a/src/hash.rs +++ b/src/hash.rs @@ -1,4 +1,5 @@ use crate::bazel::{encode_configured_input, split_configured, target_label, BazelOptions}; +use crate::cancel::{self, Cancellation}; use crate::model::{HashFileData, TargetHash}; use crate::proto::blaze_query::{target, Rule, Target}; use anyhow::{anyhow, bail, Context, Result}; @@ -153,6 +154,9 @@ pub fn generate_hashes_timed(options: &HashOptions) -> Result<(HashFileData, Gen ); })?; let bazel_query_millis = elapsed_millis(query_started); + // The query is the phase worth cancelling; a request abandoned during it stops here instead + // of hashing targets nobody will read. + options.bazel.check_cancelled()?; let (data, mut timings) = hash_targets_with_digests(options, targets, definition_digests)?; timings.bazel_query_millis = bazel_query_millis; Ok((data, timings)) @@ -171,6 +175,7 @@ fn hash_targets_with_digests( let external_resolver = ExternalRepoResolver::new(&options.bazel)?; let module_graph_started = Instant::now(); let module_graph_json = options.bazel.module_graph_json(); + options.bazel.check_cancelled()?; let module_graph_millis = elapsed_millis(module_graph_started); let data = hash_targets_with_environment( options, @@ -1131,6 +1136,8 @@ struct ExternalRepoResolver { output_base: PathBuf, startup_options: Vec, no_bazelrc: bool, + /// The owning query's cancellation token; see [`BazelOptions::cancellation`]. + cancellation: Option>, /// Memoizes resolution per apparent repository name. `resolve` is called once /// per external source file and can shell out to Bazel, so without this a /// repository with N source files costs N Bazel invocations, all of which @@ -1143,11 +1150,8 @@ struct ExternalRepoResolver { impl ExternalRepoResolver { fn new(options: &BazelOptions) -> Result { let mut command = options.command_for_hashing(); - let output = command - .arg("info") - .arg("output_base") - .stdin(Stdio::null()) - .output() + command.arg("info").arg("output_base").stdin(Stdio::null()); + let output = cancel::output(options.cancellation.as_deref(), &mut command) .context("failed to determine Bazel output_base")?; if !output.status.success() { bail!( @@ -1161,6 +1165,7 @@ impl ExternalRepoResolver { output_base: PathBuf::from(String::from_utf8_lossy(&output.stdout).trim()), startup_options: options.startup_options.clone(), no_bazelrc: options.no_bazelrc, + cancellation: options.cancellation.clone(), ..Default::default() }) } @@ -1214,15 +1219,15 @@ impl ExternalRepoResolver { return Ok(candidate); } } - let output = self - .command() + let mut command = self.command(); + command .arg("query") .arg(format!("@{repo}//...")) .arg("--keep_going") .arg("--output") .arg("location") - .stdin(Stdio::null()) - .output()?; + .stdin(Stdio::null()); + let output = cancel::output(self.cancellation.as_deref(), &mut command)?; let line = String::from_utf8_lossy(&output.stdout) .lines() .next() @@ -1245,14 +1250,13 @@ impl ExternalRepoResolver { /// (for example a WORKSPACE-only build), leaving the `bazel query` fallback. fn repo_mapping(&self) -> &HashMap { self.repo_mapping.get_or_init(|| { - let output = self - .command() + let mut command = self.command(); + command .arg("mod") .arg("dump_repo_mapping") .arg("") - .stdin(Stdio::null()) - .output(); - match output { + .stdin(Stdio::null()); + match cancel::output(self.cancellation.as_deref(), &mut command) { Ok(output) if output.status.success() => { parse_apparent_repo_mapping(&String::from_utf8_lossy(&output.stdout)) } @@ -1329,6 +1333,7 @@ mod tests { exclude_targets_query: None, no_bazelrc: true, verbose: false, + cancellation: None, } } diff --git a/src/lib.rs b/src/lib.rs index 3d21df45..dea064f4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ //! tests; its module-level API is not a stability commitment. pub mod bazel; +pub mod cancel; pub mod explain; pub mod fingerprint; pub mod hash; diff --git a/src/main.rs b/src/main.rs index 0c80a207..446fa7ee 100644 --- a/src/main.rs +++ b/src/main.rs @@ -515,6 +515,7 @@ fn bazel_options(args: &HashingArgs, verbose: bool) -> Result { exclude_targets_query: args.exclude_targets_query.clone(), no_bazelrc: false, verbose, + cancellation: None, }) } @@ -679,6 +680,7 @@ fn get_bazel_options(args: &GetImpactedTargetsArgs, verbose: bool) -> BazelOptio exclude_targets_query: None, no_bazelrc: args.no_bazelrc, verbose, + cancellation: None, } } diff --git a/src/server.rs b/src/server.rs index b7be01a3..9dd88e28 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,3 +1,5 @@ +use crate::bazel::BazelOptions; +use crate::cancel::{self, Cancellation, CANCELLED}; use crate::hash::{generate_hashes_timed, HashOptions}; use crate::model::{ filter_and_sort_labels, impacted_targets, impacted_targets_with_distances, HashFileData, @@ -17,8 +19,8 @@ use std::fmt; use std::fs; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{mpsc, Arc, Mutex}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{mpsc, Arc, Condvar, Mutex}; use std::time::{Duration, Instant, SystemTime}; use tiny_http::{Header, Method, Request, Response, Server, StatusCode}; use url::{form_urlencoded, Url}; @@ -52,15 +54,130 @@ pub struct ServerConfig { pub nexus_ca_cert: Option, } +/// How long a timed-out request waits for its cancelled query to stop before answering anyway. +/// Killing the subprocesses is immediate; this bounds the CPU-only phases (target hashing) a +/// cancel can only interrupt at their boundaries. +const CANCEL_GRACE: Duration = Duration::from_secs(60); + +/// How often a query waiting for the workspace re-checks whether it has been cancelled. +const WORKSPACE_WAIT_POLL: Duration = Duration::from_millis(50); + struct State { config: ServerConfig, ready: AtomicBool, - workspace_lock: Mutex<()>, + workspace: Workspace, + /// Queries accepted and not yet finished: waiting for the workspace, running, or abandoned by + /// a timed-out client but still running. `/metrics` reports it as `activeQueries`. + active_queries: AtomicUsize, started: Instant, fingerprint: String, remote: Option, } +impl State { + fn new(config: ServerConfig, remote: Option) -> Self { + let fingerprint = configuration_fingerprint(&config); + Self { + config, + ready: AtomicBool::new(false), + workspace: Workspace::default(), + active_queries: AtomicUsize::new(0), + started: Instant::now(), + fingerprint, + remote, + } + } + + /// The server's Bazel options bound to the query currently holding the workspace, so the + /// Bazel subprocesses it spawns die with that query. + fn bazel_options(&self) -> BazelOptions { + let mut options = self.config.hash_options.bazel.clone(); + options.cancellation = self.workspace.holder(); + options + } +} + +/// Serialises everything that mutates `--workspacePath` (git checkout, `bazel query`) and +/// remembers which operation holds it. That lets `/metrics` report `workspaceBusy`, lets the +/// subprocesses the holder spawns be killed when its query is cancelled, and lets a query that is +/// cancelled while queued give up instead of taking the workspace for nothing. +#[derive(Default)] +struct Workspace { + holder: Mutex>>, + released: Condvar, +} + +impl Workspace { + fn acquire(&self, token: &Arc) -> Result> { + let mut holder = self.holder.lock().map_err(|_| poisoned())?; + while holder.is_some() { + if token.is_cancelled() { + bail!(CANCELLED); + } + holder = self + .released + .wait_timeout(holder, WORKSPACE_WAIT_POLL) + .map_err(|_| poisoned())? + .0; + } + *holder = Some(Arc::clone(token)); + Ok(WorkspaceGuard { workspace: self }) + } + + fn holder(&self) -> Option> { + self.holder.lock().ok().and_then(|holder| holder.clone()) + } + + fn is_busy(&self) -> bool { + self.holder + .lock() + .map(|holder| holder.is_some()) + .unwrap_or(true) + } + + /// Fails once the operation holding the workspace has been cancelled: the caller is about to + /// do something (write a cache entry) that only a finished query should do. + fn check_cancelled(&self) -> Result<()> { + if let Some(token) = self.holder() { + token.check()?; + } + Ok(()) + } +} + +fn poisoned() -> anyhow::Error { + anyhow!("workspace lock poisoned") +} + +struct WorkspaceGuard<'a> { + workspace: &'a Workspace, +} + +impl Drop for WorkspaceGuard<'_> { + fn drop(&mut self) { + if let Ok(mut holder) = self.workspace.holder.lock() { + *holder = None; + } + self.workspace.released.notify_all(); + } +} + +/// Keeps `activeQueries` accurate even when the worker panics. +struct ActiveQuery(Arc); + +impl ActiveQuery { + fn start(state: &Arc) -> Self { + state.active_queries.fetch_add(1, Ordering::SeqCst); + Self(Arc::clone(state)) + } +} + +impl Drop for ActiveQuery { + fn drop(&mut self) { + self.0.active_queries.fetch_sub(1, Ordering::SeqCst); + } +} + /// The shared cache tier behind local disk. Every backend degrades the same way: a failed read is /// a miss (the revision is regenerated) and a failed write leaves the entry local-only, so an /// outage of the shared store costs throughput, never availability. @@ -451,16 +568,8 @@ pub fn parse_byte_size(value: &str) -> Result { pub fn serve(config: ServerConfig) -> Result<()> { fs::create_dir_all(&config.cache_dir)?; - let fingerprint = configuration_fingerprint(&config); let remote = RemoteCache::new(&config)?; - let state = Arc::new(State { - config, - ready: AtomicBool::new(false), - workspace_lock: Mutex::new(()), - started: Instant::now(), - fingerprint, - remote, - }); + let state = Arc::new(State::new(config, remote)); prune_cache(&state)?; start_cache_pruner(&state)?; perform_initial_fetch(&state); @@ -591,10 +700,17 @@ fn route(mut request: Request, state: &Arc) -> Result<()> { ); } let timeout = state.config.request_timeout; + let token = Arc::new(Cancellation::new()); + let active = ActiveQuery::start(state); let state_for_compute = Arc::clone(state); + let worker_token = Arc::clone(&token); let (sender, receiver) = mpsc::sync_channel(1); std::thread::spawn(move || { - let _ = sender.send(compute_query(&state_for_compute, inputs, kind)); + let result = compute_query(&state_for_compute, inputs, kind, &worker_token); + // Decrement before answering so a client that polls /metrics right after its + // response never sees its own finished query still counted. + drop(active); + let _ = sender.send(result); }); let computed = if timeout.is_zero() { receiver @@ -604,11 +720,16 @@ fn route(mut request: Request, state: &Arc) -> Result<()> { match receiver.recv_timeout(timeout) { Ok(result) => result, Err(mpsc::RecvTimeoutError::Timeout) => { + let stopped = cancel_query(&token, &receiver, timeout); return respond_json( request, 504, - &json!({"error": format!("request timed out after {}s", timeout.as_secs())}), - ) + &json!({ + "error": format!("request timed out after {}s", timeout.as_secs()), + "cancelled": stopped, + "workspaceBusy": state.workspace.is_busy(), + }), + ); } Err(_) => Err(anyhow!("query worker stopped")), } @@ -622,6 +743,39 @@ fn route(mut request: Request, state: &Arc) -> Result<()> { } } +/// Cancels the query a timed-out request abandoned and waits for its worker to stop, so the 504 +/// is sent only once the query has released the workspace and will write no cache entry. Returns +/// whether the worker stopped within [`CANCEL_GRACE`]; when it did not, `/metrics` reports the +/// moment it does through `activeQueries` and `workspaceBusy`. +fn cancel_query( + token: &Cancellation, + receiver: &mpsc::Receiver>, + timeout: Duration, +) -> bool { + let killed = token.cancel(); + eprintln!( + "[Warn] request timed out after {}s; cancelled its query and killed {killed} subprocess(es)", + timeout.as_secs() + ); + let cancelled = Instant::now(); + match receiver.recv_timeout(CANCEL_GRACE) { + Err(mpsc::RecvTimeoutError::Timeout) => { + eprintln!( + "[Warn] cancelled query is still running {}s later; /metrics reports when it stops", + CANCEL_GRACE.as_secs() + ); + false + } + Ok(_) | Err(mpsc::RecvTimeoutError::Disconnected) => { + eprintln!( + "[Info] cancelled query stopped and released the workspace after {}ms", + elapsed_millis(cancelled) + ); + true + } + } +} + fn parse_query_inputs(request: &mut Request, query: &str) -> Result { if request.method() == &Method::Post { let mut raw = String::new(); @@ -691,12 +845,14 @@ fn normalized_types(types: Option>) -> Option> { .filter(|types| !types.is_empty()) } -fn compute_query(state: &Arc, inputs: QueryInputs, kind: QueryKind) -> Result { +fn compute_query( + state: &Arc, + inputs: QueryInputs, + kind: QueryKind, + token: &Arc, +) -> Result { let started = Instant::now(); - let _guard = state - .workspace_lock - .lock() - .map_err(|_| anyhow!("workspace lock poisoned"))?; + let _guard = state.workspace.acquire(token)?; let resolve_started = Instant::now(); let (from_sha, to_sha) = resolve_both(state, &inputs.from, &inputs.to)?; let resolve_millis = resolve_started.elapsed().as_millis() as u64; @@ -718,9 +874,9 @@ fn compute_query(state: &Arc, inputs: QueryInputs, kind: QueryKind) -> Re let to = to.data; let diff_started = Instant::now(); checkout(state, &to_sha)?; - let exclude_external = state.config.hash_options.bazel.is_bzlmod_enabled(); - let impacted = - impacted_with_module_changes(&from, &to, Some(&state.config.hash_options.bazel))?; + let bazel = state.bazel_options(); + let exclude_external = bazel.is_bzlmod_enabled(); + let impacted = impacted_with_module_changes(&from, &to, Some(&bazel))?; let impacted_value = if matches!(kind, QueryKind::Distances) { let hash_impacted = impacted_targets(&from.hashes, &to.hashes, None, false)? .into_iter() @@ -780,10 +936,7 @@ fn compute_query(state: &Arc, inputs: QueryInputs, kind: QueryKind) -> Re } fn warm_revision(state: &Arc, revision: &str) -> Result<()> { - let _guard = state - .workspace_lock - .lock() - .map_err(|_| anyhow!("workspace lock poisoned"))?; + let _guard = state.workspace.acquire(&Arc::new(Cancellation::new()))?; let sha = resolve_sha(state, revision)?; get_hashes_locked(state, &sha, &BTreeSet::new())?; Ok(()) @@ -902,6 +1055,7 @@ fn get_hashes_locked( checkout(state, sha)?; let checkout_millis = elapsed_millis(checkout_started); let mut options = state.config.hash_options.clone(); + options.bazel.cancellation = state.workspace.holder(); options.modified_filepaths = modified.clone(); options.track_deps = state.config.track_deps; if modified.is_empty() && options.content_hashes.is_none() { @@ -919,6 +1073,9 @@ fn get_hashes_locked( options.content_hashes = Some(content_hashes); } let (mut data, timings) = generate_hashes_timed(&options)?; + // A generation the client stopped waiting for is not cached: the subprocess kill above may + // have left the working tree mid-checkout, and the caller may already be reusing it. + state.workspace.check_cancelled()?; if state.config.dependency_fingerprint { data.dependency_fingerprint = current_dependency_fingerprint.take().or_else(|| { match dependency_fingerprint_for_workspace(state) { @@ -1053,9 +1210,7 @@ fn short_fingerprint(value: &str) -> &str { fn dependency_fingerprint_for_workspace(state: &State) -> Result { state - .config - .hash_options - .bazel + .bazel_options() .dependency_fingerprint() .map_err(|error| { anyhow!("failed to compute dependency fingerprint for cache guard: {error:#}") @@ -1302,12 +1457,15 @@ fn git(state: &State, args: &[String]) -> Result<()> { Ok(()) } +/// Runs git in the workspace. While a query holds the workspace the subprocess belongs to it and +/// is killed if that query is cancelled; outside the lock (startup fetch) it runs plainly. fn git_output(state: &State, args: &[String]) -> Result { - Command::new(&state.config.git_path) + let mut command = Command::new(&state.config.git_path); + command .current_dir(&state.config.hash_options.bazel.workspace) .args(args) - .stdin(Stdio::null()) - .output() + .stdin(Stdio::null()); + cancel::output(state.workspace.holder().as_deref(), &mut command) .with_context(|| format!("failed to execute {}", state.config.git_path.display())) } @@ -1319,6 +1477,8 @@ fn metrics(state: &State) -> Value { "version": env!("CARGO_PKG_VERSION"), "uptimeSeconds": state.started.elapsed().as_secs(), "ready": state.ready.load(Ordering::Acquire), + "activeQueries": state.active_queries.load(Ordering::SeqCst), + "workspaceBusy": state.workspace.is_busy(), "gitEngine": "subprocess", "trackDeps": state.config.track_deps, "dependencyFingerprint": state.config.dependency_fingerprint, @@ -1475,6 +1635,7 @@ mod tests { exclude_targets_query: None, no_bazelrc: false, verbose: false, + cancellation: None, }, content_hashes: None, ignored_attributes: HashSet::new(), @@ -1565,15 +1726,13 @@ mod tests { config.track_deps = track_deps; config.hash_options.track_deps = track_deps; config.dependency_fingerprint = true; - let fingerprint = configuration_fingerprint(&config); - Arc::new(State { - config, - ready: AtomicBool::new(true), - workspace_lock: Mutex::new(()), - started: Instant::now(), - fingerprint, - remote: None, - }) + ready_state(config, None) + } + + fn ready_state(config: ServerConfig, remote: Option) -> Arc { + let state = State::new(config, remote); + state.ready.store(true, Ordering::Release); + Arc::new(state) } /// The full-content cache key `get_hashes_locked` uses for a commit: its tree, not its SHA. @@ -2149,15 +2308,7 @@ nQBeQ4AIl47nDOrFeVnvse8R9IHJ config.hash_options.bazel.workspace = repo.path().to_path_buf(); config.git_path = PathBuf::from("git"); config.cache_dir = cache.path().to_path_buf(); - let fingerprint = configuration_fingerprint(&config); - Arc::new(State { - config, - ready: AtomicBool::new(false), - workspace_lock: Mutex::new(()), - started: Instant::now(), - fingerprint, - remote: None, - }) + Arc::new(State::new(config, None)) }; let lame_duck = unready_state(false); @@ -2218,14 +2369,7 @@ nQBeQ4AIl47nDOrFeVnvse8R9IHJ let mut bad_config = state.config.clone(); bad_config.git_path = PathBuf::from("/definitely/missing/git"); - let bad = State { - config: bad_config, - ready: AtomicBool::new(true), - workspace_lock: Mutex::new(()), - started: Instant::now(), - fingerprint: state.fingerprint.clone(), - remote: None, - }; + let bad = State::new(bad_config, None); assert!(git_output(&bad, &[]).is_err()); } @@ -2280,14 +2424,7 @@ nQBeQ4AIl47nDOrFeVnvse8R9IHJ let mut bad_config = state.config.clone(); bad_config.git_path = PathBuf::from("/missing/git"); - let bad_state = State { - config: bad_config, - ready: AtomicBool::new(true), - workspace_lock: Mutex::new(()), - started: Instant::now(), - fingerprint: state.fingerprint.clone(), - remote: None, - }; + let bad_state = State::new(bad_config, None); assert!(!fetch_revision(&bad_state, "HEAD")); } @@ -2348,15 +2485,7 @@ nQBeQ4AIl47nDOrFeVnvse8R9IHJ }; let bytes = serde_json::to_vec(&data.serialized(true, false)).unwrap(); let (remote, requests, handle) = remote_cache_with_responses("", vec![(200, bytes)]); - let fingerprint = configuration_fingerprint(&config); - let state = Arc::new(State { - config, - ready: AtomicBool::new(true), - workspace_lock: Mutex::new(()), - started: Instant::now(), - fingerprint, - remote: Some(remote), - }); + let state = ready_state(config, Some(remote)); let first = get_hashes_locked(&state, &sha, &BTreeSet::new()).unwrap(); assert!(first.cache_hit); @@ -2418,7 +2547,13 @@ nQBeQ4AIl47nDOrFeVnvse8R9IHJ modified_filepaths: BTreeSet::new(), profile: true, }; - let result = compute_query(&state, inputs, QueryKind::Distances).unwrap(); + let result = compute_query( + &state, + inputs, + QueryKind::Distances, + &Arc::new(Cancellation::new()), + ) + .unwrap(); assert_eq!(result["to"], second); assert_eq!(result["impactedTargets"].as_array().unwrap().len(), 2); let retrievals = result["profile"]["hashRetrievals"].as_array().unwrap(); @@ -2530,15 +2665,7 @@ nQBeQ4AIl47nDOrFeVnvse8R9IHJ }; let bytes = serde_json::to_vec(&data.serialized(true, false)).unwrap(); let (remote, requests, handle) = remote_cache_with_responses("", vec![(200, bytes)]); - let fingerprint = configuration_fingerprint(&config); - let state = Arc::new(State { - config, - ready: AtomicBool::new(true), - workspace_lock: Mutex::new(()), - started: Instant::now(), - fingerprint, - remote: Some(remote), - }); + let state = ready_state(config, Some(remote)); let retrieval = get_hashes_locked(&state, &sha, &BTreeSet::new()).unwrap(); assert!(retrieval.cache_hit); assert_eq!(retrieval.data.hashes["//app:lib"].hash, "overall"); @@ -2763,13 +2890,147 @@ nQBeQ4AIl47nDOrFeVnvse8R9IHJ let cache = tempfile::tempdir().unwrap(); let mut state = state_for_repo(repo.path(), cache.path(), false); Arc::get_mut(&mut state).unwrap().config.request_timeout = Duration::from_millis(20); - let _guard = state.workspace_lock.lock().unwrap(); + // Another operation holds the workspace for the whole test, so the query never gets it. + let _guard = state + .workspace + .acquire(&Arc::new(Cancellation::new())) + .unwrap(); let response = request( &state, "GET /impacted_targets?from=a&to=b HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", ); assert!(response.starts_with("HTTP/1.1 504"), "{response}"); assert!(response.contains("request timed out")); + // The abandoned query stopped waiting (so it is no longer counted) and the workspace is + // reported busy because of the other holder, not because of it. + assert!(response.contains("\"cancelled\":true"), "{response}"); + assert!(response.contains("\"workspaceBusy\":true"), "{response}"); + assert_eq!(state.active_queries.load(Ordering::SeqCst), 0); + } + + #[test] + fn metrics_report_active_queries_and_workspace_busy() { + let repo = tempfile::tempdir().unwrap(); + let cache = tempfile::tempdir().unwrap(); + let state = state_for_repo(repo.path(), cache.path(), false); + let idle = metrics(&state); + assert_eq!(idle["activeQueries"], 0); + assert_eq!(idle["workspaceBusy"], false); + assert!(state.workspace.holder().is_none()); + assert!(state.bazel_options().cancellation.is_none()); + state.workspace.check_cancelled().unwrap(); + + let token = Arc::new(Cancellation::new()); + let guard = state.workspace.acquire(&token).unwrap(); + assert_eq!(metrics(&state)["workspaceBusy"], true); + assert!(Arc::ptr_eq(&state.workspace.holder().unwrap(), &token)); + // Bazel options handed out while the query holds the workspace carry its token, so the + // subprocesses they spawn are killed with it. + let bazel = state.bazel_options(); + assert!(Arc::ptr_eq(bazel.cancellation.as_ref().unwrap(), &token)); + state.workspace.check_cancelled().unwrap(); + token.cancel(); + assert_eq!( + state.workspace.check_cancelled().unwrap_err().to_string(), + CANCELLED + ); + assert_eq!(bazel.check_cancelled().unwrap_err().to_string(), CANCELLED); + drop(guard); + assert_eq!(metrics(&state)["workspaceBusy"], false); + assert!(state.workspace.holder().is_none()); + + let counted = ActiveQuery::start(&state); + assert_eq!(metrics(&state)["activeQueries"], 1); + drop(counted); + assert_eq!(metrics(&state)["activeQueries"], 0); + } + + #[test] + fn queued_query_gives_up_once_cancelled() { + let repo = tempfile::tempdir().unwrap(); + let cache = tempfile::tempdir().unwrap(); + let state = state_for_repo(repo.path(), cache.path(), false); + let _held = state + .workspace + .acquire(&Arc::new(Cancellation::new())) + .unwrap(); + + // Cancelled before it ever waits: fails on the first check. + let cancelled = Arc::new(Cancellation::new()); + cancelled.cancel(); + assert_eq!( + state + .workspace + .acquire(&cancelled) + .map(|_| ()) + .unwrap_err() + .to_string(), + CANCELLED + ); + + // Cancelled while waiting: notices within one poll interval instead of blocking until + // the holder is done. + let waiting = Arc::new(Cancellation::new()); + let waiter_state = Arc::clone(&state); + let waiter_token = Arc::clone(&waiting); + let waiter = thread::spawn(move || { + waiter_state + .workspace + .acquire(&waiter_token) + .map(|_| ()) + .unwrap_err() + .to_string() + }); + thread::sleep(WORKSPACE_WAIT_POLL); + waiting.cancel(); + assert_eq!(waiter.join().unwrap(), CANCELLED); + assert!(state.workspace.is_busy()); + } + + #[cfg(unix)] + #[test] + fn timed_out_query_kills_its_subprocess_and_releases_the_workspace() { + use std::os::unix::fs::PermissionsExt; + + let (repo, first, second) = initialize_git_repo(); + let cache = tempfile::tempdir().unwrap(); + let tools = tempfile::tempdir().unwrap(); + // A git that never returns on its own: the query's first subprocess hangs in it, so only + // the kill can end the request in time. + let slow_git = tools.path().join("slow-git"); + fs::write(&slow_git, "#!/bin/sh\nexec sleep 30\n").unwrap(); + fs::set_permissions(&slow_git, fs::Permissions::from_mode(0o755)).unwrap(); + let mut state = state_for_repo(repo.path(), cache.path(), false); + { + let config = &mut Arc::get_mut(&mut state).unwrap().config; + config.request_timeout = Duration::from_millis(100); + config.git_path = slow_git; + } + + let started = Instant::now(); + let response = request( + &state, + &format!( + "GET /impacted_targets?from={first}&to={second} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" + ), + ); + assert!(response.starts_with("HTTP/1.1 504"), "{response}"); + assert!( + response.contains("request timed out after 0s"), + "{response}" + ); + assert!(response.contains("\"cancelled\":true"), "{response}"); + assert!(response.contains("\"workspaceBusy\":false"), "{response}"); + assert!( + started.elapsed() < Duration::from_secs(10), + "the hanging git was not killed; request took {:?}", + started.elapsed() + ); + // The 504 was only sent once the query had released the workspace and stopped counting. + assert!(!state.workspace.is_busy()); + assert_eq!(state.active_queries.load(Ordering::SeqCst), 0); + // Nothing was cached for the generation the client walked away from. + assert!(cache_entries(cache.path()).is_empty()); } #[test] diff --git a/tools/readme_template.md b/tools/readme_template.md index 7144df6d..a67196a6 100644 --- a/tools/readme_template.md +++ b/tools/readme_template.md @@ -302,7 +302,11 @@ curl 'http://localhost:8080/dependency_edges?from=main&to=my-feature-branch' identity, liveness, and cache size usage without scraping logs. Unlike the query endpoints it is never gated on readiness, so it still responds on an un-ready or lame-ducked instance (the `ready` field reports the current state). The `cache` size fields are populated for the local-disk backend - and are `null` for a backend whose size is not cheaply knowable in-process. + and are `null` for a backend whose size is not cheaply knowable in-process. `activeQueries` counts + the queries accepted and not yet finished (queued, running, or abandoned by a client that stopped + waiting), and `workspaceBusy` is `true` while any operation holds `--workspacePath` (a query's + checkout and `bazel query`, or a startup warmup). A tool that shares the checkout with the server + should wait for `workspaceBusy` to be `false` before touching it. ```bash curl 'http://localhost:8080/metrics' @@ -313,6 +317,8 @@ curl 'http://localhost:8080/metrics' "version": "31.4.0", "uptimeSeconds": 3600, "ready": true, + "activeQueries": 1, + "workspaceBusy": true, "gitEngine": "subprocess", "trackDeps": false, "cache": {"directory": "/var/cache/bazel-diff", "remote": "s3://my-bucket/bazel-diff/", "entries": 128, "sizeBytes": 4823913, "sizeHuman": "4.6 MB"}, @@ -414,6 +420,16 @@ Notes and operational guidance: * The service checks out revisions inside `--workspacePath`, so point it at a dedicated clone, not a working tree you edit. All workspace-mutating work (git checkout + `bazel query`) is serialized, so a single instance answers one cold query at a time; the per-SHA cache absorbs the rest. +* `--requestTimeout ` (default `0`, no limit) bounds how long a query may run. A query that + exceeds it is cancelled, not merely abandoned: its `git` and `bazel` subprocesses are killed, the + hashes it had not finished generating are not written to the local or shared cache, and the `504` + is sent only after the query has released the workspace. The `504` body reports `cancelled` + (`true` once the query has stopped; `false` only if it was still winding down a minute later, in + which case `/metrics` `activeQueries` reports when it is done) and `workspaceBusy` (whether + another operation holds the workspace at that moment). A client that enforces its own deadline + instead cannot cancel the server-side work: it keeps running to completion, so poll `/metrics` for + `workspaceBusy` before reusing the checkout, or prefer `--requestTimeout`. A killed `git checkout` + can leave the clone mid-switch; the next query repairs it with `git checkout --force`. * Git operations (fetch and checkout) shell out to the `git` binary at `--gitPath` (default `git` on the `PATH`), so a `git` binary must be available on the host. The working tree is checked out on disk for `bazel query` to read. Because native git performs every fetch, all clone shapes are