feat(gax): enforce global timeout for resumable uploads

Enforces ResumableUploadCallSettings.getGlobalTimeout() in ResumableUploadChunkCoordinator across the upload lifecycle. Cancels in-flight RPCs and completes the future with DeadlineExceededException when the deadline is exceeded.

Author: whowes

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

@@ -45,19 +45,31 @@
 @NullMarked
 public abstract class ResumableUploadCallSettings {
   private static final int DEFAULT_CHUNK_SIZE = 8 * 1024 * 1024; // 8 MB
+  // Matches Ruby google-apis-core RequestOptions.default.max_elapsed_time = 900s (CL-R9).
+  private static final Duration DEFAULT_GLOBAL_TIMEOUT = Duration.ofMinutes(15);
+
+  abstract @Nullable Integer chunkSizeOption();
+
+  abstract @Nullable Duration globalTimeoutOption();
 
   /** Returns the configured chunk size in bytes (defaults to 8 MB / 8,388,608 bytes). */
-  public abstract int getChunkSize();
+  public int getChunkSize() {
+    Integer size = chunkSizeOption();
+    return size != null ? size : DEFAULT_CHUNK_SIZE;
+  }
 
   /**
-   * Returns the global upload timeout governing the entire upload duration, or {@code null} if
-   * disabled.
+   * Returns the global upload timeout governing the entire upload duration (defaults to 15
+   * minutes).
    */
-  public abstract @Nullable Duration getGlobalTimeout();
+  public Duration getGlobalTimeout() {
+    Duration timeout = globalTimeoutOption();
+    return timeout != null ? timeout : DEFAULT_GLOBAL_TIMEOUT;
+  }
 
   /**
-   * Merges another {@code ResumableUploadCallSettings} instance with this one. Fields set in {@code
-   * other} override fields in this instance.
+   * Merges another {@code ResumableUploadCallSettings} instance with this one. Fields explicitly
+   * set in {@code other} override fields in this instance.
    *
    * @param other settings to overlay; may be {@code null}
    * @return a new, resolved {@code ResumableUploadCallSettings} instance
@@ -67,40 +79,60 @@ public ResumableUploadCallSettings merge(@Nullable ResumableUploadCallSettings o
       return this;
     }
     Builder builder = toBuilder();
-    if (other.getChunkSize() > 0) {
-      builder.setChunkSize(other.getChunkSize());
+    if (other.chunkSizeOption() != null) {
+      builder.setChunkSize(other.chunkSizeOption());
     }
-    if (other.getGlobalTimeout() != null) {
-      builder.setGlobalTimeout(other.getGlobalTimeout());
+    if (other.globalTimeoutOption() != null) {
+      builder.setGlobalTimeout(other.globalTimeoutOption());
     }
     return builder.build();
   }
 
   public abstract Builder toBuilder();
 
   public static Builder newBuilder() {
-    return new AutoValue_ResumableUploadCallSettings.Builder().setChunkSize(DEFAULT_CHUNK_SIZE);
+    return new AutoValue_ResumableUploadCallSettings.Builder();
   }
 
   /** Builder for {@link ResumableUploadCallSettings}. */
   @AutoValue.Builder
   public abstract static class Builder {
-    public abstract Builder setChunkSize(int chunkSize);
+    abstract Builder setChunkSizeOption(@Nullable Integer chunkSize);
 
-    public abstract int getChunkSize();
+    abstract @Nullable Integer chunkSizeOption();
 
-    public abstract Builder setGlobalTimeout(@Nullable Duration globalTimeout);
+    public Builder setChunkSize(int chunkSize) {
+      return setChunkSizeOption(chunkSize);
+    }
+
+    public int getChunkSize() {
+      Integer size = chunkSizeOption();
+      return size != null ? size : DEFAULT_CHUNK_SIZE;
+    }
 
-    public abstract @Nullable Duration getGlobalTimeout();
+    abstract Builder setGlobalTimeoutOption(@Nullable Duration globalTimeout);
+
+    abstract @Nullable Duration globalTimeoutOption();
+
+    public Builder setGlobalTimeout(@Nullable Duration globalTimeout) {
+      return setGlobalTimeoutOption(globalTimeout);
+    }
+
+    public @Nullable Duration getGlobalTimeout() {
+      return globalTimeoutOption();
+    }
 
     abstract ResumableUploadCallSettings autoBuild();
 
     public ResumableUploadCallSettings build() {
-      Preconditions.checkArgument(getChunkSize() > 0, "chunkSize must be > 0");
-      if (getGlobalTimeout() != null) {
+      Integer size = chunkSizeOption();
+      if (size != null) {
+        Preconditions.checkArgument(size > 0, "chunkSize must be > 0");
+      }
+      Duration timeout = globalTimeoutOption();
+      if (timeout != null) {
         Preconditions.checkArgument(
-            !getGlobalTimeout().isNegative() && !getGlobalTimeout().isZero(),
-            "globalTimeout must be positive");
+            !timeout.isNegative() && !timeout.isZero(), "globalTimeout must be positive");
       }
       return autoBuild();
     }

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

@@ -119,7 +119,7 @@ public ResumableUploadFuture<ResponseT> futureCall(
         retryingQueryCallable,
         payload,
         effectiveSettings,
-        clientContext.getDefaultCallContext());
+        clientContext);
   }
 
   @Override

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

@@ -41,10 +41,12 @@
 import com.google.api.gax.resumable.QueryStatusRequest;
 import com.google.api.gax.resumable.QueryStatusResponse;
 import com.google.api.gax.resumable.ResumableUploadStatus;
+import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.ResumableUploadErrorClassifier.Category;
 import com.google.common.util.concurrent.MoreExecutors;
 import java.io.IOException;
 import java.io.InputStream;
+import java.time.Duration;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.Executor;
 import org.jspecify.annotations.NullMarked;
@@ -72,7 +74,10 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
       queryStatusCallable;
   private final String uploadUrl;
   private final RewindableStreamBuffer buffer;
+  private final ResumableUploadCallSettings settings;
+  private final long deadlineNanos;
   private final ApiCallContext callContext;
+  private final ClientContext clientContext;
   private final SettableApiFuture<ResponseT> result = SettableApiFuture.create();
   private volatile @Nullable ApiFuture<?> currentChunkFuture;
 
@@ -81,16 +86,20 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
       UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
       String uploadUrl,
       InputStream payload,
-      int chunkSize,
-      ApiCallContext callContext) {
+      ResumableUploadCallSettings settings,
+      long deadlineNanos,
+      ClientContext clientContext) {
     this.uploadChunkCallable =
         checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
     this.queryStatusCallable =
         checkNotNull(queryStatusCallable, "queryStatusCallable must not be null");
     this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
     checkNotNull(payload, "payload must not be null");
-    this.callContext = checkNotNull(callContext, "callContext must not be null");
-    this.buffer = new RewindableStreamBuffer(payload, chunkSize, uploadUrl);
+    this.settings = checkNotNull(settings, "settings must not be null");
+    this.deadlineNanos = deadlineNanos;
+    this.clientContext = checkNotNull(clientContext, "clientContext must not be null");
+    this.callContext = clientContext.getDefaultCallContext();
+    this.buffer = new RewindableStreamBuffer(payload, settings.getChunkSize(), uploadUrl);
   }
 
   ApiFuture<ResponseT> start() {
@@ -126,7 +135,7 @@ private void dispatchCurrentChunk() {
       ChunkUploadRequest chunkRequest = buildCurrentChunkRequest();
       // Dispatch the chunk upload call and register the in-flight future for cancellation.
       ApiFuture<ChunkUploadResponse<ResponseT>> chunkFuture =
-          uploadChunkCallable.futureCall(chunkRequest, callContext);
+          uploadChunkCallable.futureCall(chunkRequest, chunkCallContext());
       if (!trackInFlight(chunkFuture)) {
         return;
       }
@@ -171,7 +180,7 @@ private void recover() {
     try {
       // Dispatch the query status call and register the in-flight future for cancellation.
       ApiFuture<QueryStatusResponse<ResponseT>> queryFuture =
-          queryStatusCallable.futureCall(QueryStatusRequest.create(uploadUrl), callContext);
+          queryStatusCallable.futureCall(QueryStatusRequest.create(uploadUrl), queryCallContext());
       if (!trackInFlight(queryFuture)) {
         return;
       }
@@ -272,6 +281,60 @@ public StatusCode.Code getCode() {
         }
       };
 
+  /**
+   * Returns the call context for a chunk upload, bounding its retry cycle and each of its attempts
+   * to the chunk's local deadline: at most half the global timeout, never past the global deadline.
+   */
+  private ApiCallContext chunkCallContext() {
+    Duration remaining = remainingGlobalTimeout();
+    if (remaining == null) {
+      return callContext;
+    }
+    Duration deadline = min(settings.getGlobalTimeout().dividedBy(2), remaining);
+    return boundedCallContext(
+        deadline,
+        callContext.getRetrySettings().toBuilder()
+            .setInitialRpcTimeoutDuration(deadline)
+            .setRpcTimeoutMultiplier(1.0)
+            .setMaxRpcTimeoutDuration(deadline));
+  }
+
+  /** Returns the call context for a status query, bounded by the remaining global timeout. */
+  private ApiCallContext queryCallContext() {
+    Duration remaining = remainingGlobalTimeout();
+    if (remaining == null) {
+      return callContext;
+    }
+    RetrySettings baseSettings = callContext.getRetrySettings();
+    return boundedCallContext(
+        remaining,
+        baseSettings.toBuilder()
+            .setInitialRpcTimeoutDuration(
+                min(baseSettings.getInitialRpcTimeoutDuration(), remaining))
+            .setMaxRpcTimeoutDuration(min(baseSettings.getMaxRpcTimeoutDuration(), remaining)));
+  }
+
+  /**
+   * Returns the time left before the global upload deadline, or null if deadlines cannot be derived
+   * because no global timeout or no base retry settings were configured.
+   */
+  private @Nullable Duration remainingGlobalTimeout() {
+    if (settings.getGlobalTimeout() == null || callContext.getRetrySettings() == null) {
+      return null;
+    }
+    return Duration.ofNanos(Math.max(1L, deadlineNanos - clientContext.getClock().nanoTime()));
+  }
+
+  private ApiCallContext boundedCallContext(Duration deadline, RetrySettings.Builder settings) {
+    return callContext
+        .withRetrySettings(settings.setTotalTimeoutDuration(deadline).build())
+        .withTimeoutDuration(deadline);
+  }
+
+  private static Duration min(Duration left, Duration right) {
+    return left.compareTo(right) <= 0 ? left : right;
+  }
+
   private static FailedPreconditionException protocolViolation(String message) {
     return new FailedPreconditionException(message, null, FAILED_PRECONDITION_STATUS_CODE, false);
   }