feat(storage): allow explicit checksum on appendable upload finalization

Author: Dhriti Chopra

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
 
@@ -36,6 +37,7 @@ final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWri
   private boolean open;
   private long writeOffset;
   private volatile boolean nextWriteShouldFinalize;
+  private @Nullable String expectedCrc32c;
   private boolean writeCalledAtLeastOnce;
   private long lastFlushOffset;
 
@@ -53,6 +55,7 @@ final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWri
     this.open = true;
     this.writeOffset = writeOffset;
     this.nextWriteShouldFinalize = false;
+    this.expectedCrc32c = null;
     this.writeThrewError = false;
     this.lastFlushOffset = writeOffset;
   }
@@ -96,7 +99,7 @@ public void close() throws IOException {
       }
       if (nextWriteShouldFinalize) {
         //noinspection StatementWithEmptyBody
-        while (!stream.finishWrite(writeOffset)) {}
+        while (!stream.finishWrite(writeOffset, expectedCrc32c)) {}
       } else {
         //noinspection StatementWithEmptyBody
         while (!stream.closeStream(writeOffset)) {}
@@ -113,6 +116,11 @@ public void nextWriteShouldFinalize() {
     this.nextWriteShouldFinalize = true;
   }
 
+  public void nextWriteShouldFinalize(String expectedCrc32c) {
+    this.nextWriteShouldFinalize = true;
+    this.expectedCrc32c = expectedCrc32c;
+  }
+
   void flush() throws InterruptedException {
     stream.flush();
     stream.awaitAckOf(writeOffset);
@@ -155,7 +163,7 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th
       if (i < lastIdx && !shouldFlush) {
         appended = stream.append(datum);
       } else if (i == lastIdx && remainingAfterPacking == 0 && nextWriteShouldFinalize) {
-        appended = stream.appendAndFinalize(datum);
+        appended = stream.appendAndFinalize(datum, expectedCrc32c);
       } else {
         appended = stream.appendAndFlush(datum);
       }

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadStreamingStream.java

@@ -126,12 +126,13 @@ public boolean appendAndFlush(ChunkSegmenter.@NonNull ChunkSegment data) {
     }
   }
 
-  public boolean appendAndFinalize(ChunkSegmenter.@NonNull ChunkSegment data) {
+  public boolean appendAndFinalize(
+      ChunkSegmenter.@NonNull ChunkSegment data, @Nullable String expectedCrc32c) {
     lock.lock();
     try {
       boolean offered = state.offer(data);
       if (offered) {
-        finishWrite(state.getTotalSentBytes());
+        finishWrite(state.getTotalSentBytes(), expectedCrc32c);
       }
       return offered;
     } finally {
@@ -163,6 +164,10 @@ public void flush() {
   }
 
   public boolean finishWrite(long length) {
+    return finishWrite(length, null);
+  }
+
+  public boolean finishWrite(long length, @Nullable String expectedCrc32c) {
     lock.lock();
     try {
       // if we're already finalizing, ack rather than enqueueing again
@@ -172,10 +177,15 @@ public boolean finishWrite(long length) {
 
       BidiWriteObjectRequest.Builder b =
           BidiWriteObjectRequest.newBuilder().setWriteOffset(length).setFinishWrite(true);
-      Crc32cLengthKnown cumulativeCrc32c = state.getCumulativeCrc32c();
-      if (cumulativeCrc32c != null) {
-        b.setObjectChecksums(
-            ObjectChecksums.newBuilder().setCrc32C(cumulativeCrc32c.getValue()).build());
+      if (expectedCrc32c != null) {
+        int crc32cInt = Utils.crc32cCodec.decode(expectedCrc32c);
+        b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cInt).build());
+      } else {
+        Crc32cLengthKnown cumulativeCrc32c = state.getCumulativeCrc32c();
+        if (cumulativeCrc32c != null) {
+          b.setObjectChecksums(
+              ObjectChecksums.newBuilder().setCrc32C(cumulativeCrc32c.getValue()).build());
+        }
       }
       BidiWriteObjectRequest msg = b.build();
       boolean offer = state.offer(msg);

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobAppendableUpload.java

@@ -26,6 +26,7 @@
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.WritableByteChannel;
 import java.util.concurrent.TimeUnit;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * Interface representing those methods which can be used to write to and interact with an
@@ -158,6 +159,32 @@ interface AppendableUploadWriteableByteChannel extends WritableByteChannel {
     @BetaApi
     void finalizeAndClose() throws IOException;
 
+    /**
+     * <b>This method is blocking</b>
+     *
+     * <p>Finalize the upload and close this instance to further {@link #write(ByteBuffer)}ing. This
+     * will close any underlying stream and release any releasable resources once out of scope.
+     *
+     * <p>Once this method is called, and returns no more writes to the object will be allowed by
+     * GCS.
+     *
+     * <p>This method and {@link #close()} are mutually exclusive. If one of the other methods are
+     * called before this method, this method will be a no-op.
+     *
+     * @param expectedCrc32c A Base64 encoded string representing the expected CRC32c value for the
+     *     entire object. If provided, the server will validate the final object's CRC32c against
+     *     this value. If there's a mismatch, the server will return an error (such as
+     *     InvalidArgument), and this method will throw a {@link StorageException} or equivalent
+     *     exception, failing the upload.
+     * @see Storage#blobAppendableUpload(BlobInfo, BlobAppendableUploadConfig, BlobWriteOption...)
+     * @see BlobAppendableUploadConfig.CloseAction#FINALIZE_WHEN_CLOSING
+     * @see BlobAppendableUploadConfig#getCloseAction()
+     * @see BlobAppendableUploadConfig#withCloseAction(CloseAction)
+     * @since 2.51.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    void finalizeAndClose(@Nullable String expectedCrc32c) throws IOException;
+
     /**
      * <b>This method is blocking</b>
      *
@@ -197,5 +224,28 @@ interface AppendableUploadWriteableByteChannel extends WritableByteChannel {
      */
     @BetaApi
     void close() throws IOException;
+
+    /**
+     * <b>This method is blocking</b>
+     *
+     * <p>Close this instance to further {@link #write(ByteBuffer)}ing.
+     *
+     * <p>This method behaves like {@link #close()}, but allows passing an optional expected CRC32c
+     * checksum. If expectedCrc32c is null, it behaves identically to {@link #close()}.
+     *
+     * @param expectedCrc32c A Base64 encoded string representing the expected CRC32c value for the
+     *     entire object. If provided, the server will validate the final object's CRC32c against
+     *     this value. If there's a mismatch, the server will return an error (such as
+     *     InvalidArgument), and this method will throw a {@link StorageException} or equivalent
+     *     exception, failing the upload. May be null.
+     * @throws IllegalArgumentException if the stream was not configured with {@link
+     *     CloseAction#FINALIZE_WHEN_CLOSING} and expectedCrc32c is not null.
+     * @see Storage#blobAppendableUpload(BlobInfo, BlobAppendableUploadConfig, BlobWriteOption...)
+     * @see BlobAppendableUploadConfig#getCloseAction()
+     * @see BlobAppendableUploadConfig#withCloseAction(CloseAction)
+     * @since 2.51.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    void close(@Nullable String expectedCrc32c) throws IOException;
   }
 }

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobAppendableUploadImpl.java

@@ -24,6 +24,7 @@
 import java.io.InterruptedIOException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.locks.ReentrantLock;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 @BetaApi
 final class BlobAppendableUploadImpl implements BlobAppendableUpload {
@@ -132,6 +133,19 @@ public void finalizeAndClose() throws IOException {
       }
     }
 
+    @Override
+    public void finalizeAndClose(String expectedCrc32c) throws IOException {
+      lock.lock();
+      try {
+        if (buffered.isOpen()) {
+          unbuffered.nextWriteShouldFinalize(expectedCrc32c);
+          buffered.close();
+        }
+      } finally {
+        lock.unlock();
+      }
+    }
+
     @Override
     public void closeWithoutFinalizing() throws IOException {
       lock.lock();
@@ -152,5 +166,17 @@ public void close() throws IOException {
         closeWithoutFinalizing();
       }
     }
+
+    @Override
+    public void close(@Nullable String expectedCrc32c) throws IOException {
+      if (finalizeOnClose) {
+        finalizeAndClose(expectedCrc32c);
+      } else if (expectedCrc32c == null) {
+        closeWithoutFinalizing();
+      } else {
+        throw new IllegalArgumentException(
+            "expectedCrc32c can only be provided when finalizeOnClose is true.");
+      }
+    }
   }
 }

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java

@@ -2195,6 +2195,32 @@ public void finalizeAndClose() throws IOException {
         }
       }
 
+      @Override
+      @BetaApi
+      public void finalizeAndClose(String expectedCrc32c) throws IOException {
+        try (Scope ignore = openSpan.makeCurrent()) {
+          Span span = tracer.spanBuilder("finalizeAndClose").startSpan();
+          try (Scope ignore2 = span.makeCurrent()) {
+            delegate.finalizeAndClose(expectedCrc32c);
+          } catch (Throwable t) {
+            span.recordException(t);
+            span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+            throw t;
+          } finally {
+            span.end();
+          }
+        } catch (IOException | RuntimeException e) {
+          openSpan.recordException(e);
+          openSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
+          uploadSpan.recordException(e);
+          uploadSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
+          throw e;
+        } finally {
+          openSpan.end();
+          uploadSpan.end();
+        }
+      }
+
       @Override
       @BetaApi
       public void closeWithoutFinalizing() throws IOException {
@@ -2247,6 +2273,32 @@ public void close() throws IOException {
         }
       }
 
+      @Override
+      @BetaApi
+      public void close(@Nullable String expectedCrc32c) throws IOException {
+        try (Scope ignore = openSpan.makeCurrent()) {
+          Span span = tracer.spanBuilder("close").startSpan();
+          try (Scope ignore2 = span.makeCurrent()) {
+            delegate.close(expectedCrc32c);
+          } catch (Throwable t) {
+            span.recordException(t);
+            span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+            throw t;
+          } finally {
+            span.end();
+          }
+        } catch (IOException | RuntimeException e) {
+          openSpan.recordException(e);
+          openSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
+          uploadSpan.recordException(e);
+          uploadSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
+          throw e;
+        } finally {
+          openSpan.end();
+          uploadSpan.end();
+        }
+      }
+
       @Override
       public void flush() throws IOException {
         try (Scope ignore = openSpan.makeCurrent()) {