Fix OSS RE client: FindMissingCache staleness, TTL, and silent upload skip

Three related bugs in the OSS RE gRPC client cause `remote_upload_error`
when deferred-materialized outputs from one action are used as inputs to
another action.

**Bug 1: Hardcoded TTL of 0 in OSS shim**

`convert_action_result()` hardcodes `ttl: 0` for all output files and
`action_result_ttl: 0` for execute responses. The standard REv2 protocol
has no TTL field — TTL is a buck2 extension. With ttl=0, the deferred
materializer treats every blob as immediately expired, so
`get_digests_ttl()` always queries FindMissingBlobs instead of trusting
that recently-produced outputs still exist.

Fix: use `cas_ttl_secs` from runtime options (configured via
`cas_ttl_secs` buckconfig key, default 3 hours) instead of hardcoding 0.

**Bug 2: FindMissingCache caches "Missing" state**

`get_digests_ttl()` caches `DigestRemoteState::Missing` in the
FindMissingCache LRU (500K entries, 12-hour TTL). "Missing" is a
transient state — the blob may be uploaded or produced by an RE action
at any time. A stale "Missing" entry causes `get_digests_ttl()` to
return ttl=0 without sending a FindMissingBlobs RPC, even though the
blob now exists in CAS. Since deferred materialization never downloaded
the blob, buck2 cannot upload it and fails with `remote_upload_error`.

Fix: stop caching `DigestRemoteState::Missing`. Only cache
`ExistsOnRemote`. Additionally, mark top-level output file and directory
digests as `ExistsOnRemote` when processing action results from Execute,
GetActionResult, and WriteActionResult responses.

**Bug 3: Silent skip of missing CAS artifacts during upload**

When the uploader encounters a `RequiresCasDownload` file that
FindMissingBlobs reports as missing, it logs a `soft_error!("cas_missing")`
and silently skips the file (`continue`). The downstream action then
fails with incomplete inputs.

Fix: instead of silently skipping, materialize the file locally (which
downloads from CAS, bypassing FindMissingCache) and add it to the upload
list. If CAS truly doesn't have the blob, `ensure_materialized` fails
with a clear error instead of a mysterious downstream action failure.

Author: mattparks

app/buck2_execute/src/re/uploader.rs

@@ -312,9 +312,9 @@ impl Uploader {
                     }
                     Err(
                         ref err @ ArtifactNotMaterializedReason::RequiresCasDownload {
+                            ref path,
                             ref entry,
                             ref info,
-                            ..
                         },
                     ) => {
                         if let DirectoryEntry::Leaf(ActionDirectoryMember::File(file)) =
@@ -361,6 +361,19 @@ impl Uploader {
                                     quiet: true
                                 )?;
 
+                                // Materialize the file locally from CAS and add it to
+                                // the upload list. Silently skipping this file (the old
+                                // behavior) leaves the downstream action with incomplete
+                                // inputs, causing confusing failures. Materializing
+                                // bypasses FindMissingCache and talks directly to CAS;
+                                // if the blob truly is gone, we fail here with a clear
+                                // error rather than later with a mysterious one.
+                                paths_to_materialize.push(path.clone());
+                                upload_files.push(NamedDigest {
+                                    name: fs.resolve(path).as_maybe_relativized_str()?.to_owned(),
+                                    digest,
+                                    ..Default::default()
+                                });
                                 continue;
                             }
                         }

remote_execution/oss/re_grpc/src/client.rs

@@ -591,7 +591,7 @@ pub struct REClient {
     capabilities: RECapabilities,
     instance_name: InstanceName,
     // buck2 calls find_missing for same blobs
-    find_missing_cache: Mutex<FindMissingCache>,
+    find_missing_cache: Arc<Mutex<FindMissingCache>>,
     bystream_compressor: Option<Compressor>,
 }
 
@@ -669,11 +669,11 @@ impl REClient {
             grpc_clients,
             capabilities,
             instance_name,
-            find_missing_cache: Mutex::new(FindMissingCache {
+            find_missing_cache: Arc::new(Mutex::new(FindMissingCache {
                 cache: LruCache::new(NonZeroUsize::new(500_000).unwrap()),
                 ttl: Duration::from_hours(12), // 12 hours TODO: Tune this parameter
                 last_check: Instant::now(),
-            }),
+            })),
             bystream_compressor,
         }
     }
@@ -697,9 +697,12 @@ impl REClient {
             ))
             .await?;
 
+        let action_result = convert_action_result(res.into_inner(), self.runtime_opts.cas_ttl_secs)?;
+        mark_action_result_outputs_as_existing(&self.find_missing_cache, &action_result);
+
         Ok(ActionResultResponse {
-            action_result: convert_action_result(res.into_inner())?,
-            ttl: 0,
+            action_result,
+            ttl: self.runtime_opts.cas_ttl_secs,
         })
     }
 
@@ -724,9 +727,12 @@ impl REClient {
             ))
             .await?;
 
+        let action_result = convert_action_result(res.into_inner(), self.runtime_opts.cas_ttl_secs)?;
+        mark_action_result_outputs_as_existing(&self.find_missing_cache, &action_result);
+
         Ok(WriteActionResultResponse {
-            actual_action_result: convert_action_result(res.into_inner())?,
-            ttl_seconds: 0,
+            actual_action_result: action_result,
+            ttl_seconds: self.runtime_opts.cas_ttl_secs,
         })
     }
 
@@ -756,6 +762,9 @@ impl REClient {
             ..Default::default()
         };
 
+        let cas_ttl_secs = self.runtime_opts.cas_ttl_secs;
+        let find_missing_cache = self.find_missing_cache.clone();
+
         let stream = client
             .execute(with_re_metadata(
                 request,
@@ -765,7 +774,9 @@ impl REClient {
             .await?
             .into_inner();
 
-        let stream = futures::stream::try_unfold(stream, move |mut stream| async {
+        let stream = futures::stream::try_unfold(stream, move |mut stream| {
+            let find_missing_cache = find_missing_cache.clone();
+            async move {
             let msg = match stream.try_next().await.context("RE channel error")? {
                 Some(msg) => msg,
                 None => return Ok(None),
@@ -794,12 +805,13 @@ impl REClient {
                             .result
                             .with_context(|| "The action result is not defined.")?;
 
-                        let action_result = convert_action_result(action_result)?;
+                        let action_result = convert_action_result(action_result, cas_ttl_secs)?;
+                        mark_action_result_outputs_as_existing(&find_missing_cache, &action_result);
 
                         let execute_response = ExecuteResponse {
                             action_result,
                             action_result_digest: TDigest::default(),
-                            action_result_ttl: 0,
+                            action_result_ttl: cas_ttl_secs,
                             status: TStatus {
                                 code: TCode::OK,
                                 message: execute_response_grpc.message,
@@ -837,6 +849,7 @@ impl REClient {
             };
 
             anyhow::Ok(Some((status, stream)))
+            }
         });
 
         // We fill in the action digest a little later here. We do it this way so we don't have to
@@ -1019,10 +1032,13 @@ impl REClient {
                 }
 
                 for digest in &resp.missing_blob_digests.map(|d| tdigest_from(d.clone())) {
-                    // If it's present in the MissingBlobsResponse, it's expired on the remote and
-                    // needs to be refetched.
+                    // Missing is a transient state: the blob may be uploaded or
+                    // produced by an RE action at any time. Only record it in
+                    // remote_results for the current call; do NOT cache it, as a
+                    // stale "Missing" entry would prevent future get_digests_ttl
+                    // calls from sending a FindMissingBlobs RPC, causing
+                    // remote_upload_error for blobs that actually exist in CAS.
                     remote_results.insert(digest.clone(), DigestRemoteState::Missing);
-                    find_missing_cache.put(digest.clone(), DigestRemoteState::Missing);
                 }
                 digests_to_check.clear();
             }
@@ -1080,7 +1096,7 @@ impl REClient {
     }
 }
 
-fn convert_action_result(action_result: ActionResult) -> anyhow::Result<TActionResult2> {
+fn convert_action_result(action_result: ActionResult, cas_ttl_secs: i64) -> anyhow::Result<TActionResult2> {
     let execution_metadata = action_result
         .execution_metadata
         .with_context(|| "The execution metadata are not defined.")?;
@@ -1097,7 +1113,7 @@ fn convert_action_result(action_result: ActionResult) -> anyhow::Result<TActionR
             name: output_file.path,
             existed: false,
             executable: output_file.is_executable,
-            ttl: 0,
+            ttl: cas_ttl_secs,
             _dot_dot_default: (),
         })
     })?;
@@ -1176,6 +1192,24 @@ fn convert_action_result(action_result: ActionResult) -> anyhow::Result<TActionR
     Ok(action_result)
 }
 
+/// Mark all output digests from an action result as existing in the
+/// FindMissingCache. This prevents stale "Missing" entries from causing
+/// downstream actions to fail with remote_upload_error when they reference
+/// outputs produced by earlier RE actions.
+fn mark_action_result_outputs_as_existing(
+    cache: &Mutex<FindMissingCache>,
+    action_result: &TActionResult2,
+) {
+    let mut cache = cache.lock().unwrap();
+    for file in &action_result.output_files {
+        cache.put(file.digest.digest.clone(), DigestRemoteState::ExistsOnRemote);
+    }
+    for dir in &action_result.output_directories {
+        cache.put(dir.tree_digest.clone(), DigestRemoteState::ExistsOnRemote);
+        cache.put(dir.root_directory_digest.clone(), DigestRemoteState::ExistsOnRemote);
+    }
+}
+
 fn convert_t_action_result2(t_action_result: TActionResult2) -> anyhow::Result<ActionResult> {
     let t_execution_metadata = t_action_result.execution_metadata;
     let virtual_execution_duration = prost_types::Duration::try_from(