Add opt-in zstd wire compression to GrpcStore transfers

nativelink-config/src/stores.rs

@@ -1475,6 +1475,30 @@ pub struct GrpcSpec {
     /// Default: unset (disabled). When unset there is zero behavior change.
     #[serde(default)]
     pub experimental_read_batching: Option<GrpcReadBatchingConfig>,
+
+    /// Compress this store's own blob transfers on the wire with REAPI
+    /// `compressed-blobs/zstd`. Uploads and full-blob downloads of blobs at
+    /// or above 64 KiB are zstd-compressed; smaller blobs and ranged reads
+    /// keep the identity path.
+    ///
+    /// The upstream instance must have
+    /// `capabilities.remote_cache_compression` enabled, otherwise compressed
+    /// requests fail with `InvalidArgument`. This setting is what makes
+    /// NativeLink-to-NativeLink hops (for example worker to CAS) benefit
+    /// from wire compression; it is independent of what external clients
+    /// such as Bazel negotiate for themselves.
+    ///
+    /// Compressed uploads do not resume mid-stream (mirroring the REAPI
+    /// server contract): a transport failure part-way through a compressed
+    /// upload surfaces immediately to the caller instead of retrying, and
+    /// outer callers retry the whole upload.
+    ///
+    /// When combined with `experimental_chunked_uploads`, chunked uploads
+    /// take precedence for blobs at or above the chunking threshold.
+    ///
+    /// Default: false (disabled).
+    #[serde(default, deserialize_with = "convert_boolean_with_shellexpand")]
+    pub experimental_remote_cache_compression: bool,
 }
 
 /// Configuration for experimental small-blob read coalescing in a gRPC

nativelink-config/tests/deserialization_test.rs

@@ -382,6 +382,8 @@ mod optional_values_tests {
 }
 
 mod boolean_tests {
+    use nativelink_config::stores::GrpcSpec;
+
     use crate::BoolEntity;
 
     #[test]
@@ -404,6 +406,32 @@ mod boolean_tests {
             assert_eq!(deserialized.value, expected, "{input}");
         }
     }
+
+    #[test]
+    fn grpc_compression_accepts_string_and_environment_boolean() {
+        let string_spec: GrpcSpec = serde_json5::from_str(
+            r#"{
+                endpoints: [{ address: "http://localhost:1234" }],
+                store_type: "cas",
+                experimental_remote_cache_compression: "true"
+            }"#,
+        )
+        .expect("string boolean should deserialize");
+        assert!(string_spec.experimental_remote_cache_compression);
+
+        // Safety: this test uses a unique variable and does not run code that
+        // reads mutable environment state concurrently.
+        unsafe { std::env::set_var("TEST_GRPC_REMOTE_CACHE_COMPRESSION", "false") };
+        let environment_spec: GrpcSpec = serde_json5::from_str(
+            r#"{
+                endpoints: [{ address: "http://localhost:1234" }],
+                store_type: "cas",
+                experimental_remote_cache_compression: "${TEST_GRPC_REMOTE_CACHE_COMPRESSION}"
+            }"#,
+        )
+        .expect("environment boolean should deserialize");
+        assert!(!environment_spec.experimental_remote_cache_compression);
+    }
 }
 
 mod shellexpand_tests {

nativelink-service/BUILD.bazel

@@ -69,6 +69,7 @@ rust_test_suite(
         "tests/execution_server_test.rs",
         "tests/fastcdc_conformance_test.rs",
         "tests/fetch_server_test.rs",
+        "tests/grpc_wire_compression_test.rs",
         "tests/health_server_test.rs",
         "tests/push_server_test.rs",
         "tests/wire_compression_test.rs",

nativelink-service/src/bytestream_server.rs

@@ -1242,6 +1242,7 @@ impl ByteStreamServer {
         let encode_fut = Box::pin(crate::wire_compression::stream_encode_compressed_download(
             raw_rx,
             wire_compressor,
+            crate::wire_compression::ZSTD_COMPRESSION_LEVEL,
             compressed_tx,
         ));