feat: add post-upload crc32c validation for most uploads

google-cloud-storage/pom.xml

@@ -63,6 +63,10 @@
       <groupId>com.google.api</groupId>
       <artifactId>gax</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.api</groupId>
+      <artifactId>gax-httpjson</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.api</groupId>
       <artifactId>gax-grpc</artifactId>

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

@@ -300,13 +300,13 @@ static Get createGetRequest(
   }
 
   @SuppressWarnings("unchecked")
-  private static <T> T cast(Object o) {
+  static <T> T cast(Object o) {
     return (T) o;
   }
 
   @Nullable
   @SuppressWarnings("unchecked")
-  private static String getHeaderValue(@NonNull HttpHeaders headers, @NonNull String headerName) {
+  static String getHeaderValue(@NonNull HttpHeaders headers, @NonNull String headerName) {
     Object o = headers.get(headerName);
     if (o == null) {
       return null;

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

@@ -21,13 +21,17 @@
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.api.core.SettableApiFuture;
+import com.google.cloud.BaseServiceException;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.RecoveryFileManager.RecoveryVolumeSinkFactory;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
 import com.google.common.util.concurrent.MoreExecutors;
 import java.io.IOException;
 import java.io.ObjectInputStream;
@@ -215,12 +219,19 @@ private final class Flusher implements WritableByteChannel {
 
         private final WritableByteChannel delegate;
 
+        private final Hasher cumulativeCrc32c;
+        private long totalLength;
+
         private Flusher(WritableByteChannel delegate) {
           this.delegate = delegate;
+          this.cumulativeCrc32c = Hashing.crc32c().newHasher();
+          this.totalLength = 0;
         }
 
         @Override
         public int write(ByteBuffer src) throws IOException {
+          totalLength += src.remaining();
+          cumulativeCrc32c.putBytes(src.duplicate());
           return delegate.write(src);
         }
 
@@ -232,6 +243,7 @@ public boolean isOpen() {
         @Override
         public void close() throws IOException {
           delegate.close();
+          Crc32cLengthKnown expected = Crc32cValue.of(cumulativeCrc32c.hash().asInt(), totalLength);
           try (RecoveryFile rf = Factory.WriteToFileThenUpload.this.rf) {
             Path path = rf.getPath();
             long size = Files.size(path);
@@ -241,11 +253,20 @@ public void close() throws IOException {
                 size,
                 () -> {
                   BlobInfo blob = storage.internalCreateFrom(path, info, opts);
+                  Crc32cLengthKnown actual =
+                      Crc32cValue.of(Utils.crc32cCodec.decode(blob.getCrc32c()), blob.getSize());
+                  if (!expected.eqValue(actual)) {
+                    throw new ClientDetectedDataLossException(actual, expected);
+                  }
                   result.set(blob);
                 });
           } catch (StorageException | IOException e) {
             result.setException(e);
             throw e;
+          } catch (ClientDetectedDataLossException e) {
+            BaseServiceException coalesce = StorageException.coalesce(e);
+            result.setException(coalesce);
+            throw coalesce;
           }
         }
       }

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

@@ -194,4 +194,13 @@ static long copyUsingBuffer(ByteBuffer buf, ReadableByteChannel r, WritableByteC
     }
     return total;
   }
+
+  static ByteBuffer[] duplicate(ByteBuffer[] buffers, int offset, int length) {
+    ByteBuffer[] returnValue = new ByteBuffer[length - offset];
+    for (int i = offset; i < length; i++) {
+      ByteBuffer buffer = buffers[i];
+      returnValue[i] = buffer.duplicate();
+    }
+    return returnValue;
+  }
 }

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * 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.cloud.storage;
+
+import com.google.api.gax.httpjson.HttpJsonStatusCode;
+import com.google.api.gax.rpc.DataLossException;
+import com.google.api.gax.rpc.StatusCode.Code;
+import java.util.Locale;
+
+final class ClientDetectedDataLossException extends DataLossException {
+
+  public ClientDetectedDataLossException(Crc32cValue<?> actual, Crc32cValue<?> expected) {
+    super(
+        String.format(Locale.US, "actual: %s, expected: %s", actual, expected),
+        null,
+        HttpJsonStatusCode.of(Code.DATA_LOSS),
+        false);
+  }
+}

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

@@ -18,6 +18,7 @@
 
 import static java.util.Objects.requireNonNull;
 
+import com.google.api.core.ApiFunction;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.InternalApi;
@@ -27,10 +28,13 @@
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.storage.ChannelSession.BufferedWriteSession;
 import com.google.cloud.storage.ChannelSession.UnbufferedWriteSession;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.Retrying.Retrier;
 import com.google.cloud.storage.Retrying.RetrierWithAlg;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.storage.v2.Object;
 import com.google.storage.v2.QueryWriteStatusRequest;
 import com.google.storage.v2.QueryWriteStatusResponse;
 import com.google.storage.v2.ServiceConstants.Values;
@@ -43,6 +47,12 @@
 final class GapicWritableByteChannelSessionBuilder {
 
   private static final int DEFAULT_BUFFER_CAPACITY = ByteSizeConstants._16MiB;
+  private static final ApiFunction<WriteObjectResponse, Crc32cLengthKnown>
+      WRITE_OBJECT_RESPONSE_CRC32C_VALUE =
+          writeObjectResponse -> {
+            Object resource = writeObjectResponse.getResource();
+            return Crc32cValue.of(resource.getChecksums().getCrc32C(), resource.getSize());
+          };
   private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
   private Hasher hasher;
   private ByteStringStrategy byteStringStrategy;
@@ -210,11 +220,17 @@ BufferedWritableByteChannelSession<WriteObjectResponse> build() {
         return new BufferedWriteSession<>(
             ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
             lift((WriteObjectRequest start, SettableApiFuture<WriteObjectResponse> resultFuture) ->
-                    new GapicUnbufferedDirectWritableByteChannel(
-                        resultFuture,
-                        getChunkSegmenter(),
-                        write,
-                        new WriteCtx<>(WriteObjectRequestBuilderFactory.simple(start))))
+                    StorageByteChannels.writable()
+                        .validateUploadCrc32c(
+                            new GapicUnbufferedDirectWritableByteChannel(
+                                resultFuture,
+                                getChunkSegmenter(),
+                                write,
+                                new WriteCtx<>(WriteObjectRequestBuilderFactory.simple(start))),
+                            ApiFutures.transform(
+                                resultFuture,
+                                WRITE_OBJECT_RESPONSE_CRC32C_VALUE,
+                                MoreExecutors.directExecutor())))
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
@@ -293,18 +309,28 @@ UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
         return new UnbufferedWriteSession<>(
             requireNonNull(start, "start must be non null"),
             lift((ResumableWrite start, SettableApiFuture<WriteObjectResponse> result) -> {
+                  UnbufferedWritableByteChannel channel;
                   if (fsyncEvery) {
-                    return new GapicUnbufferedChunkedResumableWritableByteChannel(
-                        result,
-                        getChunkSegmenter(),
-                        write,
-                        new WriteCtx<>(start),
-                        boundRetrier,
-                        Retrying::newCallContext);
+                    channel =
+                        new GapicUnbufferedChunkedResumableWritableByteChannel(
+                            result,
+                            getChunkSegmenter(),
+                            write,
+                            new WriteCtx<>(start),
+                            boundRetrier,
+                            Retrying::newCallContext);
                   } else {
-                    return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, new WriteCtx<>(start));
+                    channel =
+                        new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+                            result, getChunkSegmenter(), write, new WriteCtx<>(start));
                   }
+                  return StorageByteChannels.writable()
+                      .validateUploadCrc32c(
+                          channel,
+                          ApiFutures.transform(
+                              result,
+                              WRITE_OBJECT_RESPONSE_CRC32C_VALUE,
+                              MoreExecutors.directExecutor()));
                 })
                 .andThen(StorageByteChannels.writable()::createSynchronized));
       }
@@ -333,18 +359,28 @@ BufferedWritableByteChannelSession<WriteObjectResponse> build() {
         return new BufferedWriteSession<>(
             requireNonNull(start, "start must be non null"),
             lift((ResumableWrite start, SettableApiFuture<WriteObjectResponse> result) -> {
+                  UnbufferedWritableByteChannel channel;
                   if (fsyncEvery) {
-                    return new GapicUnbufferedChunkedResumableWritableByteChannel(
-                        result,
-                        getChunkSegmenter(),
-                        write,
-                        new WriteCtx<>(start),
-                        retrier,
-                        Retrying::newCallContext);
+                    channel =
+                        new GapicUnbufferedChunkedResumableWritableByteChannel(
+                            result,
+                            getChunkSegmenter(),
+                            write,
+                            new WriteCtx<>(start),
+                            retrier,
+                            Retrying::newCallContext);
                   } else {
-                    return new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
-                        result, getChunkSegmenter(), write, new WriteCtx<>(start));
+                    channel =
+                        new GapicUnbufferedFinalizeOnCloseResumableWritableByteChannel(
+                            result, getChunkSegmenter(), write, new WriteCtx<>(start));
                   }
+                  return StorageByteChannels.writable()
+                      .validateUploadCrc32c(
+                          channel,
+                          ApiFutures.transform(
+                              result,
+                              WRITE_OBJECT_RESPONSE_CRC32C_VALUE,
+                              MoreExecutors.directExecutor()));
                 })
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));
@@ -436,16 +472,23 @@ BufferedWritableByteChannelSession<WriteObjectResponse> build() {
         ByteStringStrategy boundStrategy = byteStringStrategy;
         Hasher boundHasher = hasher;
         return (writeCtx, resultFuture) ->
-            new SyncAndUploadUnbufferedWritableByteChannel(
-                write,
-                query,
-                resultFuture,
-                new ChunkSegmenter(boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
-                boundRetrier,
-                alg,
-                writeCtx,
-                recoveryFile,
-                recoveryBuffer);
+            StorageByteChannels.writable()
+                .validateUploadCrc32c(
+                    new SyncAndUploadUnbufferedWritableByteChannel(
+                        write,
+                        query,
+                        resultFuture,
+                        new ChunkSegmenter(
+                            boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
+                        boundRetrier,
+                        alg,
+                        writeCtx,
+                        recoveryFile,
+                        recoveryBuffer),
+                    ApiFutures.transform(
+                        resultFuture,
+                        WRITE_OBJECT_RESPONSE_CRC32C_VALUE,
+                        MoreExecutors.directExecutor()));
       }
     }
   }

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

@@ -91,6 +91,7 @@ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
           delegate = source;
         }
       } catch (InterruptedException | ExecutionException e) {
+        Thread.currentThread().interrupt();
         throw new IOException(e);
       }
     } else if (leftovers != null && leftovers.hasRemaining()) {

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

@@ -18,21 +18,36 @@
 
 import static java.util.Objects.requireNonNull;
 
+import com.google.api.core.ApiFunction;
 import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.ChannelSession.BufferedWriteSession;
 import com.google.cloud.storage.ChannelSession.UnbufferedWriteSession;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.Retrying.RetrierWithAlg;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.util.function.BiFunction;
 import java.util.function.LongConsumer;
 import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class HttpWritableByteChannelSessionBuilder {
 
   private static final int DEFAULT_BUFFER_CAPACITY = ByteSizeConstants._16MiB;
+  public static final ApiFunction<StorageObject, Crc32cLengthKnown> STORAGE_OBJECT_TO_CRC32C_VALUE =
+      storageObject -> {
+        @Nullable BigInteger size = storageObject.getSize();
+        long length = 0;
+        if (size != null) {
+          length = size.longValueExact();
+        }
+        return Crc32cValue.of(Utils.crc32cCodec.decode(storageObject.getCrc32c()), length);
+      };
   @NonNull private final HttpClientContext httpClientContext;
 
   HttpWritableByteChannelSessionBuilder(@NonNull HttpClientContext httpClientContext) {
@@ -118,8 +133,14 @@ BufferedResumableUploadBuilder buffered(BufferHandle bufferHandle) {
       // fields.
       RetrierWithAlg boundRetrier = retrier;
       return (start, resultFuture) ->
-          new ApiaryUnbufferedWritableByteChannel(
-              httpClientContext, boundRetrier, start, resultFuture, committedBytesCallback);
+          StorageByteChannels.writable()
+              .validateUploadCrc32c(
+                  new ApiaryUnbufferedWritableByteChannel(
+                      httpClientContext, boundRetrier, start, resultFuture, committedBytesCallback),
+                  ApiFutures.transform(
+                      resultFuture,
+                      STORAGE_OBJECT_TO_CRC32C_VALUE,
+                      MoreExecutors.directExecutor()));
     }
 
     final class UnbufferedResumableUploadBuilder {

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

@@ -148,7 +148,7 @@ public void rewindTo(long offset) {
           response.ignore();
           actualSize = new BigInteger(storedContentLength, 10);
           success = true;
-          storageObject = null;
+          storageObject = jsonResumableWrite.storageObjectFromResponseHeaders(response);
         } else {
           response.ignore();
           StorageException se =