Skip to content

Commit bf605a2

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 d5c48d6 commit bf605a2

7 files changed

Lines changed: 438 additions & 46 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 & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -65,16 +65,13 @@
6565
@NullMarked
6666
final class ResumableUploadChunkCoordinator<ResponseT> {
6767

68+
// Per GAX-R7: local and per-attempt deadlines are derived from the global timeout at chunk
69+
// dispatch time; only backoff delay parameters are static.
6870
static final RetrySettings DEFAULT_CHUNK_RETRY_SETTINGS =
6971
RetrySettings.newBuilder()
7072
.setInitialRetryDelayDuration(Duration.ofMillis(100))
7173
.setRetryDelayMultiplier(1.3)
7274
.setMaxRetryDelayDuration(Duration.ofMinutes(1))
73-
.setInitialRpcTimeoutDuration(Duration.ofSeconds(30))
74-
.setRpcTimeoutMultiplier(1.0)
75-
.setMaxRpcTimeoutDuration(Duration.ofSeconds(30))
76-
.setTotalTimeoutDuration(Duration.ofMinutes(5))
77-
.setMaxAttempts(5)
7875
.build();
7976

8077
private final AtomicBoolean dispatching = new AtomicBoolean(false);
@@ -86,6 +83,8 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
8683
queryStatusCallable;
8784
private final String uploadUrl;
8885
private final RewindableStreamBuffer buffer;
86+
private final ResumableUploadCallSettings settings;
87+
private final long deadlineNanos;
8988
private final ApiCallContext callContext;
9089
private final ClientContext clientContext;
9190
private final SettableApiFuture<ResponseT> result = SettableApiFuture.create();
@@ -96,7 +95,8 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
9695
UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
9796
String uploadUrl,
9897
InputStream payload,
99-
int chunkSize,
98+
ResumableUploadCallSettings settings,
99+
long deadlineNanos,
100100
ApiCallContext callContext,
101101
ClientContext clientContext) {
102102
this.uploadChunkCallable =
@@ -105,9 +105,11 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
105105
checkNotNull(queryStatusCallable, "queryStatusCallable must not be null");
106106
this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
107107
checkNotNull(payload, "payload must not be null");
108+
this.settings = checkNotNull(settings, "settings must not be null");
109+
this.deadlineNanos = deadlineNanos;
108110
this.callContext = checkNotNull(callContext, "callContext must not be null");
109111
this.clientContext = checkNotNull(clientContext, "clientContext must not be null");
110-
this.buffer = new RewindableStreamBuffer(payload, chunkSize, uploadUrl);
112+
this.buffer = new RewindableStreamBuffer(payload, settings.getChunkSize(), uploadUrl);
111113
}
112114

113115
ApiFuture<ResponseT> start() {
@@ -175,15 +177,31 @@ private void transmitSingleChunk(long currentOffset) {
175177
.build();
176178

177179
// Dispatch the chunk upload call and register the in-flight future for cancellation.
180+
// Per GAX-R7: data-plane chunk commands use half of the original global timeout as both
181+
// the local and per-attempt deadline, trimmed to the remaining global deadline.
182+
long remainingNanos = Math.max(1L, deadlineNanos - clientContext.getClock().nanoTime());
183+
long halfGlobalNanos = settings.getGlobalTimeout().dividedBy(2).toNanos();
184+
Duration chunkDeadline = Duration.ofNanos(Math.min(halfGlobalNanos, remainingNanos));
185+
186+
RetrySettings derivedChunkRetrySettings =
187+
DEFAULT_CHUNK_RETRY_SETTINGS.toBuilder()
188+
.setTotalTimeoutDuration(chunkDeadline)
189+
.setInitialRpcTimeoutDuration(chunkDeadline)
190+
.setRpcTimeoutMultiplier(1.0)
191+
.setMaxRpcTimeoutDuration(chunkDeadline)
192+
.build();
193+
ApiCallContext chunkCallContext = callContext.withRetrySettings(derivedChunkRetrySettings);
178194
ChunkAttemptCallable<ResponseT> attemptCallable =
179195
new ChunkAttemptCallable<>(
180196
uploadChunkCallable,
181197
queryStatusCallable,
182198
buffer,
183199
uploadUrl,
184200
chunkRequest,
185-
callContext,
186-
command);
201+
chunkCallContext,
202+
command,
203+
deadlineNanos,
204+
clientContext.getClock());
187205

188206
RetryAlgorithm<ChunkUploadResponse<ResponseT>> retryAlgorithm =
189207
new RetryAlgorithm<>(
@@ -193,7 +211,7 @@ private void transmitSingleChunk(long currentOffset) {
193211
new ScheduledRetryingExecutor<>(retryAlgorithm, clientContext.getExecutor());
194212

195213
RetryingFuture<ChunkUploadResponse<ResponseT>> retryingFuture =
196-
retryingExecutor.createFuture(attemptCallable, callContext);
214+
retryingExecutor.createFuture(attemptCallable, chunkCallContext);
197215
attemptCallable.setRetryingFuture(retryingFuture);
198216
this.currentChunkFuture = retryingFuture;
199217

0 commit comments

Comments
 (0)