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

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

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

}

/// Aggressively download the digests of files and make a local folder from it. This function

/// will spawn unbounded number of futures to try and get these downloaded. The store itself

/// should be rate limited if spawning too many requests at once is an issue.

/// 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 = FuturesUnordered::new();

        for 
file4
 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 (mtime, mut unix_mode) = match file.node_properties {

                Some(properties) => (properties.mtime, properties.unix_mode),

                None => (None, None),

            };

            #[cfg_attr(target_family = "windows", allow(unused_assignments))]

            if file.is_executable {

                unix_mode = Some(unix_mode.unwrap_or(0o444) | 0o111);

            }

            futures.push(

                cas_store

                    .populate_fast_store(digest.into())

                    .and_then(move |()| async move {

                        if is_zero_digest(digest) {

                            let mut file_slot = fs::create_file(&dest).await?;

                            file_slot.write_all(&[]).await?;

                        }

                        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

                            let 
src_path4
 = 
file_entry4
.
get_file_path_locked4
(|src| async move 
{4
 
Ok(PathBuf::from(src))4
 }).
await4
?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}"))

                                    }

                                })?;

                            }

                        #[cfg(target_family = "unix")]

                        if let Some(
unix_mode1
) = unix_mode {

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

                                .await

                                .err_tip(|| 
{0

                                    format!(

                                        "Could not set unix mode in download_to_directory {dest}"

                                    )

                                })?;

                        }

                        if let Some(
mtime1
) = mtime {

                            spawn_blocking!("download_to_directory_set_mtime", move || {

                                set_file_mtime(

                                    &dest,

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

                                )

                                .err_tip(|| 
{0

                                    format!("Failed to set mtime in download_to_directory {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 
directory7
 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_node1
 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(),

            );

        }

        while futures.try_next().await
?0
.is_some() 
{}12

        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.

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 {

        match cache

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

            .await

        {

            Ok(cache_hit) => {

                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"

                );

                // Fall through to traditional path

            }

        }

    }

    // 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_init7
(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_result5
) = 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,

                        &self.action_info.input_root_digest,

                        &self.work_directory,

                    ))

                    .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| 
{9

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

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

        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 program = self.canonicalise_path(args[0], &command_proto.working_directory)
?0
;

        let mut command_builder = process::Command::new(program);

        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