feat(cas): make the BatchReadBlobs per-blob timeout configurable

nativelink-config/src/cas_server.rs

@@ -148,6 +148,17 @@ pub struct CasStoreConfig {
     /// and behavior is identical to when this option did not exist.
     #[serde(default)]
     pub experimental_chunking: Option<CasChunkingConfig>,
+
+    /// Per-blob timeout in seconds applied inside `BatchReadBlobs`. Each
+    /// blob read in a batch races this deadline so a single stalled read
+    /// cannot pin the whole batch until the client's overall deadline.
+    /// Cold reads through a slow backing tier (e.g. an object store that
+    /// populates a local cache on read) can legitimately take longer than
+    /// the default for large blobs; raise this value for such deployments.
+    ///
+    /// Default: 30 seconds.
+    #[serde(default)]
+    pub batch_read_per_blob_timeout_s: Option<u32>,
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone)]

nativelink-service/src/cas_server.rs

@@ -173,14 +173,20 @@ impl ChunkingInstance {
 #[derive(Debug)]
 pub struct CasServer {
     stores: HashMap<InstanceName, Store>,
+    /// Per-instance per-blob deadline applied inside `BatchReadBlobs`.
+    /// Defaults to `BATCH_PER_BLOB_TIMEOUT`; configurable via
+    /// `batch_read_per_blob_timeout_s`.
+    batch_read_per_blob_timeouts: HashMap<InstanceName, Duration>,
     remote_cache_compression_instances: RemoteCacheCompressionInstances,
     chunking_instances: HashMap<InstanceName, ChunkingInstance>,
     chunking_metrics: ChunkingMetrics,
 }
 
 type GetTreeStream = Pin<Box<dyn Stream<Item = Result<GetTreeResponse, Status>> + Send + 'static>>;
 
-/// Per-blob deadline applied inside `BatchReadBlobs` / `BatchUpdateBlobs`.
+/// Default per-blob deadline applied inside `BatchReadBlobs` /
+/// `BatchUpdateBlobs`. For `BatchReadBlobs` this can be overridden per
+/// instance via `batch_read_per_blob_timeout_s`.
 const BATCH_PER_BLOB_TIMEOUT: Duration = Duration::from_secs(30);
 
 /// Maximum size of a single chunk accepted in a `SpliceBlob` request.
@@ -209,11 +215,20 @@ impl CasServer {
         remote_cache_compression_instances: &RemoteCacheCompressionInstances,
     ) -> Result<Self, Error> {
         let mut stores = HashMap::with_capacity(configs.len());
+        let mut batch_read_per_blob_timeouts = HashMap::with_capacity(configs.len());
         let mut chunking_instances = HashMap::new();
         for config in configs {
             let store = store_manager.get_store(&config.cas_store).ok_or_else(|| {
                 make_input_err!("'cas_store': '{}' does not exist", config.cas_store)
             })?;
+            batch_read_per_blob_timeouts.insert(
+                config.instance_name.clone(),
+                config
+                    .batch_read_per_blob_timeout_s
+                    .map_or(BATCH_PER_BLOB_TIMEOUT, |secs| {
+                        Duration::from_secs(secs.into())
+                    }),
+            );
             if let Some(chunking_config) = &config.experimental_chunking {
                 let avg_chunk_size_bytes = chunking_config
                     .validated_avg_chunk_size_bytes()
@@ -273,6 +288,7 @@ impl CasServer {
         }
         Ok(Self {
             stores,
+            batch_read_per_blob_timeouts,
             remote_cache_compression_instances: remote_cache_compression_instances.clone(),
             chunking_instances,
             chunking_metrics: ChunkingMetrics::default(),
@@ -411,6 +427,11 @@ impl CasServer {
         }
 
         let store_ref = &store;
+        let per_blob_timeout = self
+            .batch_read_per_blob_timeouts
+            .get(instance_name)
+            .copied()
+            .unwrap_or(BATCH_PER_BLOB_TIMEOUT);
         let remote_cache_compression_enabled = self
             .remote_cache_compression_instances
             .enabled_for(instance_name);
@@ -429,7 +450,7 @@ impl CasServer {
                     // Apply a per-blob deadline so one slow read does not
                     // make the whole batch hit the client's overall deadline.
                     let result = match tokio::time::timeout(
-                        BATCH_PER_BLOB_TIMEOUT,
+                        per_blob_timeout,
                         store_ref.get_part_unchunked(digest_copy, 0, None),
                     )
                     .await
@@ -438,7 +459,7 @@ impl CasServer {
                         Err(_elapsed) => Err(make_err!(
                             Code::DeadlineExceeded,
                             "BatchReadBlobs per-blob timeout ({} s) elapsed for digest {}",
-                            BATCH_PER_BLOB_TIMEOUT.as_secs(),
+                            per_blob_timeout.as_secs(),
                             digest_copy,
                         )),
                     };

nativelink-service/tests/cas_server_test.rs

@@ -78,6 +78,7 @@ fn make_cas_server(store_manager: &StoreManager) -> Result<CasServer, Error> {
             config: nativelink_config::cas_server::CasStoreConfig {
                 cas_store: "main_cas".to_string(),
                 experimental_chunking: None,
+                batch_read_per_blob_timeout_s: None,
             },
         }],
         store_manager,
@@ -92,6 +93,7 @@ fn make_cas_server_with_zstd(store_manager: &StoreManager) -> Result<CasServer,
             config: nativelink_config::cas_server::CasStoreConfig {
                 cas_store: "main_cas".to_string(),
                 experimental_chunking: None,
+                batch_read_per_blob_timeout_s: None,
             },
         }],
         store_manager,
@@ -700,6 +702,8 @@ async fn batch_update_blobs_two_items_existence_with_third_missing()
 #[derive(Debug, MetricsComponent)]
 struct StallStore {
     delay: Duration,
+    /// Data returned by `get_part` once `delay` has elapsed.
+    data: Bytes,
 }
 
 #[async_trait]
@@ -732,12 +736,15 @@ impl StoreDriver for StallStore {
     async fn get_part(
         self: Pin<&Self>,
         _key: StoreKey<'_>,
-        _writer: &mut DropCloserWriteHalf,
+        writer: &mut DropCloserWriteHalf,
         _offset: u64,
         _length: Option<u64>,
     ) -> Result<(), Error> {
         tokio::time::sleep(self.delay).await;
-        Ok(())
+        if !self.data.is_empty() {
+            writer.send(self.data.clone()).await?;
+        }
+        writer.send_eof()
     }
 
     fn inner_store(&self, _digest: Option<StoreKey>) -> &'_ dyn StoreDriver {
@@ -760,14 +767,23 @@ impl StoreDriver for StallStore {
 default_health_status_indicator!(StallStore);
 
 fn make_cas_server_with_stall_store(delay: Duration) -> Result<CasServer, Error> {
+    make_cas_server_with_stall_store_and_timeout(delay, Bytes::new(), None)
+}
+
+fn make_cas_server_with_stall_store_and_timeout(
+    delay: Duration,
+    data: Bytes,
+    batch_read_per_blob_timeout_s: Option<u32>,
+) -> Result<CasServer, Error> {
     let store_manager = Arc::new(StoreManager::new());
-    store_manager.add_store("main_cas", Store::new(Arc::new(StallStore { delay })))?;
+    store_manager.add_store("main_cas", Store::new(Arc::new(StallStore { delay, data })))?;
     CasServer::new(
         &[WithInstanceName {
             instance_name: INSTANCE_NAME.to_string(),
             config: nativelink_config::cas_server::CasStoreConfig {
                 cas_store: "main_cas".to_string(),
                 experimental_chunking: None,
+                batch_read_per_blob_timeout_s,
             },
         }],
         &store_manager,
@@ -858,6 +874,47 @@ async fn batch_read_blobs_per_blob_timeout_returns_deadline_exceeded()
     Ok(())
 }
 
+#[nativelink_test(start_paused = true)]
+async fn batch_read_blobs_respects_configured_per_blob_timeout()
+-> Result<(), Box<dyn core::error::Error>> {
+    const VALUE: &str = "1";
+
+    // The store stalls for 2 minutes — well beyond the 30 s default that
+    // would fail the read — but the instance is configured with a 3 minute
+    // per-blob timeout, so the read must succeed.
+    let cas_server = make_cas_server_with_stall_store_and_timeout(
+        Duration::from_mins(2),
+        Bytes::from_static(VALUE.as_bytes()),
+        Some(180),
+    )?;
+
+    let digest = Digest {
+        hash: HASH1.to_string(),
+        size_bytes: VALUE.len().try_into().unwrap_or(i64::MAX),
+    };
+    let raw_response = cas_server
+        .batch_read_blobs(Request::new(BatchReadBlobsRequest {
+            instance_name: INSTANCE_NAME.to_string(),
+            digests: vec![digest.clone()],
+            acceptable_compressors: vec![compressor::Value::Identity.into()],
+            digest_function: digest_function::Value::Sha256.into(),
+        }))
+        .await;
+
+    let response = raw_response.unwrap().into_inner();
+    assert_eq!(response.responses.len(), 1);
+    let entry = &response.responses[0];
+    assert_eq!(entry.digest.as_ref(), Some(&digest));
+    let status = entry.status.as_ref().expect("status set");
+    assert_eq!(
+        status.code,
+        Code::Ok as i32,
+        "expected Ok with a raised per-blob timeout, got status: {status:?}",
+    );
+    assert_eq!(entry.data, VALUE.as_bytes());
+    Ok(())
+}
+
 #[nativelink_test]
 async fn batch_update_blobs_zstd_compressed() -> Result<(), Box<dyn core::error::Error>> {
     let store_manager = make_store_manager().await?;
@@ -1061,6 +1118,7 @@ fn make_chunking_cas_server_with_avg(
             instance_name: INSTANCE_NAME.to_string(),
             config: nativelink_config::cas_server::CasStoreConfig {
                 cas_store: "main_cas".to_string(),
+                batch_read_per_blob_timeout_s: None,
                 experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig {
                     index_store: Some("chunk_index".to_string()),
                     avg_chunk_size_bytes,
@@ -1516,6 +1574,7 @@ async fn chunking_rejects_index_store_same_as_cas_store() -> Result<(), Box<dyn
             instance_name: INSTANCE_NAME.to_string(),
             config: nativelink_config::cas_server::CasStoreConfig {
                 cas_store: "main_cas".to_string(),
+                batch_read_per_blob_timeout_s: None,
                 experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig {
                     index_store: Some("main_cas".to_string()),
                     avg_chunk_size_bytes: 0,
@@ -1575,6 +1634,7 @@ async fn chunking_on_grpc_store_forbids_index_store() -> Result<(), Box<dyn core
             instance_name: INSTANCE_NAME.to_string(),
             config: nativelink_config::cas_server::CasStoreConfig {
                 cas_store: "grpc_cas".to_string(),
+                batch_read_per_blob_timeout_s: None,
                 experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig {
                     index_store,
                     avg_chunk_size_bytes: 0,
@@ -1619,6 +1679,7 @@ async fn max_chunk_count_limits_split_and_splice() -> Result<(), Box<dyn core::e
             instance_name: INSTANCE_NAME.to_string(),
             config: nativelink_config::cas_server::CasStoreConfig {
                 cas_store: "main_cas".to_string(),
+                batch_read_per_blob_timeout_s: None,
                 experimental_chunking: Some(nativelink_config::cas_server::CasChunkingConfig {
                     index_store: Some("chunk_index".to_string()),
                     avg_chunk_size_bytes: AVG_CHUNK_SIZE,

nativelink-service/tests/grpc_wire_compression_test.rs

@@ -88,6 +88,7 @@ async fn start_real_cas_server(
             config: CasStoreConfig {
                 cas_store: "main_cas".to_string(),
                 experimental_chunking: None,
+                batch_read_per_blob_timeout_s: None,
             },
         }],
         &store_manager,

web/apps/docs/content/docs/reference/nativelink-config/main.mdx

@@ -5,7 +5,7 @@ full: true
 ---
 
 {/* AUTOGENERATED — do not edit by hand.
-   Source: nativelink-config @ main (6d6266cf)
+   Source: nativelink-config @ main (9009570165)
    Regenerate from web/: bun --filter @nativelink/docs gen:config-reference */}
 
 <ConfigVersionSwitcher />
@@ -1491,6 +1491,7 @@ eternal 3600 s), so this maps them to classes without a CAS fetch.
 | `instance_name` | string | — | `""` | Used when the config references `instance_name` in the protocol. |
 | `cas_store` | string | Yes | — | The store name referenced in the `stores` map in the main config. This store name referenced here may be reused multiple times. |
 | `experimental_chunking` | [CasChunkingConfig](#caschunkingconfig) | — | not set — chunking RPCs are rejected, nothing is advertised, | Optional and experimental: enables the REAPI `SplitBlob`/`SpliceBlob` RPCs used by content-defined chunking clients (e.g. Bazel's `--experimental_remote_cache_chunking`). When set, the capabilities service advertises blob split/splice support and `FastCDC` 2020 parameters for this instance. When `cas_store` is a grpc store the RPCs are forwarded to the backend (which must support chunking with matching parameters); otherwise they are served locally. |
+| `batch_read_per_blob_timeout_s` | integer (uint32) | — | 30 seconds | Per-blob timeout in seconds applied inside `BatchReadBlobs`. Each blob read in a batch races this deadline so a single stalled read cannot pin the whole batch until the client's overall deadline. Cold reads through a slow backing tier (e.g. an object store that populates a local cache on read) can legitimately take longer than the default for large blobs; raise this value for such deployments. |
 
 ## ActionCacheServiceConfig