remote: made CombinedCache a composition of Disk and Http Cache

Fixes bazelbuild#8052

The problem: CombinedCache is not working.
The cause: both `OnDiskBlobStore` and `HttpBlobStore` store action results differently from other actions, but not in the same way. The `OnDiskBlobStore` appends "ac_" prefix to the key, while the `HttpBlobStore` stores it in a different directory. Current implementation added "ac_" prefix for both blob stores, which is not correct for the `HttpBlobStore`.
The solution: treat actionResults' `gets` and `puts` in accordance with a particular Blob Store used.

The problem is described in bazelbuild#8052 and the root cause is described in [this](bazelbuild#8052 (comment)) and [this](bazelbuild#8052 (comment)) comments.

Closes bazelbuild#8141.

PiperOrigin-RevId: 247387377

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

@@ -186,7 +186,7 @@ public Digest uploadStream(Digest digest, InputStream in)
   }
 
   public boolean containsKey(Digest digest) throws IOException, InterruptedException {
-    return blobStore.containsKey(digest.getHash());
+    return blobStore.contains(digest.getHash());
   }
 
   @Override
@@ -206,7 +206,7 @@ private byte[] downloadActionResult(Digest digest) throws IOException, Interrupt
     }
     // This unconditionally downloads the whole blob into memory!
     ByteArrayOutputStream out = new ByteArrayOutputStream();
-    boolean success = blobStore.getActionResult(digest.getHash(), out);
+    boolean success = getFromFuture(blobStore.getActionResult(digest.getHash(), out));
     if (!success) {
       throw new CacheNotFoundException(digest, digestUtil);
     }

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

@@ -114,12 +114,16 @@ private static SimpleBlobStore createDiskCache(Path workingDirectory, PathFragme
   private static SimpleBlobStore createCombinedCache(
       Path workingDirectory, PathFragment diskCachePath, RemoteOptions options, Credentials cred)
       throws IOException {
+
     Path cacheDir =
         workingDirectory.getRelative(Preconditions.checkNotNull(diskCachePath, "diskCachePath"));
     if (!cacheDir.exists()) {
       cacheDir.createDirectoryAndParents();
     }
-    return new CombinedDiskHttpBlobStore(cacheDir, createHttp(options, cred));
+
+    OnDiskBlobStore diskCache = new OnDiskBlobStore(cacheDir);
+    SimpleBlobStore httpCache = createHttp(options, cred);
+    return new CombinedDiskHttpBlobStore(diskCache, httpCache);
   }
 
   private static boolean isDiskCache(RemoteOptions options) {

src/main/java/com/google/devtools/build/lib/remote/blobstore/CombinedDiskHttpBlobStore.java

@@ -14,14 +14,10 @@
 package com.google.devtools.build.lib.remote.blobstore;
 
 import com.google.common.base.Preconditions;
-import com.google.common.io.ByteStreams;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.SettableFuture;
 import com.google.devtools.build.lib.vfs.Path;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -31,119 +27,122 @@
 
 /**
  * A {@link SimpleBlobStore} implementation combining two blob stores. A local disk blob store and a
- * remote http blob store. If a blob isn't found in the first store, the second store is used, and
- * the blob added to the first. Put puts the blob on both stores.
+ * remote blob store. If a blob isn't found in the first store, the second store is used, and the
+ * blob added to the first. Put puts the blob on both stores.
  */
-public final class CombinedDiskHttpBlobStore extends OnDiskBlobStore {
+public final class CombinedDiskHttpBlobStore implements SimpleBlobStore {
   private static final Logger logger = Logger.getLogger(CombinedDiskHttpBlobStore.class.getName());
-  private final SimpleBlobStore bsHttp;
 
-  public CombinedDiskHttpBlobStore(Path root, SimpleBlobStore bsHttp) {
-    super(root);
-    this.bsHttp = Preconditions.checkNotNull(bsHttp);
+  private final SimpleBlobStore remoteCache;
+  private final OnDiskBlobStore diskCache;
+
+  public CombinedDiskHttpBlobStore(OnDiskBlobStore diskCache, SimpleBlobStore remoteCache) {
+    this.diskCache = Preconditions.checkNotNull(diskCache);
+    this.remoteCache = Preconditions.checkNotNull(remoteCache);
   }
 
   @Override
-  public boolean containsKey(String key) {
-    // HTTP cache does not support containsKey.
-    // Don't support it here either for predictable semantics.
-    throw new UnsupportedOperationException("HTTP Caching does not use this method.");
+  public boolean contains(String key) {
+    return diskCache.contains(key);
   }
 
   @Override
-  public ListenableFuture<Boolean> get(String key, OutputStream out) {
-    boolean foundOnDisk = super.containsKey(key);
-
-    if (foundOnDisk) {
-      return super.get(key, out);
-    } else {
-      // Write a temporary file first, and then rename, to avoid data corruption in case of a crash.
-      Path temp = toPath(UUID.randomUUID().toString());
-
-      OutputStream tempOut;
-      try {
-        tempOut = temp.getOutputStream();
-      } catch (IOException e) {
-        return Futures.immediateFailedFuture(e);
-      }
-      ListenableFuture<Boolean> chained =
-          Futures.transformAsync(
-              bsHttp.get(key, tempOut),
-              (found) -> {
-                if (!found) {
-                  return Futures.immediateFuture(false);
-                } else {
-                  Path target = toPath(key);
-                  // The following note and line is taken from OnDiskBlobStore.java
-                  // TODO(ulfjack): Fsync temp here before we rename it to avoid data loss in the
-                  // case of machine
-                  // crashes (the OS may reorder the writes and the rename).
-                  temp.renameTo(target);
-
-                  SettableFuture<Boolean> f = SettableFuture.create();
-                  try (InputStream in = target.getInputStream()) {
-                    ByteStreams.copy(in, out);
-                    f.set(true);
-                  } catch (IOException e) {
-                    f.setException(e);
-                  }
-                  return f;
-                }
-              },
-              MoreExecutors.directExecutor());
-      chained.addListener(
-          () -> {
-            try {
-              tempOut.close();
-            } catch (IOException e) {
-              // not sure what to do here, we either are here because of another exception being
-              // thrown,
-              // or we have successfully used the file we are trying (and failing) to close
-              logger.log(Level.WARNING, "Failed to close temporary file on get", e);
-            }
-          },
-          MoreExecutors.directExecutor());
-      return chained;
-    }
+  public boolean containsActionResult(String key) {
+    return diskCache.containsActionResult(key);
   }
 
   @Override
-  public boolean getActionResult(String key, OutputStream out)
+  public void put(String key, long length, InputStream in)
       throws IOException, InterruptedException {
-    if (super.getActionResult(key, out)) {
-      return true;
+    diskCache.put(key, length, in);
+    try (InputStream inFile = diskCache.toPath(key, /* actionResult= */ false).getInputStream()) {
+      remoteCache.put(key, length, inFile);
     }
+  }
 
-    try (ByteArrayOutputStream tmpOut = new ByteArrayOutputStream()) {
-      if (bsHttp.getActionResult(key, tmpOut)) {
-        byte[] tmp = tmpOut.toByteArray();
-        super.putActionResult(key, tmp);
-        ByteStreams.copy(new ByteArrayInputStream(tmp), out);
-        return true;
-      }
-    }
+  @Override
+  public void putActionResult(String key, byte[] in) throws IOException, InterruptedException {
+    diskCache.putActionResult(key, in);
+    remoteCache.putActionResult(key, in);
+  }
 
-    return false;
+  @Override
+  public void close() {
+    diskCache.close();
+    remoteCache.close();
   }
 
   @Override
-  public void put(String key, long length, InputStream in)
-      throws IOException, InterruptedException {
-    super.put(key, length, in);
-    try (InputStream inFile = toPath(key).getInputStream()) {
-      bsHttp.put(key, length, inFile);
+  public ListenableFuture<Boolean> get(String key, OutputStream out) {
+    return get(key, out, /* actionResult= */ false);
+  }
+
+  private ListenableFuture<Boolean> get(String key, OutputStream out, boolean actionResult) {
+    boolean foundOnDisk =
+        actionResult ? diskCache.containsActionResult(key) : diskCache.contains(key);
+
+    if (foundOnDisk) {
+      return getFromCache(diskCache, key, out, actionResult);
+    } else {
+      return getFromRemoteAndSaveToDisk(key, out, actionResult);
     }
   }
 
   @Override
-  public void putActionResult(String key, byte[] in) throws IOException, InterruptedException {
-    super.putActionResult(key, in);
-    bsHttp.putActionResult(key, in);
+  public ListenableFuture<Boolean> getActionResult(String key, OutputStream out) {
+    return get(key, out, /* actionResult= */ true);
   }
 
-  @Override
-  public void close() {
-    super.close();
-    bsHttp.close();
+  private ListenableFuture<Boolean> getFromRemoteAndSaveToDisk(
+      String key, OutputStream out, boolean actionResult) {
+    // Write a temporary file first, and then rename, to avoid data corruption in case of a crash.
+    Path temp = diskCache.toPath(UUID.randomUUID().toString(), /* actionResult= */ false);
+
+    OutputStream tempOut;
+    try {
+      tempOut = temp.getOutputStream();
+    } catch (IOException e) {
+      return Futures.immediateFailedFuture(e);
+    }
+    ListenableFuture<Boolean> chained =
+        Futures.transformAsync(
+            getFromCache(remoteCache, key, tempOut, actionResult),
+            (found) -> {
+              if (!found) {
+                return Futures.immediateFuture(false);
+              } else {
+                saveToDiskCache(key, temp, actionResult);
+                return getFromCache(diskCache, key, out, actionResult);
+              }
+            },
+            MoreExecutors.directExecutor());
+    chained.addListener(
+        () -> {
+          try {
+            tempOut.close();
+          } catch (IOException e) {
+            // not sure what to do here, we either are here because of another exception being
+            // thrown, or we have successfully used the file we are trying (and failing) to close
+            logger.log(Level.WARNING, "Failed to close temporary file on get", e);
+          }
+        },
+        MoreExecutors.directExecutor());
+    return chained;
+  }
+
+  private void saveToDiskCache(String key, Path temp, boolean actionResult) throws IOException {
+    Path target = diskCache.toPath(key, actionResult);
+    // TODO(ulfjack): Fsync temp here before we rename it to avoid data loss in the
+    // case of machine crashes (the OS may reorder the writes and the rename).
+    temp.renameTo(target);
+  }
+
+  private ListenableFuture<Boolean> getFromCache(
+      SimpleBlobStore blobStore, String key, OutputStream tempOut, boolean actionResult) {
+    if (!actionResult) {
+      return blobStore.get(key, tempOut);
+    } else {
+      return blobStore.getActionResult(key, tempOut);
+    }
   }
 }

src/main/java/com/google/devtools/build/lib/remote/blobstore/ConcurrentMapBlobStore.java

@@ -21,7 +21,6 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutionException;
 
 /** A {@link SimpleBlobStore} implementation using a {@link ConcurrentMap}. */
 public final class ConcurrentMapBlobStore implements SimpleBlobStore {
@@ -33,10 +32,15 @@ public ConcurrentMapBlobStore(ConcurrentMap<String, byte[]> map) {
   }
 
   @Override
-  public boolean containsKey(String key) {
+  public boolean contains(String key) {
     return map.containsKey(key);
   }
 
+  @Override
+  public boolean containsActionResult(String key) {
+    return map.containsKey(ACTION_KEY_PREFIX + key);
+  }
+
   @Override
   public ListenableFuture<Boolean> get(String key, OutputStream out) {
     byte[] data = map.get(key);
@@ -55,9 +59,8 @@ public ListenableFuture<Boolean> get(String key, OutputStream out) {
   }
 
   @Override
-  public boolean getActionResult(String key, OutputStream out)
-      throws IOException, InterruptedException {
-    return getFromFuture(get(ACTION_KEY_PREFIX + key, out));
+  public ListenableFuture<Boolean> getActionResult(String key, OutputStream out) {
+    return get(ACTION_KEY_PREFIX + key, out);
   }
 
   @Override
@@ -75,15 +78,4 @@ public void putActionResult(String key, byte[] in) {
   @Override
   public void close() {}
 
-  private static <T> T getFromFuture(ListenableFuture<T> f)
-      throws IOException, InterruptedException {
-    try {
-      return f.get();
-    } catch (ExecutionException e) {
-      if (e.getCause() instanceof IOException) {
-        throw (IOException) e.getCause();
-      }
-      throw new IOException(e.getCause());
-    }
-  }
 }