Add completeness checking store

Author: blakehatch

Cargo.lock

@@ -29,6 +29,8 @@ members = [
   "gencargo/cas_server",
   "gencargo/cas_server_test",
   "gencargo/common",
+  "gencargo/completeness_checking_store",
+  "gencargo/completeness_checking_store_test",
   "gencargo/compression_store",
   "gencargo/compression_store_test",
   "gencargo/config",
@@ -177,6 +179,8 @@ cas = { path = "gencargo/cas" }
 cas_server = { path = "gencargo/cas_server" }
 cas_server_test = { path = "gencargo/cas_server_test" }
 common = { path = "gencargo/common" }
+completeness_checking_store = { path = "gencargo/completeness_checking_store" }
+completeness_checking_store_test = { path = "gencargo/completeness_checking_store_test" }
 compression_store = { path = "gencargo/compression_store" }
 compression_store_test = { path = "gencargo/compression_store_test" }
 config = { path = "gencargo/config" }

Cargo.toml

@@ -748,6 +748,69 @@ impl From<ActionResult> for ProtoActionResult {
     }
 }
 
+impl TryFrom<ProtoActionResult> for ActionResult {
+    type Error = Error;
+
+    fn try_from(val: ProtoActionResult) -> Result<Self, Error> {
+        let output_file_symlinks: Result<Vec<_>, _> = val
+            .output_file_symlinks
+            .into_iter()
+            .map(|output_symlink| {
+                SymlinkInfo::try_from(output_symlink)
+                    .err_tip(|| "Output File Symlinks could not be converted to SymlinkInfo")
+            })
+            .collect();
+
+        let output_directory_symlinks: Result<Vec<_>, _> = val
+            .output_directory_symlinks
+            .into_iter()
+            .map(|output_symlink| {
+                SymlinkInfo::try_from(output_symlink)
+                    .err_tip(|| "Output File Symlinks could not be converted to SymlinkInfo")
+            })
+            .collect();
+
+        let output_files: Result<Vec<_>, _> = val
+            .output_files
+            .into_iter()
+            .map(|output_file| output_file.try_into().err_tip(|| "Output File could not be converted"))
+            .collect();
+
+        let output_folders: Result<Vec<_>, _> = val
+            .output_directories
+            .into_iter()
+            .map(|output_directory| {
+                output_directory
+                    .try_into()
+                    .err_tip(|| "Output File could not be converted")
+            })
+            .collect();
+
+        Ok(Self {
+            output_files: output_files?,
+            output_folders: output_folders?,
+            output_file_symlinks: output_file_symlinks?,
+            output_directory_symlinks: output_directory_symlinks?,
+            exit_code: val.exit_code,
+            stdout_digest: val
+                .stdout_digest
+                .map(|digest| digest.try_into().unwrap_or(DigestInfo::empty_digest()))
+                .unwrap_or(DigestInfo::empty_digest()),
+            stderr_digest: val
+                .stderr_digest
+                .map(|digest| digest.try_into().unwrap_or(DigestInfo::empty_digest()))
+                .unwrap_or_else(DigestInfo::empty_digest),
+            execution_metadata: val
+                .execution_metadata
+                .map(|metadata| metadata.try_into().unwrap_or(ExecutionMetadata::default()))
+                .unwrap_or_default(),
+            server_logs: Default::default(),
+            error: None,
+            message: String::new(),
+        })
+    }
+}
+
 impl TryFrom<ExecuteResponse> for ActionStage {
     type Error = Error;
 

cas/scheduler/action_messages.rs

@@ -31,6 +31,7 @@ rust_library(
         ":ref_store",
         ":s3_store",
         ":shard_store",
+        ":completeness_checking_store",
         ":size_partitioning_store",
         ":store",
         ":traits",
@@ -126,6 +127,29 @@ rust_library(
     ],
 )
 
+rust_library(
+    name = "completeness_checking_store",
+    srcs = ["completeness_checking_store.rs"],
+    proc_macro_deps = ["@crate_index//:async-trait"],
+    visibility = ["//cas:__pkg__"],
+    deps = [
+        ":traits",
+        ":ac_utils",
+        "//proto",
+        "//config",
+        "//util:buf_channel",
+        "//util:common",
+        "//util:error",
+        "//util:metrics_utils",   
+        "//cas/scheduler:action_messages",
+        "@crate_index//:futures",
+        "@crate_index//:hashbrown",
+        "@crate_index//:hex",
+        "@crate_index//:sha2",
+        "@crate_index//:tokio",
+    ],
+)
+
 rust_library(
     name = "verify_store",
     srcs = ["verify_store.rs"],
@@ -500,6 +524,27 @@ rust_test(
     ],
 )
 
+rust_test(
+    name = "completeness_checking_store_test",
+    srcs = ["tests/completeness_checking_store_test.rs"],
+    deps = [
+        ":memory_store",
+        ":traits",
+        ":completeness_checking_store",
+        ":store",
+        "//proto",
+        "//config",
+        "//util:buf_channel",
+        "//util:common",
+        "//util:error",
+        "//cas/scheduler:action_messages",
+        "@crate_index//:prost",
+        "@crate_index//:futures",
+        "@crate_index//:pretty_assertions",
+        "@crate_index//:tokio",
+    ],
+)
+
 rust_test(
     name = "filesystem_store_test",
     srcs = ["tests/filesystem_store_test.rs"],