Skip to content

Commit 6960f29

Browse files
whoweswhowes
committed
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.
1 parent 1fe75dc commit 6960f29

5 files changed

Lines changed: 426 additions & 48 deletions

File tree

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

Lines changed: 57 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,14 +31,17 @@
3131

3232
import static com.google.common.base.Preconditions.checkNotNull;
3333

34+
import com.google.api.core.ApiClock;
3435
import com.google.api.core.ApiFuture;
3536
import com.google.api.core.ApiFutureCallback;
3637
import com.google.api.core.ApiFutures;
38+
import com.google.api.core.NanoClock;
3739
import com.google.api.core.SettableApiFuture;
3840
import com.google.api.gax.resumable.ChunkUploadRequest;
3941
import com.google.api.gax.resumable.ChunkUploadResponse;
4042
import com.google.api.gax.resumable.QueryStatusRequest;
4143
import com.google.api.gax.resumable.QueryStatusResponse;
44+
import com.google.api.gax.retrying.RetrySettings;
4245
import com.google.api.gax.retrying.RetryingFuture;
4346
import com.google.common.util.concurrent.MoreExecutors;
4447
import java.time.Duration;
@@ -68,6 +71,8 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
6871
private final RewindableStreamBuffer buffer;
6972
private final String uploadUrl;
7073
private final ApiCallContext originalCallContext;
74+
private final long deadlineNanos;
75+
private final ApiClock clock;
7176

7277
private volatile ChunkUploadRequest currentRequest;
7378
private volatile ResumableUploadCommand currentCommand;
@@ -85,6 +90,28 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
8590
ChunkUploadRequest request,
8691
ApiCallContext callContext,
8792
ResumableUploadCommand command) {
93+
this(
94+
uploadChunkCallable,
95+
queryStatusCallable,
96+
buffer,
97+
uploadUrl,
98+
request,
99+
callContext,
100+
command,
101+
Long.MAX_VALUE,
102+
NanoClock.getDefaultClock());
103+
}
104+
105+
ChunkAttemptCallable(
106+
UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> uploadChunkCallable,
107+
UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
108+
RewindableStreamBuffer buffer,
109+
String uploadUrl,
110+
ChunkUploadRequest request,
111+
ApiCallContext callContext,
112+
ResumableUploadCommand command,
113+
long deadlineNanos,
114+
ApiClock clock) {
88115
this.uploadChunkCallable =
89116
checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
90117
this.queryStatusCallable =
@@ -94,6 +121,8 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
94121
this.currentRequest = checkNotNull(request, "request must not be null");
95122
this.originalCallContext = checkNotNull(callContext, "callContext must not be null");
96123
this.currentCommand = checkNotNull(command, "command must not be null");
124+
this.deadlineNanos = deadlineNanos;
125+
this.clock = checkNotNull(clock, "clock must not be null");
97126
}
98127

99128
void setRetryingFuture(RetryingFuture<ChunkUploadResponse<ResponseT>> retryingFuture) {
@@ -129,9 +158,36 @@ private void prepareAttempt(
129158
SettableApiFuture<ChunkUploadResponse<ResponseT>> attemptFuture,
130159
ApiCallContext attemptContext,
131160
RetryingFuture<ChunkUploadResponse<ResponseT>> currentRetryingFuture) {
161+
// Per GAX-R7: query uses sensible unary defaults trimmed to the remaining global deadline.
162+
long remainingNanos =
163+
deadlineNanos == Long.MAX_VALUE
164+
? Long.MAX_VALUE
165+
: Math.max(1L, deadlineNanos - clock.nanoTime());
166+
Duration queryTotal =
167+
Duration.ofNanos(
168+
Math.min(
169+
ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS
170+
.getTotalTimeoutDuration()
171+
.toNanos(),
172+
remainingNanos));
173+
Duration queryRpc =
174+
Duration.ofNanos(
175+
Math.min(
176+
ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS
177+
.getInitialRpcTimeoutDuration()
178+
.toNanos(),
179+
queryTotal.toNanos()));
180+
RetrySettings trimmedQuerySettings =
181+
ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS.toBuilder()
182+
.setTotalTimeoutDuration(queryTotal)
183+
.setInitialRpcTimeoutDuration(queryRpc)
184+
.setMaxRpcTimeoutDuration(queryRpc)
185+
.build();
186+
ApiCallContext queryContext = originalCallContext.withRetrySettings(trimmedQuerySettings);
187+
132188
QueryStatusRequest queryRequest = QueryStatusRequest.create(uploadUrl);
133189
ApiFuture<QueryStatusResponse<ResponseT>> queryFuture =
134-
queryStatusCallable.futureCall(queryRequest, attemptContext);
190+
queryStatusCallable.futureCall(queryRequest, queryContext);
135191
if (queryFuture == null) {
136192
failAttempt(
137193
attemptFuture, new IllegalStateException("queryStatusCallable returned a null future"));

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

Lines changed: 51 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -45,19 +45,31 @@
4545
@NullMarked
4646
public abstract class ResumableUploadCallSettings {
4747
private static final int DEFAULT_CHUNK_SIZE = 8 * 1024 * 1024; // 8 MB
48+
// Matches Ruby google-apis-core RequestOptions.default.max_elapsed_time = 900s (CL-R9).
49+
private static final Duration DEFAULT_GLOBAL_TIMEOUT = Duration.ofMinutes(15);
50+
51+
abstract @Nullable Integer chunkSizeOption();
52+
53+
abstract @Nullable Duration globalTimeoutOption();
4854

4955
/** Returns the configured chunk size in bytes (defaults to 8 MB / 8,388,608 bytes). */
50-
public abstract int getChunkSize();
56+
public int getChunkSize() {
57+
Integer size = chunkSizeOption();
58+
return size != null ? size : DEFAULT_CHUNK_SIZE;
59+
}
5160

5261
/**
53-
* Returns the global upload timeout governing the entire upload duration, or {@code null} if
54-
* disabled.
62+
* Returns the global upload timeout governing the entire upload duration (defaults to 15
63+
* minutes).
5564
*/
56-
public abstract @Nullable Duration getGlobalTimeout();
65+
public Duration getGlobalTimeout() {
66+
Duration timeout = globalTimeoutOption();
67+
return timeout != null ? timeout : DEFAULT_GLOBAL_TIMEOUT;
68+
}
5769

5870
/**
59-
* Merges another {@code ResumableUploadCallSettings} instance with this one. Fields set in {@code
60-
* other} override fields in this instance.
71+
* Merges another {@code ResumableUploadCallSettings} instance with this one. Fields explicitly
72+
* set in {@code other} override fields in this instance.
6173
*
6274
* @param other settings to overlay; may be {@code null}
6375
* @return a new, resolved {@code ResumableUploadCallSettings} instance
@@ -67,40 +79,60 @@ public ResumableUploadCallSettings merge(@Nullable ResumableUploadCallSettings o
6779
return this;
6880
}
6981
Builder builder = toBuilder();
70-
if (other.getChunkSize() > 0) {
71-
builder.setChunkSize(other.getChunkSize());
82+
if (other.chunkSizeOption() != null) {
83+
builder.setChunkSize(other.chunkSizeOption());
7284
}
73-
if (other.getGlobalTimeout() != null) {
74-
builder.setGlobalTimeout(other.getGlobalTimeout());
85+
if (other.globalTimeoutOption() != null) {
86+
builder.setGlobalTimeout(other.globalTimeoutOption());
7587
}
7688
return builder.build();
7789
}
7890

7991
public abstract Builder toBuilder();
8092

8193
public static Builder newBuilder() {
82-
return new AutoValue_ResumableUploadCallSettings.Builder().setChunkSize(DEFAULT_CHUNK_SIZE);
94+
return new AutoValue_ResumableUploadCallSettings.Builder();
8395
}
8496

8597
/** Builder for {@link ResumableUploadCallSettings}. */
8698
@AutoValue.Builder
8799
public abstract static class Builder {
88-
public abstract Builder setChunkSize(int chunkSize);
100+
abstract Builder setChunkSizeOption(@Nullable Integer chunkSize);
89101

90-
public abstract int getChunkSize();
102+
abstract @Nullable Integer chunkSizeOption();
91103

92-
public abstract Builder setGlobalTimeout(@Nullable Duration globalTimeout);
104+
public Builder setChunkSize(int chunkSize) {
105+
return setChunkSizeOption(chunkSize);
106+
}
107+
108+
public int getChunkSize() {
109+
Integer size = chunkSizeOption();
110+
return size != null ? size : DEFAULT_CHUNK_SIZE;
111+
}
93112

94-
public abstract @Nullable Duration getGlobalTimeout();
113+
abstract Builder setGlobalTimeoutOption(@Nullable Duration globalTimeout);
114+
115+
abstract @Nullable Duration globalTimeoutOption();
116+
117+
public Builder setGlobalTimeout(@Nullable Duration globalTimeout) {
118+
return setGlobalTimeoutOption(globalTimeout);
119+
}
120+
121+
public @Nullable Duration getGlobalTimeout() {
122+
return globalTimeoutOption();
123+
}
95124

96125
abstract ResumableUploadCallSettings autoBuild();
97126

98127
public ResumableUploadCallSettings build() {
99-
Preconditions.checkArgument(getChunkSize() > 0, "chunkSize must be > 0");
100-
if (getGlobalTimeout() != null) {
128+
Integer size = chunkSizeOption();
129+
if (size != null) {
130+
Preconditions.checkArgument(size > 0, "chunkSize must be > 0");
131+
}
132+
Duration timeout = globalTimeoutOption();
133+
if (timeout != null) {
101134
Preconditions.checkArgument(
102-
!getGlobalTimeout().isNegative() && !getGlobalTimeout().isZero(),
103-
"globalTimeout must be positive");
135+
!timeout.isNegative() && !timeout.isZero(), "globalTimeout must be positive");
104136
}
105137
return autoBuild();
106138
}

0 commit comments

Comments
 (0)