empty find_missing_blobs can return immediately (TraceMachina#2217)

Author: palfrey

nativelink-store/BUILD.bazel

@@ -118,6 +118,7 @@ rust_test_suite(
         "tests/filesystem_store_test.rs",
         "tests/gcs_client_test.rs",
         "tests/gcs_store_test.rs",
+        "tests/grpc_store_test.rs",
         "tests/memory_store_test.rs",
         "tests/mongo_store_test.rs",
         "tests/ontap_s3_existence_cache_store_test.rs",
@@ -167,6 +168,7 @@ rust_test_suite(
         "@crates//:tempfile",
         "@crates//:tokio",
         "@crates//:tokio-stream",
+        "@crates//:tonic",
         "@crates//:tracing",
         "@crates//:tracing-test",
         "@crates//:uuid",

nativelink-store/src/grpc_store.rs

@@ -141,11 +141,23 @@ impl GrpcStore {
         );
 
         let mut request = grpc_request.into_inner();
+
+        // Some builds (Chromium for example) do lots of empty requests for some reason, so shortcut them
+        if request.blob_digests.is_empty() {
+            return Ok(Response::new(FindMissingBlobsResponse {
+                missing_blob_digests: vec![],
+            }));
+        }
+
         request.instance_name.clone_from(&self.instance_name);
         self.perform_request(request, |request| async move {
             let channel = self
                 .connection_manager
-                .connection(format!("find_missing_blobs: {:?}", request.blob_digests))
+                .connection(format!(
+                    "find_missing_blobs: ({}) {:?}",
+                    request.blob_digests.len(),
+                    request.blob_digests
+                ))
                 .await
                 .err_tip(|| "in find_missing_blobs")?;
             ContentAddressableStorageClient::new(channel)

nativelink-store/tests/grpc_store_test.rs

@@ -0,0 +1,45 @@
+use core::time::Duration;
+
+use nativelink_config::stores::{GrpcEndpoint, GrpcSpec, Retry, StoreType};
+use nativelink_error::Error;
+use nativelink_macro::nativelink_test;
+use nativelink_proto::build::bazel::remote::execution::v2::{
+    FindMissingBlobsRequest, digest_function,
+};
+use nativelink_store::grpc_store::GrpcStore;
+use tokio::time::timeout;
+use tonic::Request;
+
+#[nativelink_test]
+async fn fast_find_missing_blobs() -> Result<(), Error> {
+    let spec = GrpcSpec {
+        instance_name: String::new(),
+        endpoints: vec![GrpcEndpoint {
+            address: "http://foobar".into(),
+            tls_config: None,
+            concurrency_limit: None,
+            connect_timeout_s: 0,
+            tcp_keepalive_s: 0,
+            http2_keepalive_interval_s: 0,
+            http2_keepalive_timeout_s: 0,
+        }],
+        store_type: StoreType::Cas,
+        retry: Retry::default(),
+        max_concurrent_requests: 0,
+        connections_per_endpoint: 0,
+        rpc_timeout_s: 1,
+    };
+    let store = GrpcStore::new(&spec).await?;
+    let request = Request::new(FindMissingBlobsRequest {
+        instance_name: String::new(),
+        blob_digests: vec![],
+        digest_function: digest_function::Value::Sha256.into(),
+    });
+    let res = timeout(Duration::from_secs(1), async move {
+        store.find_missing_blobs(request).await
+    })
+    .await??;
+    let inner_res = res.into_inner();
+    assert_eq!(inner_res.missing_blob_digests.len(), 0);
+    Ok(())
+}

nativelink-util/BUILD.bazel

@@ -108,6 +108,7 @@ rust_test_suite(
         "tests/proto_stream_utils_test.rs",
         "tests/resource_info_test.rs",
         "tests/retry_test.rs",
+        "tests/store_trait_test.rs",
         "tests/telemetry_test.rs",
         "tests/tls_utils_test.rs",
     ],
@@ -124,6 +125,7 @@ rust_test_suite(
         ":nativelink-util",
         "//nativelink-config",
         "//nativelink-error",
+        "//nativelink-metric",
         "//nativelink-proto",
         "@crates//:axum",
         "@crates//:bytes",

nativelink-util/src/store_trait.rs

@@ -15,6 +15,7 @@
 use core::borrow::{Borrow, BorrowMut};
 use core::convert::Into;
 use core::fmt::{self, Debug, Display};
+use core::future;
 use core::hash::{Hash, Hasher};
 use core::ops::{Bound, RangeBounds};
 use core::pin::Pin;
@@ -455,6 +456,9 @@ pub trait StoreLike: Send + Sync + Sized + Unpin + 'static {
         &'a self,
         digests: &'a [StoreKey<'a>],
     ) -> impl Future<Output = Result<Vec<Option<u64>>, Error>> + Send + 'a {
+        if digests.is_empty() {
+            return future::ready(Ok(vec![])).boxed();
+        }
         self.as_store_driver_pin().has_many(digests)
     }
 
@@ -466,6 +470,9 @@ pub trait StoreLike: Send + Sync + Sized + Unpin + 'static {
         digests: &'a [StoreKey<'a>],
         results: &'a mut [Option<u64>],
     ) -> impl Future<Output = Result<(), Error>> + Send + 'a {
+        if digests.is_empty() {
+            return future::ready(Ok(())).boxed();
+        }
         self.as_store_driver_pin()
             .has_with_results(digests, results)
     }

nativelink-util/tests/store_trait_test.rs

@@ -0,0 +1,86 @@
+use core::pin::Pin;
+use std::sync::Arc;
+
+use nativelink_error::Error;
+use nativelink_macro::nativelink_test;
+use nativelink_metric::MetricsComponent;
+use nativelink_util::buf_channel::{DropCloserReadHalf, DropCloserWriteHalf};
+use nativelink_util::default_health_status_indicator;
+use nativelink_util::health_utils::HealthStatusIndicator;
+use nativelink_util::store_trait::{
+    RemoveItemCallback, Store, StoreDriver, StoreKey, StoreLike, UploadSizeInfo,
+};
+use tonic::async_trait;
+
+#[derive(Debug, MetricsComponent)]
+struct FakeStore {}
+
+#[async_trait]
+#[allow(clippy::todo)]
+impl StoreDriver for FakeStore {
+    async fn has_with_results(
+        self: Pin<&Self>,
+        _keys: &[StoreKey<'_>],
+        _results: &mut [Option<u64>],
+    ) -> Result<(), Error> {
+        todo!();
+    }
+
+    async fn update(
+        self: Pin<&Self>,
+        _key: StoreKey<'_>,
+        _reader: DropCloserReadHalf,
+        _size_info: UploadSizeInfo,
+    ) -> Result<(), Error> {
+        todo!();
+    }
+
+    async fn get_part(
+        self: Pin<&Self>,
+        _key: StoreKey<'_>,
+        _writer: &mut DropCloserWriteHalf,
+        _offset: u64,
+        _length: Option<u64>,
+    ) -> Result<(), Error> {
+        todo!();
+    }
+
+    fn inner_store(&self, _digest: Option<StoreKey>) -> &dyn StoreDriver {
+        self
+    }
+
+    fn as_any(&self) -> &(dyn core::any::Any + Sync + Send + 'static) {
+        self
+    }
+
+    fn as_any_arc(self: Arc<Self>) -> Arc<dyn core::any::Any + Sync + Send + 'static> {
+        self
+    }
+
+    fn register_remove_callback(
+        self: Arc<Self>,
+        _callback: Arc<dyn RemoveItemCallback>,
+    ) -> Result<(), Error> {
+        todo!();
+    }
+}
+
+default_health_status_indicator!(FakeStore);
+
+#[nativelink_test]
+async fn fast_has_with_results() -> Result<(), Error> {
+    let store = Store::new(Arc::new(FakeStore {}));
+    let mut results: [Option<u64>; 0] = [];
+    store.has_with_results(&[], &mut results).await?;
+
+    Ok(())
+}
+
+#[nativelink_test]
+async fn fast_has_many() -> Result<(), Error> {
+    let store = Store::new(Arc::new(FakeStore {}));
+    let res = store.has_many(&[]).await?;
+    assert!(res.is_empty());
+
+    Ok(())
+}