/build/source/nativelink-worker/src/running_actions_manager.rs

Source
// Copyright 2024 The NativeLink Authors. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use core::cmp::min;
use core::convert::Into;
use core::fmt::Debug;
use core::pin::Pin;
use core::sync::atomic::{AtomicBool, Ordering};
use core::time::Duration;
use std::borrow::Cow;
use std::collections::vec_deque::VecDeque;
use std::collections::{HashMap, HashSet};
use std::env;
use std::ffi::{OsStr, OsString};
#[cfg(target_family = "unix")]
use std::fs::Permissions;
#[cfg(target_family = "unix")]
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{Arc, Weak};
use std::time::SystemTime;

use bytes::{Bytes, BytesMut};
use filetime::{FileTime, set_file_mtime};
use formatx::Template;
use futures::future::{
    BoxFuture, Future, FutureExt, TryFutureExt, join_all, try_join, try_join_all,
};
use futures::stream::{FuturesUnordered, StreamExt, TryStreamExt};
use nativelink_config::cas_server::{
    EnvironmentSource, UploadActionResultConfig, UploadCacheResultsStrategy,
};
use nativelink_error::{Code, Error, ResultExt, make_err, make_input_err};
use nativelink_metric::MetricsComponent;
use nativelink_proto::build::bazel::remote::execution::v2::{
    Action, ActionResult as ProtoActionResult, Command as ProtoCommand,
    Directory as ProtoDirectory, Directory, DirectoryNode, ExecuteResponse, FileNode, SymlinkNode,
    Tree as ProtoTree, UpdateActionResultRequest,
};
use nativelink_proto::com::github::trace_machina::nativelink::remote_execution::{
    ActionResourceUsage, HistoricalExecuteResponse, StartExecute,
};
use nativelink_store::ac_utils::{
    ESTIMATED_DIGEST_SIZE, compute_buf_digest, get_and_decode_digest, serialize_and_upload_message,
};
use nativelink_store::cas_utils::is_zero_digest;
use nativelink_store::fast_slow_store::FastSlowStore;
use nativelink_store::filesystem_store::{FileEntry, FileEntryImpl, FilesystemStore};
use nativelink_store::grpc_store::GrpcStore;
use nativelink_util::action_messages::{
    ActionInfo, ActionResult, DirectoryInfo, ExecutionMetadata, FileInfo, NameOrPath, OperationId,
    SymlinkInfo, to_execute_response,
};
use nativelink_util::common::{DigestInfo, fs};
use nativelink_util::digest_hasher::{DigestHasher, DigestHasherFunc};
use nativelink_util::metrics_utils::{AsyncCounterWrapper, CounterWithTime};
use nativelink_util::store_trait::{Store, StoreLike, UploadSizeInfo};
use nativelink_util::{background_spawn, spawn, spawn_blocking};
use parking_lot::Mutex;
use prost::Message;
use relative_path::RelativePath;
use scopeguard::{ScopeGuard, guard};
use serde::Deserialize;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
use tokio::process;
use tokio::sync::{Notify, oneshot, watch};
use tokio::time::Instant;
use tokio_stream::wrappers::ReadDirStream;
use tonic::Request;
use tracing::{debug, error, info, trace, warn};
use uuid::Uuid;

use crate::persistent_worker::{
    Input as PersistentWorkerInput, PersistentWorkerPool, WireFormat, WorkRequest, WorkerKey,
};

/// For simplicity we use a fixed exit code for cases when our program is terminated
/// due to a signal.
const EXIT_CODE_FOR_SIGNAL: i32 = 9;

const SUPPORTS_WORKERS_PROPERTY: &str = "supports-workers";
const REQUIRES_WORKER_PROTOCOL_PROPERTY: &str = "requires-worker-protocol";

/// Default strategy for uploading historical results.
/// Note: If this value changes the config documentation
/// should reflect it.
const DEFAULT_HISTORICAL_RESULTS_STRATEGY: UploadCacheResultsStrategy =
    UploadCacheResultsStrategy::FailuresOnly;

#[cfg(target_os = "linux")]
const RESOURCE_USAGE_SAMPLE_INTERVAL: Duration = Duration::from_millis(250);

/// What the sampler observed over an action's lifetime.
#[cfg(target_os = "linux")]
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
struct SampledResourceUsage {
    peak_memory_kb: u64,
    cpu_time_ms: u64,
}

#[cfg(target_os = "linux")]
struct ActionResourceUsageSampler {
    stop_tx: watch::Sender<bool>,
    handle: tokio::task::JoinHandle<SampledResourceUsage>,
}

#[cfg(target_os = "linux")]
fn start_action_resource_usage_sampler(pgid: u32) -> ActionResourceUsageSampler {
    let (stop_tx, stop_rx) = watch::channel(false);
    let handle = background_spawn!(
        "action_resource_usage_sampler",
        sample_action_resource_usage(pgid, stop_rx)
    );
    ActionResourceUsageSampler { stop_tx, handle }
}

#[cfg(target_os = "linux")]
async fn finish_action_resource_usage_sampler(
    sampler: ActionResourceUsageSampler,
) -> Option<SampledResourceUsage> {
    let _ = sampler.stop_tx.send(true);
    sampler.handle.await.ok()
}

#[cfg(target_os = "linux")]
async fn sample_action_resource_usage(
    pgid: u32,
    mut stop_rx: watch::Receiver<bool>,
) -> SampledResourceUsage {
    let mut peak_memory_kb = 0;
    // CPU time is cumulative per process and a process that exits stops
    // appearing, so summing the live group at the end would lose everything
    // short-lived. Keep the last figure seen for each pid and total them at
    // the end instead.
    let mut cpu_ticks_by_pid = HashMap::new();

    let sample18 = |peak_memory_kb: &mut u64, cpu_ticks_by_pid: &mut HashMap<u32, u64>| {
        let observed = sample_process_group(pgid, cpu_ticks_by_pid);
        if let Some(memory_kb16) = observed {
            *peak_memory_kb = (*peak_memory_kb).max(memory_kb);
        }
        observed
    };

    loop {
        if sample(&mut peak_memory_kb, &mut cpu_ticks_by_pid).is_none()
            && !Path::new(&format!("/proc/{pgid}")).exists()
        {
            // The group leader has been reaped and no member process remains,
            // so the action is finished.
            break;
        }

        if *stop_rx.borrow() {
            break;
        }

        tokio::select! {
            changed = stop_rx.changed() => {
                if changed.is_ok() && *stop_rx.borrow() {
                    sample(&mut peak_memory_kb, &mut cpu_ticks_by_pid);
                    break;
                }
            }
            () = tokio::time::sleep(RESOURCE_USAGE_SAMPLE_INTERVAL) => {}0
        }
    }

    SampledResourceUsage {
        peak_memory_kb,
        cpu_time_ms: ticks_to_millis(cpu_ticks_by_pid.values().sum()),
    }
}

/// Converts clock ticks from `/proc` into milliseconds.
///
/// `_SC_CLK_TCK` is 100 on every Linux we run on, but it is queryable so ask
/// rather than assume, and fall back to 100 if the query fails.
#[cfg(target_os = "linux")]
fn ticks_to_millis(ticks: u64) -> u64 {
    // SAFETY: `sysconf` takes an int and returns a long, with no memory
    // safety considerations.
    let hz = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
    let hz = if hz > 0 { hz as u64 } else { 1000 };
    ticks.saturating_mul(1_000) / hz
}

/// Sums the resident memory of every process in the action's process group.
///
/// The action is spawned as its own process-group leader (see
/// `command_builder.process_group(0)` in `inner_execute`), so `pgid` equals
/// the spawned child's pid and every descendant inherits it. Sampling by
/// process group — rather than walking the parent/child tree from the spawned
/// pid — is required because an intermediate shell frequently exits and
/// reparents the real workload to the worker (PID 1); the tree walk then sees
/// only a childless zombie and reports zero. Process-group membership is
/// inherited and survives reparenting, and excludes the worker's own group.
/// Returns `None` when no member process can be read (the group is empty).
#[cfg(target_os = "linux")]
fn sample_process_group(pgid: u32, cpu_ticks_by_pid: &mut HashMap<u32, u64>) -> Option<u64> {
    let Ok(entries) = std::fs::read_dir("/proc") else {
        return None;
    };

    let mut total_kb = 0;
    let mut found_any_process = false;
    for entry in entries36.flatten36() {
        let Ok(member_pid177) = entry.file_name().to_string_lossy().parse::<u32>() else {
            continue;
        };
        let Ok(stat) = std::fs::read_to_string(format!("/proc/{member_pid}/stat")) else {
            continue;
        };
        if parse_pgid_from_stat(&stat) != Some(pgid) {
            continue;
        }
        if let Some(ticks) = parse_cpu_ticks_from_stat(&stat) {
            // Monotonic per process, but a pid could in principle be reused
            // within one action, so never let the figure go backwards.
            let entry = cpu_ticks_by_pid.entry(member_pid).or_insert(0);
            *entry = (*entry).max(ticks);
        }0
        if let Some(memory_kb28) = read_process_rss_kb(member_pid) {
            total_kb += memory_kb;
            found_any_process = true;
        }5
    }

    found_any_process.then_some(total_kb)
}

#[cfg(target_os = "linux")]
fn read_process_rss_kb(pid: u32) -> Option<u64> {
    let status = std::fs::read_to_string(format!("/proc/{pid}/status")).ok()?0;
    status.lines()33.find_map33(|line| {
        let rest28 = line.strip_prefix("VmRSS:")?825;
        rest.split_whitespace().next()?0.parse().ok()
    })
}

/// Parses the process group id (`pgrp`, field 5) from the contents of a
/// `/proc/<pid>/stat` line.
///
/// `comm` (field 2) can contain spaces and parentheses, so fields are parsed
/// relative to the final `)` to avoid miscounting. Returns `None` when the
/// line is malformed.
#[cfg(target_os = "linux")]
pub fn parse_pgid_from_stat(stat: &str) -> Option<u32> {
    let after_comm180 = stat.rsplit_once(')')?2.1;
    // Fields after the final ')': state(0) ppid(1) pgrp(2) ...
    after_comm.split_whitespace().nth(2)?1.parse179().ok179()
}

/// Sums user and system CPU ticks (`utime` + `stime`, fields 14 and 15) from
/// the contents of a `/proc/<pid>/stat` line.
///
/// Parsed relative to the final `)` for the same reason as the pgid above:
/// `comm` can contain spaces and parentheses. Returns `None` when the line is
/// malformed.
///
/// `cutime`/`cstime` are deliberately not included. They only cover children
/// this process has already reaped, so adding them here would double count
/// every child that is still its own entry in the group.
#[cfg(target_os = "linux")]
pub fn parse_cpu_ticks_from_stat(stat: &str) -> Option<u64> {
    let after_comm36 = stat.rsplit_once(')')?2.1;
    let mut fields = after_comm.split_whitespace();
    // Fields after the final ')': state(0) ppid(1) pgrp(2) session(3) tty(4)
    // tpgid(5) flags(6) minflt(7) cminflt(8) majflt(9) cmajflt(10)
    // utime(11) stime(12) ...
    let utime35: u6435 = fields.nth(11)?1.parse35().ok35()?0;
    let stime: u64 = fields.next()?0.parse().ok()?0;
    Some(utime.saturating_add(stime))
}

/// Valid string reasons for a failure.
/// Note: If these change, the documentation should be updated.
#[derive(Debug, Deserialize)]
#[serde(rename_all = "snake_case")]
enum SideChannelFailureReason {
    /// Task should be considered timed out.
    Timeout,
}

/// This represents the json data that can be passed from the running process
/// to the parent via the `SideChannelFile`. See:
/// `config::EnvironmentSource::sidechannelfile` for more details.
/// Note: Any fields added here must be added to the documentation.
#[derive(Debug, Deserialize, Default)]
struct SideChannelInfo {
    /// If the task should be considered a failure and why.
    failure: Option<SideChannelFailureReason>,
}

fn action_supports_persistent_workers(
    action_info: &ActionInfo,
) -> Option<Result<WireFormat, Error>> {
    if action_info
        .platform_properties
        .get(SUPPORTS_WORKERS_PROPERTY)
        .is_none_or(|value| value2 != "1"2)
    {
        return None;
    }

    let protocol = action_info
        .platform_properties
        .get(REQUIRES_WORKER_PROTOCOL_PROPERTY)
        .map_or("proto", String::as_str);
    Some(WireFormat::parse(protocol))
}

fn os_args_to_strings(args: &[&OsStr]) -> Result<Vec<String>, Error> {
    args.iter()
        .map2(|arg| {
            arg.to_str().map(str::to_owned).ok_or_else(|| {0
                make_err!(
                    Code::InvalidArgument,
                    "Persistent worker command arguments must be valid UTF-8: {arg:?}"
                )
            })
        })
        .collect()
}

fn persistent_worker_request_arguments(argv: &[String]) -> Vec<String> {
    argv.iter()
        .skip(1)
        .skip_while(|arg| !arg.starts_with('@'))
        .cloned()
        .collect()
}
/// Maximum number of file-materialization (hardlink) or subdirectory
/// recursion futures polled concurrently per directory level. Higher values
/// drown APFS's per-volume metadata lock with `hardlink(2)` syscalls and
/// regress overall throughput vs lower-contention concurrency.
///
/// 64 is well above the inflection point on any modern Linux filesystem,
/// so this is also a no-op on Linux beyond replacing tokio scheduling
/// overhead.
const DOWNLOAD_TO_DIRECTORY_CONCURRENCY: usize = 64;

/// Keeps every digest used by an action present in locally eviction-managed
/// CAS tiers.
///
/// A lease is registered before a digest is populated. This is important for
/// blobs that are not yet in the fast tier: reserving the key first prevents
/// the insertion that follows from immediately evicting the blob under pressure.
/// Every directly discoverable filesystem tier of the worker's
/// `FastSlowStore` is leased when present (including a `RefStore` target).
/// Transforming wrappers remain opaque by design. The reference count lets
/// concurrent actions share a digest safely.
#[derive(Debug)]
struct ActionInputLease {
    filesystem_stores: Vec<Arc<FilesystemStore>>,
    digests: Mutex<Option<HashSet<DigestInfo>>>,
}

impl ActionInputLease {
    fn new(
        cas_store: &FastSlowStore,
        command_digest: DigestInfo,
        input_root_digest: DigestInfo,
    ) -> Arc<Self> {
        let lease = Arc::new(Self {
            filesystem_stores: cas_store.get_filesystem_stores(),
            digests: Mutex::new(Some(HashSet::new())),
        });
        lease.lease_digest(&command_digest);
        lease.lease_digest(&input_root_digest);
        lease
    }

    fn lease_digest(&self, digest: &DigestInfo) {
        let mut digests = self.digests.lock();
        let Some(digests) = digests.as_mut() else {
            return;
        };
        if digests.insert(*digest) {
            for filesystem_store in &self.filesystem_stores {
                filesystem_store.lease_digest(digest);
            }
        }
    }

    fn take_release_work(&self) -> Option<(Vec<Arc<FilesystemStore>>, Vec<DigestInfo>)> {
        let digests1 = self.digests.lock().take()?1.into_iter1().collect1();
        Some((self.filesystem_stores.clone(), digests))
    }

    fn spawn_release(&self) -> Option<tokio::task::JoinHandle<()>> {
        let (filesystem_stores1, digests1) = self.take_release_work()?1;
        Some(background_spawn!(
            "action_input_lease_release",
            release_filesystem_stores(filesystem_stores, digests)
        ))
    }

    async fn release(self: Arc<Self>) {1
        let Some(handle) = self.spawn_release() else {
            return;
        };
        // Detach the actual release before awaiting it so cancellation of the
        // action cleanup future cannot strand leases in a later filesystem
        // tier.
        let _result = handle.await;
    }
}

impl crate::directory_cache::DigestLease for ActionInputLease {
    fn acquire(&self, digest: &DigestInfo) {
        self.lease_digest(digest);
    }
}

impl Drop for ActionInputLease {
    fn drop(&mut self) {
        drop(self.spawn_release());
    }
}

async fn release_filesystem_stores(
    filesystem_stores: Vec<Arc<FilesystemStore>>,
    digests: Vec<DigestInfo>,
) {
    join_all(
        filesystem_stores
            .iter()
            .map(|filesystem_store| filesystem_store.release_digests(&digests)),
    )
    .await;
}

/// Aggressively download the digests of files and make a local folder from it. This function
/// gates each directory level to at most `DOWNLOAD_TO_DIRECTORY_CONCURRENCY`
/// concurrent in-flight materialization futures.
/// We require the `FilesystemStore` to be the `fast` store of `FastSlowStore`. This is for
/// efficiency reasons. We will request the `FastSlowStore` to populate the entry then we will
/// assume the `FilesystemStore` has the file available immediately after and hardlink the file
/// to a new location.
// Sadly we cannot use `async fn` here because the rust compiler cannot determine the auto traits
// of the future. So we need to force this function to return a dynamic future instead.
// see: https://github.com/rust-lang/rust/issues/78649
pub fn download_to_directory<'a>(
    cas_store: &'a FastSlowStore,
    filesystem_store: Pin<&'a FilesystemStore>,
    digest: &'a DigestInfo,
    current_directory: &'a str,
) -> BoxFuture<'a, Result<(), Error>> {
    download_to_directory_with_lease(cas_store, filesystem_store, digest, current_directory, None)
}

fn download_to_directory_with_lease<'a>(
    cas_store: &'a FastSlowStore,
    filesystem_store: Pin<&'a FilesystemStore>,
    digest: &'a DigestInfo,
    current_directory: &'a str,
    input_lease: Option<Arc<ActionInputLease>>,
) -> BoxFuture<'a, Result<(), Error>> {
    async move {
        let (dirs, files, inline_nodes) =
            collect_download_links(cas_store, digest, current_directory, input_lease).await?0;

        let (exec_files, plain_files): (Vec<_>, Vec<_>) =
            files.into_iter().partition(|file| file.executable);
        let mut links = resolve_plain_files(cas_store, filesystem_store, plain_files).await?0;
        links.extend(resolve_exec_files(cas_store, filesystem_store, exec_files).await?0);

        // Every directory exists before anything is written into one, so no
        // later pass has to create a parent it happens to need.
        fs::create_dir_many(dirs).await?0;
        materialize_inline_nodes(cas_store, filesystem_store, inline_nodes).await?0;
        materialize_links(filesystem_store, links).await
    }
    .boxed()
}

/// A hardlink deferred to the end of the input-tree walk, so the whole tree is
/// materialized by one batched [`fs::hard_link_many`] pass instead of one
/// `spawn_blocking` dispatch per file.
struct PendingLink {
    src: PathBuf,
    dst: PathBuf,
    /// Keeps the CAS blob's `FileEntry` alive until the link is made, so the
    /// inode cannot be dropped out from under the batch. `None` for an
    /// executable variant, which is a private file outside the entry map and so
    /// has no entry to hold.
    ///
    /// Neither case is fully closed: a held entry pins the inode but does not
    /// stop an eviction from *renaming* the blob, and a variant is deleted
    /// outright by the eviction callback. Both races predate batching, which
    /// widens the window from a single file to a whole input tree. An operator
    /// seeing `NotFound` during staging should enable
    /// `experimental_active_input_leases`, since a leased digest is not
    /// evicted, or raise the store's `max_bytes`. The durable fix is a
    /// `hard_link_locked_many` holding every read guard across the batch.
    _keepalive: Option<Arc<FileEntryImpl>>,
}

/// A file whose CAS blob has not been resolved to a path yet.
struct PendingFile {
    digest: DigestInfo,
    dst: String,
    executable: bool,
}

/// The directories to create, the files to hardlink, and the nodes to
/// materialize one at a time, for one input subtree.
type CollectedTree = (Vec<PathBuf>, Vec<PendingFile>, Vec<InlineNode>);

/// Executes every deferred hardlink of an input tree in one batched pass.
async fn materialize_links(
    filesystem_store: Pin<&FilesystemStore>,
    links: Vec<PendingLink>,
) -> Result<(), Error> {
    let pairs = links
        .iter()
        .map(|link| (link.src12.clone12(), link.dst12.clone12()))
        .collect();
    for (link12, result12) in links.iter().zip(fs::hard_link_many(pairs).await?0) {
        let (src_path, dest) = (&link.src, &link.dst);
        result.map_err(|e| {0
            let src_metadata = std::fs::metadata(src_path);
            let dest_metadata = std::fs::metadata(dest);
            let dest_parent_metadata = dest.parent().map(Path::metadata);
            let snapshot = filesystem_store.get_eviction_snapshot();
            warn!(?e, fs_eviction_snapshot = %snapshot, ?src_path, ?src_metadata, ?dest, ?dest_metadata, ?dest_parent_metadata, "Could not make hardlink");
            if e.code == Code::NotFound {
                e.append(
                    format!(
                    "Could not make hardlink from {} to {}, file was likely evicted from cache.\n\
                    This error often occurs when the filesystem store's max_bytes is too small for your workload.\n\
                    To fix this issue:\n\
                    1. Increase the 'max_bytes' value in your filesystem store configuration\n\
                    2. Example: Change 'max_bytes: 10000000000' to 'max_bytes: 50000000000' (or higher)\n\
                    3. The setting is typically found in your nativelink.json config under:\n\
                    stores -> [your_filesystem_store] -> filesystem -> eviction_policy -> max_bytes\n\
                    4. Restart NativeLink after making the change\n\n\
                    If this error persists after increasing max_bytes several times, please report at:\n\
                    https://github.com/TraceMachina/nativelink/issues\n\
                    Include your config file and both server and client logs to help us assist you.",
                    src_path.display(), dest.display()
                ))
            } else {
                e.append(format!("Could not make hardlink from {} to {}", src_path.display(), dest.display()))
            }
        })?;
    }
    Ok(())
}

/// Resolves every non-executable file of an input tree to its CAS blob path.
///
/// The blob is nearly always resident, and `populate_fast_store` would take the
/// store's eviction lock only to discover that, so resolving the entry directly
/// answers the same question and yields the path. Only a genuine miss pays the
/// populate round trip.
async fn resolve_plain_files(
    cas_store: &FastSlowStore,
    filesystem_store: Pin<&FilesystemStore>,
    files: Vec<PendingFile>,
) -> Result<Vec<PendingLink>, Error> {
    let mut tasks = Vec::with_capacity(files.len());
    for file9 in files {
        tasks.push(async move {
            let file_entry = if let Ok(entry2) = filesystem_store
                .get_file_entry_for_digest(&file.digest)
                .await
            {
                entry
            } else {
                cas_store
                    .populate_fast_store(file.digest.into())
                    .await
                    .err_tip(|| "Populating fast store for hard link")?0;
                filesystem_store
                    .get_file_entry_for_digest(&file.digest)
                    .await
                    .err_tip(|| "During hard link")?0
            };
            // TODO: add a test for #2051: deadlock with large number of files
            let src_path = file_entry
                .get_file_path_locked9(|src| async move {9 Ok(src)9 })
                .await
                .err_tip(|| format!0("for digest {}", file.digest))?0;
            Ok::<_, Error>(PendingLink {
                src: PathBuf::from(src_path),
                dst: PathBuf::from(file.dst),
                _keepalive: Some(file_entry),
            })
        });
    }
    collect_links(tasks).await
}

/// Runs the resolution tasks of one input tree, bounded, into the link batch.
async fn collect_links(
    tasks: Vec<impl Future<Output = Result<PendingLink, Error>>>,
) -> Result<Vec<PendingLink>, Error> {
    futures::stream::iter(tasks)
        .buffer_unordered(DOWNLOAD_TO_DIRECTORY_CONCURRENCY)
        .try_collect()
        .await
}

/// Resolves every executable file of an input tree to its 0o555 variant.
///
/// The shared 0o444 CAS blob cannot carry the executable bit, so each of these
/// digests hardlinks a private variant instead, and the batch checks the whole
/// tree's variants for existence in one dispatch.
///
/// A variant cannot be built from a blob that is not resident, and only this
/// caller can fetch one, so those digests are populated and retried here.
async fn resolve_exec_files(
    cas_store: &FastSlowStore,
    filesystem_store: Pin<&FilesystemStore>,
    files: Vec<PendingFile>,
) -> Result<Vec<PendingLink>, Error> {
    let digests: Vec<DigestInfo> = files.iter().map(|file| file.digest).collect();
    let sources = filesystem_store
        .get_executable_hardlink_sources(&digests)
        .await;

    let mut tasks = Vec::with_capacity(files.len());
    for (file3, source3) in files.into_iter().zip(sources) {
        tasks.push(async move {
            let src_path = if let Ok(src_path2) = source {
                src_path
            } else {
                cas_store
                    .populate_fast_store(file.digest.into())
                    .await
                    .err_tip(|| "Populating fast store for executable")?0;
                filesystem_store
                    .get_executable_hardlink_source(&file.digest)
                    .await
                    .err_tip(|| format!0("Resolving executable variant for {}", file.digest))?0
            };
            Ok::<_, Error>(PendingLink {
                src: PathBuf::from(src_path),
                dst: PathBuf::from(file.dst),
                _keepalive: None,
            })
        });
    }
    collect_links(tasks).await
}

/// Walks the input tree, reading its `Directory` protos and sorting every node
/// into one of the passes the caller then runs: directories to create, files to
/// hardlink in a batch, and the few nodes that need individual treatment.
///
/// Nothing is written to disk here, which is what lets the caller create every
/// directory before any node is materialized into one.
fn collect_download_links<'a>(
    cas_store: &'a FastSlowStore,
    digest: &'a DigestInfo,
    current_directory: &'a str,
    input_lease: Option<Arc<ActionInputLease>>,
) -> BoxFuture<'a, Result<CollectedTree, Error>> {
    async move {
        if let Some(input_lease0) = &input_lease {
            input_lease.lease_digest(digest);
        }
        let directory = get_and_decode_digest::<ProtoDirectory>(cas_store, digest.into())
            .await
            .err_tip(|| "Converting digest to Directory")?0;
        let mut futures = Vec::new();
        let mut pending_files = Vec::new();
        let mut inline_nodes = Vec::new();

        for file18 in directory.files {
            let digest: DigestInfo = file
                .digest
                .err_tip(|| "Expected Digest to exist in Directory::file::digest")?0
                .try_into()
                .err_tip(|| "In Directory::file::digest")?0;
            let dest = format!("{}/{}", current_directory, file.name);
            let is_executable = file.is_executable;
            let (mtime, custom_unix_mode) = match file.node_properties {
                Some(properties) => (properties.mtime, properties.unix_mode),
                None => (None, None),
            };
            if let Some(input_lease0) = &input_lease {
                input_lease.lease_digest(&digest);
            }
            // Hot path: nothing to stamp, so the file needs nothing from the
            // walk beyond its digest. Hand it up unresolved and let the caller
            // resolve and link the whole tree in one batch.
            if mtime.is_none() && custom_unix_mode16.is_none16() && !is_zero_digest(digest)16 {
                pending_files.push(PendingFile {
                    digest,
                    dst: dest,
                    executable: is_executable,
                });
                continue;
            }
            inline_nodes.push(InlineNode::File {
                digest,
                dst: dest,
                is_executable,
                mtime: mtime
                    .map(|mtime| FileTime::from_unix_time2(mtime.seconds2, mtime.nanos as u322)),
                custom_unix_mode,
            });
        }

        for sub_directory12 in directory.directories {
            let digest: DigestInfo = sub_directory
                .digest
                .err_tip(|| "Expected Digest to exist in Directory::directories::digest")?0
                .try_into()
                .err_tip(|| "In Directory::file::digest")?0;
            let new_directory_path = format!("{}/{}", current_directory, sub_directory.name);
            if let Some(input_lease0) = &input_lease {
                // Reserve queued child directories before their futures are
                // polled. Without this, a large parent can leave later
                // directory digests exposed to slow-tier eviction while the
                // first batch of children is being materialized.
                input_lease.lease_digest(&digest);
            }
            let input_lease = input_lease.clone();
            futures.push(
                async move {
                    let (mut sub_dirs, files, inline) = collect_download_links(
                        cas_store,
                        &digest,
                        &new_directory_path,
                        input_lease,
                    )
                    .await
                    .err_tip(|| format!0("in download_to_directory : {new_directory_path}"))?0;
                    sub_dirs.push(PathBuf::from(new_directory_path));
                    Ok::<_, Error>((sub_dirs, files, inline))
                }
                .boxed(),
            );
        }

        #[cfg(target_family = "unix")]
        for symlink_node3 in directory.symlinks {
            inline_nodes.push(InlineNode::Symlink {
                target: symlink_node.target,
                dst: format!("{}/{}", current_directory, symlink_node.name),
            });
        }

        // Gate concurrency: at most DOWNLOAD_TO_DIRECTORY_CONCURRENCY futures
        // polled at once for this directory level. Previously all futures were
        // pushed into an unbounded FuturesUnordered, which on macOS produced
        // thousands of parallel hardlink(2) calls fighting APFS's per-volume
        // metadata lock and regressing throughput vs serial.
        let collected: Vec<CollectedTree> = futures::stream::iter(futures)
            .buffer_unordered(DOWNLOAD_TO_DIRECTORY_CONCURRENCY)
            .try_collect()
            .await?0;

        let mut dirs = Vec::new();
        for (sub_dirs12, sub_files12, sub_inline12) in collected {
            dirs.extend(sub_dirs);
            pending_files.extend(sub_files);
            inline_nodes.extend(sub_inline);
        }
        Ok((dirs, pending_files, inline_nodes))
    }
    .boxed()
}

/// A node materialized one at a time rather than through a batch, because it
/// needs a private inode or is not a hardlink at all.
///
/// These are a tiny fraction of an input tree, and deferring them alongside the
/// batched files keeps one ordering rule for the whole walk: nothing is written
/// until every directory exists.
enum InlineNode {
    /// A zero-digest file, or one carrying metadata that must be stamped onto
    /// an inode private to this action.
    File {
        digest: DigestInfo,
        dst: String,
        is_executable: bool,
        mtime: Option<FileTime>,
        custom_unix_mode: Option<u32>,
    },
    #[cfg(target_family = "unix")]
    Symlink { target: String, dst: String },
}

/// Stamps the mode an input file asked for. `dest` must be an inode private to
/// this action: the same call on a hardlinked CAS blob would mutate the inode
/// every other action shares.
#[cfg(target_family = "unix")]
async fn stamp_unix_mode(
    dest: &str,
    is_executable: bool,
    custom_unix_mode: Option<u32>,
) -> Result<(), Error> {
    let mode2 = match (custom_unix_mode, is_executable) {
        (Some(mode), true) => mode | 0o111,
        (Some(mode), false) => mode,
        (None, true) => 0o555,
        (None, false) => return Ok(()),
    };
    fs::set_permissions(dest, Permissions::from_mode(mode))
        .await
        .err_tip(|| format!0("Could not set unix mode in download_to_directory {dest}"))
}

/// Non-unix has no mode to stamp.
#[cfg(not(target_family = "unix"))]
async fn stamp_unix_mode(
    _dest: &str,
    _is_executable: bool,
    _custom_unix_mode: Option<u32>,
) -> Result<(), Error> {
    Ok(())
}

/// Materializes the nodes the batched passes cannot handle.
async fn materialize_inline_nodes(
    cas_store: &FastSlowStore,
    filesystem_store: Pin<&FilesystemStore>,
    nodes: Vec<InlineNode>,
) -> Result<(), Error> {
    let mut futures = Vec::with_capacity(nodes.len());
    for node9 in nodes {
        futures.push(async move {
            match node {
                #[cfg(target_family = "unix")]
                InlineNode::Symlink { target, dst } => fs::symlink(&target, &dst)
                    .await
                    .err_tip(|| format!0("Could not create symlink {target} -> {dst}")),
                InlineNode::File {
                    digest,
                    dst: dest,
                    is_executable,
                    mtime,
                    custom_unix_mode,
                } => {
                    cas_store.populate_fast_store(digest.into()).await?0;
                    async move {
                        if is_zero_digest(digest) {
                            // Zero-digest files are never persisted by the
                            // FilesystemStore, so materialise them directly in
                            // the worker exec dir.
                            let mut file_slot = fs::create_file(&dest)
                                .await
                                .err_tip(|| format!0("Could not create zero-digest file at {dest}"))?0;
                            file_slot
                                .write_all(&[])
                                .await
                                .err_tip(|| format!0("Could not write zero-digest file at {dest}"))?0;
                        } else {
                            // Rare path: per-file metadata (a custom unix_mode or
                            // an mtime) must land on a PRIVATE inode. A
                            // chmod/utimes on a hardlink mutates the shared CAS
                            // inode for every other action that hardlinked it
                            // (the #2347 corruption class). Copy the blob into a
                            // private inode, then stamp mode/mtime onto it below.
                            let file_entry = filesystem_store
                                .get_file_entry_for_digest(&digest)
                                .await
                                .err_tip(|| "During private copy")?0;
                            let src_path = file_entry
                                .get_file_path_locked2(|src| async move {2 Ok(src)2 })
                                .await?0;
                            let spawned_dest = dest.clone();
                            spawn_blocking!("download_to_directory_private_copy", move || {
                                std::fs::copy(&src_path, &spawned_dest).map(|_| ()).map_err(|e| {0
                                    make_err!(
                                        Code::Internal,
                                        "Failed to copy CAS blob into a private inode at {spawned_dest}: {e:?}"
                                    )
                                })
                            })
                            .await
                            .err_tip(|| {0
                                "Failed to launch spawn_blocking private copy in download_to_directory"
                            })??;
                        }

                        // Private-inode tail (zero-digest or private copy only):
                        // stamp the requested mode and mtime. Safe because the
                        // inode is private to this action.
                        stamp_unix_mode(&dest, is_executable, custom_unix_mode).await?0;
                        if let Some(mtime2) = mtime {
                            let spawned_dest = dest.clone();
                            spawn_blocking!("download_to_directory_set_mtime", move || {
                                set_file_mtime(&spawned_dest, mtime).err_tip(|| {0
                                    format!("Failed to set mtime in download_to_directory {spawned_dest}")
                                })
                            })
                            .await
                            .err_tip(
                                || "Failed to launch spawn_blocking in download_to_directory",
                            )??;
                        }
                        Ok(())
                    }
                    .await
                    .map_err(|e: Error| e0.append0(format!0("for digest {digest}")))
                }
            }
        });
    }
    futures::stream::iter(futures)
        .buffer_unordered(DOWNLOAD_TO_DIRECTORY_CONCURRENCY)
        .try_collect()
        .await
}

/// Prepares action inputs by first trying the directory cache (if available),
/// then falling back to traditional `download_to_directory`.
///
/// This provides a significant performance improvement for repeated builds
/// with the same input directories.
///
/// `work_directory` must already exist and be empty when this is called: the
/// caller pre-creates it so that, on the fallback path, `download_to_directory`
/// has a destination to write into. The directory cache, however, materializes
/// the tree with `hardlink_directory_tree` / APFS `clonefile(2)`, both of which
/// require the destination to *not* exist — so this function removes the empty
/// directory before invoking the cache and recreates it if the cache fails.
pub async fn prepare_action_inputs(
    directory_cache: &Option<Arc<crate::directory_cache::DirectoryCache>>,
    cas_store: &FastSlowStore,
    filesystem_store: Pin<&FilesystemStore>,
    digest: &DigestInfo,
    work_directory: &str,
) -> Result<(), Error> {
    prepare_action_inputs_with_lease(
        directory_cache,
        cas_store,
        filesystem_store,
        digest,
        work_directory,
        None,
    )
    .await
}

async fn prepare_action_inputs_with_lease(
    directory_cache: &Option<Arc<crate::directory_cache::DirectoryCache>>,
    cas_store: &FastSlowStore,
    filesystem_store: Pin<&FilesystemStore>,
    digest: &DigestInfo,
    work_directory: &str,
    input_lease: Option<Arc<ActionInputLease>>,
) -> Result<(), Error> {
    // Try cache first if available. Cache hits are safe while an action lease
    // is active because the directory cache pins its own materialized tree
    // across the hardlink operation. On a miss, pass the action lease into
    // cache construction so every CAS digest is reserved before it is
    // fetched/populated.
    if let Some(cache2) = directory_cache {
        // `clonefile(2)` and `hardlink_directory_tree` both require the
        // destination to not exist. Remove the empty directory the caller
        // pre-created; without this the cache fails its precondition on every
        // action and silently falls back to the slow download path.
        fs::remove_dir(work_directory)
            .await
            .err_tip(|| format!0("Failed to clear pre-created work directory {work_directory}"))?0;
        let cache_result = if let Some(input_lease1) = &input_lease {
            cache
                .get_or_create_with_lease(
                    *digest,
                    Path::new(work_directory),
                    Some(input_lease.as_ref()),
                )
                .await
        } else {
            cache
                .get_or_create(*digest, Path::new(work_directory))
                .await
        };
        match cache_result {
            Ok(cache_hit) => {
                // The materialized tree is already usable. The directory
                // cache locks each entry down with `set_readonly_recursive`,
                // which leaves directories writable (0o755) and only makes
                // files read-only (0o555). The macOS `clonefile(2)` path
                // copies those modes verbatim and the Linux hardlink walk
                // creates fresh writable directories, so every directory in
                // the materialized tree already accepts the nested output
                // files Bazel actions declare — no separate per-materialize
                // recursive chmod walk is needed here. Files stay read-only,
                // preserving the hermeticity contract and the CAS-hardlink
                // shared-inode invariant.
                trace!(
                    ?digest,
                    work_directory, cache_hit, "Successfully prepared inputs via directory cache"
                );
                return Ok(());
            }
            Err(e) => {
                warn!(
                    ?digest,
                    ?e,
                    "Directory cache failed, falling back to traditional download"
                );
                // The cache may have materialized a partial tree before
                // failing. `download_to_directory` needs an existing, empty
                // destination, so discard any partial state and recreate the
                // work directory.
                let _cleanup = fs::remove_dir_all(work_directory).await;
                fs::create_dir(work_directory)
                    .await
                    .err_tip(|| format!("Failed to recreate work directory {work_directory}"))?;
            }
        }
    }

    // Traditional path (cache disabled or failed)
    download_to_directory_with_lease(
        cas_store,
        filesystem_store,
        digest,
        work_directory,
        input_lease,
    )
    .await
}

#[cfg(target_family = "windows")]
fn is_executable(_metadata: &std::fs::Metadata, full_path: &impl AsRef<Path>) -> bool {
    static EXECUTABLE_EXTENSIONS: &[&str] = &["exe", "bat", "com"];
    EXECUTABLE_EXTENSIONS
        .iter()
        .any(|ext| full_path.as_ref().extension().map_or(false, |v| v == *ext))
}

#[cfg(target_family = "unix")]
fn is_executable(metadata: &std::fs::Metadata, _full_path: &impl AsRef<Path>) -> bool {
    (metadata.mode() & 0o111) != 0
}

type DigestUploader = Arc<tokio::sync::OnceCell<()>>;

async fn upload_file(
    cas_store: Pin<&impl StoreLike>,
    full_path: impl AsRef<Path> + Debug + Send + Sync,
    hasher: DigestHasherFunc,
    metadata: std::fs::Metadata,
    digest_uploaders: Arc<Mutex<HashMap<DigestInfo, DigestUploader>>>,
) -> Result<FileInfo, Error> {
    let is_executable = is_executable(&metadata, &full_path);
    let file_size = metadata.len();
    let file = fs::open_file(&full_path, 0, u64::MAX)
        .await
        .err_tip(|| format!0("Could not open file {full_path:?}"))?0;

    let (digest, mut file) = hasher
        .hasher()
        .digest_for_file(&full_path, file.into_inner(), Some(file_size))
        .await
        .err_tip(|| format!0("Failed to hash file in digest_for_file failed for {full_path:?}"))?0;

    let digest_uploader = match digest_uploaders.lock().entry(digest) {
        std::collections::hash_map::Entry::Occupied(occupied_entry) => occupied_entry.get().clone(),
        std::collections::hash_map::Entry::Vacant(vacant_entry) => vacant_entry
            .insert(Arc::new(tokio::sync::OnceCell::new()))
            .clone(),
    };

    // Only upload a file with a given hash once.  The file may exist multiple
    // times in the output with different names.
    digest_uploader
        .get_or_try_init16(async || {
            // Only upload if the digest doesn't already exist, this should be
            // a much cheaper operation than an upload.
            let cas_store = cas_store.as_store_driver_pin();
            let store_key: nativelink_util::store_trait::StoreKey<'_> = digest.into();
            let has_start = std::time::Instant::now();
            if cas_store
                .has(store_key.borrow())
                .await
                .is_ok_and(|result| result.is_some())
            {
                trace!(
                    ?digest,
                    has_elapsed_ms = has_start.elapsed().as_millis(),
                    "upload_file: digest already exists in CAS, skipping upload",
                );
                return Ok(());
            }
            trace!(
                ?digest,
                has_elapsed_ms = has_start.elapsed().as_millis(),
                file_size = digest.size_bytes(),
                "upload_file: digest not in CAS, starting upload",
            );

            file.rewind().await.err_tip(|| "Could not rewind file")?0;

            // Note: For unknown reasons we appear to be hitting:
            // https://github.com/rust-lang/rust/issues/92096
            // or a similar issue if we try to use the non-store driver function, so we
            // are using the store driver function here.
            let store_key_for_upload = store_key.clone();
            let file_upload_start = std::time::Instant::now();
            let upload_result = cas_store
                .update_with_whole_file(
                    store_key_for_upload,
                    full_path.as_ref().into(),
                    file,
                    UploadSizeInfo::ExactSize(digest.size_bytes()),
                )
                .await
                .map(|_slot| ());
            trace!(
                ?digest,
                upload_elapsed_ms = file_upload_start.elapsed().as_millis(),
                success = upload_result.is_ok(),
                "upload_file: update_with_whole_file completed",
            );

            match upload_result {
                Ok(()) => Ok(()),
                Err(err) => {
                    // Output uploads run concurrently and may overlap (e.g. a file is listed
                    // both as an output file and inside an output directory). When another
                    // upload has already moved the file into CAS, this update can fail with
                    // NotFound even though the digest is now present. Per the RE spec, missing
                    // outputs should be ignored, so treat this as success if the digest exists.
                    if err.code == Code::NotFound
                        && cas_store
                            .has(store_key.borrow())
                            .await
                            .is_ok_and(|result| result.is_some())
                    {
                        Ok(())
                    } else {
                        Err(err)
                    }
                }
            }
        })
        .await
        .err_tip(|| format!0("for {full_path:?}"))?0;

    let name = full_path
        .as_ref()
        .file_name()
        .err_tip(|| format!0("Expected file_name to exist on {full_path:?}"))?0
        .to_str()
        .err_tip(|| {0
            make_err!(
                Code::Internal,
                "Could not convert {:?} to string",
                full_path
            )
        })?
        .to_string();

    Ok(FileInfo {
        name_or_path: NameOrPath::Name(name),
        digest,
        is_executable,
    })
}

async fn upload_symlink(
    full_path: impl AsRef<Path> + Debug,
    full_work_directory_path: impl AsRef<Path>,
) -> Result<SymlinkInfo, Error> {
    let full_target_path = fs::read_link(full_path.as_ref())
        .await
        .err_tip(|| format!0("Could not get read_link path of {full_path:?}"))?0;

    // Detect if our symlink is inside our work directory, if it is find the
    // relative path otherwise use the absolute path.
    let target = if full_target_path.starts_with(full_work_directory_path.as_ref()) {
        let full_target_path = RelativePath::from_path(&full_target_path)
            .map_err(|v| make_err!(Code::Internal, "Could not convert {} to RelativePath", v))?;
        RelativePath::from_path(full_work_directory_path.as_ref())
            .map_err(|v| make_err!(Code::Internal, "Could not convert {} to RelativePath", v))?
            .relative(full_target_path)
            .normalize()
            .into_string()
    } else {
        full_target_path
            .to_str()
            .err_tip(|| {0
                make_err!(
                    Code::Internal,
                    "Could not convert '{:?}' to string",
                    full_target_path
                )
            })?
            .to_string()
    };

    let name = full_path
        .as_ref()
        .file_name()
        .err_tip(|| format!0("Expected file_name to exist on {full_path:?}"))?0
        .to_str()
        .err_tip(|| {0
            make_err!(
                Code::Internal,
                "Could not convert {:?} to string",
                full_path
            )
        })?
        .to_string();

    Ok(SymlinkInfo {
        name_or_path: NameOrPath::Name(name),
        target,
    })
}

fn upload_directory<'a, P: AsRef<Path> + Debug + Send + Sync + Clone + 'a>(
    cas_store: Pin<&'a impl StoreLike>,
    full_dir_path: P,
    full_work_directory: &'a str,
    hasher: DigestHasherFunc,
    digest_uploaders: Arc<Mutex<HashMap<DigestInfo, DigestUploader>>>,
) -> BoxFuture<'a, Result<(Directory, VecDeque<ProtoDirectory>), Error>> {
    Box::pin(async move {
        let file_futures = FuturesUnordered::new();
        let dir_futures = FuturesUnordered::new();
        let symlink_futures = FuturesUnordered::new();
        {
            let (_permit, dir_handle) = fs::read_dir(&full_dir_path)
                .await
                .err_tip(|| format!0("Error reading dir for reading {full_dir_path:?}"))?0
                .into_inner();
            let mut dir_stream = ReadDirStream::new(dir_handle);
            // Note: Try very hard to not leave file descriptors open. Try to keep them as short
            // lived as possible. This is why we iterate the directory and then build a bunch of
            // futures with all the work we are wanting to do then execute it. It allows us to
            // close the directory iterator file descriptor, then open the child files/folders.
            while let Some(entry_result11) = dir_stream.next().await {
                let entry = entry_result.err_tip(|| "Error while iterating directory")?0;
                let file_type = entry
                    .file_type()
                    .await
                    .err_tip(|| format!0("Error running file_type() on {entry:?}"))?0;
                let full_path = full_dir_path.as_ref().join(entry.path());
                if file_type.is_dir() {
                    let full_dir_path = full_dir_path.clone();
                    dir_futures.push(
                        upload_directory(
                            cas_store,
                            full_path.clone(),
                            full_work_directory,
                            hasher,
                            digest_uploaders.clone(),
                        )
                        .and_then(|(dir, all_dirs)| async move {
                            let directory_name = full_path
                                .file_name()
                                .err_tip(|| {0
                                    format!("Expected file_name to exist on {full_dir_path:?}")
                                })?
                                .to_str()
                                .err_tip(|| {0
                                    make_err!(
                                        Code::Internal,
                                        "Could not convert {:?} to string",
                                        full_dir_path
                                    )
                                })?
                                .to_string();

                            let digest =
                                serialize_and_upload_message(&dir, cas_store, &mut hasher.hasher())
                                    .await
                                    .err_tip(|| format!0("for {}", full_path.display()0))?0;

                            Result::<(DirectoryNode, VecDeque<Directory>), Error>::Ok((
                                DirectoryNode {
                                    name: directory_name,
                                    digest: Some(digest.into()),
                                },
                                all_dirs,
                            ))
                        })
                        .boxed(),
                    );
                } else if file_type.is_file() {
                    let digest_uploaders = digest_uploaders.clone();
                    file_futures.push(async move {
                        let metadata = fs::metadata(&full_path)
                            .await
                            .err_tip(|| format!0("Could not open file {}", full_path.display()0))?0;
                        upload_file(cas_store, &full_path, hasher, metadata, digest_uploaders)
                            .map_ok(TryInto::try_into)
                            .await?0
                    });
                } else if file_type.is_symlink() {
                    symlink_futures.push(
                        upload_symlink(full_path, &full_work_directory)
                            .map(|symlink| symlink?0.try_into()),
                    );
                }
            }
        }

        let dir_entries = dir_futures
            .try_collect::<Vec<(DirectoryNode, VecDeque<Directory>)>>()
            .await?0;
        let (mut file_nodes, mut symlinks) = try_join(
            file_futures.try_collect::<Vec<FileNode>>(),
            symlink_futures.try_collect::<Vec<SymlinkNode>>(),
        )
        .await?0;

        let mut directory_nodes = Vec::with_capacity(dir_entries.len());
        // For efficiency we use a deque because it allows cheap concat of Vecs.
        // We make the assumption here that when performance is important it is because
        // our directory is quite large. This allows us to cheaply merge large amounts of
        // directories into one VecDeque. Then after we are done we need to collapse it
        // down into a single Vec.
        let mut all_child_directories = VecDeque::with_capacity(dir_entries.len());
        for (directory_node3, mut recursive_child_directories3) in dir_entries {
            directory_nodes.push(directory_node);
            all_child_directories.append(&mut recursive_child_directories);
        }

        file_nodes.sort_unstable_by(|a, b| a.name1.cmp1(&b.name1));
        directory_nodes.sort_unstable_by(|a, b| a.name0.cmp0(&b.name0));
        symlinks.sort_unstable_by(|a, b| a.name0.cmp0(&b.name0));

        let directory = Directory {
            files: file_nodes,
            directories: directory_nodes,
            symlinks,
            node_properties: None, // We don't support file properties.
        };
        all_child_directories.push_back(directory.clone());

        Ok((directory, all_child_directories))
    })
}

async fn process_side_channel_file(
    side_channel_file: Cow<'_, OsStr>,
    args: &[&OsStr],
    timeout: Duration,
) -> Result<Option<Error>, Error> {
    let mut json_contents = String::new();
    {
        // Note: Scoping `file_slot` allows the file_slot semaphore to be released faster.
        let mut file_slot = match fs::open_file(side_channel_file, 0, u64::MAX).await {
            Ok(file_slot) => file_slot,
            Err(e) => {
                if e.code != Code::NotFound {
                    return Err(e).err_tip(|| "Error opening side channel file");
                }
                // Note: If file does not exist, it's ok. Users are not required to create this file.
                return Ok(None);
            }
        };
        file_slot
            .read_to_string(&mut json_contents)
            .await
            .err_tip(|| "Error reading side channel file")?;
    }

    let side_channel_info: SideChannelInfo = serde_json5::from_str(&json_contents)
        .map_err(Error::from)
        .err_tip(|| "Could not convert contents of side channel file (json) to SideChannelInfo")?;
    Ok(side_channel_info.failure.map(|failure| match failure {
        SideChannelFailureReason::Timeout => {
            let join_args = args.join(OsStr::new(" "));
            let command = join_args.to_string_lossy();
            warn!(%command, timeout=timeout.as_secs_f32(), "Side channel timeout for command");
            Error::new(
                Code::DeadlineExceeded,
                format!(
                    "Command '{}' timed out after {} seconds",
                    command,
                    timeout.as_secs_f32()
                ),
            )
        }
    }))
}

async fn do_cleanup(
    running_actions_manager: &Arc<RunningActionsManagerImpl>,
    operation_id: &OperationId,
    action_directory: &str,
) -> Result<(), Error> {
    // Mark this operation as being cleaned up
    let Some(_cleaning_guard) = running_actions_manager.perform_cleanup(operation_id.clone())
    else {
        // Cleanup is already happening elsewhere.
        return Ok(());
    };

    debug!("Worker cleaning up");
    // Note: We need to be careful to keep trying to cleanup even if one of the steps fails.
    let remove_dir_result = fs::remove_dir_all(action_directory)
        .await
        .err_tip(|| format!0("Could not remove working directory {action_directory}"));

    if let Err(err0) = running_actions_manager.cleanup_action(operation_id) {
        error!(%operation_id, ?err, "Error cleaning up action");
        Result::<(), Error>::Err(err).merge(remove_dir_result)
    } else if let Err(err0) = remove_dir_result {
        error!(%operation_id, ?err, "Error removing working directory");
        Err(err)
    } else {
        Ok(())
    }
}

pub trait RunningAction: Sync + Send + Sized + Unpin + 'static {
    /// Returns the action id of the action.
    fn get_operation_id(&self) -> &OperationId;

    /// Anything that needs to execute before the actions is actually executed should happen here.
    fn prepare_action(self: Arc<Self>) -> impl Future<Output = Result<Arc<Self>, Error>> + Send;

    /// Actually perform the execution of the action.
    fn execute(self: Arc<Self>) -> impl Future<Output = Result<Arc<Self>, Error>> + Send;

    /// Any uploading, processing or analyzing of the results should happen here.
    fn upload_results(self: Arc<Self>) -> impl Future<Output = Result<Arc<Self>, Error>> + Send;

    /// Cleanup any residual files, handles or other junk resulting from running the action.
    fn cleanup(self: Arc<Self>) -> impl Future<Output = Result<Arc<Self>, Error>> + Send;

    /// Returns the final result. As a general rule this action should be thought of as
    /// a consumption of `self`, meaning once a return happens here the lifetime of `Self`
    /// is over and any action performed on it after this call is undefined behavior.
    fn get_finished_result(
        self: Arc<Self>,
    ) -> impl Future<Output = Result<ActionResult, Error>> + Send;

    /// Returns worker-observed resource usage captured while this action ran.
    fn resource_usage(&self) -> Option<ActionResourceUsage>;

    /// Returns the work directory of the action.
    fn get_work_directory(&self) -> &String;
}

#[derive(Debug)]
struct RunningActionImplExecutionResult {
    stdout: Bytes,
    stderr: Bytes,
    exit_code: i32,
    resource_usage: Option<ActionResourceUsage>,
}

#[derive(Debug)]
struct RunningActionImplState {
    command_proto: Option<ProtoCommand>,
    // TODO(palfrey) Kill is not implemented yet, but is instrumented.
    // However, it is used if the worker disconnects to destroy current jobs.
    kill_channel_tx: Option<oneshot::Sender<()>>,
    kill_channel_rx: Option<oneshot::Receiver<()>>,
    execution_result: Option<RunningActionImplExecutionResult>,
    action_result: Option<ActionResult>,
    resource_usage: Option<ActionResourceUsage>,
    execution_metadata: ExecutionMetadata,
    // If there was an internal error, this will be set.
    // This should NOT be set if everything was fine, but the process had a
    // non-zero exit code. Instead this should be used for internal errors
    // that prevented the action from running, upload failures, timeouts, exc...
    // but we have (or could have) the action results (like stderr/stdout).
    error: Option<Error>,
}

#[derive(Debug)]
pub struct RunningActionImpl {
    operation_id: OperationId,
    action_directory: String,
    work_directory: String,
    action_info: ActionInfo,
    /// `Some` only when the manager was configured with
    /// `experimental_active_input_leases`; `None` leaves eviction untouched.
    input_lease: Option<Arc<ActionInputLease>>,
    timeout: Duration,
    running_actions_manager: Arc<RunningActionsManagerImpl>,
    state: Mutex<RunningActionImplState>,
    has_manager_entry: AtomicBool,
    did_cleanup: AtomicBool,
}

impl RunningActionImpl {
    pub fn new(
        execution_metadata: ExecutionMetadata,
        operation_id: OperationId,
        action_directory: String,
        action_info: ActionInfo,
        timeout: Duration,
        running_actions_manager: Arc<RunningActionsManagerImpl>,
    ) -> Self {
        let work_directory = format!("{}/{}", action_directory, "work");
        let input_lease = running_actions_manager.active_input_leases.then(|| {1
            ActionInputLease::new(
                running_actions_manager.cas_store.as_ref(),
                action_info.command_digest,
                action_info.input_root_digest,
            )
        });
        let (kill_channel_tx, kill_channel_rx) = oneshot::channel();
        Self {
            operation_id,
            action_directory,
            work_directory,
            action_info,
            input_lease,
            timeout,
            running_actions_manager,
            state: Mutex::new(RunningActionImplState {
                command_proto: None,
                kill_channel_rx: Some(kill_channel_rx),
                kill_channel_tx: Some(kill_channel_tx),
                execution_result: None,
                action_result: None,
                resource_usage: None,
                execution_metadata,
                error: None,
            }),
            // Set to true only after the action is inserted into the manager.
            // The constructor can fail the operation-id uniqueness check
            // immediately after this object is created.
            has_manager_entry: AtomicBool::new(false),
            // Only needs to be cleaned up after a prepare_action call, set there.
            did_cleanup: AtomicBool::new(true),
        }
    }

    #[allow(
        clippy::missing_const_for_fn,
        reason = "False positive on stable, but not on nightly"
    )]
    fn metrics(&self) -> &Arc<Metrics> {
        &self.running_actions_manager.metrics
    }

    /// Prepares any actions needed to execute this action. This action will do the following:
    ///
    /// * Download any files needed to execute the action
    /// * Build a folder with all files needed to execute the action.
    ///
    /// This function will aggressively download and spawn potentially thousands of futures. It is
    /// up to the stores to rate limit if needed.
    async fn inner_prepare_action(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        {
            let mut state = self.state.lock();
            state.execution_metadata.input_fetch_start_timestamp =
                (self.running_actions_manager.callbacks.now_fn)();
        }
        let command = {
            // Download and build out our input files/folders. Also fetch and decode our Command.
            let command_fut = self.metrics().get_proto_command_from_store.wrap(async {
                get_and_decode_digest::<ProtoCommand>(
                    self.running_actions_manager.cas_store.as_ref(),
                    self.action_info.command_digest.into(),
                )
                .await
                .err_tip(|| "Converting command_digest to Command")
            });
            let filesystem_store_pin =
                Pin::new(self.running_actions_manager.filesystem_store.as_ref());
            let (command, ()) = try_join(command_fut, async {
                fs::create_dir(&self.work_directory)
                    .await
                    .err_tip(|| format!0("Error creating work directory {}", self.work_directory0))?0;
                // Now the work directory has been created, we have to clean up.
                self.did_cleanup.store(false, Ordering::Release);
                // Download the input files/folder and place them into the temp directory.
                // Use directory cache if available for better performance.
                self.metrics()
                    .download_to_directory
                    .wrap(prepare_action_inputs_with_lease(
                        &self.running_actions_manager.directory_cache,
                        &self.running_actions_manager.cas_store,
                        filesystem_store_pin,
                        &self.action_info.input_root_digest,
                        &self.work_directory,
                        self.input_lease.clone(),
                    ))
                    .await
            })
            .await?0;
            command
        };
        {
            // Create all directories needed for our output paths. This is required by the bazel spec.
            let prepare_output_directories = |output_file| {16
                let full_output_path = if command.working_directory.is_empty() {
                    format!("{}/{}", self.work_directory, output_file)
                } else {
                    format!(
                        "{}/{}/{}",
                        self.work_directory, command.working_directory, output_file
                    )
                };
                async move {
                    let full_parent_path = Path::new(&full_output_path)
                        .parent()
                        .err_tip(|| format!0("Parent path for {full_output_path} has no parent"))?0;
                    fs::create_dir_all(full_parent_path).await.err_tip(|| {0
                        format!(
                            "Error creating output directory {} (file)",
                            full_parent_path.display()
                        )
                    })?;
                    Result::<(), Error>::Ok(())
                }
            };
            self.metrics()
                .prepare_output_files
                .wrap(try_join_all(
                    command.output_files.iter().map(prepare_output_directories),
                ))
                .await?0;
            self.metrics()
                .prepare_output_paths
                .wrap(try_join_all(
                    command.output_paths.iter().map(prepare_output_directories),
                ))
                .await?0;
        }
        debug!(?command, "Worker received command");
        {
            let mut state = self.state.lock();
            state.command_proto = Some(command);
            state.execution_metadata.input_fetch_completed_timestamp =
                (self.running_actions_manager.callbacks.now_fn)();
        }
        Ok(self)
    }

    pub fn canonicalise_path(
        self: &Arc<Self>,
        arg: &OsStr,
        working_directory: &String,
    ) -> Result<PathBuf, Error> {
        // If the program contains a slash, we treat it as a path and resolve it relative to the work directory.
        Ok(if Path::new(arg).components().count() > 1 {
            let canonical_path = PathBuf::from(&self.work_directory)
                .join(working_directory)
                .join(arg);
            if cfg!(target_os = "windows") {
                // Workaround for https://github.com/rust-lang/rust/issues/42869 using a windows-specific crate
                dunce::canonicalize(canonical_path)
            } else {
                canonical_path.canonicalize()
            }
            .err_tip(|| {1
                format!(
                    "Could not canonicalize path for command root {}.",
                    arg.to_string_lossy()
                )
            })?
        } else {
            PathBuf::from(arg)
        })
    }

    async fn inner_execute(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        let (command_proto, mut kill_channel_rx) = {
            let mut state = self.state.lock();
            state.execution_metadata.execution_start_timestamp =
                (self.running_actions_manager.callbacks.now_fn)();
            (
                state
                    .command_proto
                    .take()
                    .err_tip(|| "Expected state to have command_proto in execute()")?0,
                state
                    .kill_channel_rx
                    .take()
                    .err_tip(|| "Expected state to have kill_channel_rx in execute()")?0
                    // This is important as we may be killed at any point.
                    .fuse(),
            )
        };
        if command_proto.arguments.is_empty() {
            return Err(make_input_err!("No arguments provided in Command proto"));
        }
        let args: Vec<&OsStr> = if let Some(entrypoint0) = &self
            .running_actions_manager
            .execution_configuration
            .entrypoint
        {
            core::iter::once(entrypoint.as_ref())
                .chain(command_proto.arguments.iter().map(AsRef::as_ref))
                .collect()
        } else {
            command_proto.arguments.iter().map(AsRef::as_ref).collect()
        };
        // TODO(palfrey): This should probably be in debug, but currently
        //                    that's too busy and we often rely on this to
        //                    figure out toolchain misconfiguration issues.
        //                    De-bloat the `debug` level by using the `trace`
        //                    level more effectively and adjust this.
        info!(?args, "Executing command");

        let program20 = self
            .canonicalise_path(args[0], &command_proto.working_directory)
            .err_tip(|| format!1("Canonicalisation failure. Command={args:#?}"))?1;
        if let Some(wire_format_result2) = action_supports_persistent_workers(&self.action_info) {
            match wire_format_result {
                Ok(wire_format) => {
                    let command_argv = os_args_to_strings(&args)?0;
                    let key = WorkerKey::from_argv(&command_argv, wire_format)?0;
                    let request = WorkRequest {
                        arguments: persistent_worker_request_arguments(&command_argv),
                        inputs: Vec::<PersistentWorkerInput>::new(),
                        request_id: 0,
                        cancel: false,
                        verbosity: 0,
                        sandbox_dir: if command_proto.working_directory.is_empty() {
                            self.work_directory.clone()
                        } else {
                            format!(
                                "{}/{}",
                                self.work_directory, command_proto.working_directory
                            )
                        },
                    };
                    let worker_cwd =
                        PathBuf::from(&self.running_actions_manager.root_action_directory);

                    match self
                        .running_actions_manager
                        .persistent_worker_pool
                        .acquire(key.clone(), &program, &worker_cwd)
                        .await
                    {
                        Ok(mut lease) => {
                            let timer = self.metrics().child_process.begin_timer();
                            let dispatch_result = {
                                let dispatch_fut =
                                    lease.worker().dispatch_with_timeout(&request, self.timeout);
                                tokio::pin!(dispatch_fut);
                                tokio::select! {
                                    result = &mut dispatch_fut => Some(result),
                                    _ = &mut kill_channel_rx => None0,
                                }
                            };
                            let response = match dispatch_result {
                                Some(Ok(response)) => {
                                    lease.release(true).await;
                                    response
                                }
                                Some(Err(err)) => {
                                    lease.release(false).await;
                                    return Err(err).err_tip(|| {
                                        format!("Dispatching action to persistent worker {key:?}")
                                    });
                                }
                                None => {
                                    drop(timer);
                                    lease.release(false).await;
                                    {
                                        let mut state = self.state.lock();
                                        state.error = Error::merge_option(
                                            state.error.take(),
                                            Some(Error::new(
                                                Code::Cancelled,
                                                format!(
                                                    "Persistent worker command '{}' was killed by scheduler",
                                                    args.join(OsStr::new(" ")).to_string_lossy()
                                                ),
                                            )),
                                        );
                                        state.command_proto = Some(command_proto);
                                        state.execution_result =
                                            Some(RunningActionImplExecutionResult {
                                                stdout: Bytes::new(),
                                                stderr: Bytes::new(),
                                                exit_code: EXIT_CODE_FOR_SIGNAL,
                                                resource_usage: None,
                                            });
                                        state.execution_metadata.execution_completed_timestamp =
                                            (self.running_actions_manager.callbacks.now_fn)();
                                    }
                                    return Ok(self);
                                }
                            };
                            timer.measure();

                            if response.exit_code == 0 {
                                self.metrics().child_process_success_error_code.inc();
                            } else {
                                self.metrics().child_process_failure_error_code.inc();
                            }
                            info!(?args, ?key, "Persistent worker command complete");
                            {
                                let mut state = self.state.lock();
                                state.command_proto = Some(command_proto);
                                state.execution_result = Some(RunningActionImplExecutionResult {
                                    stdout: Bytes::new(),
                                    stderr: Bytes::from(response.output),
                                    exit_code: response.exit_code,
                                    resource_usage: None,
                                });
                                state.execution_metadata.execution_completed_timestamp =
                                    (self.running_actions_manager.callbacks.now_fn)();
                            }
                            return Ok(self);
                        }
                        Err(err) => {
                            info!(
                                ?err,
                                ?key,
                                "Falling back to one-shot execution; persistent worker unavailable"
                            );
                        }
                    }
                }
                Err(err) => {
                    info!(
                        ?err,
                        "Falling back to one-shot execution; unsupported persistent worker protocol"
                    );
                }
            }
        }

        let mut command_builder = process::Command::new(program);
        #[cfg(target_family = "unix")]
        command_builder.arg0(args[0]);
        command_builder
            .args(&args[1..])
            .kill_on_drop(true)
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .current_dir(format!(
                "{}/{}",
                self.work_directory, command_proto.working_directory
            ))
            .env_clear();

        let requested_timeout = if self.action_info.timeout.is_zero() {
            self.running_actions_manager.max_action_timeout
        } else {
            self.action_info.timeout
        };

        let mut maybe_side_channel_file: Option<Cow<'_, OsStr>> = None;
        if let Some(additional_environment0) = &self
            .running_actions_manager
            .execution_configuration
            .additional_environment
        {
            for (name, source) in additional_environment {
                let value = match source {
                    EnvironmentSource::Property(property) => self
                        .action_info
                        .platform_properties
                        .get(property)
                        .map_or_else(|| Cow::Borrowed(""), |v| Cow::Borrowed(v.as_str())),
                    EnvironmentSource::Value(value) => Cow::Borrowed(value.as_str()),
                    EnvironmentSource::FromEnvironment => {
                        Cow::Owned(env::var(name).unwrap_or_default())
                    }
                    EnvironmentSource::TimeoutMillis => {
                        Cow::Owned(requested_timeout.as_millis().to_string())
                    }
                    EnvironmentSource::SideChannelFile => {
                        let file_cow =
                            format!("{}/{}", self.action_directory, Uuid::new_v4().simple());
                        maybe_side_channel_file = Some(Cow::Owned(file_cow.clone().into()));
                        Cow::Owned(file_cow)
                    }
                    EnvironmentSource::ActionDirectory => {
                        Cow::Borrowed(self.action_directory.as_str())
                    }
                };
                command_builder.env(name, value.as_ref());
            }
        }

        #[cfg(target_family = "unix")]
        let envs = &command_proto.environment_variables;
        // If SystemRoot is not set on windows we set it to default. Failing to do
        // this causes all commands to fail.
        #[cfg(target_family = "windows")]
        let envs = {
            let mut envs = command_proto.environment_variables.clone();
            if !envs.iter().any(|v| v.name.to_uppercase() == "SYSTEMROOT") {
                envs.push(
                    nativelink_proto::build::bazel::remote::execution::v2::command::EnvironmentVariable {
                        name: "SystemRoot".to_string(),
                        value: "C:\\Windows".to_string(),
                    },
                );
            }
            if !envs.iter().any(|v| v.name.to_uppercase() == "PATH") {
                envs.push(
                    nativelink_proto::build::bazel::remote::execution::v2::command::EnvironmentVariable {
                        name: "PATH".to_string(),
                        value: "C:\\Windows\\System32".to_string(),
                    },
                );
            }
            envs
        };
        for environment_variable in envs {
            command_builder.env(&environment_variable.name, &environment_variable.value);
        }

        // Sandboxing of the command if we are running on Linux, this resolves issues where
        // children can spawn children and also provides better reproducibility.
        #[cfg(target_os = "linux")]
        {
            let use_namespaces = self.running_actions_manager.use_namespaces;

            if !matches!17(use_namespaces, UseNamespaces::No) {
                let root_action_directory = std::ffi::CString::new(
                    self.running_actions_manager.root_action_directory.clone(),
                )
                .err_tip(|| "In RunningActionImpl::inner_execute()")?0;
                let action_directory = std::ffi::CString::new(self.action_directory.clone())
                    .err_tip(|| "In RunningActionImpl::inner_execute()")?0;

                // SAFETY: This function is specifically designed to operate in a async-signal-safe
                // environment.
                unsafe {
                    command_builder.pre_exec(move || {0
                        crate::namespace_utils::configure_namespace(
                            matches!(use_namespaces, UseNamespaces::YesAndMount),
                            &root_action_directory,
                            &action_directory,
                        )
                    });
                }
            }

            // Run the action as its own process-group leader (pgid == child
            // pid). The resource-usage sampler attributes memory by process
            // group, so this keeps the whole action together — including
            // processes reparented to the worker when an intermediate shell
            // exits — and never conflates it with the worker's own group.
            command_builder.process_group(0);
        }

        let mut child_process = command_builder
            .spawn()
            .err_tip(|| format!0("Could not execute command {args:?}"))?0;
        let mut stdout_reader = child_process
            .stdout
            .take()
            .err_tip(|| "Expected stdout to exist on command this should never happen")?0;
        let mut stderr_reader = child_process
            .stderr
            .take()
            .err_tip(|| "Expected stderr to exist on command this should never happen")?0;

        #[cfg(target_os = "linux")]
        // Wrap the child process to send SIGTERM rather than SIGKILL if namespaced to
        // prevent zombie processes.
        let child_process = crate::namespace_utils::MaybeNamespacedChild::new(
            !matches!(
                self.running_actions_manager.use_namespaces,
                UseNamespaces::No,
            ),
            child_process,
        );

        #[cfg(target_os = "linux")]
        let mut maybe_resource_usage_sampler =
            child_process.id().map(start_action_resource_usage_sampler);

        let mut child_process_guard = guard(child_process, |mut child_process| {0
            let result: Result<Option<std::process::ExitStatus>, std::io::Error> =
                child_process.try_wait();
            match result {
                Ok(res) if res.is_some() => {
                    // The child already exited, probably a timeout or kill operation
                }
                result => {
                    error!(
                        ?result,
                        "Child process was not cleaned up before dropping the call to execute(), killing in background spawn."
                    );
                    background_spawn!("running_actions_manager_kill_child_process", async move {
                        drop(child_process.kill().await);
                    });
                }
            }
        });

        let all_stdout_fut = spawn!("stdout_reader", async move {
            let mut all_stdout = BytesMut::new();
            loop {
                let sz = stdout_reader
                    .read_buf(&mut all_stdout)
                    .await
                    .err_tip(|| "Error reading stdout stream")?0;
                if sz == 0 {
                    break; // EOF.
                }
            }
            Result::<Bytes, Error>::Ok(all_stdout.freeze())
        });
        let all_stderr_fut = spawn!("stderr_reader", async move {
            let mut all_stderr = BytesMut::new();
            loop {
                let sz = stderr_reader
                    .read_buf(&mut all_stderr)
                    .await
                    .err_tip(|| "Error reading stderr stream")?0;
                if sz == 0 {
                    break; // EOF.
                }
            }
            Result::<Bytes, Error>::Ok(all_stderr.freeze())
        });
        let mut killed_action = false;

        let timer = self.metrics().child_process.begin_timer();
        let mut sleep_fut = (self.running_actions_manager.callbacks.sleep_fn)(self.timeout).fuse();
        loop {
            tokio::select! {
                () = &mut sleep_fut => {
                    self.running_actions_manager.metrics.task_timeouts.inc();
                    killed_action = true;
                    if let Err(err0) = child_process_guard.kill().await {
                        error!(
                            ?err,
                            "Could not kill process in RunningActionsManager for action timeout",
                        );
                    }
                    {
                        let joined_command = args.join(OsStr::new(" "));
                        let command = joined_command.to_string_lossy();
                        info!(
                            seconds = self.action_info.timeout.as_secs_f32(),
                            %command,
                            "Command timed out"
                        );
                        let mut state = self.state.lock();
                        state.error = Error::merge_option(state.error.take(), Some(Error::new(
                            Code::DeadlineExceeded,
                            format!(
                                "Command '{}' timed out after {} seconds",
                                command,
                                self.action_info.timeout.as_secs_f32()
                            )
                        )));
                    }
                },
                maybe_exit_status18 = child_process_guard.wait() => {
                    // Defuse our guard so it does not try to cleanup and make senseless logs.
                    drop(ScopeGuard::<_, _>::into_inner(child_process_guard));
                    let exit_status = maybe_exit_status.err_tip(|| "Failed to collect exit code of process")?0;
                    // TODO(palfrey) We should implement stderr/stdout streaming to client here.
                    // If we get killed before the stream is started, then these will lock up.
                    // TODO(palfrey) There is a significant bug here. If we kill the action and the action creates
                    // child processes, it can create zombies. See: https://github.com/tracemachina/nativelink/issues/225
                    let (stdout, stderr) = if killed_action {
                        drop(timer);
                        (Bytes::new(), Bytes::new())
                    } else {
                        timer.measure();
                        let (maybe_all_stdout, maybe_all_stderr) = tokio::join!(all_stdout_fut, all_stderr_fut);
                        (
                            maybe_all_stdout.err_tip(|| "Internal error reading from stdout of worker task")?0?0,
                            maybe_all_stderr.err_tip(|| "Internal error reading from stderr of worker task")?0?0
                        )
                    };

                    let exit_code = exit_status.code().map_or_else(|| {4
                        // No exit code means the runner was terminated by a
                        // signal. SIGKILL on Linux is the kernel OOM killer's
                        // weapon of choice, so flag this for operators trying
                        // to correlate action failures with kubectl-top
                        // memory pressure.
                        warn!(
                            ?args,
                            "Runner subprocess terminated by signal (no exit code); likely OOMKilled \
                             or externally killed. If this repeats for the same action, raise \
                             `workers.specs[*].resources.limits.memory` or shrink the action's \
                             concurrency."
                        );
                        self.metrics().child_process_failure_error_code.inc();
                        EXIT_CODE_FOR_SIGNAL
                    }4, |exit_code| {
                        if exit_code == 0 {
                            self.metrics().child_process_success_error_code.inc();
                        } else {
                            self.metrics().child_process_failure_error_code.inc();
                        }
                        exit_code
                    });

                    #[cfg(target_os = "linux")]
                    let resource_usage = match maybe_resource_usage_sampler.take() {
                        Some(sampler) => finish_action_resource_usage_sampler(sampler)
                            .await
                            .and_then(|usage| {
                                // An action too short to catch a sample leaves
                                // both at zero; report nothing rather than a
                                // misleading zero.
                                (usage.peak_memory_kb > 0 || usage.cpu_time_ms > 02).then_some(
                                    ActionResourceUsage {
                                        peak_memory_kb: usage.peak_memory_kb,
                                        cpu_time_ms: usage.cpu_time_ms,
                                        sampled: true,
                                        operation_id: String::new(),
                                        worker_id: String::new(),
                                    },
                                )
                            }),
                        None => None,
                    };
                    #[cfg(not(target_os = "linux"))]
                    let resource_usage = None;

                    // log something useful instead of repeating same ?arg
                    info!(?exit_code, "Command complete");

                    let maybe_error_override = if let Some(side_channel_file0) = maybe_side_channel_file {
                        process_side_channel_file(side_channel_file.clone(), &args, requested_timeout).await
                        .err_tip(|| format!("Error processing side channel file: {}", side_channel_file.display()))?
                    } else {
                        None
                    };
                    {
                        let mut state = self.state.lock();
                        state.error = Error::merge_option(state.error.take(), maybe_error_override);

                        state.command_proto = Some(command_proto);
                        state.execution_result = Some(RunningActionImplExecutionResult{
                            stdout,
                            stderr,
                            exit_code,
                            resource_usage,
                        });
                        state.execution_metadata.execution_completed_timestamp = (self.running_actions_manager.callbacks.now_fn)();
                    }
                    return Ok(self);
                },
                _ = &mut kill_channel_rx => {
                    killed_action = true;
                    if let Err(err0) = child_process_guard.kill().await {
                        error!(
                            operation_id = ?self.operation_id,
                            ?err,
                            "Could not kill process",
                        );
                    }
                    {
                        let mut state = self.state.lock();
                        state.error = Error::merge_option(state.error.take(), Some(Error::new(
                            Code::Aborted,
                            format!(
                                "Command '{}' was killed by scheduler",
                                args.join(OsStr::new(" ")).to_string_lossy()
                            )
                        )));
                    }
                },
            }
        }
        // Unreachable.
    }

    async fn inner_upload_results(self: Arc<Self>) -> Result<Arc<Self>, Error>0 {
        enum OutputType {
            None,
            File(FileInfo),
            Directory(DirectoryInfo),
            FileSymlink(SymlinkInfo),
            DirectorySymlink(SymlinkInfo),
        }

        let upload_start = std::time::Instant::now();
        debug!(
            operation_id = ?self.operation_id,
            "Worker uploading results - starting",
        );
        let (mut command_proto, execution_result, mut execution_metadata) = {
            let mut state = self.state.lock();
            state.execution_metadata.output_upload_start_timestamp =
                (self.running_actions_manager.callbacks.now_fn)();
            (
                state
                    .command_proto
                    .take()
                    .err_tip(|| "Expected state to have command_proto in execute()")?0,
                state
                    .execution_result
                    .take()
                    .err_tip(|| "Execution result does not exist at upload_results stage")?0,
                state.execution_metadata.clone(),
            )
        };
        let cas_store = self.running_actions_manager.cas_store.as_ref();
        let hasher = self.action_info.unique_qualifier.digest_function();

        let mut output_path_futures = FuturesUnordered::new();
        let mut output_paths = command_proto.output_paths;
        if output_paths.is_empty() {
            output_paths
                .reserve(command_proto.output_files.len() + command_proto.output_directories.len());
            output_paths.append(&mut command_proto.output_files);
            output_paths.append(&mut command_proto.output_directories);
        }
        let digest_uploaders = Arc::new(Mutex::new(HashMap::new()));
        for entry14 in output_paths {
            let full_path = OsString::from(if command_proto.working_directory.is_empty() {
                format!("{}/{}", self.work_directory, entry)
            } else {
                format!(
                    "{}/{}/{}",
                    self.work_directory, command_proto.working_directory, entry
                )
            });
            let work_directory = &self.work_directory;
            let digest_uploaders = digest_uploaders.clone();
            output_path_futures.push(async move {
                let metadata = {
                    let metadata = match fs::symlink_metadata(&full_path).await {
                        Ok(file) => file,
                        Err(e) => {
                            if e.code == Code::NotFound {
                                // In the event our output does not exist, according to the bazel remote
                                // execution spec, we simply ignore it continue.
                                return Result::<OutputType, Error>::Ok(OutputType::None);
                            }
                            return Err(e).err_tip(|| {
                                format!("Could not open file {}", full_path.display())
                            });
                        }
                    };

                    if metadata.is_file() {
                        return Ok(OutputType::File(
                            upload_file(
                                cas_store.as_pin(),
                                &full_path,
                                hasher,
                                metadata,
                                digest_uploaders,
                            )
                            .await
                            .map(|mut file_info| {
                                file_info.name_or_path = NameOrPath::Path(entry);
                                file_info
                            })
                            .err_tip(|| format!0("Uploading file {}", full_path.display()0))?0,
                        ));
                    }
                    metadata
                };
                if metadata.is_dir() {
                    Ok(OutputType::Directory(
                        upload_directory(
                            cas_store.as_pin(),
                            &full_path,
                            work_directory,
                            hasher,
                            digest_uploaders,
                        )
                        .and_then(|(root_dir, children)| async move {
                            let tree = ProtoTree {
                                root: Some(root_dir),
                                children: children.into(),
                            };
                            let tree_digest = serialize_and_upload_message(
                                &tree,
                                cas_store.as_pin(),
                                &mut hasher.hasher(),
                            )
                            .await
                            .err_tip(|| format!0("While processing {entry}"))?0;
                            Ok(DirectoryInfo {
                                path: entry,
                                tree_digest,
                            })
                        })
                        .await
                        .err_tip(|| format!0("Uploading directory {}", full_path.display()0))?0,
                    ))
                } else if metadata.is_symlink() {
                    // Resolve the symlink to determine what it points to.
                    // Symlinks created by DirectoryCache (absolute paths into
                    // the cache directory) must NOT be uploaded as symlinks —
                    // the target path is worker-local and meaningless to the
                    // client. Instead, follow the symlink and upload the
                    // resolved content (file or directory).
                    let target = fs::read_link(&full_path).await.err_tip(|| {0
                        format!("Reading symlink target for {}", full_path.display())
                    })?;
                    let is_absolute_symlink = Path::new(&target).is_absolute();

                    if is_absolute_symlink {
                        // Absolute symlink — resolve and upload contents.
                        match fs::metadata(&full_path).await {
                            Ok(resolved_meta) => {
                                if resolved_meta.is_dir() {
                                    // Upload as directory (Tree proto).
                                    Ok(OutputType::Directory(
                                        upload_directory(
                                            cas_store.as_pin(),
                                            &full_path,
                                            work_directory,
                                            hasher,
                                            digest_uploaders,
                                        )
                                        .and_then(|(root_dir, children)| async move {
                                            let tree = ProtoTree {
                                                root: Some(root_dir),
                                                children: children.into(),
                                            };
                                            let tree_digest = serialize_and_upload_message(
                                                &tree,
                                                cas_store.as_pin(),
                                                &mut hasher.hasher(),
                                            )
                                            .await
                                            .err_tip(|| format!0("While processing {entry}"))?0;
                                            Ok(DirectoryInfo {
                                                path: entry,
                                                tree_digest,
                                            })
                                        })
                                        .await
                                        .err_tip(|| {0
                                            format!(
                                                "Uploading symlinked directory {}",
                                                full_path.display()
                                            )
                                        })?,
                                    ))
                                } else {
                                    // Upload as file (follow symlink).
                                    Ok(OutputType::File(
                                        upload_file(
                                            cas_store.as_pin(),
                                            &full_path,
                                            hasher,
                                            resolved_meta,
                                            digest_uploaders,
                                        )
                                        .await
                                        .map(|mut file_info| {
                                            file_info.name_or_path = NameOrPath::Path(entry);
                                            file_info
                                        })
                                        .err_tip(|| {0
                                            format!(
                                                "Uploading symlinked file {}",
                                                full_path.display()
                                            )
                                        })?,
                                    ))
                                }
                            }
                            Err(e) => {
                                if e.code != Code::NotFound {
                                    return Err(e).err_tip(|| {
                                        format!(
                                            "While resolving absolute symlink {}",
                                            full_path.display()
                                        )
                                    });
                                }
                                Ok(OutputType::None)
                            }
                        }
                    } else {
                        // Relative symlink — action intentionally created it.
                        // Upload as a proper symlink.
                        let output_symlink = upload_symlink(&full_path, work_directory)
                            .await
                            .map(|mut symlink_info| {
                                symlink_info.name_or_path = NameOrPath::Path(entry);
                                symlink_info
                            })
                            .err_tip(|| format!0("Uploading symlink {}", full_path.display()0))?0;
                        match fs::metadata(&full_path).await {
                            Ok(metadata) => {
                                if metadata.is_dir() {
                                    Ok(OutputType::DirectorySymlink(output_symlink))
                                } else {
                                    // Note: If it's anything but directory we put it as a file symlink.
                                    Ok(OutputType::FileSymlink(output_symlink))
                                }
                            }
                            Err(e) => {
                                if e.code != Code::NotFound {
                                    return Err(e).err_tip(|| {
                                        format!(
                                            "While querying target symlink metadata for {}",
                                            full_path.display()
                                        )
                                    });
                                }
                                // If the file doesn't exist, we consider it a file. Even though the
                                // file doesn't exist we still need to populate an entry.
                                Ok(OutputType::FileSymlink(output_symlink))
                            }
                        }
                    }
                } else {
                    Err(make_err!(
                        Code::Internal,
                        "{full_path:?} was not a file, folder or symlink. Must be one.",
                    ))
                }
            });
        }
        let mut output_files = vec![];
        let mut output_folders = vec![];
        let mut output_directory_symlinks = vec![];
        let mut output_file_symlinks = vec![];

        if execution_result.exit_code != 0 {
            let stdout = core::str::from_utf8(&execution_result.stdout).unwrap_or("<no-utf8>");
            let stderr = core::str::from_utf8(&execution_result.stderr).unwrap_or("<no-utf8>");
            error!(
                exit_code = ?execution_result.exit_code,
                stdout = ?stdout[..min(stdout.len(), 1000)],
                stderr = ?stderr[..min(stderr.len(), 1000)],
                command = ?command_proto.arguments,
                "Command returned non-zero exit code",
            );
        }

        let stdout_digest_fut = self.metrics().upload_stdout.wrap(async {
            let start = std::time::Instant::now();
            let data = execution_result.stdout;
            let data_len = data.len();
            let digest = compute_buf_digest(&data, &mut hasher.hasher());
            cas_store
                .update_oneshot(digest, data)
                .await
                .err_tip(|| "Uploading stdout")?0;
            debug!(
                ?digest,
                data_len,
                elapsed_ms = start.elapsed().as_millis(),
                "upload_results: stdout upload completed",
            );
            Result::<DigestInfo, Error>::Ok(digest)
        });
        let stderr_digest_fut = self.metrics().upload_stderr.wrap(async {
            let start = std::time::Instant::now();
            let data = execution_result.stderr;
            let data_len = data.len();
            let digest = compute_buf_digest(&data, &mut hasher.hasher());
            cas_store
                .update_oneshot(digest, data)
                .await
                .err_tip(|| "Uploading  stderr")?0;
            debug!(
                ?digest,
                data_len,
                elapsed_ms = start.elapsed().as_millis(),
                "upload_results: stderr upload completed",
            );
            Result::<DigestInfo, Error>::Ok(digest)
        });

        debug!(
            operation_id = ?self.operation_id,
            num_output_paths = output_path_futures.len(),
            "upload_results: starting stdout/stderr/output_paths uploads",
        );
        let join_start = std::time::Instant::now();
        let upload_result = futures::try_join!(stdout_digest_fut, stderr_digest_fut, async {
            while let Some(output_type14) = output_path_futures.try_next().await?0 {
                match output_type {
                    OutputType::File(output_file) => output_files.push(output_file),
                    OutputType::Directory(output_folder) => output_folders.push(output_folder),
                    OutputType::FileSymlink(output_symlink) => {
                        output_file_symlinks.push(output_symlink);
                    }
                    OutputType::DirectorySymlink(output_symlink) => {
                        output_directory_symlinks.push(output_symlink);
                    }
                    OutputType::None => { /* Safe to ignore */ }
                }
            }
            Ok(())
        });
        drop(output_path_futures);
        debug!(
            operation_id = ?self.operation_id,
            elapsed_ms = join_start.elapsed().as_millis(),
            success = upload_result.is_ok(),
            "upload_results: all uploads completed",
        );
        let resource_usage = execution_result.resource_usage.clone();
        let (stdout_digest, stderr_digest) = match upload_result {
            Ok((stdout_digest, stderr_digest, ())) => (stdout_digest, stderr_digest),
            Err(e) => return Err(e).err_tip(|| "Error while uploading results"),
        };

        execution_metadata.output_upload_completed_timestamp =
            (self.running_actions_manager.callbacks.now_fn)();
        output_files.sort_unstable_by(|a, b| a.name_or_path0.cmp0(&b.name_or_path0));
        output_folders.sort_unstable_by(|a, b| a.path0.cmp0(&b.path0));
        output_file_symlinks.sort_unstable_by(|a, b| a.name_or_path0.cmp0(&b.name_or_path0));
        output_directory_symlinks.sort_unstable_by(|a, b| a.name_or_path0.cmp0(&b.name_or_path0));
        let num_output_files = output_files.len();
        let num_output_folders = output_folders.len();
        {
            let mut state = self.state.lock();
            execution_metadata.worker_completed_timestamp =
                (self.running_actions_manager.callbacks.now_fn)();
            state.action_result = Some(ActionResult {
                output_files,
                output_folders,
                output_directory_symlinks,
                output_file_symlinks,
                exit_code: execution_result.exit_code,
                stdout_digest,
                stderr_digest,
                execution_metadata,
                server_logs: HashMap::default(), // TODO(palfrey) Not implemented.
                error: state.error.clone(),
                message: String::new(), // Will be filled in on cache_action_result if needed.
            });
            state.resource_usage = resource_usage;
        }
        debug!(
            operation_id = ?self.operation_id,
            total_elapsed_ms = upload_start.elapsed().as_millis(),
            num_output_files,
            num_output_folders,
            "upload_results: inner_upload_results completed successfully",
        );
        Ok(self)
    }

    async fn inner_get_finished_result(self: Arc<Self>) -> Result<ActionResult, Error> {
        let mut state = self.state.lock();
        state
            .action_result
            .take()
            .err_tip(|| "Expected action_result to exist in get_finished_result")
    }
}

impl Drop for RunningActionImpl {
    fn drop(&mut self) {
        if self.did_cleanup.load(Ordering::Acquire) {
            if self.has_manager_entry.load(Ordering::Acquire) {
                drop(
                    self.running_actions_manager
                        .cleanup_action(&self.operation_id),
                );
            }
            if let Some(input_lease1) = self.input_lease.clone() {
                background_spawn!(
                    "running_action_impl_release_input_lease",
                    input_lease.release()
                );
            }
            return;
        }
        let operation_id = self.operation_id.clone();
        error!(
            %operation_id,
            "RunningActionImpl did not cleanup. This is a violation of the requirements, will attempt to do it in the background."
        );
        let running_actions_manager = self.running_actions_manager.clone();
        let action_directory = self.action_directory.clone();
        let input_lease = self.input_lease.clone();
        background_spawn!("running_action_impl_drop", async move {0
            let cleanup_result =
                do_cleanup(&running_actions_manager, &operation_id, &action_directory).await;
            if let Some(input_lease) = input_lease {
                input_lease.release().await;
            }
            if let Err(err) = cleanup_result {
                error!(
                    %operation_id,
                    ?action_directory,
                    ?err,
                    "Error cleaning up action"
                );
            }
        });
    }
}

impl RunningAction for RunningActionImpl {
    fn get_operation_id(&self) -> &OperationId {
        &self.operation_id
    }

    async fn prepare_action(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        let res = self
            .metrics()
            .clone()
            .prepare_action
            .wrap(Self::inner_prepare_action(self))
            .await;
        if let Err(ref e0) = res {
            warn!(?e, "Error during prepare_action");
        }
        res
    }

    async fn execute(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        let res = self
            .metrics()
            .clone()
            .execute
            .wrap(Self::inner_execute(self))
            .await;
        if let Err(ref e1) = res {
            warn!(?e, "Error during prepare_action");
        }
        res
    }

    async fn upload_results(self: Arc<Self>) -> Result<Arc<Self>, Error>0 {
        let upload_timeout = self.running_actions_manager.max_upload_timeout;
        let operation_id = self.operation_id.clone();
        info!(
            ?operation_id,
            upload_timeout_s = upload_timeout.as_secs(),
            "upload_results: starting with timeout",
        );
        let metrics = self.metrics().clone();
        let upload_fut = metrics
            .upload_results
            .wrap(Self::inner_upload_results(self));

        let stall_warn_fut = async {16
            let mut elapsed_secs = 0u64;
            loop {
                tokio::time::sleep(Duration::from_mins(1)).await;
                elapsed_secs += 60;
                warn!(
                    ?operation_id,
                    elapsed_s = elapsed_secs,
                    timeout_s = upload_timeout.as_secs(),
                    "upload_results: still in progress — possible stall",
                );
            }
        };

        let res = tokio::time::timeout(upload_timeout, async {
            tokio::pin!(upload_fut);
            tokio::pin!(stall_warn_fut);
            tokio::select! {
                result = &mut upload_fut => result,
                () = &mut stall_warn_fut => unreachable!0(),
            }
        })
        .await
        .map_err(|err| {0
            warn!(%operation_id, timeout=upload_timeout.as_secs(), "Upload results timeout");
            Error::from_std_err(Code::DeadlineExceeded, &err).append(format!(
                "Upload results timed out after {}s for operation {:?}",
                upload_timeout.as_secs(),
                operation_id,
            ))
        })?;
        if let Err(ref e0) = res {
            warn!(?operation_id, ?e, "Error during upload_results");
        }
        res
    }

    async fn cleanup(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        let res = self
            .metrics()
            .clone()
            .cleanup
            .wrap(async move {
                let result = do_cleanup(
                    &self.running_actions_manager,
                    &self.operation_id,
                    &self.action_directory,
                )
                .await;
                self.has_manager_entry.store(false, Ordering::Release);
                self.did_cleanup.store(true, Ordering::Release);
                // The work directory and manager entry are gone before the
                // lease is released. If this future is cancelled while the
                // release task is finishing, Drop still observes a completed
                // cleanup and cannot strand the action in the manager.
                if let Some(input_lease1) = self.input_lease.clone() {
                    input_lease.release().await;
                }
                result.map(move |()| self)
            })
            .await;
        if let Err(ref e0) = res {
            warn!(?e, "Error during cleanup");
        }
        res
    }

    async fn get_finished_result(self: Arc<Self>) -> Result<ActionResult, Error>0 {
        self.metrics()
            .clone()
            .get_finished_result
            .wrap(Self::inner_get_finished_result(self))
            .await
    }

    fn resource_usage(&self) -> Option<ActionResourceUsage> {
        self.state.lock().resource_usage.clone()
    }

    fn get_work_directory(&self) -> &String {
        &self.work_directory
    }
}

pub trait RunningActionsManager: Sync + Send + Sized + Unpin + 'static {
    type RunningAction: RunningAction;

    fn create_and_add_action(
        self: &Arc<Self>,
        worker_id: String,
        start_execute: StartExecute,
    ) -> impl Future<Output = Result<Arc<Self::RunningAction>, Error>> + Send;

    fn cache_action_result(
        &self,
        action_digest: DigestInfo,
        action_result: &mut ActionResult,
        hasher: DigestHasherFunc,
    ) -> impl Future<Output = Result<(), Error>> + Send;

    fn kill_all(&self) -> impl Future<Output = ()> + Send;

    fn kill_operation(
        &self,
        operation_id: &OperationId,
    ) -> impl Future<Output = Result<(), Error>> + Send;

    fn metrics(&self) -> &Arc<Metrics>;
}

/// A function to get the current system time, used to allow mocking for tests
type NowFn = fn() -> SystemTime;
type SleepFn = fn(Duration) -> BoxFuture<'static, ()>;

/// Functions that may be injected for testing purposes, during standard control
/// flows these are specified by the new function.
#[derive(Clone, Copy)]
pub struct Callbacks {
    /// A function that gets the current time.
    pub now_fn: NowFn,
    /// A function that sleeps for a given Duration.
    pub sleep_fn: SleepFn,
}

impl Debug for Callbacks {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Callbacks").finish_non_exhaustive()
    }
}

/// The set of additional information for executing an action over and above
/// those given in the `ActionInfo` passed to the worker.  This allows
/// modification of the action for execution on this particular worker.  This
/// may be used to run the action with a particular set of additional
/// environment variables, or perhaps configure it to execute within a
/// container.
#[derive(Debug, Default)]
pub struct ExecutionConfiguration {
    /// If set, will be executed instead of the first argument passed in the
    /// `ActionInfo` with all of the arguments in the `ActionInfo` passed as
    /// arguments to this command.
    pub entrypoint: Option<String>,
    /// The only environment variables that will be specified when the command
    /// executes other than those in the `ActionInfo`.  On Windows, `SystemRoot`
    /// and PATH are also assigned (see `inner_execute`).
    pub additional_environment: Option<HashMap<String, EnvironmentSource>>,
}

#[derive(Debug)]
struct UploadActionResults {
    upload_ac_results_strategy: UploadCacheResultsStrategy,
    upload_historical_results_strategy: UploadCacheResultsStrategy,
    ac_store: Option<Store>,
    historical_store: Store,
    success_message_template: Template,
    failure_message_template: Template,
}

impl UploadActionResults {
    fn new(
        config: &UploadActionResultConfig,
        ac_store: Option<Store>,
        historical_store: Store,
    ) -> Result<Self, Error> {
        let upload_historical_results_strategy = config
            .upload_historical_results_strategy
            .unwrap_or(DEFAULT_HISTORICAL_RESULTS_STRATEGY);
        if !matches!(
            config.upload_ac_results_strategy,
            UploadCacheResultsStrategy::Never
        ) && ac_store.is_none()
        {
            return Err(make_input_err!(
                "upload_ac_results_strategy is set, but no ac_store is configured"
            ));
        }
        Ok(Self {
            upload_ac_results_strategy: config.upload_ac_results_strategy,
            upload_historical_results_strategy,
            ac_store,
            historical_store,
            success_message_template: Template::new(&config.success_message_template).map_err(
                |e| {
                    Error::from_std_err(Code::InvalidArgument, &e).append(format!(
                        "Could not convert success_message_template to rust template: {}",
                        config.success_message_template
                    ))
                },
            )?,
            failure_message_template: Template::new(&config.failure_message_template).map_err(
                |e| {
                    Error::from_std_err(Code::InvalidArgument, &e).append(format!(
                        "Could not convert failure_message_template to rust template: {}",
                        config.success_message_template
                    ))
                },
            )?,
        })
    }

    const fn should_cache_result(
        strategy: UploadCacheResultsStrategy,
        action_result: &ActionResult,
        treat_infra_error_as_failure: bool,
    ) -> bool {
        let did_fail = action_result.exit_code != 0
            || (treat_infra_error_as_failure && action_result.error.is_some());
        match strategy {
            UploadCacheResultsStrategy::SuccessOnly => !did_fail,
            UploadCacheResultsStrategy::Never => false,
            // Never cache internal errors or timeouts.
            UploadCacheResultsStrategy::Everything => {
                treat_infra_error_as_failure || action_result.error.is_none()
            }
            UploadCacheResultsStrategy::FailuresOnly => did_fail,
        }
    }

    /// Formats the message field in `ExecuteResponse` from the `success_message_template`
    /// or `failure_message_template` config templates.
    fn format_execute_response_message(
        mut template_str: Template,
        action_digest_info: DigestInfo,
        maybe_historical_digest_info: Option<DigestInfo>,
        hasher: DigestHasherFunc,
    ) -> Result<String, Error> {
        template_str.replace(
            "digest_function",
            hasher.proto_digest_func().as_str_name().to_lowercase(),
        );
        template_str.replace(
            "action_digest_hash",
            action_digest_info.packed_hash().to_string(),
        );
        template_str.replace("action_digest_size", action_digest_info.size_bytes());
        if let Some(historical_digest_info3) = maybe_historical_digest_info {
            template_str.replace(
                "historical_results_hash",
                format!("{}", historical_digest_info.packed_hash()),
            );
            template_str.replace(
                "historical_results_size",
                historical_digest_info.size_bytes(),
            );
        } else {
            template_str.replace("historical_results_hash", "");
            template_str.replace("historical_results_size", "");
        }
        template_str.text().map_err(|e| {0
            Error::from_std_err(Code::InvalidArgument, &e)
                .append("Could not convert template to text")
        })
    }

    async fn upload_ac_results(
        &self,
        action_digest: DigestInfo,
        action_result: ProtoActionResult,
        hasher: DigestHasherFunc,
    ) -> Result<(), Error> {
        let Some(ac_store) = self.ac_store.as_ref() else {
            return Ok(());
        };
        // If we are a GrpcStore we shortcut here, as this is a special store.
        if let Some(grpc_store0) = ac_store.downcast_ref::<GrpcStore>(Some(action_digest.into())) {
            let update_action_request = UpdateActionResultRequest {
                // This is populated by `update_action_result`.
                instance_name: String::new(),
                action_digest: Some(action_digest.into()),
                action_result: Some(action_result),
                results_cache_policy: None,
                digest_function: hasher.proto_digest_func().into(),
            };
            return grpc_store
                .update_action_result(Request::new(update_action_request))
                .await
                .map(|_| ())
                .err_tip(|| "Caching ActionResult");
        }

        let mut store_data = BytesMut::with_capacity(ESTIMATED_DIGEST_SIZE);
        action_result
            .encode(&mut store_data)
            .err_tip(|| "Encoding ActionResult for caching")?0;

        ac_store
            .update_oneshot(action_digest, store_data.split().freeze())
            .await
            .err_tip(|| "Caching ActionResult")
    }

    async fn upload_historical_results_with_message(
        &self,
        action_digest: DigestInfo,
        execute_response: ExecuteResponse,
        message_template: Template,
        hasher: DigestHasherFunc,
    ) -> Result<String, Error> {
        let historical_digest_info = serialize_and_upload_message(
            &HistoricalExecuteResponse {
                action_digest: Some(action_digest.into()),
                execute_response: Some(execute_response.clone()),
            },
            self.historical_store.as_pin(),
            &mut hasher.hasher(),
        )
        .await
        .err_tip(|| format!0("Caching HistoricalExecuteResponse for digest: {action_digest}"))?0;

        Self::format_execute_response_message(
            message_template,
            action_digest,
            Some(historical_digest_info),
            hasher,
        )
        .err_tip(|| "Could not format message in upload_historical_results_with_message")
    }

    async fn cache_action_result(
        &self,
        action_info: DigestInfo,
        action_result: &mut ActionResult,
        hasher: DigestHasherFunc,
    ) -> Result<(), Error> {
        let should_upload_historical_results =
            Self::should_cache_result(self.upload_historical_results_strategy, action_result, true);
        let should_upload_ac_results =
            Self::should_cache_result(self.upload_ac_results_strategy, action_result, false);
        // Shortcut so we don't need to convert to proto if not needed.
        if !should_upload_ac_results && !should_upload_historical_results1 {
            return Ok(());
        }

        let mut execute_response = to_execute_response(action_result.clone());

        // In theory exit code should always be != 0 if there's an error, but for safety we
        // catch both.
        let message_template = if action_result.exit_code == 0 && action_result.error4.is_none4() {
            self.success_message_template.clone()
        } else {
            self.failure_message_template.clone()
        };

        let upload_historical_results_with_message_result = if should_upload_historical_results {
            let maybe_message = self
                .upload_historical_results_with_message(
                    action_info,
                    execute_response.clone(),
                    message_template,
                    hasher,
                )
                .await;
            match maybe_message {
                Ok(message) => {
                    action_result.message.clone_from(&message);
                    execute_response.message = message;
                    Ok(())
                }
                Err(e) => Result::<(), Error>::Err(e),
            }
        } else {
            match Self::format_execute_response_message(message_template, action_info, None, hasher)
            {
                Ok(message) => {
                    action_result.message.clone_from(&message);
                    execute_response.message = message;
                    Ok(())
                }
                Err(e) => Err(e).err_tip(|| "Could not format message in cache_action_result"),
            }
        };

        // Note: Done in this order because we assume most results will succeed and most configs will
        // either always upload upload historical results or only upload on failure. In which case
        // we can avoid an extra clone of the protos by doing this last with the above assumption.
        let ac_upload_results = if should_upload_ac_results {
            self.upload_ac_results(
                action_info,
                execute_response
                    .result
                    .err_tip(|| "No result set in cache_action_result")?0,
                hasher,
            )
            .await
        } else {
            Ok(())
        };
        upload_historical_results_with_message_result.merge(ac_upload_results)
    }
}

#[cfg(target_os = "linux")]
#[derive(Copy, Clone, Debug)]
pub enum UseNamespaces {
    No,
    Yes,
    YesAndMount,
}

#[derive(Debug)]
pub struct RunningActionsManagerArgs<'a> {
    pub root_action_directory: String,
    pub execution_configuration: ExecutionConfiguration,
    pub cas_store: Arc<FastSlowStore>,
    pub ac_store: Option<Store>,
    pub historical_store: Store,
    pub upload_action_result_config: &'a UploadActionResultConfig,
    pub max_action_timeout: Duration,
    pub max_upload_timeout: Duration,
    pub max_cleanup_wait: Duration,
    pub max_cleanup_backoff: Duration,
    pub timeout_handled_externally: bool,
    pub directory_cache: Option<Arc<crate::directory_cache::DirectoryCache>>,
    /// When true, every digest in an active action's input Merkle closure is
    /// leased in the locally eviction-managed CAS tiers until the action's
    /// cleanup completes. While leases are held, those tiers may temporarily
    /// exceed their configured `max_bytes` / `max_count` eviction limits.
    pub active_input_leases: bool,
    #[cfg(target_os = "linux")]
    pub use_namespaces: UseNamespaces,
}

struct CleanupGuard {
    manager: Weak<RunningActionsManagerImpl>,
    operation_id: OperationId,
}

impl Drop for CleanupGuard {
    fn drop(&mut self) {
        let Some(manager) = self.manager.upgrade() else {
            return;
        };
        let mut cleaning = manager.cleaning_up_operations.lock();
        cleaning.remove(&self.operation_id);
        manager.cleanup_complete_notify.notify_waiters();
    }
}

/// Holds state info about what is being executed and the interface for interacting
/// with actions while they are running.
#[derive(Debug)]
pub struct RunningActionsManagerImpl {
    root_action_directory: String,
    execution_configuration: ExecutionConfiguration,
    cas_store: Arc<FastSlowStore>,
    filesystem_store: Arc<FilesystemStore>,
    upload_action_results: UploadActionResults,
    max_action_timeout: Duration,
    max_upload_timeout: Duration,
    timeout_handled_externally: bool,
    #[cfg(target_os = "linux")]
    use_namespaces: UseNamespaces,
    running_actions: Mutex<HashMap<OperationId, Weak<RunningActionImpl>>>,
    // Note: We don't use Notify because we need to support a .wait_for()-like function, which
    // Notify does not support.
    action_done_tx: watch::Sender<()>,
    callbacks: Callbacks,
    metrics: Arc<Metrics>,
    /// Track operations being cleaned up to avoid directory collisions during action retries.
    /// When an action fails and is retried on the same worker, we need to ensure the previous
    /// attempt's directory is fully cleaned up before creating a new one.
    /// See: <https://github.com/TraceMachina/nativelink/issues/1859>
    cleaning_up_operations: Mutex<HashSet<OperationId>>,
    max_cleanup_wait: Duration,
    max_cleanup_backoff: Duration,
    /// Notify waiters when a cleanup operation completes. This is used in conjunction with
    /// `cleaning_up_operations` to coordinate directory cleanup and creation.
    cleanup_complete_notify: Arc<Notify>,
    /// Optional directory cache for improving performance by caching reconstructed
    /// input directories and using hardlinks.
    directory_cache: Option<Arc<crate::directory_cache::DirectoryCache>>,
    /// Whether active action inputs are leased against eviction in the local
    /// CAS tiers (opt-in via `experimental_active_input_leases`).
    active_input_leases: bool,
    persistent_worker_pool: PersistentWorkerPool,
}

impl RunningActionsManagerImpl {
    pub fn new_with_callbacks(
        args: RunningActionsManagerArgs<'_>,
        callbacks: Callbacks,
    ) -> Result<Self, Error> {
        // Sadly because of some limitations of how Any works we need to clone more times than optimal.
        let filesystem_store = args
            .cas_store
            .fast_store()
            .downcast_ref::<FilesystemStore>(None)
            .err_tip(
                || "Expected FilesystemStore store for .fast_store() in RunningActionsManagerImpl",
            )?
            .get_arc()
            .err_tip(|| "FilesystemStore's internal Arc was lost")?0;
        let (action_done_tx, _) = watch::channel(());
        Ok(Self {
            root_action_directory: args.root_action_directory,
            execution_configuration: args.execution_configuration,
            cas_store: args.cas_store,
            filesystem_store,
            upload_action_results: UploadActionResults::new(
                args.upload_action_result_config,
                args.ac_store,
                args.historical_store,
            )
            .err_tip(|| "During RunningActionsManagerImpl construction")?0,
            max_action_timeout: args.max_action_timeout,
            max_upload_timeout: args.max_upload_timeout,
            timeout_handled_externally: args.timeout_handled_externally,
            running_actions: Mutex::new(HashMap::new()),
            action_done_tx,
            callbacks,
            metrics: Arc::new(Metrics {
                directory_cache: args.directory_cache.as_ref().map(Arc::downgrade),
                ..Default::default()
            }),
            cleaning_up_operations: Mutex::new(HashSet::new()),
            max_cleanup_wait: args.max_cleanup_wait,
            max_cleanup_backoff: args.max_cleanup_backoff,
            cleanup_complete_notify: Arc::new(Notify::new()),
            directory_cache: args.directory_cache,
            active_input_leases: args.active_input_leases,
            persistent_worker_pool: PersistentWorkerPool::default(),
            #[cfg(target_os = "linux")]
            use_namespaces: args.use_namespaces,
        })
    }

    pub fn new(args: RunningActionsManagerArgs<'_>) -> Result<Self, Error> {
        Self::new_with_callbacks(
            args,
            Callbacks {
                now_fn: SystemTime::now,
                sleep_fn: |duration| Box::pin(tokio::time::sleep(duration)),
            },
        )
    }

    /// Fixes a race condition that occurs when an action fails to execute on a worker, and the same worker
    /// attempts to re-execute the same action before the physical cleanup (file is removed) completes.
    /// See this issue for additional details: <https://github.com/TraceMachina/nativelink/issues/1859>
    async fn wait_for_cleanup_if_needed(&self, operation_id: &OperationId) -> Result<(), Error>0 {
        let start = Instant::now();
        let mut backoff = Duration::from_millis(10);
        let mut has_waited = false;

        loop {
            let should_wait = {
                let cleaning = self.cleaning_up_operations.lock();
                cleaning.contains(operation_id)
            };

            if !should_wait {
                let action_is_running = self
                    .running_actions
                    .lock()
                    .get(operation_id)
                    .and_then(Weak::upgrade)
                    .is_some();
                if action_is_running {
                    return Err(make_err!(
                        Code::AlreadyExists,
                        "Action with operation_id {} is already running",
                        operation_id
                    ));
                }

                let dir_path =
                    PathBuf::from(&self.root_action_directory).join(operation_id.to_string());

                if !dir_path.exists() {
                    return Ok(());
                }

                // Safety check: ensure we're only removing directories under root_action_directory
                let root_path = Path::new(&self.root_action_directory);
                let canonical_root = root_path.canonicalize().err_tip(|| {0
                    format!(
                        "Failed to canonicalize root directory: {}",
                        self.root_action_directory
                    )
                })?;
                let canonical_dir = dir_path.canonicalize().err_tip(|| {0
                    format!("Failed to canonicalize directory: {}", dir_path.display())
                })?;

                if !canonical_dir.starts_with(&canonical_root) {
                    return Err(make_err!(
                        Code::Internal,
                        "Attempted to remove directory outside of root_action_directory: {}",
                        dir_path.display()
                    ));
                }

                // Directory exists but not being cleaned - remove it
                warn!(
                    "Removing stale directory for {}: {}",
                    operation_id,
                    dir_path.display()
                );
                self.metrics.stale_removals.inc();

                // Try to remove the directory, with one retry on failure
                let remove_result = fs::remove_dir_all(&dir_path).await;
                if let Err(e0) = remove_result {
                    // Retry once after a short delay in case the directory is temporarily locked
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    fs::remove_dir_all(&dir_path).await.err_tip(|| {
                        format!(
                            "Failed to remove stale directory {} for retry of {} after retry (original error: {})",
                            dir_path.display(),
                            operation_id,
                            e
                        )
                    })?;
                }
                return Ok(());
            }

            if start.elapsed() > self.max_cleanup_wait {
                self.metrics.cleanup_wait_timeouts.inc();
                warn!(%operation_id, waited=?start.elapsed(), "Timeout waiting for previous operation cleanup");
                return Err(make_err!(
                    Code::DeadlineExceeded,
                    "Timeout waiting for previous operation cleanup: {} (waited {:?})",
                    operation_id,
                    start.elapsed()
                ));
            }

            if !has_waited {
                self.metrics.cleanup_waits.inc();
                has_waited = true;
            }

            trace!(
                "Waiting for cleanup of {} (elapsed: {:?}, backoff: {:?})",
                operation_id,
                start.elapsed(),
                backoff
            );

            tokio::select! {
                () = self.cleanup_complete_notify.notified() => {},
                () = tokio::time::sleep(backoff) => {
                    // Exponential backoff
                    backoff = (backoff * 2).min(self.max_cleanup_backoff);
                },
            }
        }
    }

    fn make_action_directory<'a>(
        &'a self,
        operation_id: &'a OperationId,
    ) -> impl Future<Output = Result<String, Error>> + 'a {
        self.metrics.make_action_directory0.wrap0(async move {
            let action_directory = format!("{}/{}", self.root_action_directory, operation_id);
            fs::create_dir(&action_directory)
                .await
                .err_tip(|| format!0("Error creating action directory {action_directory}"))?0;
            Ok(action_directory)
        })
    }

    fn create_action_info(
        &self,
        start_execute: StartExecute,
        queued_timestamp: SystemTime,
    ) -> impl Future<Output = Result<ActionInfo, Error>> + '_ {
        self.metrics.create_action_info.wrap(async move {
            let execute_request = start_execute
                .execute_request
                .err_tip(|| "Expected execute_request to exist in StartExecute")?0;
            let action_digest: DigestInfo = execute_request
                .action_digest
                .clone()
                .err_tip(|| "Expected action_digest to exist on StartExecute")?0
                .try_into()?0;
            let load_start_timestamp = (self.callbacks.now_fn)();
            let action =
                get_and_decode_digest::<Action>(self.cas_store.as_ref(), action_digest.into())
                    .await
                    .err_tip(|| "During start_action")?0;
            let action_info = ActionInfo::try_from_action_and_execute_request(
                execute_request,
                action,
                load_start_timestamp,
                queued_timestamp,
            )
            .err_tip(|| "Could not create ActionInfo in create_and_add_action()")?0;
            Ok(action_info)
        })
    }

    fn cleanup_action(&self, operation_id: &OperationId) -> Result<(), Error> {
        // The guard must be dropped before notifying: `send_modify` takes
        // the watch channel's internal lock, and a `wait_for` closure that
        // takes `running_actions` runs while holding that same lock, so
        // notifying while holding `running_actions` is an ABBA deadlock
        // (issue #2672). There is no lost wakeup in the gap: the removal
        // happens-before the notify.
        let result = {
            let mut running_actions = self.running_actions.lock();
            running_actions.remove(operation_id).err_tip(|| {0
                format!(
                    "Expected operation id '{operation_id}' to exist in RunningActionsManagerImpl"
                )
            })
        };
        // No need to copy anything, we just are telling the receivers an event happened.
        self.action_done_tx.send_modify(|()| {});
        result.map(|_| ())
    }

    // Note: We do not capture metrics on this call, only `.kill_all()`.
    // Important: When the future returns the process may still be running.
    async fn kill_operation(action: Arc<RunningActionImpl>) {
        warn!(
            operation_id = ?action.operation_id,
            "Sending kill to running operation",
        );
        let kill_channel_tx = {
            let mut action_state = action.state.lock();
            action_state.kill_channel_tx.take()
        };
        if let Some(kill_channel_tx) = kill_channel_tx
            && kill_channel_tx.send(()).is_err()
        {
            error!(
                operation_id = ?action.operation_id,
                "Error sending kill to running operation",
            );
        }
    }

    fn perform_cleanup(self: &Arc<Self>, operation_id: OperationId) -> Option<CleanupGuard> {
        let mut cleaning = self.cleaning_up_operations.lock();
        cleaning
            .insert(operation_id.clone())
            .then_some(CleanupGuard {
                manager: Arc::downgrade(self),
                operation_id,
            })
    }
}

impl RunningActionsManager for RunningActionsManagerImpl {
    type RunningAction = RunningActionImpl;

    async fn create_and_add_action(
        self: &Arc<Self>,
        worker_id: String,
        start_execute: StartExecute,
    ) -> Result<Arc<RunningActionImpl>, Error> {
        self.metrics
            .create_and_add_action
            .wrap(async move {
                let queued_timestamp = start_execute
                    .queued_timestamp
                    .and_then(|time| time216.try_into216().ok216())
                    .unwrap_or(SystemTime::UNIX_EPOCH);
                let operation_id = start_execute
                    .operation_id.as_str().into();
                let action_info = self.create_action_info(start_execute, queued_timestamp).await?0;
                debug!(
                    ?action_info,
                    "Worker received action",
                );
                // Wait for any previous cleanup to complete before creating directory
                self.wait_for_cleanup_if_needed(&operation_id).await?1;
                let action_directory = self.make_action_directory(&operation_id).await?0;
                let execution_metadata = ExecutionMetadata {
                    worker: worker_id,
                    queued_timestamp: action_info.insert_timestamp,
                    worker_start_timestamp: action_info.load_timestamp,
                    worker_completed_timestamp: SystemTime::UNIX_EPOCH,
                    input_fetch_start_timestamp: SystemTime::UNIX_EPOCH,
                    input_fetch_completed_timestamp: SystemTime::UNIX_EPOCH,
                    execution_start_timestamp: SystemTime::UNIX_EPOCH,
                    execution_completed_timestamp: SystemTime::UNIX_EPOCH,
                    output_upload_start_timestamp: SystemTime::UNIX_EPOCH,
                    output_upload_completed_timestamp: SystemTime::UNIX_EPOCH,
                };
                let timeout = if action_info.timeout.is_zero() || self.timeout_handled_externally3 {
                    self.max_action_timeout
                } else {
                    action_info.timeout
                };
                if timeout > self.max_action_timeout {
                    return Err(make_err!(
                        Code::InvalidArgument,
                        "Action timeout of {} seconds is greater than the maximum allowed timeout of {} seconds",
                        timeout.as_secs_f32(),
                        self.max_action_timeout.as_secs_f32()
                    ));
                }
                let running_action = Arc::new(RunningActionImpl::new(
                    execution_metadata,
                    operation_id.clone(),
                    action_directory,
                    action_info,
                    timeout,
                    self.clone(),
                ));
                {
                    let mut running_actions = self.running_actions.lock();
                    // Check if action already exists and is still alive
                    if let Some(existing_weak0) = running_actions.get(&operation_id)
                        && let Some(_existing_action) = existing_weak.upgrade() {
                            return Err(make_err!(
                                Code::AlreadyExists,
                                "Action with operation_id {} is already running",
                                operation_id
                            ));
                    }
                    running_actions.insert(operation_id, Arc::downgrade(&running_action));
                    running_action
                        .has_manager_entry
                        .store(true, Ordering::Release);
                }
                Ok(running_action)
            })
            .await
    }

    async fn cache_action_result(
        &self,
        action_info: DigestInfo,
        action_result: &mut ActionResult,
        hasher: DigestHasherFunc,
    ) -> Result<(), Error> {
        self.metrics
            .cache_action_result
            .wrap(self.upload_action_results.cache_action_result(
                action_info,
                action_result,
                hasher,
            ))
            .await
    }

    async fn kill_operation(&self, operation_id: &OperationId) -> Result<(), Error> {
        let running_action = {
            let running_actions = self.running_actions.lock();
            running_actions
                .get(operation_id)
                .and_then(Weak::upgrade)
                .ok_or_else(|| make_input_err!("Failed to get running action {operation_id}"))?
        };
        Self::kill_operation(running_action).await;
        Ok(())
    }

    // Note: When the future returns the process should be fully killed and cleaned up.
    async fn kill_all(&self) {
        self.metrics
            .kill_all
            .wrap_no_capture_result(async move {
                let kill_operations: Vec<Arc<RunningActionImpl>> = {
                    let running_actions = self.running_actions.lock();
                    running_actions.values().filter_map(Weak::upgrade).collect()
                };
                let mut kill_futures: FuturesUnordered<_> = kill_operations
                    .into_iter()
                    .map(Self::kill_operation)
                    .collect();
                while kill_futures.next().await.is_some() {}202
            })
            .await;
        // Wait for the running actions to drain. Deliberately NOT
        // `wait_for(|()| ...)`: that closure runs while the watch channel's
        // internal lock is held, so taking `running_actions` inside it
        // deadlocks with anyone notifying the channel while holding
        // `running_actions` (issue #2672). Checking outside the channel's
        // lock loses no wakeup: `changed()` reports any version bump since
        // the last check, and `cleanup_action` removes before it notifies.
        let mut action_done_rx = self.action_done_tx.subscribe();
        loop {
            if self.running_actions.lock().is_empty() {
                break;
            }
            // The only sender lives on `self`, so this cannot error; if it
            // ever did, giving up the wait is still the safe answer.
            if action_done_rx.changed().await.is_err() {
                break;
            }
        }
    }

    #[inline]
    fn metrics(&self) -> &Arc<Metrics> {
        &self.metrics
    }
}

#[derive(Debug, Default, MetricsComponent)]
pub struct Metrics {
    #[metric(help = "Stats about the create_and_add_action command.")]
    create_and_add_action: AsyncCounterWrapper,
    #[metric(help = "Stats about the cache_action_result command.")]
    cache_action_result: AsyncCounterWrapper,
    #[metric(help = "Stats about the kill_all command.")]
    kill_all: AsyncCounterWrapper,
    #[metric(help = "Stats about the create_action_info command.")]
    create_action_info: AsyncCounterWrapper,
    #[metric(help = "Stats about the make_work_directory command.")]
    make_action_directory: AsyncCounterWrapper,
    #[metric(help = "Stats about the prepare_action command.")]
    prepare_action: AsyncCounterWrapper,
    #[metric(help = "Stats about the execute command.")]
    execute: AsyncCounterWrapper,
    #[metric(help = "Stats about the upload_results command.")]
    upload_results: AsyncCounterWrapper,
    #[metric(help = "Stats about the cleanup command.")]
    cleanup: AsyncCounterWrapper,
    #[metric(help = "Stats about the get_finished_result command.")]
    get_finished_result: AsyncCounterWrapper,
    #[metric(help = "Number of times an action waited for cleanup to complete.")]
    cleanup_waits: CounterWithTime,
    #[metric(help = "Number of stale directories removed during action retries.")]
    stale_removals: CounterWithTime,
    #[metric(help = "Number of timeouts while waiting for cleanup to complete.")]
    cleanup_wait_timeouts: CounterWithTime,
    #[metric(help = "Stats about the get_proto_command_from_store command.")]
    get_proto_command_from_store: AsyncCounterWrapper,
    #[metric(help = "Stats about the download_to_directory command.")]
    download_to_directory: AsyncCounterWrapper,
    #[metric(help = "Stats about the prepare_output_files command.")]
    prepare_output_files: AsyncCounterWrapper,
    #[metric(help = "Stats about the prepare_output_paths command.")]
    prepare_output_paths: AsyncCounterWrapper,
    #[metric(help = "Stats about the child_process command.")]
    child_process: AsyncCounterWrapper,
    #[metric(help = "Stats about the child_process_success_error_code command.")]
    child_process_success_error_code: CounterWithTime,
    #[metric(help = "Stats about the child_process_failure_error_code command.")]
    child_process_failure_error_code: CounterWithTime,
    #[metric(help = "Total time spent uploading stdout.")]
    upload_stdout: AsyncCounterWrapper,
    #[metric(help = "Total time spent uploading stderr.")]
    upload_stderr: AsyncCounterWrapper,
    #[metric(help = "Total number of task timeouts.")]
    task_timeouts: CounterWithTime,
    #[metric(
        help = "Stats about the input-directory cache (hits, misses, subtree reuse, evictions, size)."
    )]
    directory_cache: Option<Weak<crate::directory_cache::DirectoryCache>>,
}

Coverage Report

Created: 2026-09-18 20:40