Replace a local process cache entry whose outputs are gone

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

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

src/rust/process_execution/src/cache_tests.rs

@@ -42,10 +42,7 @@ fn create_local_runner() -> (Box<dyn CommandRunnerTrait>, Store, TempDir) {
     (runner, store, base_dir)
 }
 
-fn create_cached_runner(
-    local: Box<dyn CommandRunnerTrait>,
-    store: Store,
-) -> (Box<dyn CommandRunnerTrait>, 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<dyn CommandRunnerTrait>,
+    cache: PersistentCache,
+    store: Store,
+    cache_read: bool,
+) -> Box<dyn CommandRunnerTrait> {
+    Box::new(crate::cache::CommandRunner::new(
+        local,
         cache,
         store,
-        true,
+        cache_read,
         CacheContentBehavior::Fetch,
         None,
-    ));
+    ))
+}
 
+fn create_cached_runner(
+    local: Box<dyn CommandRunnerTrait>,
+    store: Store,
+) -> (Box<dyn CommandRunnerTrait>, 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<dyn CommandRunnerTrait> = 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()
+    );
+}

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`.

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