Fix disk cache failures on concurrent read-write access on Windows

This applies the fix made to the download cache in 753dc97 to the disk cache.

Author: fmeum

src/main/java/com/google/devtools/build/lib/bazel/repository/cache/BUILD

@@ -23,6 +23,7 @@ java_library(
     deps = [
         "//src/main/java/com/google/devtools/build/lib/server:idle_task",
         "//src/main/java/com/google/devtools/build/lib/util:file_system_lock",
+        "//src/main/java/com/google/devtools/build/lib/util:os",
         "//src/main/java/com/google/devtools/build/lib/vfs",
         "//third_party:guava",
         "//third_party:jsr305",

src/main/java/com/google/devtools/build/lib/bazel/repository/cache/DownloadCache.java

@@ -21,6 +21,7 @@
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hasher;
 import com.google.common.io.BaseEncoding;
+import com.google.devtools.build.lib.util.OS;
 import com.google.devtools.build.lib.vfs.DigestHashFunction;
 import com.google.devtools.build.lib.vfs.FileAccessException;
 import com.google.devtools.build.lib.vfs.FileSystemUtils;
@@ -251,7 +252,7 @@ private void storeCacheValue(
       // that another thread won the race to place the file in the cache. As the exception is rather
       // generic and could result from other failure types, we rethrow the exception if the cache
       // entry hasn't been created.
-      if (!cacheValue.exists()) {
+      if (OS.getCurrent() != OS.WINDOWS || !cacheValue.exists()) {
         throw e;
       }
     }

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

@@ -24,6 +24,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib/remote/util:digest_utils",
         "//src/main/java/com/google/devtools/build/lib/server:idle_task",
         "//src/main/java/com/google/devtools/build/lib/util:file_system_lock",
+        "//src/main/java/com/google/devtools/build/lib/util:os",
         "//src/main/java/com/google/devtools/build/lib/vfs",
         "//third_party:flogger",
         "//third_party:guava",

src/main/java/com/google/devtools/build/lib/remote/disk/DiskCacheClient.java

@@ -35,6 +35,8 @@
 import com.google.devtools.build.lib.remote.util.DigestOutputStream;
 import com.google.devtools.build.lib.remote.util.DigestUtil;
 import com.google.devtools.build.lib.remote.util.Utils;
+import com.google.devtools.build.lib.util.OS;
+import com.google.devtools.build.lib.vfs.FileAccessException;
 import com.google.devtools.build.lib.vfs.Path;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.ExtensionRegistryLite;
@@ -142,7 +144,20 @@ public void captureFile(Path src, Digest digest, Store store) throws IOException
     }
 
     target.getParentDirectory().createDirectoryAndParents();
-    src.renameTo(target);
+    try {
+      src.renameTo(target);
+    } catch (FileAccessException e) {
+      // On Windows, atomically replacing a file that is currently opened (e.g. due to a
+      // concurrent get on the cache) results in renameTo throwing this exception, which wraps an
+      // AccessDeniedException. This case is benign since if the target path already exists, we
+      // know that another thread won the race to place the file in the cache. As the exception is
+      // rather generic and could result from other failure types, we rethrow the exception if the
+      // cache entry hasn't been created.
+      if (OS.getCurrent() != OS.WINDOWS || !target.exists()) {
+        throw e;
+      }
+      src.delete();
+    }
   }
 
   private ListenableFuture<Void> download(Digest digest, OutputStream out, Store store) {
@@ -333,7 +348,20 @@ public void saveFile(Digest digest, Store store, InputStream in) throws IOExcept
         }
       }
       path.getParentDirectory().createDirectoryAndParents();
-      temp.renameTo(path);
+      try {
+        temp.renameTo(path);
+      } catch (FileAccessException e) {
+        // On Windows, atomically replacing a file that is currently opened (e.g. due to a
+        // concurrent get on the cache) results in renameTo throwing this exception, which wraps an
+        // AccessDeniedException. This case is benign since if the target path already exists, we
+        // know that another thread won the race to place the file in the cache. As the exception is
+        // rather generic and could result from other failure types, we rethrow the exception if the
+        // cache entry hasn't been created.
+        if (OS.getCurrent() != OS.WINDOWS || !path.exists()) {
+          throw e;
+        }
+        temp.delete();
+      }
     } catch (IOException e) {
       try {
         temp.delete();

src/test/java/com/google/devtools/build/lib/remote/disk/BUILD

@@ -33,6 +33,7 @@ java_test(
         "//src/test/java/com/google/devtools/build/lib:test_runner",
         "//src/test/java/com/google/devtools/build/lib/testutil:TestUtils",
         "//src/test/java/com/google/devtools/build/lib/testutil:external_file_system_lock",
+        "//src/test/java/com/google/devtools/build/lib/vfs/util:util_internal",
         "//third_party:error_prone_annotations",
         "//third_party:guava",
         "//third_party:junit4",

src/test/java/com/google/devtools/build/lib/remote/disk/DiskCacheClientTest.java

@@ -32,17 +32,25 @@
 import com.google.devtools.build.lib.remote.common.OutputDigestMismatchException;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient.ActionKey;
 import com.google.devtools.build.lib.remote.util.DigestUtil;
+import com.google.devtools.build.lib.testutil.TestUtils;
 import com.google.devtools.build.lib.vfs.DigestHashFunction;
 import com.google.devtools.build.lib.vfs.FileSystem;
 import com.google.devtools.build.lib.vfs.FileSystemUtils;
 import com.google.devtools.build.lib.vfs.Path;
 import com.google.devtools.build.lib.vfs.SyscallCache;
 import com.google.devtools.build.lib.vfs.bazel.BazelHashFunctions;
 import com.google.devtools.build.lib.vfs.inmemoryfs.InMemoryFileSystem;
+import com.google.devtools.build.lib.vfs.util.FileSystems;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Message;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -346,6 +354,86 @@ public void downloadActionResult_withReferencedTreeFileMissing_returnsNull() thr
     assertThat(result).isNull();
   }
 
+  @Test
+  public void concurrentUploadDownload()
+      throws IOException, ExecutionException, InterruptedException {
+    var nativeDiskCacheDir = TestUtils.createUniqueTmpDir(FileSystems.getNativeFileSystem());
+    var nativeClient =
+        new DiskCacheClient(nativeDiskCacheDir, DIGEST_UTIL, /* verifyDownloads= */ false);
+    var tasks = new ArrayList<Future<?>>();
+    // Use 1 MB blobs to increase the window for concurrent access during write/rename.
+    var contentSize = 1024 * 1024;
+    var numConcurrentOps = 10;
+    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
+      for (int attempt = 0; attempt < 100; attempt++) {
+        var contentArray = new byte[contentSize];
+        // Fill with a pattern based on the attempt number.
+        for (int i = 0; i < contentSize; i++) {
+          contentArray[i] = (byte) (attempt + i);
+        }
+        var contentBytes = ByteString.copyFrom(contentArray);
+        var contentDigest = DIGEST_UTIL.compute(contentArray);
+        // Use a latch to ensure all concurrent tasks start at roughly the same time.
+        var startLatch = new CountDownLatch(numConcurrentOps);
+        // Half the tasks do uploads, half do downloads with a slow OutputStream to keep the file
+        // open longer. This maximizes the chance of a rename failing because a download has the
+        // file open.
+        for (int concurrentOp = 0; concurrentOp < numConcurrentOps; concurrentOp++) {
+          boolean isUploader = concurrentOp % 2 == 0;
+          tasks.add(
+              executor.submit(
+                  () -> {
+                    // Signal ready and wait for all tasks to be ready.
+                    startLatch.countDown();
+                    startLatch.await();
+                    if (isUploader) {
+                      getFromFuture(nativeClient.uploadBlob(contentDigest, contentBytes));
+                    } else {
+                      // Use a slow OutputStream that pauses periodically to keep the file open
+                      // longer during download.
+                      var out =
+                          new OutputStream() {
+                            private int bytesWritten = 0;
+
+                            @Override
+                            public void write(int b) throws IOException {
+                              bytesWritten++;
+                              maybeSleep();
+                            }
+
+                            @Override
+                            public void write(byte[] b, int off, int len) throws IOException {
+                              bytesWritten += len;
+                              maybeSleep();
+                            }
+
+                            private void maybeSleep() {
+                              // Sleep every 64KB to slow down the download.
+                              if (bytesWritten % (64 * 1024) < 100) {
+                                try {
+                                  Thread.sleep(1);
+                                } catch (InterruptedException e) {
+                                  Thread.currentThread().interrupt();
+                                }
+                              }
+                            }
+                          };
+                      try {
+                        getFromFuture(nativeClient.downloadBlob(contentDigest, out));
+                      } catch (CacheNotFoundException ignored) {
+                        // File not yet uploaded by another task.
+                      }
+                    }
+                    return null;
+                  }));
+        }
+      }
+      for (var task : tasks) {
+        task.get();
+      }
+    }
+  }
+
   private Tree getTreeWithFile(Digest fileDigest) {
     return Tree.newBuilder()
         .addChildren(Directory.newBuilder().addFiles(FileNode.newBuilder().setDigest(fileDigest)))