Add opt-in CDC chunked uploads to GrpcStore

.github/styles/config/vocabularies/TraceMachina/accept.txt

@@ -14,6 +14,9 @@ Colab
 composable
 CPUs
 [Dd]eduplication
+[Dd]emotes?
+[Dd]emoted
+[Dd]emotion
 eviction_policy
 ELB
 Eskandar

nativelink-config/examples/chunking_cas.json5

@@ -7,6 +7,11 @@
 // before and does not advertise chunking support. When enabled, clients
 // upload and download large blobs as content-defined chunks, so small
 // changes to large outputs only transfer the chunks that changed.
+//
+// This example configures the CAS service for CDC-aware external clients.
+// NativeLink worker/StoreDriver uploads use a separate opt-in:
+// `experimental_chunked_uploads` on the worker's upstream grpc CAS store.
+// Ordinary external ByteStream uploads remain unchanged.
 {
   stores: [
     {

nativelink-config/src/stores.rs

@@ -1475,6 +1475,105 @@ pub struct GrpcSpec {
     /// Default: unset (disabled). When unset there is zero behavior change.
     #[serde(default)]
     pub experimental_read_batching: Option<GrpcReadBatchingConfig>,
+
+    /// Experimental: upload large blobs from the worker/`StoreDriver` path as
+    /// content-defined chunks. Blobs at or above `min_blob_size_bytes` are
+    /// split locally with `FastCDC` 2020, only the chunks the backend reports
+    /// missing are transferred, and the blob is assembled remotely with
+    /// `SpliceBlob`. On incremental changes to large artifacts this transfers
+    /// only the changed chunks.
+    ///
+    /// The backend MUST support the REAPI chunking extension (a `NativeLink`
+    /// CAS with `experimental_chunking` configured, or another compatible
+    /// server). During startup, `NativeLink` calls `GetCapabilities` and
+    /// requires the configured upstream instance to advertise `SplitBlob`,
+    /// `SpliceBlob`, and `FastCDC` 2020 with the same average chunk size and
+    /// seed 0. Startup fails with a configuration error when these
+    /// requirements are not met. This check is validation, not negotiation:
+    /// `NativeLink` never changes the configured parameters to match the
+    /// backend.
+    ///
+    /// If a chunk is evicted from the backend between its upload and the
+    /// final `SpliceBlob`, the upload fails with a retryable ABORTED error.
+    /// This option does not change the `ByteStream.Write` proxy path: external
+    /// `ByteStream` uploads remain ordinary streaming writes. If a chunked
+    /// worker upload fails after its input stream is consumed, the failure
+    /// surfaces to the caller's higher-level retry (normally action retry).
+    ///
+    /// Takes precedence over `experimental_remote_cache_compression` for
+    /// blobs at or above `min_blob_size_bytes` (chunks are transferred
+    /// uncompressed; the two features do not compose on the chunked path).
+    ///
+    /// WARNING (CAS sizing): a chunked upload stores each large blob twice
+    /// in the backend CAS — the chunk blobs (retained for future
+    /// incremental deduplication) plus the assembled blob. On a
+    /// size-capped CAS, budget roughly 2x the large-output working set as
+    /// headroom, and prefer a backend with post-splice chunk demotion so
+    /// chunks are evicted before primary blobs. A CAS whose eviction can
+    /// outpace a build's working set risks evicting still-referenced blobs
+    /// under build-without-the-bytes regardless of chunking; chunking
+    /// increases that pressure.
+    ///
+    /// Default: unset (disabled). When unset there is zero behavior change.
+    #[serde(default)]
+    pub experimental_chunked_uploads: Option<GrpcChunkedUploadsConfig>,
+}
+
+/// Configuration for experimental chunked uploads in a gRPC store.
+/// See [`GrpcSpec::experimental_chunked_uploads`].
+#[derive(Serialize, Deserialize, Debug, Clone, Copy)]
+#[serde(deny_unknown_fields)]
+#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
+pub struct GrpcChunkedUploadsConfig {
+    /// Only blobs at or above this size (in bytes) are uploaded as chunks.
+    /// Smaller blobs always use the plain `ByteStream` `Write` path.
+    ///
+    /// Default: 8388608 (8 MiB).
+    #[serde(
+        default = "default_chunked_uploads_min_blob_size_bytes",
+        deserialize_with = "convert_data_size_with_shellexpand"
+    )]
+    pub min_blob_size_bytes: u64,
+
+    /// The average `FastCDC` 2020 chunk size in bytes. The minimum and
+    /// maximum chunk sizes are derived from this value (avg / 4 and
+    /// avg * 4). MUST match the average chunk size the backend uses for its
+    /// own chunking so worker-uploaded and server-split chunks share digests.
+    /// The worker path currently caps this at 768 KiB so a largest possible
+    /// chunk fits under its `BatchUpdateBlobs` request budget. Must be between
+    /// 1 KiB and 768 KiB for worker uploads.
+    ///
+    /// Default: 524288 (512 KiB), the REAPI-recommended value.
+    #[serde(
+        default = "default_chunked_uploads_avg_chunk_size_bytes",
+        deserialize_with = "convert_data_size_with_shellexpand"
+    )]
+    pub avg_chunk_size_bytes: u64,
+
+    /// Blobs that could produce more than this many chunks (at the minimum
+    /// chunk size) use the plain streaming path instead, since a chunked
+    /// upload cannot fall back once the stream is partially consumed.
+    /// Should not exceed the `max_chunk_count` of the backend.
+    ///
+    /// Default: 50000 (matches the backend default; ~25 GiB at the default
+    /// average chunk size).
+    #[serde(
+        default = "default_chunked_uploads_max_chunk_count",
+        deserialize_with = "convert_numeric_with_shellexpand"
+    )]
+    pub max_chunk_count: u64,
+}
+
+const fn default_chunked_uploads_min_blob_size_bytes() -> u64 {
+    8 * 1024 * 1024 // 8 MiB.
+}
+
+const fn default_chunked_uploads_avg_chunk_size_bytes() -> u64 {
+    512 * 1024 // 512 KiB.
+}
+
+const fn default_chunked_uploads_max_chunk_count() -> u64 {
+    50_000
 }
 
 /// Configuration for experimental small-blob read coalescing in a gRPC
@@ -1876,3 +1975,44 @@ impl Retry {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{GrpcChunkedUploadsConfig, GrpcSpec};
+
+    #[test]
+    fn grpc_chunked_uploads_defaults_are_backward_compatible() {
+        let config: GrpcChunkedUploadsConfig = serde_json5::from_str("{}").unwrap();
+        assert_eq!(config.min_blob_size_bytes, 8 * 1024 * 1024);
+        assert_eq!(config.avg_chunk_size_bytes, 512 * 1024);
+        assert_eq!(config.max_chunk_count, 50_000);
+
+        let spec: GrpcSpec = serde_json5::from_str(
+            r#"{
+                endpoints: [{ address: "http://127.0.0.1:50051" }],
+                store_type: "cas"
+            }"#,
+        )
+        .unwrap();
+        assert!(spec.experimental_chunked_uploads.is_none());
+    }
+
+    #[test]
+    fn grpc_chunked_uploads_parse_data_sizes_and_reject_unknown_fields() {
+        let config: GrpcChunkedUploadsConfig = serde_json5::from_str(
+            r#"{
+                min_blob_size_bytes: "16MiB",
+                avg_chunk_size_bytes: "256KiB",
+                max_chunk_count: 1234
+            }"#,
+        )
+        .unwrap();
+        assert_eq!(config.min_blob_size_bytes, 16 * 1024 * 1024);
+        assert_eq!(config.avg_chunk_size_bytes, 256 * 1024);
+        assert_eq!(config.max_chunk_count, 1234);
+
+        let error =
+            serde_json5::from_str::<GrpcChunkedUploadsConfig>(r"{ unknown: true }").unwrap_err();
+        assert!(error.to_string().contains("unknown"));
+    }
+}

nativelink-service/src/cas_server.rs

@@ -232,6 +232,15 @@ impl CasServer {
                         "'experimental_chunking.index_store' of instance '{}' must not be set when 'cas_store' is a grpc store: SplitBlob/SpliceBlob are forwarded to the backend",
                         config.instance_name
                     );
+                    if let Some(grpc_store) = store.downcast_ref::<GrpcStore>(None)
+                        && let Some(upload_avg) = grpc_store.chunked_upload_avg_chunk_size_bytes()
+                    {
+                        error_if!(
+                            upload_avg != avg_chunk_size_bytes,
+                            "'experimental_chunking.avg_chunk_size_bytes' of instance '{}' ({avg_chunk_size_bytes}) must match the grpc store's 'experimental_chunked_uploads.avg_chunk_size_bytes' ({upload_avg})",
+                            config.instance_name
+                        );
+                    }
                     // No ChunkingInstance: the forwarding shortcut in the
                     // handlers takes over before local chunking is reached.
                     stores.insert(config.instance_name.clone(), store);

nativelink-service/tests/cas_server_test.rs

@@ -1561,6 +1561,7 @@ async fn chunking_on_grpc_store_forbids_index_store() -> Result<(), Box<dyn core
                 use_legacy_resource_names: false,
                 headers: std::collections::HashMap::new(),
                 forward_headers: vec![],
+                experimental_chunked_uploads: None,
                 experimental_read_batching: None,
             }),
             &store_manager,

nativelink-store/BUILD.bazel

@@ -74,6 +74,7 @@ rust_library(
         "@crates//:byteorder",
         "@crates//:bytes",
         "@crates//:const_format",
+        "@crates//:fastcdc",
         "@crates//:futures",
         "@crates//:gcloud-auth",
         "@crates//:gcloud-storage",
@@ -135,6 +136,7 @@ rust_test_suite(
         "tests/filesystem_store_test.rs",
         "tests/gcs_client_test.rs",
         "tests/gcs_store_test.rs",
+        "tests/grpc_chunked_upload_test.rs",
         "tests/grpc_read_batching_test.rs",
         "tests/grpc_store_test.rs",
         "tests/memory_store_test.rs",

nativelink-store/Cargo.toml

@@ -43,6 +43,7 @@ blake3 = { version = "1.8.0", default-features = false }
 byteorder = { version = "1.5.0", default-features = false }
 bytes = { version = "1.10.1", default-features = false }
 const_format = { version = "0.2.34", default-features = false }
+fastcdc = { version = "3.2.1", default-features = false, features = ["tokio"] }
 futures = { version = "0.3.31", default-features = false, features = ["std"] }
 gcloud-auth = { version = "1.3", default-features = false, features = [
   "jwt-rust-crypto",
@@ -114,7 +115,7 @@ tokio = { version = "1.52.2", features = [
 tokio-stream = { version = "0.1.17", features = [
   "fs",
 ], default-features = false }
-tokio-util = { version = "0.7.14", default-features = false }
+tokio-util = { version = "0.7.14", features = ["io"], default-features = false }
 tonic = { version = "0.14.0", features = [
   "tls-ring",
   "transport",