Bytestreams bugfixes (buildfarm#240)

* Write reset interface addition

Write resets required a new interface, close() was not sufficient to
correctly handle a write request with a 0 offset with the intention of
resetting.

* Logging cleanup for ByteStreamService

* Call readBlob in ByteStreamService for blobs

ByteStreamService now uses the readBlob layer to interpret special
unlimited 'limit' of 0.

* Memory getOperationStreamWrite implementation

Expand the ByteStringStreamSource to support committedSize and a
completion future for close.

* ByteStreamService queryWriteStatus implementation

Present write query interface for progressive client implementations

* Prevent write processing of completed blobs

An attempt to upload a previously completed blob will be handled
immediately by the write listener, and must not attempt to write into
its output stream.

* Executor write support for stdout/stderr

The (renamed) ByteStringWriteReader now waits for a successful write
completion in getData(), uses ByteString.Output, and try-with-resources.
In the event of an empty operation stream for stdout/err, a null Write
is used that completes immediately upon close.
The Executor no longer interrupts the readers to preserve remote request
safety - they are expected to close safely.

Author: werkt

src/main/java/build/buildfarm/BUILD

@@ -110,6 +110,7 @@ java_library(
     name = "server",
     srcs = glob(["server/**/*.java"]),
     deps = [
+        ":cas",
         ":common",
         ":common-grpc",
         ":instance",

src/main/java/build/buildfarm/cas/DigestMismatchException.java

@@ -4,7 +4,7 @@
 import build.buildfarm.common.DigestUtil;
 import java.io.IOException;
 
-class DigestMismatchException extends IOException {
+public class DigestMismatchException extends IOException {
   private final Digest actual;
   private final Digest expected;
 

src/main/java/build/buildfarm/cas/MemoryWriteOutputStream.java

@@ -17,7 +17,7 @@ class MemoryWriteOutputStream extends OutputStream implements Write {
   private final Digest digest;
   private final ListenableFuture<ByteString> writtenFuture;
   private final ByteString.Output out;
-  private final HashingOutputStream hashOut;
+  private HashingOutputStream hashOut;
 
   MemoryWriteOutputStream(ContentAddressableStorage storage, Digest digest, ListenableFuture<ByteString> writtenFuture) {
     this.storage = storage;
@@ -95,6 +95,12 @@ public OutputStream getOutput() {
     return this;
   }
 
+  @Override
+  public void reset() {
+    out.reset();
+    hashOut = DigestUtil.forDigest(digest).newHashingOutputStream(out);
+  }
+
   @Override
   public void addListener(Runnable onCompleted, Executor executor) {
     writtenFuture.addListener(onCompleted, executor);

src/main/java/build/buildfarm/cas/Writes.java

@@ -65,6 +65,10 @@ public OutputStream getOutput() {
       return nullOutputStream();
     }
 
+    @Override
+    public void reset() {
+    }
+
     @Override
     public void addListener(Runnable onCompleted, Executor executor) {
       executor.execute(onCompleted);

src/main/java/build/buildfarm/common/Write.java

@@ -25,6 +25,8 @@ public interface Write {
 
   OutputStream getOutput() throws IOException;
 
+  void reset();
+
   /** add a callback to be invoked when blob has been completed */
   void addListener(Runnable onCompleted, Executor executor);
 }

src/main/java/build/buildfarm/common/grpc/StubWriteOutputStream.java

@@ -30,10 +30,17 @@ public class StubWriteOutputStream extends OutputStream implements Write {
   private final long expectedSize;
   private final boolean autoflush;
   private final byte buf[];
+  private boolean wasReset = false;
   private final Supplier<QueryWriteStatusResponse> writeStatus = Suppliers.memoize(
       new Supplier() {
         @Override
         public QueryWriteStatusResponse get() {
+          if (wasReset) {
+            return QueryWriteStatusResponse.newBuilder()
+                .setCommittedSize(0)
+                .setComplete(false)
+                .build();
+          }
           return bsBlockingStub.get()
               .queryWriteStatus(QueryWriteStatusRequest.newBuilder()
                   .setResourceName(resourceName)
@@ -194,6 +201,13 @@ public OutputStream getOutput() {
     return this;
   }
 
+  @Override
+  public void reset() {
+    wasReset = true;
+    offset = 0;
+    writtenBytes = 0;
+  }
+
   @Override
   public void addListener(Runnable onCompleted, Executor executor) {
     writeFuture.addListener(onCompleted, executor);

src/main/java/build/buildfarm/instance/memory/ByteStringStreamSource.java

@@ -15,12 +15,15 @@
 package build.buildfarm.instance.memory;
 
 import com.google.protobuf.ByteString;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import java.io.InputStream;
 import java.io.OutputStream;
 
 class ByteStringStreamSource {
   private final Runnable onClose;
   private final OutputStream outputStream;
+  private final SettableFuture<Void> closedFuture = SettableFuture.create();
 
   private final Object bufferSync;
   private ByteString buffer;
@@ -55,6 +58,7 @@ public void close() {
         synchronized (bufferSync) {
           closed = true;
           bufferSync.notifyAll();
+          closedFuture.set(null);
         }
         onClose.run();
       }
@@ -71,6 +75,14 @@ public OutputStream getOutputStream() {
     return outputStream;
   }
 
+  public long getCommittedSize() {
+    return buffer.size();
+  }
+
+  public ListenableFuture<Void> getClosedFuture() {
+    return closedFuture;
+  }
+
   public InputStream openStream() {
     return new InputStream() {
       private int offset = 0;

src/main/java/build/buildfarm/instance/memory/MemoryInstance.java

@@ -80,6 +80,7 @@
 import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Predicate;
 import java.util.logging.Logger;
@@ -272,15 +273,33 @@ private ByteStringStreamSource getSource(String name) {
 
   @Override
   public Write getOperationStreamWrite(String name) {
-    throw new UnsupportedOperationException(); // needs source->write conversion
-  }
+    return new Write() {
+      @Override
+      public long getCommittedSize() {
+        return getSource(name).getCommittedSize();
+      }
 
-  /*
-  @Override
-  public OutputStream getStreamOutput(String name) {
-    return getSource(name).getOutputStream();
+      @Override
+      public boolean isComplete() {
+        return getSource(name).isClosed();
+      }
+
+      @Override
+      public OutputStream getOutput() {
+        return getSource(name).getOutputStream();
+      }
+
+      @Override
+      public void reset() {
+        streams.remove(name);
+      }
+
+      @Override
+      public void addListener(Runnable onCompleted, Executor executor) {
+        getSource(name).getClosedFuture().addListener(onCompleted, executor);
+      }
+    };
   }
-  */
 
   @Override
   public InputStream newOperationStreamInput(String name, long offset) throws IOException {