feat(gax): enforce global timeout for resumable uploads

Enforces ResumableUploadCallSettings.getGlobalTimeout() across the
resumable upload lifecycle.

A scheduled task is registered at future initiation using the client
context executor, covering both session initiation and chunk upload
streaming. On deadline expiry, any in-flight attempt future is cancelled,
the payload stream is closed, and the session future fails with a
non-retryable DeadlineExceededException. The scheduled timeout task is cancelled
cleanly across all terminal outcomes (success, failure, cancellation).

Single payload closure is guaranteed via AtomicBoolean compare-and-set,
with close errors attached via addSuppressed on failure paths and logged
on success paths. Buffer safety is maintained during in-flight timeout
cancellation without stream advancement or buffer corruption.

Author: whowes

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java

@@ -52,6 +52,9 @@
 import java.io.InputStream;
 import java.time.Duration;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import org.jspecify.annotations.NullMarked;
 import org.jspecify.annotations.Nullable;
 
@@ -84,6 +87,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
   private final UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>>
       queryStatusCallable;
   private final InputStream payload;
+  private final ResumableUploadCallSettings settings;
   private final int chunkSize;
   private final ApiCallContext callContext;
   private final ClientContext clientContext;
@@ -101,6 +105,9 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
   @GuardedBy("lock")
   private @Nullable ApiFuture<?> inFlightFuture;
 
+  @GuardedBy("lock")
+  private @Nullable ScheduledFuture<?> timeoutFuture;
+
   ResumableUploadChunkCoordinator(
       SettableApiFuture<ResponseT> result,
       ApiFuture<ResumableUploadSession> startFuture,
@@ -117,7 +124,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
     this.queryStatusCallable =
         checkNotNull(queryStatusCallable, "queryStatusCallable must not be null");
     this.payload = checkNotNull(payload, "payload must not be null");
-    checkNotNull(settings, "settings must not be null");
+    this.settings = checkNotNull(settings, "settings must not be null");
     checkArgument(settings.getChunkSize() > 0, "chunkSize must be > 0");
     this.chunkSize = settings.getChunkSize();
     this.callContext = checkNotNull(callContext, "callContext must not be null");
@@ -129,6 +136,19 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
   }
 
   void start() {
+    Duration timeout = settings.getGlobalTimeout();
+    if (timeout != null && !timeout.isZero() && !timeout.isNegative()) {
+      ScheduledExecutorService executor = clientContext.getExecutor();
+      if (executor != null) {
+        synchronized (lock) {
+          if (!done) {
+            this.timeoutFuture =
+                executor.schedule(this::onTimeout, timeout.toMillis(), TimeUnit.MILLISECONDS);
+          }
+        }
+      }
+    }
+
     ApiFutures.addCallback(
         startFuture,
         new ApiFutureCallback<ResumableUploadSession>() {
@@ -155,6 +175,32 @@ public void onFailure(Throwable t) {
         MoreExecutors.directExecutor());
   }
 
+  private static final StatusCode TIMEOUT_STATUS_CODE =
+      new StatusCode() {
+        @Override
+        public StatusCode.Code getCode() {
+          return StatusCode.Code.DEADLINE_EXCEEDED;
+        }
+
+        @Override
+        public @Nullable Object getTransportCode() {
+          return null;
+        }
+      };
+
+  private void onTimeout() {
+    synchronized (lock) {
+      if (done) {
+        return;
+      }
+    }
+    String message =
+        uploadSessionUrl != null
+            ? "Resumable upload timed out for session: " + uploadSessionUrl
+            : "Resumable upload timed out before session initiation completed";
+    finish(null, new DeadlineExceededException(message, null, TIMEOUT_STATUS_CODE, false));
+  }
+
   @Nullable String getUploadSessionUrl() {
     return uploadSessionUrl;
   }
@@ -163,7 +209,7 @@ void setInFlightFuture(ApiFuture<?> future) {
     boolean shouldCancel = false;
     synchronized (lock) {
       if (done) {
-        shouldCancel = result.isCancelled();
+        shouldCancel = true;
       } else {
         this.inFlightFuture = future;
       }
@@ -174,6 +220,7 @@ void setInFlightFuture(ApiFuture<?> future) {
   }
 
   void cancel(boolean mayInterruptIfRunning) {
+    ScheduledFuture<?> timeout;
     ApiFuture<?> inFlight;
     synchronized (lock) {
       if (done) {
@@ -182,6 +229,11 @@ void cancel(boolean mayInterruptIfRunning) {
       done = true;
       inFlight = this.inFlightFuture;
       this.inFlightFuture = null;
+      timeout = this.timeoutFuture;
+      this.timeoutFuture = null;
+    }
+    if (timeout != null) {
+      timeout.cancel(false);
     }
     if (inFlight != null) {
       inFlight.cancel(mayInterruptIfRunning);
@@ -190,12 +242,23 @@ void cancel(boolean mayInterruptIfRunning) {
   }
 
   private void finish(@Nullable ResponseT response, @Nullable Throwable error) {
+    ScheduledFuture<?> timeout;
+    ApiFuture<?> inFlight;
     synchronized (lock) {
       if (done) {
         return;
       }
       done = true;
+      inFlight = inFlightFuture;
       inFlightFuture = null;
+      timeout = timeoutFuture;
+      timeoutFuture = null;
+    }
+    if (timeout != null) {
+      timeout.cancel(false);
+    }
+    if (inFlight != null) {
+      inFlight.cancel(true);
     }
     IOException closeError = closePayload();
     if (error == null) {