[9.1.0] Support remote cache CDC

src/main/java/com/google/devtools/build/lib/remote/BUILD

@@ -7,7 +7,8 @@ package(
 
 filegroup(
     name = "srcs",
-    srcs = glob(["*"]) + [
+    srcs = glob(["**"]) + [
+        "//src/main/java/com/google/devtools/build/lib/remote/chunking:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/circuitbreaker:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/common:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/disk:srcs",
@@ -97,6 +98,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib/exec/local",
         "//src/main/java/com/google/devtools/build/lib/packages/semantics",
         "//src/main/java/com/google/devtools/build/lib/profiler",
+        "//src/main/java/com/google/devtools/build/lib/remote/chunking",
         "//src/main/java/com/google/devtools/build/lib/remote/circuitbreaker",
         "//src/main/java/com/google/devtools/build/lib/remote/common",
         "//src/main/java/com/google/devtools/build/lib/remote/common:bulk_transfer_exception",

src/main/java/com/google/devtools/build/lib/remote/ChunkedBlobDownloader.java

@@ -0,0 +1,74 @@
+// Copyright 2026 The Bazel Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.devtools.build.lib.remote;
+
+import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
+
+import build.bazel.remote.execution.v2.Digest;
+import build.bazel.remote.execution.v2.SplitBlobResponse;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
+import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.List;
+
+/** Downloads blobs by sequentially fetching chunks via the SplitBlob API. */
+public class ChunkedBlobDownloader {
+  private final GrpcCacheClient grpcCacheClient;
+  private final CombinedCache combinedCache;
+
+  public ChunkedBlobDownloader(GrpcCacheClient grpcCacheClient, CombinedCache combinedCache) {
+    this.grpcCacheClient = grpcCacheClient;
+    this.combinedCache = combinedCache;
+  }
+
+  /**
+   * Downloads a blob using chunked download via the SplitBlob API. This should be called with
+   * virtual threads, as it blocks on futures via {@link
+   * com.google.devtools.build.lib.remote.util.Utils#getFromFuture}.
+   */
+  public void downloadChunked(
+      RemoteActionExecutionContext context, Digest blobDigest, OutputStream out)
+      throws IOException, InterruptedException {
+    List<Digest> chunkDigests = getChunkDigests(context, blobDigest);
+    downloadAndReassembleChunks(context, chunkDigests, out);
+  }
+
+  private List<Digest> getChunkDigests(RemoteActionExecutionContext context, Digest blobDigest)
+      throws IOException, InterruptedException {
+    if (blobDigest.getSizeBytes() == 0) {
+      return List.of();
+    }
+    ListenableFuture<SplitBlobResponse> splitResponseFuture =
+        grpcCacheClient.splitBlob(context, blobDigest);
+    if (splitResponseFuture == null) {
+      throw new CacheNotFoundException(blobDigest);
+    }
+    List<Digest> chunkDigests = getFromFuture(splitResponseFuture).getChunkDigestsList();
+    if (chunkDigests.isEmpty()) {
+      throw new CacheNotFoundException(blobDigest);
+    }
+    return chunkDigests;
+  }
+
+  private void downloadAndReassembleChunks(
+      RemoteActionExecutionContext context, List<Digest> chunkDigests, OutputStream out)
+      throws IOException, InterruptedException {
+    for (Digest chunkDigest : chunkDigests) {
+      getFromFuture(combinedCache.downloadBlob(context, chunkDigest, out));
+    }
+  }
+}

src/main/java/com/google/devtools/build/lib/remote/ChunkedBlobUploader.java

@@ -0,0 +1,121 @@
+// Copyright 2026 The Bazel Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.devtools.build.lib.remote;
+
+import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
+
+import build.bazel.remote.execution.v2.Digest;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.io.ByteStreams;
+import com.google.devtools.build.lib.remote.chunking.ChunkingConfig;
+import com.google.devtools.build.lib.remote.chunking.FastCdcChunker;
+import com.google.devtools.build.lib.remote.common.RemoteActionExecutionContext;
+import com.google.devtools.build.lib.remote.util.DigestUtil;
+import com.google.devtools.build.lib.vfs.Path;
+import com.google.protobuf.ByteString;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Uploads blobs in chunks using Content-Defined Chunking with FastCDC 2020.
+ *
+ * <p>Upload flow for blobs above threshold:
+ *
+ * <ol>
+ *   <li>Chunk file with FastCDC
+ *   <li>Call findMissingDigests on chunk digests
+ *   <li>Upload only missing chunks
+ *   <li>Call SpliceBlob to register the blob as the concatenation of chunks
+ * </ol>
+ */
+public class ChunkedBlobUploader {
+
+  private final GrpcCacheClient grpcCacheClient;
+  private final CombinedCache combinedCache;
+  private final FastCdcChunker chunker;
+  private final long chunkingThreshold;
+
+  /**
+   * Creates a new uploader with the given chunking configuration.
+   *
+   * @param grpcCacheClient client used for {@code FindMissingDigests} and {@code SpliceBlob} RPCs
+   * @param combinedCache cache used to upload individual chunks
+   * @param config chunking parameters negotiated from server capabilities
+   * @param digestUtil utility for computing chunk digests
+   */
+  public ChunkedBlobUploader(
+      GrpcCacheClient grpcCacheClient,
+      CombinedCache combinedCache,
+      ChunkingConfig config,
+      DigestUtil digestUtil) {
+    this.grpcCacheClient = grpcCacheClient;
+    this.combinedCache = combinedCache;
+    this.chunker = new FastCdcChunker(config, digestUtil);
+    this.chunkingThreshold = config.chunkingThreshold();
+  }
+
+  /** Returns the minimum blob size for chunked upload. */
+  public long getChunkingThreshold() {
+    return chunkingThreshold;
+  }
+
+  /**
+   * Uploads a blob in content-defined chunks. The file is chunked with FastCDC, missing chunks are
+   * uploaded, and {@code SpliceBlob} is called to register the blob as the concatenation of its
+   * chunks.
+   */
+  public void uploadChunked(RemoteActionExecutionContext context, Digest blobDigest, Path file)
+      throws IOException, InterruptedException {
+    List<Digest> chunkDigests;
+    try (InputStream input = file.getInputStream()) {
+      chunkDigests = chunker.chunkToDigests(input);
+    }
+    if (chunkDigests.isEmpty()) {
+      return;
+    }
+
+    ImmutableSet<Digest> missingDigests =
+        getFromFuture(grpcCacheClient.findMissingDigests(context, chunkDigests));
+    uploadMissingChunks(context, missingDigests, chunkDigests, file);
+    getFromFuture(grpcCacheClient.spliceBlob(context, blobDigest, chunkDigests));
+  }
+
+  private void uploadMissingChunks(
+      RemoteActionExecutionContext context,
+      ImmutableSet<Digest> missingDigests,
+      List<Digest> chunkDigests,
+      Path file)
+      throws IOException, InterruptedException {
+    if (missingDigests.isEmpty()) {
+      return;
+    }
+
+    Set<Digest> uploaded = new HashSet<>();
+    try (InputStream input = file.getInputStream()) {
+      for (Digest chunkDigest : chunkDigests) {
+        if (missingDigests.contains(chunkDigest) && uploaded.add(chunkDigest)) {
+          ByteString.Output out = ByteString.newOutput((int) chunkDigest.getSizeBytes());
+          ByteStreams.limit(input, chunkDigest.getSizeBytes()).transferTo(out);
+          getFromFuture(combinedCache.uploadBlob(context, chunkDigest, out.toByteString()));
+        } else {
+          input.skipNBytes(chunkDigest.getSizeBytes());
+        }
+      }
+    }
+  }
+}

src/main/java/com/google/devtools/build/lib/remote/CombinedCache.java

@@ -32,9 +32,12 @@
 import com.google.common.flogger.GoogleLogger;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.devtools.build.lib.concurrent.ThreadSafety;
 import com.google.devtools.build.lib.exec.SpawnCheckingCacheEvent;
 import com.google.devtools.build.lib.exec.SpawnProgressEvent;
+import com.google.devtools.build.lib.remote.chunking.ChunkingConfig;
 import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
 import com.google.devtools.build.lib.remote.common.LazyFileOutputStream;
 import com.google.devtools.build.lib.remote.common.MaybePathBacked;
@@ -65,6 +68,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -94,23 +98,75 @@ public class CombinedCache extends AbstractReferenceCounted {
   private final CountDownLatch closeCountDownLatch = new CountDownLatch(1);
   protected final AsyncTaskCache.NoResult<Digest> casUploadCache = AsyncTaskCache.NoResult.create();
 
+  @SuppressWarnings("AllowVirtualThreads")
+  private final ListeningExecutorService virtualThreadExecutor =
+      MoreExecutors.listeningDecorator(
+          Executors.newThreadPerTaskExecutor(
+              Thread.ofVirtual().name("combined-cache-", 0).factory()));
+
   @Nullable protected final RemoteCacheClient remoteCacheClient;
   @Nullable protected final DiskCacheClient diskCacheClient;
   @Nullable protected final String symlinkTemplate;
   protected final DigestUtil digestUtil;
+  private final boolean chunkingEnabled;
+
+  // Delays the initialization of the chunking support logic until first use to avoid blocking on
+  // a server capabilities check at construction time.
+  private final class Chunking {
+    private ChunkingConfig config = null;
+    private ChunkedBlobDownloader downloader = null;
+    private ChunkedBlobUploader uploader = null;
+    private volatile boolean initialized = false;
+
+    boolean supported() throws IOException {
+      if (!chunkingEnabled) {
+        return false;
+      }
+      if (!(remoteCacheClient instanceof GrpcCacheClient grpcClient)) {
+        return false;
+      }
+      if (!initialized) {
+        synchronized (this) {
+          config = ChunkingConfig.fromServerCapabilities(getRemoteServerCapabilities());
+          if (config != null) {
+            downloader = new ChunkedBlobDownloader(grpcClient, CombinedCache.this);
+            uploader = new ChunkedBlobUploader(grpcClient, CombinedCache.this, config, digestUtil);
+          }
+          initialized = true;
+        }
+      }
+      return config != null;
+    }
+
+    ChunkingConfig config() {
+      return checkNotNull(config, "must not call config() unless supported() is true");
+    }
+
+    ChunkedBlobDownloader downloader() {
+      return checkNotNull(downloader, "must not call downloader() unless supported() is true");
+    }
+
+    ChunkedBlobUploader uploader() {
+      return checkNotNull(uploader, "must not call uploader() unless supported() is true");
+    }
+  }
+
+  private final Chunking chunking = new Chunking();
 
   public CombinedCache(
       @Nullable RemoteCacheClient remoteCacheClient,
       @Nullable DiskCacheClient diskCacheClient,
       @Nullable String symlinkTemplate,
-      DigestUtil digestUtil) {
+      DigestUtil digestUtil,
+      boolean chunkingEnabled) {
     checkArgument(
         remoteCacheClient != null || diskCacheClient != null,
         "remoteCacheClient and diskCacheClient cannot be null at the same time");
     this.remoteCacheClient = remoteCacheClient;
     this.diskCacheClient = diskCacheClient;
     this.symlinkTemplate = symlinkTemplate;
     this.digestUtil = digestUtil;
+    this.chunkingEnabled = chunkingEnabled;
   }
 
   public CacheCapabilities getRemoteCacheCapabilities() throws IOException {
@@ -314,15 +370,31 @@ protected ListenableFuture<Void> uploadFile(
       diskCacheFuture = diskCacheClient.uploadFile(digest, file);
     }
 
+    boolean chunkingSupported;
+    try {
+      chunkingSupported = chunking.supported();
+    } catch (IOException e) {
+      return immediateFailedFuture(e);
+    }
+
     ListenableFuture<Void> remoteCacheFuture = Futures.immediateVoidFuture();
     if (remoteCacheClient != null && context.getWriteCachePolicy().allowRemoteCache()) {
-      Completable upload =
-          casUploadCache.execute(
-              digest,
-              RxFutures.toCompletable(
-                  () -> remoteCacheClient.uploadFile(context, digest, file), directExecutor()),
-              force);
-      remoteCacheFuture = RxFutures.toListenableFuture(upload);
+      if (chunkingSupported && digest.getSizeBytes() > chunking.config().chunkingThreshold()) {
+        remoteCacheFuture =
+            virtualThreadExecutor.submit(
+                () -> {
+                  chunking.uploader().uploadChunked(context, digest, file);
+                  return null;
+                });
+      } else {
+        Completable upload =
+            casUploadCache.execute(
+                digest,
+                RxFutures.toCompletable(
+                    () -> remoteCacheClient.uploadFile(context, digest, file), directExecutor()),
+                force);
+        remoteCacheFuture = RxFutures.toListenableFuture(upload);
+      }
     }
 
     return Futures.whenAllSucceed(diskCacheFuture, remoteCacheFuture)
@@ -417,7 +489,7 @@ private ListenableFuture<Void> downloadBlob(
         directExecutor());
   }
 
-  private ListenableFuture<Void> downloadBlob(
+  ListenableFuture<Void> downloadBlob(
       RemoteActionExecutionContext context, Digest digest, OutputStream out) {
     ListenableFuture<Void> future = immediateFailedFuture(new CacheNotFoundException(digest));
 
@@ -441,6 +513,34 @@ private ListenableFuture<Void> downloadBlobFromRemote(
       RemoteActionExecutionContext context, Digest digest, OutputStream out) {
     checkState(remoteCacheClient != null && context.getReadCachePolicy().allowRemoteCache());
 
+    boolean chunkingSupported;
+    try {
+      chunkingSupported = chunking.supported();
+    } catch (IOException e) {
+      return immediateFailedFuture(e);
+    }
+
+    if (chunkingSupported && digest.getSizeBytes() > chunking.config().chunkingThreshold()) {
+      ListenableFuture<Void> chunkedDownloadFuture =
+          virtualThreadExecutor.submit(
+              () -> {
+                chunking.downloader().downloadChunked(context, digest, out);
+                return null;
+              });
+      return Futures.catchingAsync(
+          chunkedDownloadFuture,
+          CacheNotFoundException.class,
+          (e) -> regularDownloadBlobFromRemote(context, digest, out),
+          directExecutor());
+    }
+
+    return regularDownloadBlobFromRemote(context, digest, out);
+  }
+
+  private ListenableFuture<Void> regularDownloadBlobFromRemote(
+      RemoteActionExecutionContext context, Digest digest, OutputStream out) {
+    checkState(remoteCacheClient != null && context.getReadCachePolicy().allowRemoteCache());
+
     if (diskCacheClient != null && context.getWriteCachePolicy().allowDiskCache()) {
       Path tempPath = diskCacheClient.getTempPath();
       LazyFileOutputStream tempOut = new LazyFileOutputStream(tempPath);