From 54cd7b8a5ebbc06c7aea08881ce20ad3508f6f17 Mon Sep 17 00:00:00 2001 From: jgranstrom Date: Sat, 31 Oct 2026 17:45:27 +0100 Subject: [PATCH] Replace a local process cache entry whose outputs are gone PersistentCache::store wrote with NO_OVERWRITE, so an entry whose outputs garbage collection had deleted was never replaced. A process whose outputs differ per run then re-ran on every later run and its result was dropped. It now writes through ShardedLmdb::overwrite_bytes, which replaces the existing value. --- docs/notes/2.34.x.md | 2 + src/rust/cache/src/lib.rs | 2 +- src/rust/process_execution/src/cache_tests.rs | 175 ++++++++++++++---- src/rust/sharded_lmdb/src/lib.rs | 27 +++ src/rust/sharded_lmdb/src/tests.rs | 15 ++ 5 files changed, 183 insertions(+), 38 deletions(-) diff --git a/docs/notes/2.34.x.md b/docs/notes/2.34.x.md index 439b1b3f576..60477921d7c 100644 --- a/docs/notes/2.34.x.md +++ b/docs/notes/2.34.x.md @@ -69,6 +69,8 @@ An idle `pantsd` no longer re-expands and rewrites the lease of every store entr 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. +Fixed a process re-running on every run after garbage collection removed its cached outputs. The local process cache now replaces such an entry with the new result. + ### Goals ### Backends diff --git a/src/rust/cache/src/lib.rs b/src/rust/cache/src/lib.rs index 6a3a1cb2985..561a64ff9b0 100644 --- a/src/rust/cache/src/lib.rs +++ b/src/rust/cache/src/lib.rs @@ -51,7 +51,7 @@ impl PersistentCache { // rather than storing the value under its _own_ Fingerprint, the value is stored under the // Fingerprint of the CacheKey. let fingerprint = Digest::of_bytes(&key.to_bytes()).hash; - self.store.store_bytes(fingerprint, value, false).await?; + self.store.overwrite_bytes(fingerprint, value).await?; Ok(()) } diff --git a/src/rust/process_execution/src/cache_tests.rs b/src/rust/process_execution/src/cache_tests.rs index 6a28c9be91f..a1b4130f22e 100644 --- a/src/rust/process_execution/src/cache_tests.rs +++ b/src/rust/process_execution/src/cache_tests.rs @@ -42,10 +42,7 @@ fn create_local_runner() -> (Box, Store, TempDir) { (runner, store, base_dir) } -fn create_cached_runner( - local: Box, - store: Store, -) -> (Box, TempDir) { +fn create_cache() -> (PersistentCache, TempDir) { let runtime = task_executor::Executor::new(); let cache_dir = TempDir::new().unwrap(); let max_lmdb_size = 50 * 1024 * 1024; //50 MB - I didn't pick that number but it seems reasonable. @@ -59,15 +56,31 @@ fn create_cached_runner( ) .unwrap(); - let runner = Box::new(crate::cache::CommandRunner::new( - local.into(), + (cache, cache_dir) +} + +fn create_runner_with_cache( + local: Arc, + cache: PersistentCache, + store: Store, + cache_read: bool, +) -> Box { + Box::new(crate::cache::CommandRunner::new( + local, cache, store, - true, + cache_read, CacheContentBehavior::Fetch, None, - )); + )) +} +fn create_cached_runner( + local: Box, + store: Store, +) -> (Box, TempDir) { + let (cache, cache_dir) = create_cache(); + let runner = create_runner_with_cache(local.into(), cache, store, true); (runner, cache_dir) } @@ -94,6 +107,50 @@ fn create_script(script_exit_code: i8) -> (Process, PathBuf, TempDir) { (process, script_path, script_dir) } +fn create_sandbox_path_script() -> (Process, PathBuf, TempDir) { + let script_dir = TempDir::new().unwrap(); + let script_path = script_dir.path().join("script"); + std::fs::write(&script_path, "pwd > roland").unwrap(); + + let process = Process::new(vec![ + testutil::path::find_bash(), + format!("{}", script_path.display()), + ]) + .output_files(relative_paths(&["roland"]).collect()); + + (process, script_path, script_dir) +} + +async fn remove_first_output_file(store: &Store, result: &FallibleProcessResultWithPlatform) { + let output_dir_digest = result.output_directory.clone(); + store + .ensure_directory_digest_persisted(output_dir_digest.clone()) + .await + .unwrap(); + let output_dir = store + .load_directory(output_dir_digest.as_digest()) + .await + .unwrap(); + let output_child_digest = output_dir + .files + .first() + .unwrap() + .digest + .as_ref() + .unwrap() + .try_into() + .unwrap(); + let removed = store.remove_file(output_child_digest).await.unwrap(); + assert!(removed); + assert!( + store + .contents_for_directory(output_dir_digest) + .await + .err() + .is_some() + ) +} + async fn run_roundtrip(script_exit_code: i8, workunit: &mut RunningWorkunit) -> RoundtripResults { let (local, store, _local_runner_dir) = create_local_runner(); let (process, script_path, _script_dir) = create_script(script_exit_code); @@ -154,35 +211,7 @@ async fn recover_from_missing_store_contents() { // Delete the first child of the output directory parent to confirm that we ensure that more // than just the root of the output is present when hitting the cache. - { - let output_dir_digest = first_result.output_directory; - store - .ensure_directory_digest_persisted(output_dir_digest.clone()) - .await - .unwrap(); - let output_dir = store - .load_directory(output_dir_digest.as_digest()) - .await - .unwrap(); - let output_child_digest = output_dir - .files - .first() - .unwrap() - .digest - .as_ref() - .unwrap() - .try_into() - .unwrap(); - let removed = store.remove_file(output_child_digest).await.unwrap(); - assert!(removed); - assert!( - store - .contents_for_directory(output_dir_digest) - .await - .err() - .is_some() - ) - } + remove_first_output_file(&store, &first_result).await; // Ensure that we don't fail if we re-run. let second_result = caching @@ -199,3 +228,75 @@ async fn recover_from_missing_store_contents() { .is_some() ) } + +#[tokio::test] +async fn replace_entry_with_missing_store_contents() { + let (_, mut workunit) = WorkunitStore::setup_for_tests(); + + let (local, store, _local_runner_dir) = create_local_runner(); + let (caching, _cache_dir) = create_cached_runner(local, store.clone()); + let (process, script_path, _script_dir) = create_sandbox_path_script(); + + let first_result = caching + .run(Context::default(), &mut workunit, process.clone()) + .await + .unwrap(); + remove_first_output_file(&store, &first_result).await; + + let second_result = caching + .run(Context::default(), &mut workunit, process.clone()) + .await + .unwrap(); + assert_ne!( + first_result.output_directory.as_digest(), + second_result.output_directory.as_digest() + ); + + // Without the script, only a cache hit can succeed. + std::fs::remove_file(&script_path).unwrap(); + let third_result = caching + .run(Context::default(), &mut workunit, process) + .await + .unwrap(); + assert_eq!(third_result.exit_code, 0); + assert_eq!( + second_result.output_directory.as_digest(), + third_result.output_directory.as_digest() + ); +} + +#[tokio::test] +async fn replace_entry_from_runner_without_cache_reads() { + let (_, mut workunit) = WorkunitStore::setup_for_tests(); + + let (local, store, _local_runner_dir) = create_local_runner(); + let local: Arc = local.into(); + let (cache, _cache_dir) = create_cache(); + let reading = create_runner_with_cache(local.clone(), cache.clone(), store.clone(), true); + // Backtracking re-runs a process through a runner that writes the cache without reading it. + let writing = create_runner_with_cache(local, cache, store.clone(), false); + let (process, script_path, _script_dir) = create_sandbox_path_script(); + + let first_result = reading + .run(Context::default(), &mut workunit, process.clone()) + .await + .unwrap(); + remove_first_output_file(&store, &first_result).await; + + let second_result = writing + .run(Context::default(), &mut workunit, process.clone()) + .await + .unwrap(); + + // Without the script, only a cache hit can succeed. + std::fs::remove_file(&script_path).unwrap(); + let third_result = reading + .run(Context::default(), &mut workunit, process) + .await + .unwrap(); + assert_eq!(third_result.exit_code, 0); + assert_eq!( + second_result.output_directory.as_digest(), + third_result.output_directory.as_digest() + ); +} diff --git a/src/rust/sharded_lmdb/src/lib.rs b/src/rust/sharded_lmdb/src/lib.rs index cfd2cb162b8..0f40e9a7565 100644 --- a/src/rust/sharded_lmdb/src/lib.rs +++ b/src/rust/sharded_lmdb/src/lib.rs @@ -587,6 +587,33 @@ impl ShardedLmdb { Ok(fingerprint) } + /// + /// Stores the given Bytes under the given Fingerprint, replacing any value already stored + /// under it. + /// + pub async fn overwrite_bytes( + &self, + fingerprint: Fingerprint, + bytes: Bytes, + ) -> Result<(), String> { + let store = self.clone(); + self.executor + .spawn_blocking(move || { + let effective_key = + VersionedFingerprint::new(fingerprint, ShardedLmdb::SCHEMA_VERSION); + let (env, db, _) = store.get(&fingerprint); + env.begin_rw_txn() + .and_then(|mut txn| { + txn.put(db, &effective_key, &bytes, WriteFlags::empty())?; + txn.commit() + }) + .map_err(|e| format!("Error overwriting {fingerprint:?}: {e}")) + }) + .await + .map_err(|e| format!("`overwrite_bytes` task failed: {e}")) + .flatten() + } + /// /// Store the given Bytes instances under the given Fingerprints, or under their computed /// Fingerprints. For large/streaming usecases, prefer `Self::store`. diff --git a/src/rust/sharded_lmdb/src/tests.rs b/src/rust/sharded_lmdb/src/tests.rs index 9726171d1fd..8212ba3567d 100644 --- a/src/rust/sharded_lmdb/src/tests.rs +++ b/src/rust/sharded_lmdb/src/tests.rs @@ -98,6 +98,21 @@ async fn store_failure() { assert!(result.is_err()); } +#[tokio::test] +async fn overwrite_bytes() { + let (s, _tempdir) = new_store(1); + let fingerprint = Digest::of_bytes(&bytes(0)).hash; + s.store_bytes(fingerprint, bytes(1), false).await.unwrap(); + + s.overwrite_bytes(fingerprint, bytes(2)).await.unwrap(); + + let stored = s + .load_bytes_with(fingerprint, |b| Ok(Bytes::copy_from_slice(b))) + .await + .unwrap(); + assert_eq!(stored, Some(bytes(2))); +} + #[tokio::test] async fn stores_on_one_directory_share_environments() { let tempdir = TempDir::new().unwrap();