From 9fc9e0f0ad2d18e8cd2db6a7efce2b6dac8982fe Mon Sep 17 00:00:00 2001 From: jgranstrom Date: Sun, 1 Nov 2026 17:26:22 +0100 Subject: [PATCH] Backtrack when the sandboxer cannot find a process input With the sandboxer, a missing input reached prepare_workdir as text, so the run failed instead of backtracking, and the sandboxer was restarted. The sandboxer now answers a missing digest with NotFound, on which Sandboxer does not restart it. prepare_workdir then asks Store::ensure_downloaded for the typed MissingDigest that backtracking needs, or retries once if nothing is missing. --- docs/notes/2.34.x.md | 2 + .../process_execution/sandboxer/src/lib.rs | 88 +++++++++++++------ .../process_execution/sandboxer/src/tests.rs | 53 ++++++++++- .../sandboxer/tests/integration_test.rs | 43 +++++++++ src/rust/process_execution/src/local.rs | 36 ++++++-- 5 files changed, 187 insertions(+), 35 deletions(-) diff --git a/docs/notes/2.34.x.md b/docs/notes/2.34.x.md index 161da74f163..e993011ec4b 100644 --- a/docs/notes/2.34.x.md +++ b/docs/notes/2.34.x.md @@ -65,6 +65,8 @@ Pants now logs a warning when it cannot delete a local process execution sandbox An idle `pantsd` no longer re-expands and rewrites the lease of every store entry its graph references every 72 seconds. A lease with more than half of its time left is no longer rewritten, and a digest leased within the last quarter of the lease time (30 minutes) is not expanded again, which removes most of the CPU time and disk writes an idle daemon spent extending leases ([#23694](https://github.com/pantsbuild/pants/issues/23694)). +With the `sandboxer`, a process input missing from the local store now makes Pants re-run the process that produced it, as it does without the `sandboxer`. Previously every run needing that input failed until `pantsd` restarted. + ### Goals ### Backends diff --git a/src/rust/process_execution/sandboxer/src/lib.rs b/src/rust/process_execution/sandboxer/src/lib.rs index 9b6422369da..4f82f5d7cdc 100644 --- a/src/rust/process_execution/sandboxer/src/lib.rs +++ b/src/rust/process_execution/sandboxer/src/lib.rs @@ -24,7 +24,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use std::{collections::BTreeSet, io}; use std::{fs, thread}; -use store::{Store, StoreCliOpt}; +use store::{Store, StoreCliOpt, StoreError}; use tokio::fs::OpenOptions; use tokio::net::{UnixListener, UnixStream}; use tokio::sync::RwLock; @@ -32,7 +32,7 @@ use tokio::sync::oneshot; use tokio::{process::Command, sync::RwLockWriteGuard}; use tokio_stream::wrappers::UnixListenerStream; use tonic::transport::{Channel, Endpoint, Server, Uri}; -use tonic::{Request, Response, Status}; +use tonic::{Code, Request, Response, Status}; use tower::service_fn; #[cfg(test)] @@ -233,11 +233,14 @@ impl SandboxerGrpcImpl { async fn do_materialize_directory( &self, request: Request, - ) -> Result, String> { + ) -> Result, StoreError> { let r = request.into_inner(); debug!("Received materialize_directory() request: {:#?}", r); let destination_root: PathBuf = r.destination_root.into(); - let digest: Digest = r.digest.ok_or("No digest provided")?.try_into()?; + let digest: Digest = r + .digest + .ok_or_else(|| "No digest provided".to_owned())? + .try_into()?; let dir_digest = DirectoryDigest::from_persisted_digest(digest); let mutable_paths = r .mutable_paths @@ -253,8 +256,7 @@ impl SandboxerGrpcImpl { &mutable_paths, Permissions::Writable, ) - .await - .map_err(|e| e.to_string())?; + .await?; Ok(Response::new(MaterializeDirectoryResponse {})) } } @@ -269,7 +271,10 @@ impl SandboxerGrpc for SandboxerGrpcImpl { let ret = self .do_materialize_directory(request) .await - .map_err(Status::failed_precondition); + .map_err(|e| match e { + StoreError::MissingDigest(..) => Status::not_found(e.to_string()), + StoreError::Unclassified(_) => Status::failed_precondition(e.to_string()), + }); debug!("materialize_directory() result: {:#?}", ret); ret } @@ -336,20 +341,36 @@ impl SandboxerClient { dir_digest: &DirectoryDigest, mutable_paths: &[String], ) -> Result<(), String> { + self.request_materialize_directory(destination, destination_root, dir_digest, mutable_paths) + .await + .map_err(|status| status.to_string()) + } + + async fn request_materialize_directory( + &mut self, + destination: &Path, + destination_root: &Path, + dir_digest: &DirectoryDigest, + mutable_paths: &[String], + ) -> Result<(), Status> { let request = tonic::Request::new(MaterializeDirectoryRequest { destination: destination .to_str() - .ok_or(format!( - "workdir_path is not a valid str: {}", - destination.display() - ))? + .ok_or_else(|| { + Status::invalid_argument(format!( + "workdir_path is not a valid str: {}", + destination.display() + )) + })? .to_string(), destination_root: destination_root .to_str() - .ok_or(format!( - "workdir_root_path is not a valid str: {}", - destination_root.display() - ))? + .ok_or_else(|| { + Status::invalid_argument(format!( + "workdir_root_path is not a valid str: {}", + destination_root.display() + )) + })? .to_string(), digest: Some(dir_digest.as_digest().into()), mutable_paths: mutable_paths.to_vec(), @@ -359,7 +380,6 @@ impl SandboxerClient { .materialize_directory(request) .await .map(|_| ()) - .map_err(|e| e.to_string()) } } @@ -567,20 +587,32 @@ impl Sandboxer { let ret = self .client() .await? - .materialize_directory(destination, destination_root, dir_digest, &mutable_paths) + .request_materialize_directory( + destination, + destination_root, + dir_digest, + &mutable_paths, + ) .await; - if ret.is_err() { - // The server might be down for some reason (e.g., it has not been initially started - // yet, or the socket file was deleted by the user). If this happens, ensure the - // server is properly killed, and retry once. - self.kill().await; - return self - .client() - .await? - .materialize_directory(destination, destination_root, dir_digest, &mutable_paths) - .await; + match ret { + Err(status) if status.code() != Code::NotFound => { + // The server might be down for some reason (e.g., it has not been initially started + // yet, or the socket file was deleted by the user). If this happens, ensure the + // server is properly killed, and retry once. A missing input is not a server + // failure. + self.kill().await; + self.client() + .await? + .materialize_directory( + destination, + destination_root, + dir_digest, + &mutable_paths, + ) + .await + } + ret => ret.map_err(|status| status.to_string()), } - ret } } diff --git a/src/rust/process_execution/sandboxer/src/tests.rs b/src/rust/process_execution/sandboxer/src/tests.rs index f577827a795..1c7c3b6b4ff 100644 --- a/src/rust/process_execution/sandboxer/src/tests.rs +++ b/src/rust/process_execution/sandboxer/src/tests.rs @@ -1,11 +1,15 @@ // Copyright 2025 Pants project contributors (see CONTRIBUTORS.md). // Licensed under the Apache License, Version 2.0 (see LICENSE). -use ::fs::DirectoryDigest; +use ::fs::{DigestTrie, DirectoryDigest, TypedPath}; +use hashing::Digest; +use std::collections::HashMap; use std::fs; use std::io::Write; use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; use tempfile::TempDir; +use tonic::Code; use crate::test_util::prep_store; use crate::{SandboxerClient, SandboxerService, ensure_socket_file_removed}; @@ -70,3 +74,50 @@ async fn test_sandboxer_service() { fs::read_to_string(sandbox_path.join("subdir").join("greeting.txt")).unwrap(); assert_eq!("Hello, world!", materialized_content); } + +#[tokio::test] +async fn test_sandboxer_service_missing_digest() { + let dir = TempDir::new().unwrap(); + let dir_path = dir.path(); + let (store_cli_opt, _) = prep_store(dir_path).await; + let socket_path = dir_path.join("sandboxer.sock"); + let sandbox_path = dir_path.join("sandbox"); + + let dir_digest = { + let store = store_cli_opt + .create_store(task_executor::Executor::new()) + .await + .unwrap(); + let file_digests = HashMap::from([( + PathBuf::from("missing.txt"), + Digest::of_bytes(b"not in the store"), + )]); + let tree = DigestTrie::from_unique_paths( + vec![TypedPath::File { + path: Path::new("missing.txt"), + is_executable: false, + }], + &file_digests, + ) + .unwrap(); + store.record_digest_trie(tree, true).await.unwrap() + }; + + 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) + .await + .unwrap(); + let status = client + .request_materialize_directory( + &sandbox_path, + &sandbox_path, + &DirectoryDigest::from_persisted_digest(dir_digest.as_digest()), + &[], + ) + .await + .unwrap_err(); + + assert_eq!(Code::NotFound, status.code(), "{status:?}"); +} diff --git a/src/rust/process_execution/sandboxer/tests/integration_test.rs b/src/rust/process_execution/sandboxer/tests/integration_test.rs index a4a90e8f8be..3f1022eaecf 100644 --- a/src/rust/process_execution/sandboxer/tests/integration_test.rs +++ b/src/rust/process_execution/sandboxer/tests/integration_test.rs @@ -2,10 +2,12 @@ // Licensed under the Apache License, Version 2.0 (see LICENSE). use ::fs::DirectoryDigest; +use hashing::Digest; use sandboxer::SandboxerService; use std::collections::BTreeSet; use std::env; use std::fs; +use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::thread; @@ -135,3 +137,44 @@ async fn test_orphaned_sandboxer_exits() { }); assert!(!socket_path.exists()); } + +#[tokio::test] +async fn test_sandboxer_process_keeps_running_on_missing_digest() { + let dir = TempDir::new().unwrap(); + let dir_path = dir.path(); + let (store_cli_opt, dir_digest) = test_util::prep_store(dir_path).await; + let sandbox_path = dir_path.join("sandbox"); + + let sandboxer = Sandboxer::new(sandboxer_bin(), dir_path.join("workdir"), store_cli_opt) + .await + .unwrap(); + sandboxer + .materialize_directory( + &sandbox_path, + &sandbox_path, + &DirectoryDigest::from_persisted_digest(dir_digest.as_digest()), + &BTreeSet::new(), + ) + .await + .unwrap(); + let socket_file = socket_file_identity(sandboxer.socket_path()); + + let res = sandboxer + .materialize_directory( + &sandbox_path, + &sandbox_path, + &DirectoryDigest::from_persisted_digest(Digest::of_bytes(b"not in the store")), + &BTreeSet::new(), + ) + .await; + + assert!(res.is_err()); + // A respawned sandboxer would have bound a new socket file. + assert_eq!(socket_file, socket_file_identity(sandboxer.socket_path())); +} + +// Freed inode numbers are reused, so the change time is compared too. +fn socket_file_identity(path: &Path) -> (u64, i64, i64) { + let metadata = fs::metadata(path).unwrap(); + (metadata.ino(), metadata.ctime(), metadata.ctime_nsec()) +} diff --git a/src/rust/process_execution/src/local.rs b/src/rust/process_execution/src/local.rs index 9c4597748ff..7c83de4fbd1 100644 --- a/src/rust/process_execution/src/local.rs +++ b/src/rust/process_execution/src/local.rs @@ -717,6 +717,10 @@ pub async fn prepare_workdir( let mut mutable_paths = req.output_files.clone(); mutable_paths.extend(req.output_directories.clone()); + let materialize_error = |se: StoreError| { + CapturedWorkdirError::from(se.enrich(format!("An error occurred when attempting to materialize a working directory at {workdir_path:#?}").as_str())) + }; + if let Some(sandboxer) = sandboxer { debug!( "Materializing via sandboxer to {:?}: {:#?}", @@ -733,7 +737,7 @@ pub async fn prepare_workdir( .record_digest_trie(digest_trie, true) .await?; } - sandboxer + if sandboxer .materialize_directory( &workdir_path, workdir_root_path, @@ -741,11 +745,31 @@ pub async fn prepare_workdir( &mutable_paths, ) .await - .map_err(|e| { - format!( - "materialize_directory() request to sandboxer process failed: {e}" + .is_err() + { + // The sandboxer's error is text, so ask the store for the typed `MissingDigest` + // that backtracking needs. If nothing is missing, the input was stored since the + // failure, so retry once. + if let Err(se @ StoreError::MissingDigest(..)) = store + .ensure_downloaded(HashSet::new(), HashSet::from([persisted_digest.clone()])) + .await + { + return Err(materialize_error(se)); + } + sandboxer + .materialize_directory( + &workdir_path, + workdir_root_path, + &persisted_digest, + &mutable_paths, ) - })?; + .await + .map_err(|e| { + format!( + "materialize_directory() request to sandboxer process failed: {e}" + ) + })?; + } } else { debug!( "Materializing directly to {:?}: {:#?}", @@ -761,7 +785,7 @@ pub async fn prepare_workdir( Permissions::Writable, ) .await - .map_err(|se| CapturedWorkdirError::from(se.enrich(format!("An error occurred when attempting to materialize a working directory at {workdir_path:#?}").as_str())))?; + .map_err(materialize_error)?; } if let Some(executable_path) = maybe_executable_path {