From 7c78ff5fffcb1ad9b066e4bfb36103552f550802 Mon Sep 17 00:00:00 2001 From: b7r6 Date: Sat, 3 Oct 2026 00:26:26 -0400 Subject: [PATCH 1/2] feat(cas): make the BatchReadBlobs per-blob timeout configurable BatchReadBlobs applies a hardcoded 30 s deadline to each blob read so a single stalled read cannot pin the whole batch until the client's overall deadline. Under cold large-tree reads against a slow backing tier -- e.g. an R2/S3 object store behind a fast store doing writeback backfill on read -- large blobs legitimately take longer than 30 s to materialize, and the hardcoded deadline fails the read with DeadlineExceeded even though the transfer is making progress, failing the entire batch. Add an optional per-instance `batch_read_per_blob_timeout_s` field to the CAS service config. Unset preserves today's 30 s default exactly; deployments with slow cold tiers can raise it. The simplest defensible design -- a configurable base rather than a size-scaled formula -- was chosen deliberately: the operator knows their tier's throughput floor, and a scaling heuristic baked into the server is hard to tune and harder to remove later. BatchUpdateBlobs keeps the fixed default. The new regression test stalls the store for 2 minutes and shows the read succeeds with a configured 3 minute timeout; with the fix reverted the configured value is ignored and the read fails with DeadlineExceeded at 30 s (the existing default-timeout test continues to cover the unset case). Co-Authored-By: Claude Opus 4.8 --- nativelink-config/src/cas_server.rs | 11 +++ nativelink-service/src/cas_server.rs | 27 +++++++- nativelink-service/tests/cas_server_test.rs | 67 ++++++++++++++++++- .../tests/grpc_wire_compression_test.rs | 1 + 4 files changed, 100 insertions(+), 6 deletions(-) diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs index 1b4aac223fe..bc1561a294b 100644 --- a/nativelink-config/src/cas_server.rs +++ b/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, + + /// 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, } #[derive(Deserialize, Serialize, Debug, Clone)] diff --git a/nativelink-service/src/cas_server.rs b/nativelink-service/src/cas_server.rs index 953c2bc6e5a..fd3d4832896 100644 --- a/nativelink-service/src/cas_server.rs +++ b/nativelink-service/src/cas_server.rs @@ -173,6 +173,10 @@ impl ChunkingInstance { #[derive(Debug)] pub struct CasServer { stores: HashMap, + /// 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, remote_cache_compression_instances: RemoteCacheCompressionInstances, chunking_instances: HashMap, chunking_metrics: ChunkingMetrics, @@ -180,7 +184,9 @@ pub struct CasServer { type GetTreeStream = Pin> + 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 { 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, )), }; diff --git a/nativelink-service/tests/cas_server_test.rs b/nativelink-service/tests/cas_server_test.rs index 055bfe9a884..3ee5477c25f 100644 --- a/nativelink-service/tests/cas_server_test.rs +++ b/nativelink-service/tests/cas_server_test.rs @@ -78,6 +78,7 @@ fn make_cas_server(store_manager: &StoreManager) -> Result { 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, _key: StoreKey<'_>, - _writer: &mut DropCloserWriteHalf, + writer: &mut DropCloserWriteHalf, _offset: u64, _length: Option, ) -> 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) -> &'_ 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 { + 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, +) -> Result { 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> { + 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> { 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 Result<(), Box Result<(), Box Date: Wed, 21 Oct 2026 17:30:19 +0000 Subject: [PATCH 2/2] docs(config-reference): regenerate main for `batch_read_per_blob_timeout_s` Regenerated web/apps/docs/content/docs/reference/nativelink-config/main.mdx from the nativelink-config crate (build-schema + schema-to-mdx), as AGENTS.md requires when a config field or its doc comment changes. --- .../docs/content/docs/reference/nativelink-config/main.mdx | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx index 5d97a6a58b5..3ac09ba144d 100644 --- a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx +++ b/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 */} @@ -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