[Do not Submit] fix(storage): only reset appendable upload retry budget on progress

java-storage/google-cloud-storage/pom.xml

@@ -459,6 +459,8 @@
         <configuration>
           <forkCount>1</forkCount>
           <reuseForks>false</reuseForks>
+          <!-- Each IT class runs in its own JVM; kill a stuck class instead of hanging the job -->
+          <forkedProcessTimeoutInSeconds>900</forkedProcessTimeoutInSeconds>
           <systemPropertyVariables>
             <logback.statusListenerClass>ch.qos.logback.core.status.NopStatusListener</logback.statusListenerClass>
           </systemPropertyVariables>

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

@@ -0,0 +1,114 @@
+/*
+ * 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.storage.v2.BidiWriteObjectRequest;
+import com.google.storage.v2.BidiWriteObjectResponse;
+import io.grpc.Status;
+import java.util.concurrent.atomic.AtomicLong;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * DIAGNOSTIC ONLY (do not submit). Logs every message sent / received on appendable upload streams
+ * so a CI run shows exactly where an upload waits.
+ */
+final class AppendableDiag {
+  static final Logger LOG = LoggerFactory.getLogger("com.google.cloud.storage.DIAG");
+  private static final AtomicLong IDS = new AtomicLong();
+
+  private AppendableDiag() {}
+
+  static long nextId() {
+    return IDS.incrementAndGet();
+  }
+
+  /** Never throws: diagnostics must not change behavior (test stubs may not implement getters). */
+  static Object safe(java.util.function.Supplier<?> s) {
+    try {
+      return s.get();
+    } catch (Throwable t) {
+      return "n/a";
+    }
+  }
+
+  static String req(BidiWriteObjectRequest r) {
+    StringBuilder sb = new StringBuilder();
+    if (r.hasWriteObjectSpec()) {
+      sb.append("spec=write(")
+          .append(r.getWriteObjectSpec().getResource().getName())
+          .append(", appendable=")
+          .append(r.getWriteObjectSpec().getAppendable())
+          .append(") ");
+    } else if (r.hasAppendObjectSpec()) {
+      sb.append("spec=append(")
+          .append(r.getAppendObjectSpec().getObject())
+          .append(", gen=")
+          .append(r.getAppendObjectSpec().getGeneration())
+          .append(", handle=")
+          .append(r.getAppendObjectSpec().hasWriteHandle())
+          .append(", routingToken=")
+          .append(r.getAppendObjectSpec().hasRoutingToken())
+          .append(") ");
+    } else if (!r.getUploadId().isEmpty()) {
+      sb.append("uploadId=").append(r.getUploadId()).append(' ');
+    }
+    sb.append("offset=").append(r.getWriteOffset());
+    if (r.hasChecksummedData()) {
+      sb.append(" data=").append(r.getChecksummedData().getContent().size());
+    }
+    sb.append(" flush=").append(r.getFlush());
+    sb.append(" lookup=").append(r.getStateLookup());
+    sb.append(" finish=").append(r.getFinishWrite());
+    if (r.hasObjectChecksums()) {
+      sb.append(" objectCrc32c=").append(r.getObjectChecksums().hasCrc32C());
+    }
+    return sb.toString();
+  }
+
+  static String resp(BidiWriteObjectResponse r) {
+    StringBuilder sb = new StringBuilder();
+    if (r.hasPersistedSize()) {
+      sb.append("persistedSize=").append(r.getPersistedSize());
+    }
+    if (r.hasResource()) {
+      sb.append("resource(size=")
+          .append(r.getResource().getSize())
+          .append(", gen=")
+          .append(r.getResource().getGeneration())
+          .append(", finalized=")
+          .append(r.getResource().hasFinalizeTime())
+          .append(")");
+    }
+    sb.append(" writeHandle=").append(r.hasWriteHandle());
+    return sb.toString();
+  }
+
+  static String err(@Nullable Throwable t) {
+    if (t == null) {
+      return "null";
+    }
+    Status s = Status.fromThrowable(t);
+    String msg = String.valueOf(t.getMessage());
+    int nl = msg.indexOf('\n');
+    if (nl > -1) {
+      msg = msg.substring(0, nl);
+    }
+    return t.getClass().getSimpleName() + "(grpcStatus=" + s.getCode() + "): " + msg;
+  }
+}

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

@@ -97,13 +97,29 @@ public void close() throws IOException {
       if (!writeCalledAtLeastOnce) {
         stream.flush();
       }
+      // DIAGNOSTIC ONLY
+      long loops = 0;
+      AppendableDiag.LOG.info(
+          "DIAG channel CLOSE begin writeOffset={} finalize={} expectedCrc32c={}",
+          writeOffset,
+          nextWriteShouldFinalize,
+          expectedCrc32c != null);
       if (nextWriteShouldFinalize) {
         //noinspection StatementWithEmptyBody
-        while (!stream.finishWrite(writeOffset, expectedCrc32c)) {}
+        while (!stream.finishWrite(writeOffset, expectedCrc32c)) {
+          loops++;
+        }
       } else {
         //noinspection StatementWithEmptyBody
-        while (!stream.closeStream(writeOffset)) {}
+        while (!stream.closeStream(writeOffset)) {
+          loops++;
+        }
       }
+      AppendableDiag.LOG.info(
+          "DIAG channel CLOSE enqueued {} after {} retry loop(s); waiting for ack of {}",
+          nextWriteShouldFinalize ? "finish_write" : "final flush",
+          loops,
+          writeOffset);
 
       awaitResultFuture();
     } finally {
@@ -122,8 +138,16 @@ public void nextWriteShouldFinalize(@Nullable String expectedCrc32c) {
   }
 
   void flush() throws InterruptedException {
+    // DIAGNOSTIC ONLY
+    long start = System.nanoTime();
+    AppendableDiag.LOG.info("DIAG channel FLUSH begin writeOffset={}", writeOffset);
     stream.flush();
     stream.awaitAckOf(writeOffset);
+    AppendableDiag.LOG.info(
+        "DIAG channel FLUSH end writeOffset={} waitedMs={} resultDone={}",
+        writeOffset,
+        TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
+        stream.getResultFuture().isDone());
   }
 
   private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
@@ -191,16 +215,30 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th
   }
 
   private void awaitResultFuture() throws IOException {
+    // DIAGNOSTIC ONLY
+    long start = System.nanoTime();
     try {
       stream.awaitAckOf(writeOffset);
+      AppendableDiag.LOG.info(
+          "DIAG channel ACK-WAIT end writeOffset={} waitedMs={} resultDone={}",
+          writeOffset,
+          TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
+          stream.getResultFuture().isDone());
       stream.getResultFuture().get(10_717, TimeUnit.MILLISECONDS);
+      AppendableDiag.LOG.info(
+          "DIAG channel RESULT OK totalMs={}",
+          TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
       InterruptedIOException ioe = new InterruptedIOException();
       ioe.initCause(e);
       writeThrewError = true;
       throw ioe;
     } catch (ExecutionException e) {
+      AppendableDiag.LOG.info(
+          "DIAG channel RESULT FAILED totalMs={} cause={}",
+          TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
+          AppendableDiag.err(e.getCause()));
       BaseServiceException coalesce = StorageException.coalesce(e.getCause());
       String message = coalesce.getMessage();
       String ioExceptionMessage = message;
@@ -215,6 +253,9 @@ private void awaitResultFuture() throws IOException {
       writeThrewError = true;
       throw ioException;
     } catch (TimeoutException e) {
+      AppendableDiag.LOG.info(
+          "DIAG channel RESULT TIMEOUT (ack arrived, result not set in 10.7s) totalMs={}",
+          TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
       writeThrewError = true;
       throw new IOException(e);
     }