diff --git a/docs/notes/2.34.x.md b/docs/notes/2.34.x.md index c913a74115b..85a9b0f646c 100644 --- a/docs/notes/2.34.x.md +++ b/docs/notes/2.34.x.md @@ -33,6 +33,8 @@ Thank you to [Klaviyo](https://www.klaviyo.com/) and Depot (https://depot.dev/) The [`sandboxer` sidecar binary](https://www.pantsbuild.org/blog/2025/06/29/introducing-the-sandboxer) is now enabled by default. If this causes any issues you can disable it by setting `sandboxer = false` in the `[GLOBAL]` section of `pants.toml`, but please [report](https://github.com/pantsbuild/pants/issues/new/choose) the issues to us. +Concurrent Pants processes in the same repo (e.g., multiple `--no-pantsd` runs, or a pantsd alongside a `--no-pantsd` run) no longer preempt each other's sandboxer process, which previously caused failed runs and leaked LMDB reader slots. + Fixed a bug where `SingleSourceField` could not be hydrated for generated targets because files were validated before code generation ran. Fixed an issue where `pants --changed-since` would unnecessarily invalidate all targets in a `BUILD` file when only whitespace or comment lines were modified. diff --git a/src/rust/Cargo.lock b/src/rust/Cargo.lock index af112f73431..1ee98100509 100644 --- a/src/rust/Cargo.lock +++ b/src/rust/Cargo.lock @@ -4554,6 +4554,7 @@ dependencies = [ "bytes", "hashing", "lmdb-rkv", + "lmdb-rkv-sys", "log", "parking_lot 0.12.5", "task_executor", diff --git a/src/rust/Cargo.toml b/src/rust/Cargo.toml index 0f5df91690d..cfc57c300b1 100644 --- a/src/rust/Cargo.toml +++ b/src/rust/Cargo.toml @@ -168,6 +168,7 @@ internment = { version = "0.8", features = ["deepsize"] } itertools = "0.14" libc = "0.2.182" lmdb-rkv = { git = "https://github.com/pantsbuild/lmdb-rs.git", rev = "6ae7a552aa2c932c3ddf652a68cdde2fed547cbc" } +lmdb-rkv-sys = { git = "https://github.com/pantsbuild/lmdb-rs.git", rev = "6ae7a552aa2c932c3ddf652a68cdde2fed547cbc" } log = "0.4.29" maplit = "1.0.1" nails = "0.13" diff --git a/src/rust/fs/store/src/local_tests.rs b/src/rust/fs/store/src/local_tests.rs index 565d8037a1e..e1ae0ff81c2 100644 --- a/src/rust/fs/store/src/local_tests.rs +++ b/src/rust/fs/store/src/local_tests.rs @@ -799,11 +799,13 @@ async fn prime_store_with_file_bytes(store: &ByteStore, bytes: Bytes) -> Digest fn get_directory_size(path: &Path) -> usize { let mut len: usize = 0; for entry in WalkDir::new(path) { - len += entry - .expect("Error walking directory") - .metadata() - .expect("Error reading metadata") - .len() as usize; + let entry = entry.expect("Error walking directory"); + // LMDB lock files are a fixed size (proportional to the max readers), unrelated to the + // amount of data stored. + if entry.file_name() == "lock.mdb" { + continue; + } + len += entry.metadata().expect("Error reading metadata").len() as usize; } len } diff --git a/src/rust/process_execution/sandboxer/bin/server.rs b/src/rust/process_execution/sandboxer/bin/server.rs index fadd56143b3..d9d35a25e9f 100644 --- a/src/rust/process_execution/sandboxer/bin/server.rs +++ b/src/rust/process_execution/sandboxer/bin/server.rs @@ -14,6 +14,11 @@ struct Opt { #[arg(long)] socket_path: PathBuf, + /// The pid of the spawning process. The sandboxer exits if this process exits. + /// If unspecified, the sandboxer's parent process at startup is used. + #[arg(long)] + parent_pid: Option, + #[command(flatten)] store_options: StoreCliOpt, } @@ -41,7 +46,7 @@ async fn main() -> Result<(), Box> { std::env::var("RUST_LOG").unwrap_or("".into()), args ); - SandboxerService::new(args.socket_path, args.store_options) + SandboxerService::new(args.socket_path, args.store_options, args.parent_pid) .serve() .await } diff --git a/src/rust/process_execution/sandboxer/src/lib.rs b/src/rust/process_execution/sandboxer/src/lib.rs index d2875c670c1..9b6422369da 100644 --- a/src/rust/process_execution/sandboxer/src/lib.rs +++ b/src/rust/process_execution/sandboxer/src/lib.rs @@ -15,7 +15,7 @@ use protos::pb::pants::sandboxer::{ use std::process::Stdio; use std::{ env, - os::unix::{ffi::OsStrExt, net::SocketAddr}, + os::unix::{ffi::OsStrExt, fs::MetadataExt, net::SocketAddr, process::parent_id}, path::{Path, PathBuf}, }; @@ -69,7 +69,11 @@ mod test_util; // - The SandboxerProcess struct, which encapsulates a server process and a client connection to it. // - The Sandboxer struct, which manages a SandboxerProcess, respawning it as needed. // -// We use a Unix domain socket primarily because it's easy to pick a unique socket path per repo. +// We use a Unix domain socket primarily because it's easy to pick a unique socket path. +// +// The socket path is unique per controlling Pants process (e.g., pantsd, or a --no-pantsd run), +// so that concurrent Pants processes in the same repo each get their own sandboxer, rather +// than preempting each other's sandboxer (and aborting its in-flight requests). fn ensure_socket_file_removed(socket_path: &Path) -> Result<(), String> { let mut ret = fs::remove_file(socket_path); @@ -83,17 +87,28 @@ fn ensure_socket_file_removed(socket_path: &Path) -> Result<(), String> { ret.map_err(|e| e.to_string()) } +// Remove the socket file, but only if it's still the one with the given inode, i.e., the one +// that we bound, and not one that some newer sandboxer process has since bound at that path. +fn remove_socket_file_if_owned(socket_path: &Path, ino: u64) { + if fs::metadata(socket_path).is_ok_and(|m| m.ino() == ino) { + let _ = fs::remove_file(socket_path); + } +} + fn ensure_parent_dir(path: &Path) -> Result<(), String> { std::fs::create_dir_all(path.parent().unwrap()).map_err(|e| e.to_string()) } // The service that the spawned sandboxer process runs. // The process exits if anything goes wrong, since it's easy and cheap to respawn it. -// It also exits if it's been idle for a while, so that orphaned processes don't pile up. +// It also exits if its parent process exits, or if it's been idle for a while, so that orphaned +// processes don't pile up. #[derive(Debug)] pub struct SandboxerService { socket_path: PathBuf, store_cli_opt: StoreCliOpt, + // The pid of the process that spawned us, if known. + parent_pid: Option, } impl SandboxerService { @@ -106,10 +121,11 @@ impl SandboxerService { const ALLOWED_IDLE_BEATS: usize = Self::ALLOWED_IDLE_TIME.div_duration_f32(Self::POLLING_INTERVAL) as usize; - pub fn new(socket_path: PathBuf, store_cli_opt: StoreCliOpt) -> Self { + pub fn new(socket_path: PathBuf, store_cli_opt: StoreCliOpt, parent_pid: Option) -> Self { Self { socket_path, store_cli_opt, + parent_pid, } } @@ -125,6 +141,11 @@ impl SandboxerService { ensure_parent_dir(&self.socket_path)?; let listener = UnixListener::bind(&self.socket_path)?; + let socket_ino = fs::metadata(&self.socket_path)?.ino(); + // If our parent exits we'll be reparented, so a change in parent pid means we're orphaned. + // The spawner should tell us its pid, since if it exits before we get here then our + // current parent will already be the one we were reparented to. + let initial_parent_id = self.parent_pid.unwrap_or_else(parent_id); // `bind()` creates the socket file, so now we can poll it to check if it has // been deleted by the user. If it has, existing connections will continue to // work (via the underlying inode) but new connections will not be possible, @@ -150,6 +171,11 @@ impl SandboxerService { let _ = graceful_shutdown_tx.send(()); break; } + if parent_id() != initial_parent_id { + warn!("Parent process {initial_parent_id} exited. Exiting."); + let _ = graceful_shutdown_tx.send(()); + break; + } if inactivity_counter_ref.fetch_add(1, Ordering::Relaxed) > Self::ALLOWED_IDLE_BEATS { warn!("Idle time limit exceeded. Exiting."); @@ -186,9 +212,11 @@ impl SandboxerService { drop(graceful_shutdown_rx.await); }) => {} _ = abrupt_shutdown_rx => { + remove_socket_file_if_owned(&self.socket_path, socket_ino); std::process::exit(22); } } + remove_socket_file_if_owned(&self.socket_path, socket_ino); Ok(()) } } @@ -275,6 +303,11 @@ impl SandboxerClient { }) } + // How long to wait for a newly spawned server to start accepting connections. Startup is + // usually fast, but can take a second or more on a loaded machine (or, e.g., while macOS + // scans a newly built binary on its first exec), so we're generous here. + const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + pub async fn connect_with_retries(socket_path: &Path) -> Result { // TODO: POLLING_INTERVAL seems to be a good time quantum to use when waiting // for server startup (since we know the server sleeps that long on startup) @@ -283,7 +316,7 @@ impl SandboxerClient { let mut slept = Duration::ZERO; let mut maybe_client: Result = Err("Not connected yet".to_string()); - for _ in 0..20 { + while slept < Self::CONNECT_TIMEOUT { tokio::time::sleep(sleep_time).await; slept += sleep_time; debug!("Waited {slept:?} to connect to sandboxer at {socket_path:?}"); @@ -367,7 +400,9 @@ impl Sandboxer { } else { base.join(&workdir_hash) }; - let socket_path = run_dir.join("sandboxer.sock"); + // Each Pants process gets its own sandboxer, so that concurrent Pants processes + // don't preempt each other's sandboxer. + let socket_path = run_dir.join(format!("sandboxer-{}.sock", std::process::id())); debug!("Trying sandboxer socket path {}", socket_path.display()); match SocketAddr::from_pathname(&socket_path) { @@ -449,6 +484,8 @@ impl Sandboxer { let mut cmd = Command::new(&self.sandboxer_bin); cmd.arg("--socket-path"); cmd.arg(self.socket_path.as_os_str()); + cmd.arg("--parent-pid"); + cmd.arg(std::process::id().to_string()); cmd.args(self.store_cli_opt.to_cli_args()); // Pass the calling code's global log level to the sandboxer process. // TODO: Check for a relevant per-target log level? @@ -555,4 +592,5 @@ impl Sandboxer { // attempt to kill it asynchronously for us. This may leave a zombie process, as tokio doesn't // guarantee timely reaping. But since we expect the Sandboxer to be dropped only when pantsd // (or any other controlling process) is about to exit, at which point any of its child zombies -// will be reaped by the system, this should be fine. +// will be reaped by the system, this should be fine. And if the controlling process exits without +// killing the sandboxer, the sandboxer will notice that it's been orphaned, and exit. diff --git a/src/rust/process_execution/sandboxer/src/tests.rs b/src/rust/process_execution/sandboxer/src/tests.rs index 0b850adfc2a..f577827a795 100644 --- a/src/rust/process_execution/sandboxer/src/tests.rs +++ b/src/rust/process_execution/sandboxer/src/tests.rs @@ -50,7 +50,7 @@ async fn test_sandboxer_service() { let socket_path = dir_path.join("sandboxer.sock"); let sandbox_path = dir_path.join("sandbox"); - let mut sandboxer_service = SandboxerService::new(socket_path.clone(), store_cli_opt); + let mut sandboxer_service = SandboxerService::new(socket_path.clone(), store_cli_opt, None); tokio::spawn(async move { sandboxer_service.serve().await.unwrap() }); let mut client = SandboxerClient::connect_with_retries(&socket_path) diff --git a/src/rust/process_execution/sandboxer/tests/integration_test.rs b/src/rust/process_execution/sandboxer/tests/integration_test.rs index 9687890b8ee..a4a90e8f8be 100644 --- a/src/rust/process_execution/sandboxer/tests/integration_test.rs +++ b/src/rust/process_execution/sandboxer/tests/integration_test.rs @@ -6,7 +6,8 @@ use sandboxer::SandboxerService; use std::collections::BTreeSet; use std::env; use std::fs; -use std::path::Path; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; use std::thread; use std::time::Duration; use tempfile::TempDir; @@ -18,17 +19,33 @@ use sandboxer::Sandboxer; #[path = "../src/test_util.rs"] mod test_util; -#[tokio::test] -async fn test_sandboxer_process() { - // Tests the process management wrapping. Implemented as an integration test under `tests/` - // because cargo will automatically build the sandboxer binary, which this test needs, - // for integration tests, but not for unit tests under `src/`. - let sandboxer_bin = env::current_exe() +fn sandboxer_bin() -> PathBuf { + // Cargo will automatically build the sandboxer binary for integration tests under `tests/`, + // but not for unit tests under `src/`. + env::current_exe() .unwrap() .parent() .and_then(Path::parent) .unwrap() - .join("sandboxer"); + .join("sandboxer") +} + +fn wait_for(desc: &str, cond: impl Fn() -> bool) { + let mut n = 100; + while !cond() { + n -= 1; + thread::sleep(SandboxerService::POLLING_INTERVAL); + if n == 0 { + panic!("Waited too long for {desc}."); + } + } +} + +#[tokio::test] +async fn test_sandboxer_process() { + // Tests the process management wrapping. Implemented as an integration test under `tests/` + // because it needs the sandboxer binary. + let sandboxer_bin = sandboxer_bin(); let dir = TempDir::new().unwrap(); let dir_path = dir.path(); @@ -82,3 +99,39 @@ async fn test_sandboxer_process() { assert!(!sandboxer.is_alive().await.unwrap()); } + +#[tokio::test] +async fn test_orphaned_sandboxer_exits() { + let dir = TempDir::new().unwrap(); + let dir_path = dir.path(); + let (store_cli_opt, _) = test_util::prep_store(dir_path).await; + let socket_path = dir_path.join("sandboxer.sock"); + + // Spawn the sandboxer via an intermediate shell that exits shortly after, orphaning it. + // Note that the shell may exit before the sandboxer has even started up. + let shell = Command::new("sh") + .arg("-c") + .arg("\"$@\" --parent-pid $$ & echo $!; sleep 1") + .arg("sh") + .arg(sandboxer_bin()) + .arg("--socket-path") + .arg(&socket_path) + .args(store_cli_opt.to_cli_args()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + let output = shell.wait_with_output().unwrap(); + let sandboxer_pid = String::from_utf8(output.stdout).unwrap().trim().to_owned(); + + // The orphaned sandboxer should notice its parent exited, shut down, and clean up its socket. + wait_for("orphaned sandboxer to exit", || { + !Command::new("kill") + .arg("-0") + .arg(&sandboxer_pid) + .stderr(Stdio::null()) + .status() + .unwrap() + .success() + }); + assert!(!socket_path.exists()); +} diff --git a/src/rust/sharded_lmdb/Cargo.toml b/src/rust/sharded_lmdb/Cargo.toml index 8651d3c80f4..0d37831da0a 100644 --- a/src/rust/sharded_lmdb/Cargo.toml +++ b/src/rust/sharded_lmdb/Cargo.toml @@ -8,6 +8,7 @@ edition = "2024" bytes = { workspace = true } hashing = { path = "../hashing" } lmdb-rkv = { workspace = true } +lmdb-rkv-sys = { workspace = true } log = { workspace = true } task_executor = { path = "../task_executor" } tempfile = { workspace = true } diff --git a/src/rust/sharded_lmdb/src/lib.rs b/src/rust/sharded_lmdb/src/lib.rs index 325cde3d998..fa432648475 100644 --- a/src/rust/sharded_lmdb/src/lib.rs +++ b/src/rust/sharded_lmdb/src/lib.rs @@ -2,6 +2,7 @@ // Licensed under the Apache License, Version 2.0 (see LICENSE). use std::collections::{HashMap, HashSet}; +use std::ffi::c_int; use std::fmt; use std::fmt::Debug; use std::io::{self, Read}; @@ -15,7 +16,7 @@ use lmdb::{ self, Cursor, Database, DatabaseFlags, Environment, EnvironmentCopyFlags, EnvironmentFlags, RwTransaction, Transaction, WriteFlags, }; -use log::trace; +use log::{debug, trace, warn}; use tempfile::TempDir; /// @@ -106,6 +107,12 @@ impl ShardedLmdb { // with a different version of pants on a different schema version. pub const SCHEMA_VERSION: u8 = 2; + // See the MAX_READERS note in `make_env`. Each reader slot costs 64 bytes of lock file. + // Each process is capped at 120 readers (see MAX_BLOCKING_THREADS_LMDB_SAFE in + // task_executor/src/lib.rs) so this allows ~34 concurrent processes in a single + // build root, which is probably a lot more than anyone should ever need. + const MAX_READERS: u32 = 4096; + // max_size is the maximum size the databases together will be allowed to grow to. // When calling this function, we will attempt to allocate that much virtual (not resident) memory // for the mmap; in theory it should be possible not to bound this, but in practice we see travis @@ -209,7 +216,7 @@ impl ShardedLmdb { } fn make_env(dir: &Path, max_size_per_shard: usize) -> Result { - Environment::new() + let env = Environment::new() // NO_SYNC // ======= // @@ -243,9 +250,41 @@ impl ShardedLmdb { .set_flags(EnvironmentFlags::NO_SYNC | EnvironmentFlags::NO_TLS) // 2 DBs; one for file contents, one for leases. .set_max_dbs(2) + // MAX_READERS + // =========== + // + // The reader table is shared by every process using the store: pantsd and any + // --no-pantsd runs, plus one sandboxer per such process. Each process may have up + // to MAX_BLOCKING_THREADS_LMDB_SAFE concurrent reads, so LMDB's default of 126 slots + // is easily exhausted (MDB_READERS_FULL) by a handful of concurrent Pants processes. + // NB: The table is sized by whichever process creates the lock file, so this only + // takes effect for a store that no process has open at the time. + .set_max_readers(ShardedLmdb::MAX_READERS) .set_map_size(max_size_per_shard) .open(dir) - .map_err(|e| format!("Error making env for store at {dir:?}: {e}")) + .map_err(|e| format!("Error making env for store at {dir:?}: {e}"))?; + // Reclaim any reader slots left behind by processes that died without releasing them + // (e.g., due to SIGKILL), so that they don't accumulate across runs. + match Self::reader_check(&env) { + Ok(0) => {} + Ok(dead) => debug!("Cleared {dead} stale reader slots in store at {dir:?}"), + Err(e) => warn!("Failed to check for stale reader slots in store at {dir:?}: {e}"), + } + Ok(env) + } + + // Clears any reader slots in the given Environment's reader table that are held by dead + // processes, and returns the number of slots cleared. + fn reader_check(env: &Environment) -> Result { + let mut dead: c_int = 0; + // SAFETY: The env pointer is valid for the lifetime of `env`, and `dead` is a valid out + // pointer. + let rc = unsafe { lmdb_sys::mdb_reader_check(env.env(), &mut dead) }; + if rc == 0 { + Ok(dead as usize) + } else { + Err(lmdb::Error::from_err_code(rc)) + } } // First Database is content, second is leases.