Restructure LocalWorker for easier testing

Abstracts away most of the LocalWorker implementation classes
to enable easier unit testing.

cas/worker/BUILD

@@ -6,7 +6,7 @@ rust_library(
     name = "local_worker",
     srcs = ["local_worker.rs"],
     deps = [
-         "//cas/store",
+        "//cas/store",
         "//config",
         "//proto",
         "//third_party:futures",
@@ -65,24 +65,51 @@ rust_library(
 
 rust_library(
     name = "local_worker_test_utils",
-    srcs = ["local_worker_test_utils.rs"],
+    srcs = ["tests/utils/local_worker_test_utils.rs"],
     testonly = True,
     deps = [
         "//config",
         "//proto",
-        "//third_party:fast_async_mutex",
         "//third_party:hyper",
         "//third_party:tokio",
         "//third_party:tonic",
         "//util:common",
         "//util:error",
         ":local_worker",
-        ":running_actions_manager",
+        ":mock_running_actions_manager",
+        ":mock_worker_api_client",
+    ],
+)
+
+
+rust_library(
+    name = "mock_worker_api_client",
+    srcs = ["tests/utils/mock_worker_api_client.rs"],
+    testonly = True,
+    deps = [
+        "//proto",
+        "//third_party:fast_async_mutex",
+        "//third_party:tokio",
+        "//third_party:tonic",
         ":worker_api_client_wrapper",
     ],
     proc_macro_deps = ["//third_party:async_trait"],
 )
 
+rust_library(
+    name = "mock_running_actions_manager",
+    srcs = ["tests/utils/mock_running_actions_manager.rs"],
+    testonly = True,
+    deps = [
+        "//proto",
+        "//third_party:fast_async_mutex",
+        "//third_party:tokio",
+        "//util:error",
+        ":running_actions_manager",
+    ],
+    proc_macro_deps = ["//third_party:async_trait"],
+)
+
 rust_test(
     name = "local_worker_test",
     srcs = ["tests/local_worker_test.rs"],
@@ -99,6 +126,7 @@ rust_test(
         "//util:common",
         "//util:error",
         ":local_worker_test_utils",
+        ":mock_running_actions_manager",
     ],
-    proc_macro_deps = ["//third_party:async_trait", "//third_party:ctor"],
+    proc_macro_deps = ["//third_party:ctor"],
 )

cas/worker/local_worker.rs

@@ -4,7 +4,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-use futures::{future::BoxFuture, select, stream::FuturesUnordered, FutureExt, StreamExt};
+use futures::{future::BoxFuture, select, stream::FuturesUnordered, FutureExt, StreamExt, TryFutureExt};
 use tokio::sync::mpsc;
 use tokio::time::sleep;
 use tokio_stream::wrappers::UnboundedReceiverStream;
@@ -17,7 +17,7 @@ use proto::com::github::allada::turbo_cache::remote_execution::{
     execute_result, update_for_worker::Update, worker_api_client::WorkerApiClient, ExecuteFinishedResult,
     ExecuteResult, KeepAliveRequest, UpdateForWorker,
 };
-use running_actions_manager::{RunningActionsManager, RunningActionsManagerImpl};
+use running_actions_manager::{RunningAction, RunningActionsManager, RunningActionsManagerImpl};
 use store::Store;
 use worker_api_client_wrapper::{WorkerApiClientTrait, WorkerApiClientWrapper};
 use worker_utils::make_supported_properties;
@@ -118,7 +118,15 @@ impl<'a, T: WorkerApiClientTrait, U: RunningActionsManager> LocalWorkerImpl<'a,
                         Update::StartAction(start_execute) => {
                             let add_future_channel = add_future_channel.clone();
                             let mut grpc_client = self.grpc_client.clone();
-                            let start_action_fut = self.running_actions_manager.clone().start_action(start_execute);
+                            let start_action_fut = self
+                                .running_actions_manager
+                                .clone()
+                                .create_and_add_action(start_execute)
+                                .and_then(|action| action.prepare_action())
+                                .and_then(|action| action.execute())
+                                .and_then(|action| action.upload_results())
+                                .and_then(|action| action.cleanup())
+                                .and_then(|action| action.get_finished_result());
 
                             let make_publish_future = move |res: Result<ExecuteFinishedResult, Error>| async move {
                                 match res {