From 315211317d5e1af76d40875aedf079f21e933c43 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 04:57:56 +0000 Subject: [PATCH] serve: cancel timed-out queries and expose workspace state in /metrics Fixes Tinder/bazel-diff#518. A query that outlived `--requestTimeout` kept running after the 504: it held the workspace, checked out revisions, ran `bazel query`, and then wrote hashes to the local and shared cache, with no way for the client to tell when the checkout was free again. A client that touched the checkout early could get a hash of a changed tree cached under its SHA and served fleet-wide. Both approaches the issue proposes are implemented: * `/metrics` reports `activeQueries` (queries accepted and not yet finished, including abandoned ones) and `workspaceBusy` (whether any operation holds `--workspacePath`), so a client can wait for the workspace before reusing it. * A timed-out query is cancelled rather than abandoned. A new `cancel::Cancellation` token runs every git and Bazel subprocess the query spawns and kills them on cancel, the generation it did not finish is never written to any cache tier, and the 504 is sent only after the query has released the workspace. The response body carries `cancelled` and `workspaceBusy`. `State.workspace_lock` becomes a `Workspace` that records its holder's token: `git_output` and the server's `BazelOptions` pick the token up from it, and a query cancelled while still queued gives up instead of taking the workspace for nothing. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01EhN3zsapeda1JRboWAeS8w --- README.md | 18 +- src/bazel.rs | 65 ++++-- src/cancel.rs | 307 +++++++++++++++++++++++++++ src/hash.rs | 33 +-- src/lib.rs | 1 + src/main.rs | 2 + src/server.rs | 435 +++++++++++++++++++++++++++++++-------- tools/readme_template.md | 18 +- 8 files changed, 761 insertions(+), 118 deletions(-) create mode 100644 src/cancel.rs 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