feat(storage): adding disabling option for bidi reads

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

@@ -150,7 +150,10 @@ private AccumulatingRead(
         IOAutoCloseable onCloseCallback) {
       super(rangeSpec, retryContext, onCloseCallback);
       this.readId = readId;
-      this.hasher = hasher;
+      this.hasher =
+          (rangeSpec.begin() == 0)
+              ? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
+              : hasher;
       this.complete = SettableApiFuture.create();
       this.childRefs = Collections.synchronizedList(new ArrayList<>());
     }
@@ -280,7 +283,10 @@ static class StreamingRead extends BaseObjectReadSessionStreamRead<ScatteringByt
         IOAutoCloseable onCloseCallback) {
       super(rangeSpec, retryContext, onCloseCallback);
       this.readId = new AtomicLong(readId);
-      this.hasher = hasher;
+      this.hasher =
+          (rangeSpec.begin() == 0)
+              ? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
+              : hasher;
       this.closed = false;
       this.failFuture = SettableApiFuture.create();
       this.queue = new ArrayBlockingQueue<>(2);

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

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2026 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.grpc.GrpcStatusCode;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.Object;
+import io.grpc.Status.Code;
+import java.nio.ByteBuffer;
+import java.util.Locale;
+import java.util.OptionalLong;
+import java.util.function.Supplier;
+
+/**
+ * A wrapper around hasher that accumulates checksums and validates them at the end of the read if
+ * it was a full object read.
+ */
+final class CumulativeHasher implements Hasher {
+  private final Hasher delegate;
+  private final long startOffset;
+  private final OptionalLong limit;
+  private Crc32cLengthKnown cumulativeHash;
+
+  CumulativeHasher(Hasher delegate, long startOffset, OptionalLong limit) {
+    this.delegate = delegate;
+    this.startOffset = startOffset;
+    this.limit = limit;
+    this.cumulativeHash = delegate.initialValue();
+  }
+
+  @Override
+  public Crc32cLengthKnown hash(ByteBuffer b) {
+    return delegate.hash(b);
+  }
+
+  @Override
+  public Crc32cLengthKnown hash(ByteString byteString) {
+    return delegate.hash(byteString);
+  }
+
+  @Override
+  public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b)
+      throws ChecksumMismatchException {
+    ByteBuffer byteBuffer = b.get();
+    Crc32cLengthKnown actual = delegate.hash(byteBuffer);
+    if (actual != null) {
+      if (expected != null && !actual.eqValue(expected)) {
+        throw new ChecksumMismatchException(expected, actual);
+      }
+      accumulate(actual);
+    }
+  }
+
+  @Override
+  public void validate(Crc32cValue<?> expected, ByteString byteString)
+      throws ChecksumMismatchException {
+    Crc32cLengthKnown actual = delegate.hash(byteString);
+    if (actual != null) {
+      if (expected != null && !actual.eqValue(expected)) {
+        throw new ChecksumMismatchException(expected, actual);
+      }
+      accumulate(actual);
+    }
+  }
+
+  @Override
+  public void validateUnchecked(Crc32cValue<?> expected, ByteString byteString)
+      throws UncheckedChecksumMismatchException {
+    Crc32cLengthKnown actual = delegate.hash(byteString);
+    if (actual != null) {
+      if (expected != null && !actual.eqValue(expected)) {
+        throw new UncheckedChecksumMismatchException(expected, actual);
+      }
+      accumulate(actual);
+    }
+  }
+
+  @Override
+  public <C extends Crc32cValue<?>> C nullSafeConcat(C r1, Crc32cLengthKnown r2) {
+    return delegate.nullSafeConcat(r1, r2);
+  }
+
+  @Override
+  public Crc32cLengthKnown initialValue() {
+    return delegate.initialValue();
+  }
+
+  // Checks if it was a full object read.
+  boolean qualifiesForVerification(Object metadata) {
+    return startOffset == 0
+        && metadata != null
+        && metadata.hasChecksums()
+        && metadata.getChecksums().hasCrc32C()
+        && (!limit.isPresent() || limit.getAsLong() >= metadata.getSize());
+  }
+
+  void validateCumulativeChecksum(Object metadata)
+      throws UncheckedCumulativeChecksumMismatchException {
+    if (qualifiesForVerification(metadata)) {
+      Crc32cLengthKnown actual = getCumulativeHash();
+      if (actual == null) {
+        return;
+      }
+      Crc32cValue<?> expected = Crc32cValue.of(metadata.getChecksums().getCrc32C());
+      if (!actual.eqValue(expected)) {
+        throw new UncheckedCumulativeChecksumMismatchException(expected, actual);
+      }
+    }
+  }
+
+  private void accumulate(Crc32cLengthKnown actual) {
+    cumulativeHash = nullSafeConcat(cumulativeHash, actual);
+  }
+
+  Crc32cLengthKnown getCumulativeHash() {
+    return cumulativeHash;
+  }
+}
+
+final class UncheckedCumulativeChecksumMismatchException
+    extends com.google.api.gax.rpc.DataLossException {
+  private static final GrpcStatusCode STATUS_CODE = GrpcStatusCode.of(Code.DATA_LOSS);
+  private final Crc32cValue<?> expected;
+  private final Crc32cLengthKnown actual;
+
+  UncheckedCumulativeChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
+    super(
+        String.format(
+            Locale.US,
+            "Mismatch cumulative checksum value. Expected %s actual %s",
+            expected.debugString(),
+            actual.debugString()),
+        /* cause= */ null,
+        STATUS_CODE,
+        /* retryable= */ false);
+    this.expected = expected;
+    this.actual = actual;
+  }
+
+  Crc32cValue<?> getExpected() {
+    return expected;
+  }
+
+  Crc32cLengthKnown getActual() {
+    return actual;
+  }
+}

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

@@ -49,6 +49,7 @@
 import java.nio.channels.ScatteringByteChannel;
 import java.util.List;
 import java.util.Locale;
+import java.util.OptionalLong;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
@@ -91,7 +92,15 @@ final class GapicUnbufferedReadableByteChannel
     this.result = result;
     this.read = read;
     this.req = req;
-    this.hasher = hasher;
+    this.hasher =
+        (req.getReadOffset() == 0)
+            ? new CumulativeHasher(
+                hasher,
+                0,
+                req.getReadLimit() <= 0
+                    ? OptionalLong.empty()
+                    : OptionalLong.of(req.getReadLimit()))
+            : hasher;
     this.fetchOffset = new AtomicLong(req.getReadOffset());
     this.blobOffset = req.getReadOffset();
     this.retrier = retrier;
@@ -174,6 +183,7 @@ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
       }
       if (take == EOF_MARKER) {
         complete = true;
+        validateCumulativeChecksum();
         break;
       }
 
@@ -311,6 +321,17 @@ private IOException createError(String message) throws IOException {
     return new IOException(message, cause);
   }
 
+  private void validateCumulativeChecksum() throws IOException {
+    if (hasher instanceof CumulativeHasher) {
+      CumulativeHasher cumulativeHasher = (CumulativeHasher) hasher;
+      try {
+        cumulativeHasher.validateCumulativeChecksum(metadata);
+      } catch (UncheckedCumulativeChecksumMismatchException exception) {
+        throw new IOException(StorageException.coalesce(exception));
+      }
+    }
+  }
+
   private final class ReadObjectObserver extends StateCheckingResponseObserver<ReadObjectResponse> {
 
     private final SettableApiFuture<Void> open = SettableApiFuture.create();

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

@@ -212,7 +212,7 @@ final class ChecksumMismatchException extends IOException {
     private final Crc32cValue<?> expected;
     private final Crc32cLengthKnown actual;
 
-    private ChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
+    ChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
       super(
           String.format(
               Locale.US,
@@ -237,7 +237,7 @@ final class UncheckedChecksumMismatchException extends DataLossException {
     private final Crc32cValue<?> expected;
     private final Crc32cLengthKnown actual;
 
-    private UncheckedChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
+    UncheckedChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
       super(
           String.format(
               "Mismatch checksum value. Expected %s actual %s",

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

@@ -351,6 +351,12 @@ public void onResponse(BidiReadObjectResponse response) {
             executor.execute(
                 StorageException.liftToRunnable(
                     () -> {
+                      try {
+                        validateCumulativeChecksum(read);
+                      } catch (UncheckedCumulativeChecksumMismatchException e) {
+                        state.removeOutstandingReadOnFailure(id, read::fail).onFailure(e);
+                        return;
+                      }
                       read.eof();
                       // don't remove the outstanding read until the future has been resolved
                       state.removeOutstandingRead(id);
@@ -545,6 +551,15 @@ public void onComplete() {
     }
   }
 
+  private void validateCumulativeChecksum(ObjectReadSessionStreamRead<?> read) {
+    Hasher hasher = read.hasher();
+    if (hasher instanceof CumulativeHasher) {
+      CumulativeHasher cumulativeHasher = (CumulativeHasher) hasher;
+      com.google.storage.v2.Object metadata = state.getMetadata();
+      cumulativeHasher.validateCumulativeChecksum(metadata);
+    }
+  }
+
   static ObjectReadSessionStream create(
       ScheduledExecutorService executor,
       ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable,

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

@@ -46,7 +46,7 @@
 @BetaApi
 @Immutable
 public final class ReadAsChannel extends BaseConfig<ScatteringByteChannel, StreamingRead> {
-  static final ReadAsChannel INSTANCE = new ReadAsChannel(RangeSpec.all(), Hasher.enabled());
+  static final ReadAsChannel INSTANCE = new ReadAsChannel(RangeSpec.all(), Hasher.defaultHasher());
 
   private final RangeSpec range;
   private final Hasher hasher;
@@ -97,7 +97,7 @@ public ReadAsChannel withRangeSpec(RangeSpec range) {
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -111,7 +111,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsChannel withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;

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

@@ -46,7 +46,7 @@ public final class ReadAsFutureByteString
     extends BaseConfig<ApiFuture<DisposableByteString>, AccumulatingRead<DisposableByteString>> {
 
   static final ReadAsFutureByteString INSTANCE =
-      new ReadAsFutureByteString(RangeSpec.all(), Hasher.enabled());
+      new ReadAsFutureByteString(RangeSpec.all(), Hasher.defaultHasher());
 
   private final RangeSpec range;
   private final Hasher hasher;
@@ -97,7 +97,7 @@ public ReadAsFutureByteString withRangeSpec(RangeSpec range) {
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -111,7 +111,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsFutureByteString withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;

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

@@ -41,7 +41,7 @@ public final class ReadAsFutureBytes
     extends BaseConfig<ApiFuture<byte[]>, AccumulatingRead<byte[]>> {
 
   static final ReadAsFutureBytes INSTANCE =
-      new ReadAsFutureBytes(RangeSpec.all(), Hasher.enabled());
+      new ReadAsFutureBytes(RangeSpec.all(), Hasher.defaultHasher());
 
   private final RangeSpec range;
   private final Hasher hasher;
@@ -92,7 +92,7 @@ public ReadAsFutureBytes withRangeSpec(RangeSpec range) {
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -106,7 +106,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsFutureBytes withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;

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

@@ -46,7 +46,7 @@
 public final class ReadAsSeekableChannel extends ReadProjectionConfig<SeekableByteChannel> {
 
   static final ReadAsSeekableChannel INSTANCE =
-      new ReadAsSeekableChannel(Hasher.enabled(), LinearExponentialRangeSpecFunction.INSTANCE);
+      new ReadAsSeekableChannel(Hasher.defaultHasher(), LinearExponentialRangeSpecFunction.INSTANCE);
 
   private final Hasher hasher;
   private final RangeSpecFunction rangeSpecFunction;
@@ -94,7 +94,7 @@ public ReadAsSeekableChannel withRangeSpecFunction(RangeSpecFunction rangeSpecFu
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -108,7 +108,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsSeekableChannel withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;