chore: update BlobDescriptorStreamRead to track whether it is ready to be sent as a request

BenWhitehead · BenWhitehead · commit f437c0a427d2 · 2025-03-14T16:50:07.000-04:00
Also update state tracking to track if tombstoned, like when withNewReadId is called
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorState.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorState.java
@@ -116,6 +116,7 @@ BlobDescriptorStreamRead assignNewReadId(long oldReadId) {
   List<ReadRange> getOutstandingReads() {
     synchronized (this) {
       return outstandingReads.values().stream()
+          .filter(BlobDescriptorStreamRead::readyToSend)
           .map(BlobDescriptorStreamRead::makeReadRange)
           .collect(ImmutableList.toImmutableList());
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorStreamRead.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorStreamRead.java
@@ -41,6 +41,7 @@ abstract class BlobDescriptorStreamRead implements AutoCloseable, Closeable {
   protected final List<ChildRef> childRefs;
   protected final RetryContext retryContext;
   protected boolean closed;
+  protected boolean tombstoned;
 
   private BlobDescriptorStreamRead(long readId, ReadCursor readCursor, RetryContext retryContext) {
     this(readId, readCursor, Collections.synchronizedList(new ArrayList<>()), retryContext, false);
@@ -57,6 +58,7 @@ private BlobDescriptorStreamRead(
     this.childRefs = childRefs;
     this.retryContext = retryContext;
     this.closed = closed;
+    this.tombstoned = false;
   }
 
   ReadCursor getReadCursor() {
@@ -110,7 +112,9 @@ public void close() throws IOException {
     }
   }
 
-  abstract void recordError(Throwable t, OnSuccess onSuccess, OnFailure onFailure);
+  abstract <T extends Throwable> void recordError(T t, OnSuccess onSuccess, OnFailure<T> onFailure);
+
+  public abstract boolean readyToSend();
 
   static AccumulatingRead<byte[]> createByteArrayAccumulatingRead(
       long readId,
@@ -154,22 +158,30 @@ private AccumulatingRead(
 
     @Override
     boolean acceptingBytes() {
-      return !complete.isDone() && readCursor.hasRemaining();
+      return !complete.isDone() && !tombstoned && readCursor.hasRemaining();
     }
 
     @Override
-    void fail(Throwable t) throws IOException {
+    void fail(Throwable t) {
       try {
-        complete.setException(t);
-      } finally {
+        tombstoned = true;
         close();
+      } catch (IOException e) {
+        t.addSuppressed(t);
+      } finally {
+        complete.setException(t);
       }
     }
 
     @Override
-    void recordError(Throwable t, OnSuccess onSuccess, OnFailure onFailure) {
+    <T extends Throwable> void recordError(T t, OnSuccess onSuccess, OnFailure<T> onFailure) {
       retryContext.recordError(t, onSuccess, onFailure);
     }
+
+    @Override
+    public boolean readyToSend() {
+      return !tombstoned && !retryContext.inBackoff();
+    }
   }
 
   /**
@@ -235,6 +247,7 @@ void eof() throws IOException {
 
     @Override
     ByteArrayAccumulatingRead withNewReadId(long newReadId) {
+      this.tombstoned = true;
       return new ByteArrayAccumulatingRead(
           newReadId, readCursor, childRefs, retryContext, closed, complete);
     }
@@ -291,6 +304,7 @@ void eof() throws IOException {
 
     @Override
     ZeroCopyByteStringAccumulatingRead withNewReadId(long newReadId) {
+      this.tombstoned = true;
       return new ZeroCopyByteStringAccumulatingRead(
           newReadId, readCursor, childRefs, retryContext, closed, complete, byteString);
     }

Original file line number	Diff line number	Diff line change
`@@ -116,6 +116,7 @@ BlobDescriptorStreamRead assignNewReadId(long oldReadId) {`
`116`	`116`	`List<ReadRange> getOutstandingReads() {`
`117`	`117`	`synchronized (this) {`
`118`	`118`	`return outstandingReads.values().stream()`
	`119`	`+ .filter(BlobDescriptorStreamRead::readyToSend)`
`119`	`120`	`.map(BlobDescriptorStreamRead::makeReadRange)`
`120`	`121`	`.collect(ImmutableList.toImmutableList());`
`121`	`122`	`}`