Move the SimpleBlobStore and implementations to a subpackage

Also extend the API to throw exceptions rather than having to wrap or swallow.

This is in preparation for adding yet another implementation using an on-disk
cache. Having the RemoteWorker keep everything in memory is not good for my
sanity.

PiperOrigin-RevId: 160871586

10 files changed, 231 insertions, 190 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/remote/BUILD b/src/main/java/com/google/devtools/build/lib/remote/BUILD
index 61fd521ae6..6d42b4320f 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/BUILD
+++ b/src/main/java/com/google/devtools/build/lib/remote/BUILD
@@ -4,7 +4,7 @@ package(
 
 java_library(
     name = "remote",
-    srcs = glob(["*.java"]),
+    srcs = glob(["**/*.java"]),
     tags = ["bazel"],
     runtime_deps = [
         # This is required for client TLS.
diff --git a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java
index 6fd37b0a61..559d55d3fc 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java
@@ -20,6 +20,7 @@ import com.google.devtools.build.lib.actions.cache.VirtualActionInput;
 import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
 import com.google.devtools.build.lib.remote.Digests.ActionKey;
 import com.google.devtools.build.lib.remote.TreeNodeRepository.TreeNode;
+import com.google.devtools.build.lib.remote.blobstore.SimpleBlobStore;
 import com.google.devtools.build.lib.util.Preconditions;
 import com.google.devtools.build.lib.util.io.FileOutErr;
 import com.google.devtools.build.lib.vfs.FileSystemUtils;
@@ -76,7 +77,7 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
   }
 
   public void downloadTree(Digest rootDigest, Path rootLocation)
-      throws IOException, CacheNotFoundException {
+      throws IOException, CacheNotFoundException, InterruptedException {
     Directory directory = Directory.parseFrom(downloadBlob(rootDigest));
     for (FileNode file : directory.getFilesList()) {
       downloadFileContents(
@@ -110,7 +111,7 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
 
   @Override
   public void download(ActionResult result, Path execRoot, FileOutErr outErr)
-      throws IOException, CacheNotFoundException {
+      throws IOException, CacheNotFoundException, InterruptedException {
     for (OutputFile file : result.getOutputFilesList()) {
       if (!file.getContent().isEmpty()) {
         createFile(
@@ -129,7 +130,7 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
   }
 
   private void downloadOutErr(ActionResult result, FileOutErr outErr)
-          throws IOException, CacheNotFoundException {
+          throws IOException, CacheNotFoundException, InterruptedException {
     if (!result.getStdoutRaw().isEmpty()) {
       result.getStdoutRaw().writeTo(outErr.getOutputStream());
       outErr.getOutputStream().flush();
@@ -187,7 +188,7 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
   }
 
   public void uploadOutErr(ActionResult.Builder result, byte[] stdout, byte[] stderr)
-      throws InterruptedException {
+      throws IOException, InterruptedException {
     if (stdout.length <= MAX_BLOB_SIZE_FOR_INLINE) {
       result.setStdoutRaw(ByteString.copyFrom(stdout));
     } else if (stdout.length > 0) {
@@ -201,7 +202,7 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
   }
 
   private void downloadFileContents(Digest digest, Path dest, boolean executable)
-      throws IOException, CacheNotFoundException {
+      throws IOException, CacheNotFoundException, InterruptedException {
     // This unconditionally downloads the whole file into memory first!
     createFile(downloadBlob(digest), dest, executable);
   }
@@ -222,11 +223,11 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
         MAX_MEMORY_KBYTES);
   }
 
-  public Digest uploadBlob(byte[] blob) throws InterruptedException {
+  public Digest uploadBlob(byte[] blob) throws IOException, InterruptedException {
     return uploadBlob(blob, Digests.computeDigest(blob));
   }
 
-  private Digest uploadBlob(byte[] blob, Digest digest) throws InterruptedException {
+  private Digest uploadBlob(byte[] blob, Digest digest) throws IOException, InterruptedException {
     int blobSizeKBytes = blob.length / 1024;
     checkBlobSize(blobSizeKBytes, "Upload");
     uploadMemoryAvailable.acquire(blobSizeKBytes);
@@ -238,7 +239,8 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
     return digest;
   }
 
-  public byte[] downloadBlob(Digest digest) throws CacheNotFoundException {
+  public byte[] downloadBlob(Digest digest)
+      throws IOException, CacheNotFoundException, InterruptedException {
     if (digest.getSizeBytes() == 0) {
       return new byte[0];
     }
@@ -251,12 +253,13 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
     return data;
   }
 
-  public boolean containsKey(Digest digest) {
+  public boolean containsKey(Digest digest) throws IOException, InterruptedException {
     return blobStore.containsKey(digest.getHash());
   }
 
   @Override
-  public ActionResult getCachedActionResult(ActionKey actionKey) {
+  public ActionResult getCachedActionResult(ActionKey actionKey)
+      throws IOException, InterruptedException {
     byte[] data = blobStore.get(actionKey.getDigest().getHash());
     if (data == null) {
       return null;
@@ -269,7 +272,7 @@ public final class SimpleBlobStoreActionCache implements RemoteActionCache {
   }
 
   public void setCachedActionResult(ActionKey actionKey, ActionResult result)
-      throws InterruptedException {
+      throws IOException, InterruptedException {
     blobStore.put(actionKey.getDigest().getHash(), result.toByteArray());
   }
 
diff --git a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java
index 9e906dee08..fd05ef84d3 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java
@@ -14,6 +14,9 @@
 
 package com.google.devtools.build.lib.remote;
 
+import com.google.devtools.build.lib.remote.blobstore.ConcurrentMapBlobStore;
+import com.google.devtools.build.lib.remote.blobstore.RestBlobStore;
+import com.google.devtools.build.lib.remote.blobstore.SimpleBlobStore;
 import com.hazelcast.client.HazelcastClient;
 import com.hazelcast.client.config.ClientConfig;
 import com.hazelcast.client.config.ClientNetworkConfig;
@@ -21,19 +24,8 @@ import com.hazelcast.client.config.XmlClientConfigBuilder;
 import com.hazelcast.config.Config;
 import com.hazelcast.core.Hazelcast;
 import com.hazelcast.core.HazelcastInstance;
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.concurrent.ConcurrentMap;
-import org.apache.http.HttpResponse;
-import org.apache.http.HttpStatus;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.ResponseHandler;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpHead;
-import org.apache.http.client.methods.HttpPut;
-import org.apache.http.entity.ByteArrayEntity;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 
 /**
  * A factory class for providing a {@link SimpleBlobStore} to be used with {@link
@@ -45,33 +37,6 @@ public final class SimpleBlobStoreFactory {
 
   private SimpleBlobStoreFactory() {}
 
-  /** A {@link SimpleBlobStore} implementation using a {@link ConcurrentMap}. */
-  public static class ConcurrentMapBlobStore implements SimpleBlobStore {
-    private final ConcurrentMap<String, byte[]> map;
-
-    public ConcurrentMapBlobStore(ConcurrentMap<String, byte[]> map) {
-      this.map = map;
-    }
-
-    @Override
-    public boolean containsKey(String key) {
-      return map.containsKey(key);
-    }
-
-    @Override
-    public byte[] get(String key) {
-      return map.get(key);
-    }
-
-    @Override
-    public void put(String key, byte[] value) {
-      map.put(key, value);
-    }
-
-    @Override
-    public void close() {}
-  }
-
   /** Construct a {@link SimpleBlobStore} using Hazelcast's version of {@link ConcurrentMap} */
   public static SimpleBlobStore createHazelcast(RemoteOptions options) {
     HazelcastInstance instance;
@@ -105,120 +70,6 @@ public final class SimpleBlobStoreFactory {
     return new ConcurrentMapBlobStore(instance.<String, byte[]>getMap(HAZELCAST_CACHE_NAME));
   }
 
-  /**
-   * Implementation of {@link SimpleBlobStore} with a REST service. The REST service needs to
-   * support the following HTTP methods.
-   *
-   * <p>PUT /cache/1234 HTTP/1.1 PUT method is used to upload a blob with a base16 key. In this
-   * example the key is 1234. Valid status codes are 200, 201, 202 and 204.
-   *
-   * <p>GET /cache/1234 HTTP/1.1 GET method fetches a blob with the specified key. In this example
-   * the key is 1234. A status code of 200 should be followed by the content of blob. Status code of
-   * 404 or 204 means the key cannot be found.
-   *
-   * <p>HEAD /cache/1234 HTTP/1.1 HEAD method checks to see if the specified key exists in the blob
-   * store. A status code of 200 indicates the key is found in the blob store. A status code of 404
-   * indicates the key is not found in the blob store.
-   */
-  private static class RestBlobStore implements SimpleBlobStore {
-
-    private final String baseUrl;
-    private final PoolingHttpClientConnectionManager connMan;
-    private final HttpClientBuilder clientFactory;
-
-    RestBlobStore(String baseUrl, int poolSize) {
-      this.baseUrl = baseUrl;
-      connMan = new PoolingHttpClientConnectionManager();
-      connMan.setDefaultMaxPerRoute(poolSize);
-      connMan.setMaxTotal(poolSize);
-      clientFactory = HttpClientBuilder.create();
-      clientFactory.setConnectionManager(connMan);
-      clientFactory.setConnectionManagerShared(true);
-    }
-
-    @Override
-    public void close() {
-      connMan.close();
-    }
-
-    @Override
-    public boolean containsKey(String key) {
-      try {
-        HttpClient client = clientFactory.build();
-        HttpHead head = new HttpHead(baseUrl + "/" + key);
-        return client.execute(
-            head,
-            new ResponseHandler<Boolean>() {
-              @Override
-              public Boolean handleResponse(HttpResponse response) {
-                int statusCode = response.getStatusLine().getStatusCode();
-                return HttpStatus.SC_OK == statusCode;
-              }
-            });
-      } catch (IOException e) {
-        throw new RuntimeException(e);
-      }
-    }
-
-    @Override
-    public byte[] get(String key) {
-      try {
-        HttpClient client = clientFactory.build();
-        HttpGet get = new HttpGet(baseUrl + "/" + key);
-        return client.execute(
-            get,
-            new ResponseHandler<byte[]>() {
-              @Override
-              public byte[] handleResponse(HttpResponse response) throws IOException {
-                int statusCode = response.getStatusLine().getStatusCode();
-                if (HttpStatus.SC_NOT_FOUND == statusCode
-                    || HttpStatus.SC_NO_CONTENT == statusCode) {
-                  return null;
-                }
-                if (HttpStatus.SC_OK != statusCode) {
-                  throw new RuntimeException("GET failed with status code " + statusCode);
-                }
-                ByteArrayOutputStream buffer = new ByteArrayOutputStream();
-                response.getEntity().writeTo(buffer);
-                buffer.flush();
-                return buffer.toByteArray();
-              }
-            });
-      } catch (IOException e) {
-        throw new RuntimeException(e);
-      }
-    }
-
-    @Override
-    public void put(String key, byte[] value) {
-      try {
-        HttpClient client = clientFactory.build();
-        HttpPut put = new HttpPut(baseUrl + "/" + key);
-        put.setEntity(new ByteArrayEntity(value));
-        put.setHeader("Content-Type", "application/octet-stream");
-        client.execute(
-            put,
-            new ResponseHandler<Void>() {
-              @Override
-              public Void handleResponse(HttpResponse response) {
-                int statusCode = response.getStatusLine().getStatusCode();
-
-                // Accept more than SC_OK to be compatible with Nginx WebDav module.
-                if (HttpStatus.SC_OK != statusCode
-                    && HttpStatus.SC_ACCEPTED != statusCode
-                    && HttpStatus.SC_CREATED != statusCode
-                    && HttpStatus.SC_NO_CONTENT != statusCode) {
-                  throw new RuntimeException("PUT failed with status code " + statusCode);
-                }
-                return null;
-              }
-            });
-      } catch (IOException e) {
-        throw new RuntimeException(e);
-      }
-    }
-  }
-
   public static SimpleBlobStore createRest(RemoteOptions options) {
     return new RestBlobStore(options.remoteRestCache, options.restCachePoolSize);
   }
diff --git a/src/main/java/com/google/devtools/build/lib/remote/blobstore/ConcurrentMapBlobStore.java b/src/main/java/com/google/devtools/build/lib/remote/blobstore/ConcurrentMapBlobStore.java
new file mode 100644
index 0000000000..59b5a117b0
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/remote/blobstore/ConcurrentMapBlobStore.java
@@ -0,0 +1,43 @@
+// Copyright 2017 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.blobstore;
+
+import java.util.concurrent.ConcurrentMap;
+
+/** A {@link SimpleBlobStore} implementation using a {@link ConcurrentMap}. */
+public final class ConcurrentMapBlobStore implements SimpleBlobStore {
+  private final ConcurrentMap<String, byte[]> map;
+
+  public ConcurrentMapBlobStore(ConcurrentMap<String, byte[]> map) {
+    this.map = map;
+  }
+
+  @Override
+  public boolean containsKey(String key) {
+    return map.containsKey(key);
+  }
+
+  @Override
+  public byte[] get(String key) {
+    return map.get(key);
+  }
+
+  @Override
+  public void put(String key, byte[] value) {
+    map.put(key, value);
+  }
+
+  @Override
+  public void close() {}
+}
\ No newline at end of file
diff --git a/src/main/java/com/google/devtools/build/lib/remote/blobstore/RestBlobStore.java b/src/main/java/com/google/devtools/build/lib/remote/blobstore/RestBlobStore.java
new file mode 100644
index 0000000000..898bd84c8a
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/remote/blobstore/RestBlobStore.java
@@ -0,0 +1,122 @@
+// Copyright 2017 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.blobstore;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpHead;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+
+/**
+ * Implementation of {@link SimpleBlobStore} with a REST service. The REST service needs to
+ * support the following HTTP methods.
+ *
+ * <p>PUT /cache/1234 HTTP/1.1 PUT method is used to upload a blob with a base16 key. In this
+ * example the key is 1234. Valid status codes are 200, 201, 202 and 204.
+ *
+ * <p>GET /cache/1234 HTTP/1.1 GET method fetches a blob with the specified key. In this example
+ * the key is 1234. A status code of 200 should be followed by the content of blob. Status code of
+ * 404 or 204 means the key cannot be found.
+ *
+ * <p>HEAD /cache/1234 HTTP/1.1 HEAD method checks to see if the specified key exists in the blob
+ * store. A status code of 200 indicates the key is found in the blob store. A status code of 404
+ * indicates the key is not found in the blob store.
+ */
+public final class RestBlobStore implements SimpleBlobStore {
+
+  private final String baseUrl;
+  private final PoolingHttpClientConnectionManager connMan;
+  private final HttpClientBuilder clientFactory;
+
+  /**
+   * Creates a new instance.
+   *
+   * @param baseUrl base URL for the remote cache
+   * @param poolSize maximum number of simultaneous connections
+   */
+  public RestBlobStore(String baseUrl, int poolSize) {
+    this.baseUrl = baseUrl;
+    connMan = new PoolingHttpClientConnectionManager();
+    connMan.setDefaultMaxPerRoute(poolSize);
+    connMan.setMaxTotal(poolSize);
+    clientFactory = HttpClientBuilder.create();
+    clientFactory.setConnectionManager(connMan);
+    clientFactory.setConnectionManagerShared(true);
+  }
+
+  @Override
+  public void close() {
+    connMan.close();
+  }
+
+  @Override
+  public boolean containsKey(String key) throws IOException {
+    HttpClient client = clientFactory.build();
+    HttpHead head = new HttpHead(baseUrl + "/" + key);
+    return client.execute(
+        head,
+        response -> {
+          int statusCode = response.getStatusLine().getStatusCode();
+          return HttpStatus.SC_OK == statusCode;
+        });
+  }
+
+  @Override
+  public byte[] get(String key) throws IOException {
+    HttpClient client = clientFactory.build();
+    HttpGet get = new HttpGet(baseUrl + "/" + key);
+    return client.execute(
+        get,
+        response -> {
+          int statusCode = response.getStatusLine().getStatusCode();
+          if (HttpStatus.SC_NOT_FOUND == statusCode
+              || HttpStatus.SC_NO_CONTENT == statusCode) {
+            return null;
+          }
+          if (HttpStatus.SC_OK != statusCode) {
+            throw new IOException("GET failed with status code " + statusCode);
+          }
+          ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+          response.getEntity().writeTo(buffer);
+          return buffer.toByteArray();
+        });
+  }
+
+  @Override
+  public void put(String key, byte[] value) throws IOException {
+    HttpClient client = clientFactory.build();
+    HttpPut put = new HttpPut(baseUrl + "/" + key);
+    put.setEntity(new ByteArrayEntity(value));
+    put.setHeader("Content-Type", "application/octet-stream");
+    client.execute(
+        put,
+        (response) -> {
+          int statusCode = response.getStatusLine().getStatusCode();
+          // Accept more than SC_OK to be compatible with Nginx WebDav module.
+          if (HttpStatus.SC_OK != statusCode
+              && HttpStatus.SC_ACCEPTED != statusCode
+              && HttpStatus.SC_CREATED != statusCode
+              && HttpStatus.SC_NO_CONTENT != statusCode) {
+            throw new IOException("PUT failed with status code " + statusCode);
+          }
+          return null;
+        });
+  }
+}
\ No newline at end of file
diff --git a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStore.java b/src/main/java/com/google/devtools/build/lib/remote/blobstore/SimpleBlobStore.java
index 66b80139dc..f3088b9d60 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStore.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/blobstore/SimpleBlobStore.java
@@ -12,27 +12,29 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.google.devtools.build.lib.remote;
+package com.google.devtools.build.lib.remote.blobstore;
+
+import java.io.IOException;
 
 /**
  * A simple interface for storing blobs (in the form of byte arrays) each one indexed by a
  * hexadecimal string. Implementation must be thread-safe.
  */
 public interface SimpleBlobStore {
-  /** Returns true if the provided {@param key} is stored in the blob store. */
-  boolean containsKey(String key);
+  /** Returns true if the provided {@code key} is stored in the blob store. */
+  boolean containsKey(String key) throws IOException, InterruptedException;
 
   /**
-   * Returns the blob (in the form of a byte array) indexed by {@param key}. Returns null if the
-   * {@pram key} cannot be found.
+   * Returns the blob (in the form of a byte array) indexed by {@code key}. Returns null if the
+   * {@code key} cannot be found.
    */
-  byte[] get(String key);
+  byte[] get(String key) throws IOException, InterruptedException;
 
   /**
-   * Uploads a blob (as {@param value}) indexed by {@param key} to the blob store. Existing blob
-   * indexed by the same {@param key} will be overwritten.
+   * Uploads a blob (as {@code value}) indexed by {@code key} to the blob store. Existing blob
+   * indexed by the same {@code key} will be overwritten.
    */
-  void put(String key, byte[] value);
+  void put(String key, byte[] value) throws IOException, InterruptedException;
 
   /** Close resources associated with the blob store. */
   void close();
diff --git a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/ByteStreamServer.java b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/ByteStreamServer.java
index c090ce8553..87ca590ca0 100644
--- a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/ByteStreamServer.java
+++ b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/ByteStreamServer.java
@@ -68,11 +68,6 @@ final class ByteStreamServer extends ByteStreamImplBase {
               "Failed parsing digest from resource_name:" + request.getResourceName()));
     }
 
-    if (!cache.containsKey(digest)) {
-      responseObserver.onError(StatusUtils.notFoundError(digest));
-      return;
-    }
-
     try {
       // This still relies on the blob size to be small enough to fit in memory.
       // TODO(olaola): refactor to fix this if the need arises.
@@ -83,7 +78,6 @@ final class ByteStreamServer extends ByteStreamImplBase {
       }
       responseObserver.onCompleted();
     } catch (CacheNotFoundException e) {
-      // This can only happen if an item gets evicted right after we check.
       responseObserver.onError(StatusUtils.notFoundError(digest));
     } catch (Exception e) {
       logger.log(WARNING, "Read request failed.", e);
diff --git a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/CasServer.java b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/CasServer.java
index 037c8bd45b..13e42b239d 100644
--- a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/CasServer.java
+++ b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/CasServer.java
@@ -24,6 +24,7 @@ import com.google.devtools.remoteexecution.v1test.FindMissingBlobsResponse;
 import com.google.devtools.remoteexecution.v1test.UpdateBlobRequest;
 import com.google.rpc.Code;
 import io.grpc.stub.StreamObserver;
+import java.io.IOException;
 
 /** A basic implementation of a {@link ContentAddressableStorageImplBase} service. */
 final class CasServer extends ContentAddressableStorageImplBase {
@@ -38,14 +39,23 @@ final class CasServer extends ContentAddressableStorageImplBase {
       FindMissingBlobsRequest request, StreamObserver<FindMissingBlobsResponse> responseObserver) {
     FindMissingBlobsResponse.Builder response = FindMissingBlobsResponse.newBuilder();
 
-    for (Digest digest : request.getBlobDigestsList()) {
-      if (!cache.containsKey(digest)) {
-        response.addMissingBlobDigests(digest);
+    try {
+      for (Digest digest : request.getBlobDigestsList()) {
+        try {
+          if (!cache.containsKey(digest)) {
+            response.addMissingBlobDigests(digest);
+          }
+         } catch (InterruptedException e) {
+          responseObserver.onError(StatusUtils.interruptedError(digest));
+          Thread.currentThread().interrupt();
+          return;
+         }
       }
+      responseObserver.onNext(response.build());
+      responseObserver.onCompleted();
+    } catch (IOException e) {
+      responseObserver.onError(StatusUtils.internalError(e));
     }
-
-    responseObserver.onNext(response.build());
-    responseObserver.onCompleted();
   }
 
   @Override
diff --git a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/RemoteWorker.java b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/RemoteWorker.java
index efc8927bb7..d9056e9932 100644
--- a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/RemoteWorker.java
+++ b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/RemoteWorker.java
@@ -24,9 +24,10 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.ByteStreams;
 import com.google.devtools.build.lib.remote.RemoteOptions;
-import com.google.devtools.build.lib.remote.SimpleBlobStore;
 import com.google.devtools.build.lib.remote.SimpleBlobStoreActionCache;
 import com.google.devtools.build.lib.remote.SimpleBlobStoreFactory;
+import com.google.devtools.build.lib.remote.blobstore.ConcurrentMapBlobStore;
+import com.google.devtools.build.lib.remote.blobstore.SimpleBlobStore;
 import com.google.devtools.build.lib.shell.Command;
 import com.google.devtools.build.lib.shell.CommandException;
 import com.google.devtools.build.lib.shell.CommandResult;
@@ -49,7 +50,9 @@ import io.grpc.netty.NettyServerBuilder;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.PrintWriter;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.logging.Logger;
 
@@ -118,8 +121,10 @@ public final class RemoteWorker {
     }
 
     final Path pidFile = getFileSystem().getPath(workerOptions.pidFile);
-    try (PrintWriter printWriter = new PrintWriter(pidFile.getPathFile())) {
-      printWriter.println(ProcessUtils.getpid());
+    try (Writer writer =
+        new OutputStreamWriter(pidFile.getOutputStream(), StandardCharsets.UTF_8)) {
+      writer.write(Integer.toString(ProcessUtils.getpid()));
+      writer.write("\n");
     }
 
     Runtime.getRuntime()
@@ -164,7 +169,7 @@ public final class RemoteWorker {
     SimpleBlobStore blobStore =
         usingRemoteCache
             ? SimpleBlobStoreFactory.create(remoteOptions)
-            : new SimpleBlobStoreFactory.ConcurrentMapBlobStore(
+            : new ConcurrentMapBlobStore(
                 new ConcurrentHashMap<String, byte[]>());
 
     RemoteWorker worker =
diff --git a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/StatusUtils.java b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/StatusUtils.java
index c3b2e73f73..cc860dbd72 100644
--- a/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/StatusUtils.java
+++ b/src/tools/remote_worker/src/main/java/com/google/devtools/build/remote/StatusUtils.java
@@ -49,6 +49,17 @@ final class StatusUtils {
         .build();
   }
 
+  static StatusRuntimeException interruptedError(Digest digest) {
+    return StatusProto.toStatusRuntimeException(interruptedStatus(digest));
+  }
+
+  static com.google.rpc.Status interruptedStatus(Digest digest) {
+    return Status.newBuilder()
+        .setCode(Code.CANCELLED.getNumber())
+        .setMessage("Server operation was interrupted for " + digest)
+        .build();
+  }
+
   static StatusRuntimeException invalidArgumentError(String field, String desc) {
     return StatusProto.toStatusRuntimeException(invalidArgumentStatus(field, desc));
   }