// Copyright 2024 The NativeLink Authors. All rights reserved.

//

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

//    http://www.apache.org/licenses/LICENSE-2.0

//

// 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 std::borrow::Cow;

use std::cmp::min;

use std::collections::vec_deque::VecDeque;

use std::collections::HashMap;

use std::convert::Into;

use std::ffi::{OsStr, OsString};

use std::fmt::Debug;

#[cfg(target_family = "unix")]

use std::fs::Permissions;

#[cfg(target_family = "unix")]

use std::os::unix::fs::{MetadataExt, PermissionsExt};

use std::path::Path;

use std::pin::Pin;

use std::process::Stdio;

use std::sync::atomic::{AtomicBool, Ordering};

use std::sync::{Arc, Weak};

use std::time::{Duration, SystemTime};

use bytes::{Bytes, BytesMut};

use filetime::{set_file_mtime, FileTime};

use formatx::Template;

use futures::future::{

    try_join, try_join3, try_join_all, BoxFuture, Future, FutureExt, TryFutureExt,

};

use futures::stream::{FuturesUnordered, StreamExt, TryStreamExt};

use nativelink_config::cas_server::{

    EnvironmentSource, UploadActionResultConfig, UploadCacheResultsStrategy,

};

use nativelink_error::{make_err, make_input_err, Code, Error, ResultExt};

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

    compute_buf_digest, get_and_decode_digest, serialize_and_upload_message, ESTIMATED_DIGEST_SIZE,

};

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

    to_execute_response, ActionInfo, ActionResult, DirectoryInfo, ExecutionMetadata, FileInfo,

    NameOrPath, OperationId, SymlinkInfo,

};

use nativelink_util::common::{fs, DigestInfo};

use nativelink_util::digest_hasher::{DigestHasher, DigestHasherFunc};

use nativelink_util::metrics_utils::{AsyncCounterWrapper, CounterWithTime};

use nativelink_util::shutdown_guard::ShutdownGuard;

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::{guard, ScopeGuard};

use serde::Deserialize;

use tokio::io::{AsyncReadExt, AsyncSeekExt};

use tokio::process;

use tokio::sync::{oneshot, watch};

use tokio_stream::wrappers::ReadDirStream;

use tonic::Request;

use tracing::{enabled, event, Level};

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::failures_only;

/// Valid string reasons for a failure.

/// Note: If these change, the documentation should be updated.

#[allow(non_camel_case_types)]

#[derive(Debug, Deserialize)]

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
)
?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
)
?0

                .try_into()

                .err_tip(|| 
"In Directory::file::digest"0
)
?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 {

  Branch (145:16): [True: 1, False: 3]

  Branch (145:16): [Folded - Ignored]

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

            }

            futures.push(

                cas_store

                    .populate_fast_store(digest.into())

                    .and_then(move |()| async move {

                        let file_entry = filesystem_store

                            .get_file_entry_for_digest(&digest)

                            .await

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

                        file_entry

                            .get_file_path_locked(|src| fs::hard_link(src, &dest))

                            .await

                            .map_err(|e| {

                                make_err!(Code::Internal, "Could not make hardlink, {e:?} : {dest}")

                            })
?0
;

                        #[cfg(target_family = "unix")]

                        if let Some(
unix_mode1
) = unix_mode {

  Branch (163:32): [True: 1, False: 3]

  Branch (163:32): [Folded - Ignored]

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

                                .await

                                .err_tip(|| {

                                    format!(

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

                                    )

                                })
?0
;

                        }

                        if let Some(
mtime1
) = mtime {

  Branch (172:32): [True: 1, False: 3]

  Branch (172:32): [Folded - Ignored]

                            spawn_blocking!("download_to_directory_set_mtime", move || {

                                set_file_mtime(

                                    &dest,

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

                                )

                                .err_tip(|| {

                                    format!("Failed to set mtime in download_to_directory {dest}")

                                })

                            })

                            .await

                            .err_tip(|| {

                                "Failed to launch spawn_blocking in download_to_directory"

                            })
?0
?0
;

                        }

                        Ok(())

                    })

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

                    .boxed(),

            );

        }

        for 
directory7
 in directory.directories {

            let digest: DigestInfo = directory

                .digest

                .err_tip(|| 
"Expected Digest to exist in Directory::directories::digest"0
)
?0

                .try_into()

                .err_tip(|| 
"In Directory::file::digest"0
)
?0
;

            let new_directory_path = format!("{}/{}", current_directory, directory.name);

            futures.push(

                async move {

                    fs::create_dir(&new_directory_path)

                        .await

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

                    download_to_directory(

                        cas_store,

                        filesystem_store,

                        &digest,

                        &new_directory_path,

                    )

                    .await

                    .err_tip(|| 
format!("in download_to_directory : {new_directory_path}")0
)
?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(|| {

                        format!(

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

                            symlink_node.target, dest

                        )

                    })
?0
;

                    Ok(())

                }

                .boxed(),

            );

        }

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

  Branch (237:15): [True: 12, False: 25]

  Branch (237:15): [Folded - Ignored]

        Ok(())

    }

    .boxed()

}

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

}

async fn upload_file(

    cas_store: Pin<&impl StoreLike>,

    full_path: impl AsRef<Path> + Debug,

    hasher: DigestHasherFunc,

    metadata: std::fs::Metadata,

) -> 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!("Could not open file {full_path:?}")0
)
?0
;

    let (digest, mut file) = hasher

        .hasher()

        .digest_for_file(&full_path, file.into_inner(), Some(file_size))

        .await

        .err_tip(|| 
format!("Failed to hash file in digest_for_file failed for {full_path:?}")0
)
?0
;

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

    // Note: For unknown reasons we appear to be hitting:

    // https://github.com/rust-lang/rust/issues/92096

    // or a smiliar issue if we try to use the non-store driver function, so we

    // are using the store driver function here.

    cas_store

        .as_store_driver_pin()

        .update_with_whole_file(

            digest.into(),

            full_path.as_ref().into(),

            file,

            UploadSizeInfo::ExactSize(digest.size_bytes()),

        )

        .await

        .err_tip(|| 
format!("for {full_path:?}")0
)
?0
;

    let name = full_path

        .as_ref()

        .file_name()

        .err_tip(|| 
format!("Expected file_name to exist on {full_path:?}")0
)
?0

        .to_str()

        .err_tip(|| {

            make_err!(

                Code::Internal,

                "Could not convert {:?} to string",

                full_path

            )

        })
?0

        .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!("Could not get read_link path of {full_path:?}")0
)
?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()) {

  Branch (322:21): [Folded - Ignored]

  Branch (322:21): [Folded - Ignored]

  Branch (322:21): [True: 0, False: 1]

  Branch (322:21): [True: 0, False: 1]

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

                make_err!(

                    Code::Internal,

                    "Could not convert '{:?}' to string",

                    full_target_path

                )

            })
?0

            .to_string()

    };

    let name = full_path

        .as_ref()

        .file_name()

        .err_tip(|| 
format!("Expected file_name to exist on {full_path:?}")0
)
?0

        .to_str()

        .err_tip(|| {

            make_err!(

                Code::Internal,

                "Could not convert {:?} to string",

                full_path

            )

        })
?0

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

) -> 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!("Error reading dir for reading {full_dir_path:?}")0
)
?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 {

  Branch (383:23): [Folded - Ignored]

  Branch (383:23): [Folded - Ignored]

  Branch (383:23): [True: 1, False: 1]

  Branch (383:23): [True: 4, False: 2]

                let entry = entry_result.err_tip(|| 
"Error while iterating directory"0
)
?0
;

                let file_type = entry

                    .file_type()

                    .await

                    .err_tip(|| 
format!("Error running file_type() on {entry:?}")0
)
?0
;

                let full_path = full_dir_path.as_ref().join(entry.path());

                if file_type.is_dir() {

  Branch (390:20): [Folded - Ignored]

  Branch (390:20): [Folded - Ignored]

  Branch (390:20): [True: 0, False: 1]

  Branch (390:20): [True: 1, False: 3]

                    let full_dir_path = full_dir_path.clone();

                    dir_futures.push(

                        upload_directory(cas_store, full_path.clone(), full_work_directory, hasher)

                            .and_then(|(dir, all_dirs)| async move {

                                let directory_name = full_path

                                    .file_name()

                                    .err_tip(|| {

                                        format!("Expected file_name to exist on {full_dir_path:?}")

                                    })
?0

                                    .to_str()

                                    .err_tip(|| {

                                        make_err!(

                                            Code::Internal,

                                            "Could not convert {:?} to string",

                                            full_dir_path

                                        )

                                    })
?0

                                    .to_string();

                                let digest = serialize_and_upload_message(

                                    &dir,

                                    cas_store,

                                    &mut hasher.hasher(),

                                )

                                .await

                                .err_tip(|| 
format!("for {full_path:?}")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() {

  Branch (428:27): [Folded - Ignored]

  Branch (428:27): [Folded - Ignored]

  Branch (428:27): [True: 0, False: 1]

  Branch (428:27): [True: 3, False: 0]

                    file_futures.push(async move {

                        let metadata = fs::metadata(&full_path)

                            .await

                            .err_tip(|| 
format!("Could not open file {full_path:?}")0
)
?0
;

                        upload_file(cas_store, &full_path, hasher, metadata)

                            .map_ok(Into::into)

                            .await

                    });

                } else if 
file_type.is_symlink()1
 {

  Branch (437:27): [Folded - Ignored]

  Branch (437:27): [Folded - Ignored]

  Branch (437:27): [True: 1, False: 0]

  Branch (437:27): [True: 0, False: 0]

                    symlink_futures

                        .push(upload_symlink(full_path, &full_work_directory).map_ok(Into::into));

                
}0

            }

        }

        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_node, 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.name.cmp(&b.name)1
);

        directory_nodes.sort_unstable_by(|a, b| 
a.name.cmp(&b.name)0
);

        symlinks.sort_unstable_by(|a, b| 
a.name.cmp(&b.name)0
);

        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 {

  Branch (490:20): [Folded - Ignored]

  Branch (490:20): [Folded - Ignored]

  Branch (490:20): [True: 0, False: 0]

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

            make_input_err!(

                "Could not convert contents of side channel file (json) to SideChannelInfo : {e:?}"

            )

        })?;

    Ok(side_channel_info.failure.map(|failure| match failure {

        SideChannelFailureReason::timeout => Error::new(

            Code::DeadlineExceeded,

            format!(

                "Command '{}' timed out after {} seconds",

                args.join(OsStr::new(" ")).to_string_lossy(),

                timeout.as_secs_f32()

            ),

        ),

    }))

}

async fn do_cleanup(

    running_actions_manager: &RunningActionsManagerImpl,

    operation_id: &OperationId,

    action_directory: &str,

) -> Result<(), Error> {

    event!(Level::INFO, 
"Worker cleaning up"0
);

    // 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!("Could not remove working directory {action_directory}")0
);

    if let Err(
err0
) = running_actions_manager.cleanup_action(operation_id) {

  Branch (531:12): [True: 0, False: 0]

  Branch (531:12): [Folded - Ignored]

  Branch (531:12): [True: 0, False: 15]

        event!(

            Level::ERROR,

            ?operation_id,

            ?err,

            "Error cleaning up action"

        );

        return Result::<(), Error>::Err(err).merge(remove_dir_result);

    }

    if let Err(
err0
) = remove_dir_result {

  Branch (540:12): [True: 0, False: 0]

  Branch (540:12): [Folded - Ignored]

  Branch (540:12): [True: 0, False: 15]

        event!(

            Level::ERROR,

            ?operation_id,

            ?err,

            "Error removing working directory"

        );

        return Err(err);

    }

    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;

}

struct RunningActionImplExecutionResult {

    stdout: Bytes,

    stderr: Bytes,

    exit_code: i32,

}

struct RunningActionImplState {

    command_proto: Option<ProtoCommand>,

    // TODO(allada) 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>,

}

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

    did_cleanup: AtomicBool,

}

impl RunningActionImpl {

    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,

            }),

            did_cleanup: AtomicBool::new(false),

        }

    }

    fn metrics(&self) -> &Arc<Metrics> {

        &self.running_actions_manager.metrics

    }

    /// Prepares any actions needed to execution 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"0
)

            });

            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!("Error creating work directory {}", self.work_directory)0
)
?0
;

                // Download the input files/folder and place them into the temp directory.

                self.metrics()

                    .download_to_directory

                    .wrap(download_to_directory(

                        &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() {

  Branch (694:43): [Folded - Ignored]

  Branch (694:43): [Folded - Ignored]

  Branch (694:43): [True: 1, False: 8]

                    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!("Parent path for {full_output_path} has no parent")0
)
?0
;

                    fs::create_dir_all(full_parent_path).await.err_tip(|| {

                        format!(

                            "Error creating output directory {} (file)",

                            full_parent_path.display()

                        )

                    })
?0
;

                    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
;

        }

        event!(Level::INFO, ?command, 
"Worker received command"0
,);

        {

            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)

    }

    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
)
?0
,

                state

                    .kill_channel_rx

                    .take()

                    .err_tip(|| 
"Expected state to have kill_channel_rx in execute()"0
)
?0

                    // This is important as we may be killed at any point.

                    .fuse(),

            )

        };

        if command_proto.arguments.is_empty() {

  Branch (756:12): [Folded - Ignored]

  Branch (756:12): [Folded - Ignored]

  Branch (756:12): [True: 0, False: 13]

            return Err(make_input_err!("No arguments provided in Command proto"));

        }

        let args: Vec<&OsStr> = if let Some(
entrypoint0
) = &self

  Branch (759:40): [Folded - Ignored]

  Branch (759:40): [Folded - Ignored]

  Branch (759:40): [True: 0, False: 13]

            .running_actions_manager

            .execution_configuration

            .entrypoint

        {

            std::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()

        };

        event!(Level::INFO, ?args, 
"Executing command"0
,);

        let mut command_builder = process::Command::new(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() {

  Branch (784:36): [Folded - Ignored]

  Branch (784:36): [Folded - Ignored]

  Branch (784:36): [True: 11, False: 2]

            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

  Branch (791:16): [Folded - Ignored]

  Branch (791:16): [Folded - Ignored]

  Branch (791:16): [True: 0, False: 13]

            .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::timeout_millis => {

                        Cow::Owned(requested_timeout.as_millis().to_string())

                    }

                    EnvironmentSource::side_channel_file => {

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

                        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_variable13
 in envs {

            command_builder.env(&environment_variable.name, &environment_variable.value);

        }

        let mut child_process = command_builder

            .spawn()

            .err_tip(|| 
format!("Could not execute command {args:?}")0
)
?0
;

        let mut stdout_reader = child_process

            .stdout

            .take()

            .err_tip(|| 
"Expected stdout to exist on command this should never happen"0
)
?0
;

        let mut stderr_reader = child_process

            .stderr

            .take()

            .err_tip(|| 
"Expected stderr to exist on command this should never happen"0
)
?0
;

        let mut child_process_guard = guard(child_process, |mut child_process| {

            event!(

                Level::ERROR,

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

                child_process.kill().await

            });

        
}0
);

        let all_stdout_fut = spawn!("stdout_reader", async move {

            let mut all_stdout = BytesMut::new();

            loop {

                let 
sz13
 = stdout_reader

                    .read_buf(&mut all_stdout)

                    .await

                    .err_tip(|| 
"Error reading stdout stream"0
)
?0
;

                if sz == 0 {

  Branch (879:20): [Folded - Ignored]

  Branch (879:20): [Folded - Ignored]

  Branch (879:20): [True: 10, False: 3]

                    break; // EOF.

                }

            }

            Result::<Bytes, Error>::Ok(all_stdout.freeze())

        
}10
);

        let all_stderr_fut = spawn!("stderr_reader", async move {

            let mut all_stderr = BytesMut::new();

            loop {

                let 
sz14
 = stderr_reader

                    .read_buf(&mut all_stderr)

                    .await

                    .err_tip(|| 
"Error reading stderr stream"0
)
?0
;

                if sz == 0 {

  Branch (892:20): [Folded - Ignored]

  Branch (892:20): [Folded - Ignored]

  Branch (892:20): [True: 11, False: 3]

                    break; // EOF.

                }

            }

            Result::<Bytes, Error>::Ok(all_stderr.freeze())

        
}11
);

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

  Branch (907:28): [Folded - Ignored]

  Branch (907:28): [Folded - Ignored]

  Branch (907:28): [True: 0, False: 2]

                        event!(

                            Level::ERROR,

                            ?err,

                            "Could not kill process in RunningActionsManager for action timeout",

                        );

                    }

                    {

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

                                args.join(OsStr::new(" ")).to_string_lossy(),

                                self.action_info.timeout.as_secs_f32()

                            )

                        )));

                    }

                },

                
maybe_exit_status13
 = child_process_guard.wait() => {

                    // Defuse our guard so it does not try to cleanup and make nessless logs.