support remote cache chunking using FastCDC 2020

Author: tyler-french

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

@@ -8,6 +8,7 @@ package(
 filegroup(
     name = "srcs",
     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",
@@ -28,7 +29,9 @@ java_library(
     srcs = glob(
         ["*.java"],
         exclude = [
+            "ChunkingConfig.java",
             "ExecutionStatusException.java",
+            "FastCDCChunker.java",
             "ReferenceCountedChannel.java",
             "ChannelConnectionWithServerCapabilitiesFactory.java",
             "RemoteRetrier.java",
@@ -53,6 +56,7 @@ java_library(
         ":Retrier",
         ":abstract_action_input_prefetcher",
         ":lease_service",
+        "//src/main/java/com/google/devtools/build/lib/concurrent:task_deduplicator",
         ":remote_important_output_handler",
         ":remote_output_checker",
         ":scrubber",
@@ -98,6 +102,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,77 @@
+// 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.OutputDigestMismatchException;
@@ -64,6 +67,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;
@@ -93,9 +97,17 @@ 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;
+  @Nullable private final ChunkingConfig chunkingConfig;
+  @Nullable private final ChunkedBlobDownloader chunkedDownloader;
+  @Nullable private final ChunkedBlobUploader chunkedUploader;
   protected final DigestUtil digestUtil;
 
   public CombinedCache(
@@ -110,6 +122,18 @@ public CombinedCache(
     this.diskCacheClient = diskCacheClient;
     this.symlinkTemplate = symlinkTemplate;
     this.digestUtil = digestUtil;
+
+    if (remoteCacheClient instanceof GrpcCacheClient grpcClient
+        && grpcClient.getChunkingConfig() != null) {
+      ChunkingConfig config = grpcClient.getChunkingConfig();
+      this.chunkingConfig = config;
+      this.chunkedDownloader = new ChunkedBlobDownloader(grpcClient, this);
+      this.chunkedUploader = new ChunkedBlobUploader(grpcClient, this, config, digestUtil);
+    } else {
+      this.chunkingConfig = null;
+      this.chunkedDownloader = null;
+      this.chunkedUploader = null;
+    }
   }
 
   public CacheCapabilities getRemoteCacheCapabilities() throws IOException {
@@ -130,6 +154,11 @@ public ServerCapabilities getRemoteServerCapabilities() throws IOException {
     return remoteCacheClient.getServerCapabilities();
   }
 
+  @Nullable
+  public ChunkingConfig getChunkingConfig() {
+    return chunkingConfig;
+  }
+
   /**
    * Class to keep track of which cache (disk or remote) a given [cached] ActionResult comes from.
    */
@@ -315,13 +344,21 @@ protected ListenableFuture<Void> uploadFile(
 
     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 (chunkedUploader != null
+          && digest.getSizeBytes() > chunkingConfig.chunkingThreshold()) {
+        remoteCacheFuture = virtualThreadExecutor.submit(() -> {
+          chunkedUploader.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)
@@ -416,7 +453,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));
 
@@ -440,6 +477,27 @@ private ListenableFuture<Void> downloadBlobFromRemote(
       RemoteActionExecutionContext context, Digest digest, OutputStream out) {
     checkState(remoteCacheClient != null && context.getReadCachePolicy().allowRemoteCache());
 
+    if (chunkedDownloader != null
+        && digest.getSizeBytes() > chunkingConfig.chunkingThreshold()) {
+      ListenableFuture<Void> chunkedDownloadFuture =
+          virtualThreadExecutor.submit(() -> {
+            chunkedDownloader.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);