Per-process sandboxer socket path

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.

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"

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
 }

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<u32>,
+
     #[command(flatten)]
     store_options: StoreCliOpt,
 }
@@ -41,7 +46,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         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
 }

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<u32>,
 }
 
 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<u32>) -> 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<Self, String> {
         // 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<Self, String> = 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.

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)

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());
+}

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 }