Skip to content

Commit c36f0cd

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 292921f commit c36f0cd

6 files changed

Lines changed: 431 additions & 36 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,15 +31,18 @@
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;
4244
import com.google.api.gax.resumable.ResumableUploadStatus;
45+
import com.google.api.gax.retrying.RetrySettings;
4346
import com.google.api.gax.retrying.RetryingFuture;
4447
import com.google.common.util.concurrent.MoreExecutors;
4548
import java.time.Duration;
@@ -69,6 +72,8 @@ class ChunkAttemptCallable<ResponseT> implements Callable<ChunkUploadResponse<Re
6972
private final RewindableStreamBuffer buffer;
7073
private final String uploadUrl;
7174
private final ApiCallContext originalCallContext;
75+
private final long deadlineNanos;
76+
private final ApiClock clock;
7277

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

100129
void setRetryingFuture(RetryingFuture<ChunkUploadResponse<ResponseT>> retryingFuture) {
@@ -135,9 +164,36 @@ private void prepareAttempt(
135164
ApiCallContext attemptContext,
136165
RetryingFuture<ChunkUploadResponse<ResponseT>> currentRetryingFuture) {
137166
this.currentCommand = ResumableUploadCommand.QUERY;
167+
// Per GAX-R7: query uses sensible unary defaults trimmed to the remaining global deadline.
168+
long remainingNanos =
169+
deadlineNanos == Long.MAX_VALUE
170+
? Long.MAX_VALUE
171+
: Math.max(1L, deadlineNanos - clock.nanoTime());
172+
Duration queryTotal =
173+
Duration.ofNanos(
174+
Math.min(
175+
ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS
176+
.getTotalTimeoutDuration()
177+
.toNanos(),
178+
remainingNanos));
179+
Duration queryRpc =
180+
Duration.ofNanos(
181+
Math.min(
182+
ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS
183+
.getInitialRpcTimeoutDuration()
184+
.toNanos(),
185+
queryTotal.toNanos()));
186+
RetrySettings trimmedQuerySettings =
187+
ResumableUploadCallableImpl.DEFAULT_QUERY_RETRY_SETTINGS.toBuilder()
188+
.setTotalTimeoutDuration(queryTotal)
189+
.setInitialRpcTimeoutDuration(queryRpc)
190+
.setMaxRpcTimeoutDuration(queryRpc)
191+
.build();
192+
ApiCallContext queryContext = originalCallContext.withRetrySettings(trimmedQuerySettings);
193+
138194
QueryStatusRequest queryRequest = QueryStatusRequest.create(uploadUrl);
139195
ApiFuture<QueryStatusResponse<ResponseT>> queryFuture =
140-
queryStatusCallable.futureCall(queryRequest, attemptContext);
196+
queryStatusCallable.futureCall(queryRequest, queryContext);
141197
if (queryFuture == null) {
142198
failAttempt(
143199
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
}

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

Lines changed: 28 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,8 @@
6464
@NullMarked
6565
final class ResumableUploadChunkCoordinator<ResponseT> {
6666

67+
// Per GAX-R7: local and per-attempt deadlines are derived from the global timeout at chunk
68+
// dispatch time; only backoff delay parameters are static.
6769
private static final RetrySettings CHUNK_RETRY_SETTINGS =
6870
RetrySettings.newBuilder()
6971
.setInitialRetryDelayDuration(Duration.ofMillis(100))
@@ -81,6 +83,8 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
8183
queryStatusCallable;
8284
private final String uploadUrl;
8385
private final RewindableStreamBuffer buffer;
86+
private final ResumableUploadCallSettings settings;
87+
private final long deadlineNanos;
8488
private final ApiCallContext callContext;
8589
private final ClientContext clientContext;
8690
private final SettableApiFuture<ResponseT> result = SettableApiFuture.create();
@@ -91,17 +95,20 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
9195
UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
9296
String uploadUrl,
9397
InputStream payload,
94-
int chunkSize,
98+
ResumableUploadCallSettings settings,
99+
long deadlineNanos,
95100
ClientContext clientContext) {
96101
this.uploadChunkCallable =
97102
checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
98103
this.queryStatusCallable =
99104
checkNotNull(queryStatusCallable, "queryStatusCallable must not be null");
100105
this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
101106
checkNotNull(payload, "payload must not be null");
107+
this.settings = checkNotNull(settings, "settings must not be null");
108+
this.deadlineNanos = deadlineNanos;
102109
this.clientContext = checkNotNull(clientContext, "clientContext must not be null");
103110
this.callContext = clientContext.getDefaultCallContext();
104-
this.buffer = new RewindableStreamBuffer(payload, chunkSize, uploadUrl);
111+
this.buffer = new RewindableStreamBuffer(payload, settings.getChunkSize(), uploadUrl);
105112
}
106113

107114
ApiFuture<ResponseT> start() {
@@ -152,15 +159,31 @@ private void transmitChunk(long currentOffset) {
152159
.build();
153160

154161
// Dispatch the chunk upload call and register the in-flight future for cancellation.
162+
// Per GAX-R7: data-plane chunk commands use half of the original global timeout as both
163+
// the local and per-attempt deadline, trimmed to the remaining global deadline.
164+
long remainingNanos = Math.max(1L, deadlineNanos - clientContext.getClock().nanoTime());
165+
long halfGlobalNanos = settings.getGlobalTimeout().dividedBy(2).toNanos();
166+
Duration chunkDeadline = Duration.ofNanos(Math.min(halfGlobalNanos, remainingNanos));
167+
168+
RetrySettings derivedChunkRetrySettings =
169+
CHUNK_RETRY_SETTINGS.toBuilder()
170+
.setTotalTimeoutDuration(chunkDeadline)
171+
.setInitialRpcTimeoutDuration(chunkDeadline)
172+
.setRpcTimeoutMultiplier(1.0)
173+
.setMaxRpcTimeoutDuration(chunkDeadline)
174+
.build();
175+
ApiCallContext chunkCallContext = callContext.withRetrySettings(derivedChunkRetrySettings);
155176
ChunkAttemptCallable<ResponseT> attemptCallable =
156177
new ChunkAttemptCallable<>(
157178
uploadChunkCallable,
158179
queryStatusCallable,
159180
buffer,
160181
uploadUrl,
161182
chunkRequest,
162-
callContext,
163-
command);
183+
chunkCallContext,
184+
command,
185+
deadlineNanos,
186+
clientContext.getClock());
164187

165188
RetryAlgorithm<ChunkUploadResponse<ResponseT>> retryAlgorithm =
166189
new RetryAlgorithm<>(
@@ -170,7 +193,7 @@ private void transmitChunk(long currentOffset) {
170193
new ScheduledRetryingExecutor<>(retryAlgorithm, clientContext.getExecutor());
171194

172195
RetryingFuture<ChunkUploadResponse<ResponseT>> retryingFuture =
173-
retryingExecutor.createFuture(attemptCallable, callContext);
196+
retryingExecutor.createFuture(attemptCallable, chunkCallContext);
174197
attemptCallable.setRetryingFuture(retryingFuture);
175198
this.currentChunkFuture = retryingFuture;
176199

0 commit comments

Comments
 (0)