// 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, try_join, try_join_all, try_join3,

};

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::{

    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, 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;

/// 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;

/// 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>> {

    async move {

        let directory = get_and_decode_digest::<ProtoDirectory>(cas_store, digest.into())

            .await

            .err_tip(|| "Converting digest to Directory")
?0
;

        let mut futures = Vec::new();

        for 
file12
 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),

            };

            futures.push(

                cas_store

                    .populate_fast_store(digest.into())

                    .and_then(move |()| 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 if custom_unix_mode.is_some() || 
mtime8
.
is_some8
() {

                            // 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_locked1
(|src| async move 
{1
 
Ok(src)1
 })

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

                            })??;

                        } else {

                            // Hot path: hardlink only — no writable fd is ever

                            // opened for the materialized inode, so a concurrent

                            // `execve` of an executable input cannot hit ETXTBSY

                            // ("Text file busy"). Executables hardlink a

                            // per-digest 0o555 variant created once off the hot

                            // path (the 0o444 CAS blob is shared and cannot carry

                            // +x); non-executables hardlink the 0o444 CAS blob.

                            let src_path = if is_executable {

                                filesystem_store

                                    .get_executable_hardlink_source(&digest)

                                    .await

                                    .err_tip(|| "Resolving executable hardlink source")
?0

                            } else {

                                let file_entry = filesystem_store

                                    .get_file_entry_for_digest(&digest)

                                    .await

                                    .err_tip(|| "During hard link")
?0
;

                                // TODO: add a test for #2051: deadlock with large number of files

                                file_entry

                                    .
get_file_path_locked6
(|src| async move 
{6
 
Ok(src)6
 })

                                    .await
?0

                            };

                            fs::hard_link(&src_path, &dest)

                                .await

                                .map_err(|e| 
{0

                                    if e.code == Code::NotFound {

                                        e.append(

                                            format!(

                                            "Could not make hardlink to {dest}, 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."

                                        ))

                                    } else {

                                        e.append(format!("Could not make hardlink to {dest}"))

                                    }

                                })?;

                            // Hardlinked inodes are already correct (the 0o444

                            // blob or the 0o555 executable variant) and carry no

                            // per-file metadata, so there is nothing to stamp.

                            return Ok(());

                        }

                        // Private-inode tail (zero-digest or private copy only):

                        // stamp the requested mode and mtime. Safe because the

                        // inode is private to this action.

                        #[cfg(target_family = "unix")]

                        {

                            let mode = if let Some(
mode1
) = custom_unix_mode {

                                Some(if is_executable { mode | 0o111 } else { 
mode0
 })

                            } else if is_executable {

                                Some(0o555)

                            } else {

                                None

                            };

                            if let Some(
mode1
) = mode {

                                fs::set_permissions(&dest, Permissions::from_mode(mode))

                                    .await

                                    .err_tip(|| 
{0

                                        format!("Could not set unix mode in download_to_directory {dest}")

                                    })?;

                            }

                        }

                        if let Some(
mtime1
) = mtime {

                            let spawned_dest = dest.clone();

                            spawn_blocking!("download_to_directory_set_mtime", move || {

                                set_file_mtime(

                                    &spawned_dest,

                                    FileTime::from_unix_time(mtime.seconds, mtime.nanos as u32),

                                )

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

                    })

                    .map_err(move |e| 
e0
.
append0
(
format!0
("for digest {digest}")))

                    .boxed(),

            );

        }

        for 
directory8
 in directory.directories {

            let digest: DigestInfo = 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, directory.name);

            futures.push(

                async move {

                    fs::create_dir(&new_directory_path)

                        .await

                        .err_tip(|| 
format!0
("Could not create directory {new_directory_path}"))
?0
;

                    download_to_directory(

                        cas_store,

                        filesystem_store,

                        &digest,

                        &new_directory_path,

                    )

                    .await

                    .err_tip(|| 
format!0
("in download_to_directory : {new_directory_path}"))
?0
;

                    Ok(())

                }

                .boxed(),

            );

        }

        #[cfg(target_family = "unix")]

        for 
symlink_node2
 in directory.symlinks {

            let dest = format!("{}/{}", current_directory, symlink_node.name);

            futures.push(

                async move {

                    fs::symlink(&symlink_node.target, &dest).await.err_tip(|| 
{0

                        format!(

                            "Could not create symlink {} -> {}",

                            symlink_node.target, dest

                        )

                    })?;

                    Ok(())

                }

                .boxed(),

            );

        }

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

        futures::stream::iter(futures)

            .buffer_unordered(DOWNLOAD_TO_DIRECTORY_CONCURRENCY)

            .try_collect::<Vec<_>>()

            .await
?0
;

        Ok(())

    }

    .boxed()

}

/// 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> {

    // Try cache first if available

    if let Some(
cache0
) = 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!("Failed to clear pre-created work directory {work_directory}"))?;

        match cache

            .get_or_create(*digest, Path::new(work_directory))

            .await

        {

            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(cas_store, filesystem_store, digest, work_directory).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_init13
(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_result6
) = 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 (mut file_nodes, dir_entries, mut symlinks) = try_join3(

            file_futures.try_collect::<Vec<FileNode>>(),

            dir_futures.try_collect::<Vec<(DirectoryNode, VecDeque<Directory>)>>(),

            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_node1
, 
mut recursive_child_directories1
) 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 the work directory of the action.

    fn get_work_directory(&self) -> &String;

}

#[derive(Debug)]

struct RunningActionImplExecutionResult {

    stdout: Bytes,

    stderr: Bytes,

    exit_code: i32,

}

#[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>,

    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,

    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 (kill_channel_tx, kill_channel_rx) = oneshot::channel();

        Self {

            operation_id,

            action_directory,

            work_directory,

            action_info,

            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,

                execution_metadata,

                error: None,

            }),

            // Always need to ensure that we're removed from the manager on Drop.

            has_manager_entry: AtomicBool::new(true),

            // 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>0
 {

        {

            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(

                        &self.running_actions_manager.directory_cache,

                        &self.running_actions_manager.cas_store,

                        filesystem_store_pin,