fix: update gRPC ReadObject retry to avoid double retry (#2765)

BenWhitehead · web-flow · commit 1fc57b9b4fe2 · 2024-10-10T16:27:01.000-04:00
Prevent adding an unavailable error to the queue if the retrying loop is still active.

Also, cleanup unused retry config for gRPC ReadObject.
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java
@@ -47,6 +47,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class GapicUnbufferedReadableByteChannel
     implements UnbufferedReadableByteChannel, ScatteringByteChannel {
@@ -258,6 +259,11 @@ ApiFuture<Object> getResult() {
 
   private void ensureStreamOpen() {
     if (readObjectObserver == null) {
+      java.lang.Object peek = queue.peek();
+      if (peek instanceof Throwable || peek == EOF_MARKER) {
+        // If our queue has an error or EOF, do not send another request
+        return;
+      }
       readObjectObserver =
           Retrying.run(
               retryingDeps,
@@ -326,13 +332,15 @@ protected void onResponseImpl(ReadObjectResponse response) {
 
     @Override
     protected void onErrorImpl(Throwable t) {
-      open.setException(t);
-      if (!alg.shouldRetry(t, null)) {
-        result.setException(StorageException.coalesce(t));
-      }
       if (t instanceof CancellationException) {
         cancellation.set(t);
       }
+      if (!open.isDone()) {
+        open.setException(t);
+        if (!alg.shouldRetry(t, null)) {
+          result.setException(StorageException.coalesce(t));
+        }
+      }
       try {
         queue.offer(t);
       } catch (InterruptedException e) {
@@ -369,6 +377,11 @@ public boolean nonEmpty() {
       return !queue.isEmpty();
     }
 
+    @Nullable
+    public T peek() {
+      return queue.peek();
+    }
+
     @NonNull
     public T poll() throws InterruptedException {
       return queue.take();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -723,8 +723,7 @@ public GrpcBlobReadChannel reader(String bucket, String blob, BlobSourceOption..
   public GrpcBlobReadChannel reader(BlobId blob, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts);
     ReadObjectRequest request = getReadObjectRequest(blob, opts);
-    Set<StatusCode.Code> codes = resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(request));
-    GrpcCallContext grpcCallContext = Retrying.newCallContext().withRetryableCodes(codes);
+    GrpcCallContext grpcCallContext = Retrying.newCallContext();
 
     return new GrpcBlobReadChannel(
         storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext),
@@ -1708,10 +1707,7 @@ private UnbufferedReadableByteChannelSession<Object> unbufferedReadSession(
 
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts);
     ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts);
-    Set<StatusCode.Code> codes =
-        resultRetryAlgorithmToCodes(retryAlgorithmManager.getFor(readObjectRequest));
-    GrpcCallContext grpcCallContext =
-        opts.grpcMetadataMapper().apply(Retrying.newCallContext().withRetryableCodes(codes));
+    GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(Retrying.newCallContext());
     return ResumableMedia.gapic()
         .read()
         .byteChannel(
