feat(gax): enforce global timeout for resumable uploads

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

@@ -31,14 +31,17 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.google.api.core.ApiClock;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
+import com.google.api.core.NanoClock;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.resumable.ChunkUploadRequest;
 import com.google.api.gax.resumable.ChunkUploadResponse;
 import com.google.api.gax.resumable.QueryStatusRequest;
 import com.google.api.gax.resumable.QueryStatusResponse;
+import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.retrying.RetryingFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import java.time.Duration;
@@ -68,6 +71,8 @@
   private final RewindableStreamBuffer buffer;
   private final String uploadUrl;
   private final ApiCallContext originalCallContext;
+  private final long deadlineNanos;
+  private final ApiClock clock;
 
   private volatile ChunkUploadRequest currentRequest;
   private volatile ResumableUploadCommand currentCommand;
@@ -85,6 +90,28 @@
       ChunkUploadRequest request,
       ApiCallContext callContext,
       ResumableUploadCommand command) {
+    this(
+        uploadChunkCallable,
+        queryStatusCallable,
+        buffer,
+        uploadUrl,
+        request,
+        callContext,
+        command,
+        Long.MAX_VALUE,
+        NanoClock.getDefaultClock());
+  }
+
+  ChunkAttemptCallable(
+      UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> uploadChunkCallable,
+      UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
+      RewindableStreamBuffer buffer,
+      String uploadUrl,
+      ChunkUploadRequest request,
+      ApiCallContext callContext,
+      ResumableUploadCommand command,
+      long deadlineNanos,
+      ApiClock clock) {
     this.uploadChunkCallable =
         checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
     this.queryStatusCallable =
@@ -94,6 +121,8 @@
     this.currentRequest = checkNotNull(request, "request must not be null");
     this.originalCallContext = checkNotNull(callContext, "callContext must not be null");
     this.currentCommand = checkNotNull(command, "command must not be null");
+    this.deadlineNanos = deadlineNanos;
+    this.clock = checkNotNull(clock, "clock must not be null");
   }
 
   void setRetryingFuture(RetryingFuture<ChunkUploadResponse<ResponseT>> retryingFuture) {
@@ -129,10 +158,37 @@
       SettableApiFuture<ChunkUploadResponse<ResponseT>> attemptFuture,
       ApiCallContext attemptContext,
       RetryingFuture<ChunkUploadResponse<ResponseT>> currentRetryingFuture) {
+    // Per GAX-R7: query uses sensible unary defaults trimmed to the remaining global deadline.
+    long remainingNanos =
+        deadlineNanos == Long.MAX_VALUE
+            ? Long.MAX_VALUE
+            : Math.max(1L, deadlineNanos - clock.nanoTime());
+    Duration queryTotal =
+        Duration.ofNanos(
+            Math.min(
+                ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS
+                    .getTotalTimeoutDuration()
+                    .toNanos(),
+                remainingNanos));
+    Duration queryRpc =
+        Duration.ofNanos(
+            Math.min(
+                ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS
+                    .getInitialRpcTimeoutDuration()
+                    .toNanos(),
+                queryTotal.toNanos()));
+    RetrySettings trimmedQuerySettings =
+        ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS.toBuilder()
+            .setTotalTimeoutDuration(queryTotal)
+            .setInitialRpcTimeoutDuration(queryRpc)
+            .setMaxRpcTimeoutDuration(queryRpc)
+            .build();
+    ApiCallContext queryContext = originalCallContext.withRetrySettings(trimmedQuerySettings);
+
     QueryStatusRequest queryRequest = QueryStatusRequest.create(uploadUrl);
     ApiFuture<QueryStatusResponse<ResponseT>> queryFuture =
-        queryStatusCallable.futureCall(queryRequest, attemptContext);
+        queryStatusCallable.futureCall(queryRequest, queryContext);
     if (queryFuture == null) {
       failAttempt(
           attemptFuture, new IllegalStateException("queryStatusCallable returned a null future"));
       return;

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 @@
       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();
     }