Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/notes/2.34.x.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
88 changes: 60 additions & 28 deletions src/rust/process_execution/sandboxer/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -24,15 +24,15 @@ 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;
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)]
Expand Down Expand Up @@ -233,11 +233,14 @@ impl SandboxerGrpcImpl {
async fn do_materialize_directory(
&self,
request: Request<MaterializeDirectoryRequest>,
) -> Result<Response<MaterializeDirectoryResponse>, String> {
) -> Result<Response<MaterializeDirectoryResponse>, 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
Expand All @@ -253,8 +256,7 @@ impl SandboxerGrpcImpl {
&mutable_paths,
Permissions::Writable,
)
.await
.map_err(|e| e.to_string())?;
.await?;
Ok(Response::new(MaterializeDirectoryResponse {}))
}
}
Expand All @@ -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
}
Expand Down Expand Up @@ -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(),
Expand All @@ -359,7 +380,6 @@ impl SandboxerClient {
.materialize_directory(request)
.await
.map(|_| ())
.map_err(|e| e.to_string())
}
}

Expand Down Expand Up @@ -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
}
}

Expand Down
53 changes: 52 additions & 1 deletion src/rust/process_execution/sandboxer/src/tests.rs
Original file line number Diff line number Diff line change
@@ -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};
Expand Down Expand Up @@ -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:?}");
}
43 changes: 43 additions & 0 deletions src/rust/process_execution/sandboxer/tests/integration_test.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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())
}
36 changes: 30 additions & 6 deletions src/rust/process_execution/src/local.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {:?}: {:#?}",
Expand All @@ -733,19 +737,39 @@ pub async fn prepare_workdir(
.record_digest_trie(digest_trie, true)
.await?;
}
sandboxer
if 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}"
.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 {:?}: {:#?}",
Expand All @@ -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 {
Expand Down