chore: ensure BlobDescriptor opening can be retried

BenWhitehead · BenWhitehead · commit 36160978c767 · 2025-03-14T16:50:50.000-04:00
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorStream.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorStream.java
@@ -44,6 +44,7 @@
 import com.google.storage.v2.ReadRange;
 import com.google.storage.v2.ReadRangeError;
 import io.grpc.Status.Code;
+import io.grpc.StatusRuntimeException;
 import java.io.IOException;
 import java.nio.channels.AsynchronousCloseException;
 import java.util.List;
@@ -441,15 +442,28 @@ public void onResponse(BidiReadObjectResponse response) {
     @Override
     public void onError(Throwable t) {
       delegate.onError(t);
-      blobDescriptorResolveFuture.setException(t);
       openSignal.setException(t);
       closeSignal.setException(t);
     }
 
     @Override
     public void onComplete() {
       delegate.onComplete();
-      blobDescriptorResolveFuture.set(null);
+      if (state.getMetadata() == null) {
+        StatusRuntimeException cause =
+            Code.UNAVAILABLE
+                .toStatus()
+                .withDescription("onComplete without prior onNext")
+                .asRuntimeException();
+        ApiException apiException =
+            ApiExceptionFactory.createException(cause, GrpcStatusCode.of(Code.UNAVAILABLE), false);
+        StorageException storageException =
+            new StorageException(0, cause.getMessage(), apiException);
+        streamRetryContext.recordError(
+            storageException,
+            BlobDescriptorStream.this::restart,
+            blobDescriptorResolveFuture::setException);
+      }
       openSignal.set(null);
       closeSignal.set(null);
     }
@@ -489,6 +503,7 @@ public void onError(Throwable t) {
         // bubble all the way up to the invoker we'll be able to see it in a bug report.
         t.addSuppressed(new MaxRedirectsExceededException(maxRedirectsAllowed, redirectCount));
         delegate.onError(t);
+        blobDescriptorResolveFuture.setException(t);
         return;
       }
       if (error.hasReadHandle()) {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITBlobDescriptorFakeTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITBlobDescriptorFakeTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.ByteSizeConstants._2MiB;
+import static com.google.cloud.storage.TestUtils.apiException;
 import static com.google.cloud.storage.TestUtils.assertAll;
 import static com.google.cloud.storage.TestUtils.getChecksummedData;
 import static com.google.cloud.storage.TestUtils.xxd;
@@ -28,6 +29,7 @@
 import com.google.api.core.ApiFutures;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.AbortedException;
+import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.DataLossException;
 import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.api.gax.rpc.UnavailableException;
@@ -1067,6 +1069,74 @@ public void streaming() throws Exception {
     }
   }
 
+  @Test
+  public void retryableErrorWhileOpeningIsRetried() throws Exception {
+    AtomicInteger reqCounter = new AtomicInteger(0);
+    FakeStorage fake =
+        FakeStorage.of(
+            ImmutableMap.of(
+                REQ_OPEN,
+                respond -> {
+                  int i = reqCounter.incrementAndGet();
+                  if (i <= 1) {
+                    ApiException apiException =
+                        apiException(Code.UNAVAILABLE, String.format("{unavailable %d}", i));
+                    respond.onError(apiException);
+                  } else {
+                    respond.onNext(RES_OPEN);
+                  }
+                }));
+
+    try (FakeServer fakeServer = FakeServer.of(fake);
+        Storage storage =
+            fakeServer
+                .getGrpcStorageOptions()
+                .toBuilder()
+                .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(3).build())
+                .build()
+                .getService()) {
+
+      BlobId id = BlobId.of("b", "o");
+      ApiFuture<BlobDescriptor> futureBlobDescriptor = storage.getBlobDescriptor(id);
+      try (BlobDescriptor bd = futureBlobDescriptor.get(20, TimeUnit.SECONDS)) {
+        assertThat(bd).isNotNull();
+      }
+    }
+  }
+
+  @Test
+  public void onCompleteWithoutAValue() throws Exception {
+    // I'm not sure if this is something that can actually happen in practice, but is being here
+    // to ensure it's at least accounted for, rather than a null pointer exception or something else
+    // equally cryptic.
+    FakeStorage fake = FakeStorage.of(ImmutableMap.of(REQ_OPEN, StreamObserver::onCompleted));
+
+    try (FakeServer fakeServer = FakeServer.of(fake);
+        Storage storage =
+            fakeServer
+                .getGrpcStorageOptions()
+                .toBuilder()
+                .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(3).build())
+                .build()
+                .getService()) {
+
+      BlobId id = BlobId.of("b", "o");
+      ApiFuture<BlobDescriptor> futureBlobDescriptor = storage.getBlobDescriptor(id);
+      ExecutionException ee =
+          assertThrows(
+              ExecutionException.class, () -> futureBlobDescriptor.get(20, TimeUnit.SECONDS));
+      assertAll(
+          () -> assertThat(ee).hasCauseThat().isInstanceOf(StorageException.class),
+          () ->
+              assertThat(ee).hasCauseThat().hasCauseThat().isInstanceOf(UnavailableException.class),
+          () -> assertThat(((StorageException) ee.getCause()).getCode()).isEqualTo(0),
+          () -> {
+            String messages = TestUtils.messagesToText(ee.getCause());
+            assertThat(messages).contains("Unretryable error");
+          });
+    }
+  }
+
   private static void runTestAgainstFakeServer(
       FakeStorage fakeStorage, RangeSpec range, ChecksummedTestContent expected) throws Exception {
 
