From c36f0cdf211541ba9bcef3249be379e29de7f243 Mon Sep 17 00:00:00 2001 From: whowes Date: Sat, 12 Sep 2026 16:54:21 +0000 Subject: [PATCH] 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. --- .../api/gax/rpc/ChunkAttemptCallable.java | 58 ++++- .../gax/rpc/ResumableUploadCallSettings.java | 70 ++++-- .../rpc/ResumableUploadChunkCoordinator.java | 33 ++- .../gax/rpc/ResumableUploadFutureImpl.java | 66 +++++- .../rpc/ResumableUploadCallSettingsTest.java | 28 ++- .../rpc/ResumableUploadCallableImplTest.java | 212 +++++++++++++++++- 6 files changed, 431 insertions(+), 36 deletions(-) diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ChunkAttemptCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ChunkAttemptCallable.java index 1960e9735ba0..0a6004029084 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ChunkAttemptCallable.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ChunkAttemptCallable.java @@ -31,15 +31,18 @@ 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.resumable.ResumableUploadStatus; +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; @@ -69,6 +72,8 @@ class ChunkAttemptCallable implements Callable implements Callable> uploadChunkCallable, + UnaryCallable> 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 = @@ -95,6 +122,8 @@ class ChunkAttemptCallable implements Callable> retryingFuture) { @@ -135,9 +164,36 @@ private void prepareAttempt( ApiCallContext attemptContext, RetryingFuture> currentRetryingFuture) { this.currentCommand = ResumableUploadCommand.QUERY; + // 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> queryFuture = - queryStatusCallable.futureCall(queryRequest, attemptContext); + queryStatusCallable.futureCall(queryRequest, queryContext); if (queryFuture == null) { failAttempt( attemptFuture, new IllegalStateException("queryStatusCallable returned a null future")); diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallSettings.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallSettings.java index 82e6838a52c7..4821a9e285fa 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallSettings.java +++ b/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,11 +79,11 @@ 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(); } @@ -79,28 +91,48 @@ public ResumableUploadCallSettings merge(@Nullable ResumableUploadCallSettings o 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(); } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java index ce539ffd183c..a21aa7cc0b09 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java @@ -64,6 +64,8 @@ @NullMarked final class ResumableUploadChunkCoordinator { + // Per GAX-R7: local and per-attempt deadlines are derived from the global timeout at chunk + // dispatch time; only backoff delay parameters are static. private static final RetrySettings CHUNK_RETRY_SETTINGS = RetrySettings.newBuilder() .setInitialRetryDelayDuration(Duration.ofMillis(100)) @@ -81,6 +83,8 @@ final class ResumableUploadChunkCoordinator { 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 result = SettableApiFuture.create(); @@ -91,7 +95,8 @@ final class ResumableUploadChunkCoordinator { UnaryCallable> queryStatusCallable, String uploadUrl, InputStream payload, - int chunkSize, + ResumableUploadCallSettings settings, + long deadlineNanos, ClientContext clientContext) { this.uploadChunkCallable = checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null"); @@ -99,9 +104,11 @@ final class ResumableUploadChunkCoordinator { checkNotNull(queryStatusCallable, "queryStatusCallable must not be null"); this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null"); checkNotNull(payload, "payload must not be null"); + 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, chunkSize, uploadUrl); + this.buffer = new RewindableStreamBuffer(payload, settings.getChunkSize(), uploadUrl); } ApiFuture start() { @@ -152,6 +159,20 @@ private void transmitChunk(long currentOffset) { .build(); // Dispatch the chunk upload call and register the in-flight future for cancellation. + // Per GAX-R7: data-plane chunk commands use half of the original global timeout as both + // the local and per-attempt deadline, trimmed to the remaining global deadline. + long remainingNanos = Math.max(1L, deadlineNanos - clientContext.getClock().nanoTime()); + long halfGlobalNanos = settings.getGlobalTimeout().dividedBy(2).toNanos(); + Duration chunkDeadline = Duration.ofNanos(Math.min(halfGlobalNanos, remainingNanos)); + + RetrySettings derivedChunkRetrySettings = + CHUNK_RETRY_SETTINGS.toBuilder() + .setTotalTimeoutDuration(chunkDeadline) + .setInitialRpcTimeoutDuration(chunkDeadline) + .setRpcTimeoutMultiplier(1.0) + .setMaxRpcTimeoutDuration(chunkDeadline) + .build(); + ApiCallContext chunkCallContext = callContext.withRetrySettings(derivedChunkRetrySettings); ChunkAttemptCallable attemptCallable = new ChunkAttemptCallable<>( uploadChunkCallable, @@ -159,8 +180,10 @@ private void transmitChunk(long currentOffset) { buffer, uploadUrl, chunkRequest, - callContext, - command); + chunkCallContext, + command, + deadlineNanos, + clientContext.getClock()); RetryAlgorithm> retryAlgorithm = new RetryAlgorithm<>( @@ -170,7 +193,7 @@ private void transmitChunk(long currentOffset) { new ScheduledRetryingExecutor<>(retryAlgorithm, clientContext.getExecutor()); RetryingFuture> retryingFuture = - retryingExecutor.createFuture(attemptCallable, callContext); + retryingExecutor.createFuture(attemptCallable, chunkCallContext); attemptCallable.setRetryingFuture(retryingFuture); this.currentChunkFuture = retryingFuture; diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java index d041f137225f..0fe2ceb7caaa 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java @@ -45,9 +45,12 @@ import com.google.errorprone.annotations.concurrent.GuardedBy; import java.io.IOException; import java.io.InputStream; +import java.time.Duration; import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.jspecify.annotations.NullMarked; @@ -72,13 +75,18 @@ final class ResumableUploadFutureImpl implements ResumableUploadFutur private final InputStream payload; private final ResumableUploadCallSettings settings; private final ClientContext clientContext; + private final ScheduledExecutorService executor; private final SettableApiFuture resultFuture = SettableApiFuture.create(); private volatile @Nullable String uploadSessionUrl; + private volatile long deadlineNanos; @GuardedBy("lock") private @Nullable ApiFuture inFlightFuture; + @GuardedBy("lock") + private @Nullable ScheduledFuture timeoutFuture; + static ResumableUploadFutureImpl create( ApiFuture startFuture, UnaryCallable> uploadChunkCallable, @@ -118,10 +126,17 @@ private ResumableUploadFutureImpl( this.settings = checkNotNull(settings, "settings must not be null"); checkArgument(settings.getChunkSize() > 0, "chunkSize must be > 0"); this.clientContext = checkNotNull(clientContext, "clientContext must not be null"); + this.executor = checkNotNull(clientContext.getExecutor(), "executor must not be null"); this.inFlightFuture = startFuture; } private void start() { + Duration timeout = settings.getGlobalTimeout(); + this.deadlineNanos = clientContext.getClock().nanoTime() + timeout.toNanos(); + synchronized (lock) { + this.timeoutFuture = + executor.schedule(this::onTimeout, timeout.toMillis(), TimeUnit.MILLISECONDS); + } ApiFutures.addCallback( startFuture, new ApiFutureCallback() { @@ -137,7 +152,8 @@ public void onSuccess(ResumableUploadSession session) { queryStatusCallable, uploadSessionUrl, payload, - settings.getChunkSize(), + settings, + deadlineNanos, clientContext); ApiFuture uploadFuture; try { @@ -182,17 +198,44 @@ public void onFailure(Throwable t) { MoreExecutors.directExecutor()); } + private void onTimeout() { + String message; + if (uploadSessionUrl != null) { + message = "Resumable upload timed out for session: " + uploadSessionUrl; + } else { + message = "Resumable upload timed out before session initiation completed"; + } + fail(new DeadlineExceededException(message, null, TIMEOUT_STATUS_CODE, false)); + } + private void succeed(@Nullable ResponseT result) { + ScheduledFuture timeout; synchronized (lock) { inFlightFuture = null; + timeout = this.timeoutFuture; + this.timeoutFuture = null; + } + if (timeout != null) { + timeout.cancel(false); } closePayload(); resultFuture.set(result); } private void fail(Throwable t) { + ScheduledFuture timeout; + ApiFuture inFlight; synchronized (lock) { - inFlightFuture = null; + inFlight = this.inFlightFuture; + this.inFlightFuture = null; + timeout = this.timeoutFuture; + this.timeoutFuture = null; + } + if (timeout != null) { + timeout.cancel(false); + } + if (inFlight != null) { + inFlight.cancel(true); } closePayload(); resultFuture.setException(t); @@ -220,10 +263,16 @@ public void addListener(Runnable listener, Executor executor) { public boolean cancel(boolean mayInterruptIfRunning) { boolean cancelled; ApiFuture inFlight; + ScheduledFuture timeout; synchronized (lock) { cancelled = resultFuture.cancel(mayInterruptIfRunning); inFlight = this.inFlightFuture; this.inFlightFuture = null; + timeout = this.timeoutFuture; + this.timeoutFuture = null; + } + if (timeout != null) { + timeout.cancel(false); } if (inFlight != null) { inFlight.cancel(mayInterruptIfRunning); @@ -252,4 +301,17 @@ public ResponseT get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return resultFuture.get(timeout, unit); } + + private static final StatusCode TIMEOUT_STATUS_CODE = + new StatusCode() { + @Override + public StatusCode.Code getCode() { + return StatusCode.Code.DEADLINE_EXCEEDED; + } + + @Override + public @Nullable Object getTransportCode() { + return null; + } + }; } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallSettingsTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallSettingsTest.java index 8bc80d5d7d0f..cf3a21a2d795 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallSettingsTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallSettingsTest.java @@ -38,16 +38,24 @@ public class ResumableUploadCallSettingsTest { + @Test + public void testDefaultSettings() { + ResumableUploadCallSettings settings = ResumableUploadCallSettings.newBuilder().build(); + + assertEquals(8 * 1024 * 1024, settings.getChunkSize()); + assertEquals(Duration.ofMinutes(15), settings.getGlobalTimeout()); + } + @Test public void testCustomSettingsAndToBuilder() { ResumableUploadCallSettings settings = ResumableUploadCallSettings.newBuilder() .setChunkSize(16 * 1024 * 1024) - .setGlobalTimeout(Duration.ofMinutes(15)) + .setGlobalTimeout(Duration.ofMinutes(20)) .build(); assertEquals(16 * 1024 * 1024, settings.getChunkSize()); - assertEquals(Duration.ofMinutes(15), settings.getGlobalTimeout()); + assertEquals(Duration.ofMinutes(20), settings.getGlobalTimeout()); assertEquals(settings, settings.toBuilder().build()); } @@ -103,7 +111,7 @@ public void testMerge_overridesChunkSizeAndGlobalTimeout() { } @Test - public void testMerge_nullGlobalTimeoutDoesNotOverride() { + public void testMerge_chunkSizeOnlyOverlayDoesNotOverrideGlobalTimeout() { ResumableUploadCallSettings stubSettings = ResumableUploadCallSettings.newBuilder().setGlobalTimeout(Duration.ofMinutes(10)).build(); @@ -115,4 +123,18 @@ public void testMerge_nullGlobalTimeoutDoesNotOverride() { assertEquals(32 * 1024 * 1024, merged.getChunkSize()); assertEquals(Duration.ofMinutes(10), merged.getGlobalTimeout()); } + + @Test + public void testMerge_timeoutOnlyOverlayDoesNotOverrideChunkSize() { + ResumableUploadCallSettings stubSettings = + ResumableUploadCallSettings.newBuilder().setChunkSize(32 * 1024 * 1024).build(); + + ResumableUploadCallSettings perRequestSettings = + ResumableUploadCallSettings.newBuilder().setGlobalTimeout(Duration.ofMinutes(30)).build(); + + ResumableUploadCallSettings merged = stubSettings.merge(perRequestSettings); + + assertEquals(32 * 1024 * 1024, merged.getChunkSize()); + assertEquals(Duration.ofMinutes(30), merged.getGlobalTimeout()); + } } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java index ce0f08dd4c42..1524ef999021 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java @@ -32,6 +32,9 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.atLeast; +import static org.mockito.Mockito.clearInvocations; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -54,6 +57,7 @@ import java.io.IOException; import java.io.InputStream; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.Arrays; import java.util.List; import java.util.concurrent.CancellationException; @@ -61,6 +65,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.AfterEach; @@ -81,6 +86,7 @@ class ResumableUploadCallableImplTest { private ResumableUploadCallSettings defaultSettings; private FakeCallContext callContext; private ScheduledExecutorService executor; + private ClientContext clientContext; private ResumableUploadCallableImpl callable; @BeforeEach @@ -98,7 +104,7 @@ void setUp() { defaultSettings = ResumableUploadCallSettings.newBuilder().setChunkSize(8).build(); callContext = FakeCallContext.createDefault(); executor = Executors.newScheduledThreadPool(2); - ClientContext clientContext = + clientContext = ClientContext.newBuilder().setDefaultCallContext(callContext).setExecutor(executor).build(); callable = new ResumableUploadCallableImpl<>(mockClient, defaultSettings, clientContext); } @@ -287,7 +293,7 @@ void testUploadCallable_closesPayloadOnSuccess() throws Exception { TrackableStream stream = new TrackableStream("data"); callable.futureCall("resource-path", stream, null).get(); - assertThat(stream.closed).isTrue(); + assertThat(stream.closeCount).isEqualTo(1); } @Test @@ -299,7 +305,7 @@ void testUploadCallable_closesPayloadOnFailure() { ResumableUploadFuture future = callable.futureCall("resource-path", stream, null); assertThrows(ExecutionException.class, future::get); - assertThat(stream.closed).isTrue(); + assertThat(stream.closeCount).isEqualTo(1); } @Test @@ -459,16 +465,18 @@ void testChunkRetry_transientFailureExhaustion_surfacesLastError() { .thenReturn( ApiFutures.immediateFailedFuture(createApiException(503, StatusCode.Code.UNAVAILABLE))); + // 300ms global timeout -> 150ms derived chunk local deadline (per GAX-R7) + ResumableUploadCallSettings settings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(300)).build(); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("hello"), null); + callable.futureCall("resource-path", streamOf("hello"), settings); ExecutionException exception = assertThrows(ExecutionException.class, future::get); assertThat(exception.getCause()).isInstanceOf(ApiException.class); assertThat(((ApiException) exception.getCause()).getStatusCode().getTransportCode()) .isEqualTo(503); - // Default chunk retry settings has maxAttempts = 5 - verify(mockChunkCallable, times(5)).futureCall(any(), any()); + verify(mockChunkCallable, atLeast(2)).futureCall(any(), any()); } @Test @@ -868,6 +876,179 @@ void testRecovery_transientErrorOnQuery_isRetried() throws Exception { verify(mockChunkCallable, times(2)).futureCall(any(), any()); } + @Test + void testGlobalTimeout_firesAndFailsSessionWithDeadlineExceeded() throws Exception { + stubStartSession("https://upload.url/timeout-fire"); + SettableApiFuture> hungChunk = SettableApiFuture.create(); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())).thenReturn(hungChunk); + + ResumableUploadCallSettings timeoutSettings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(100)).build(); + + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("hello"), timeoutSettings); + + ExecutionException exception = assertThrows(ExecutionException.class, future::get); + assertThat(exception.getCause()).isInstanceOf(DeadlineExceededException.class); + DeadlineExceededException cause = (DeadlineExceededException) exception.getCause(); + assertThat(cause.getStatusCode().getCode()).isEqualTo(StatusCode.Code.DEADLINE_EXCEEDED); + assertThat(cause.getMessage()).contains("https://upload.url/timeout-fire"); + assertThat(future.isDone()).isTrue(); + assertThat(future.isCancelled()).isFalse(); + } + + @Test + void testGlobalTimeout_cancelledCleanlyOnSuccess() throws Exception { + stubStartSession("https://upload.url/timeout-success"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn( + ApiFutures.immediateFuture( + ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "ok"))); + + ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class); + ScheduledFuture mockScheduledFuture = mock(ScheduledFuture.class); + when(mockExecutor.schedule(any(Runnable.class), anyLong(), any())) + .thenAnswer(inv -> mockScheduledFuture); + + ClientContext customClientContext = clientContext.toBuilder().setExecutor(mockExecutor).build(); + ResumableUploadCallableImpl customCallable = + new ResumableUploadCallableImpl<>(mockClient, defaultSettings, customClientContext); + + ResumableUploadCallSettings timeoutSettings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofSeconds(60)).build(); + + ResumableUploadFuture future = + customCallable.futureCall("resource-path", streamOf("hello"), timeoutSettings); + + assertThat(future.get()).isEqualTo("ok"); + verify(mockScheduledFuture).cancel(false); + } + + @Test + void testGlobalTimeout_cancelledCleanlyOnFailure() throws Exception { + when(mockStartCallable.futureCall(any(), any())) + .thenReturn( + ApiFutures.immediateFailedFuture( + createApiException(401, StatusCode.Code.UNAUTHENTICATED))); + + ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class); + ScheduledFuture mockScheduledFuture = mock(ScheduledFuture.class); + when(mockExecutor.schedule(any(Runnable.class), anyLong(), any())) + .thenAnswer(inv -> mockScheduledFuture); + + ClientContext customClientContext = clientContext.toBuilder().setExecutor(mockExecutor).build(); + ResumableUploadCallableImpl customCallable = + new ResumableUploadCallableImpl<>(mockClient, defaultSettings, customClientContext); + + ResumableUploadCallSettings timeoutSettings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofSeconds(60)).build(); + + ResumableUploadFuture future = + customCallable.futureCall("resource-path", streamOf("hello"), timeoutSettings); + + assertThrows(ExecutionException.class, future::get); + verify(mockScheduledFuture).cancel(false); + } + + @Test + void testGlobalTimeout_cancelledCleanlyOnUserCancel() throws Exception { + stubStartSession("https://upload.url/timeout-cancel"); + SettableApiFuture> hungChunk = SettableApiFuture.create(); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())).thenReturn(hungChunk); + + ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class); + ScheduledFuture mockScheduledFuture = mock(ScheduledFuture.class); + when(mockExecutor.schedule(any(Runnable.class), anyLong(), any())) + .thenAnswer(inv -> mockScheduledFuture); + + ClientContext customClientContext = clientContext.toBuilder().setExecutor(mockExecutor).build(); + ResumableUploadCallableImpl customCallable = + new ResumableUploadCallableImpl<>(mockClient, defaultSettings, customClientContext); + + ResumableUploadCallSettings timeoutSettings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofSeconds(60)).build(); + + ResumableUploadFuture future = + customCallable.futureCall("resource-path", streamOf("hello"), timeoutSettings); + + assertThat(future.cancel(true)).isTrue(); + verify(mockScheduledFuture).cancel(false); + } + + @Test + void testGlobalTimeout_derivesChunkLocalAndAttemptDeadlines() throws Exception { + stubStartSession("https://upload.url/derived-deadlines"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn( + ApiFutures.immediateFuture( + ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "ok"))); + + // 1. 60s global timeout yields 30s chunk budget + ResumableUploadCallSettings settings60s = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofSeconds(60)).build(); + callable.futureCall("resource-path", streamOf("hello"), settings60s).get(); + + ArgumentCaptor captor60s = ArgumentCaptor.forClass(ApiCallContext.class); + verify(mockChunkCallable).futureCall(any(), captor60s.capture()); + assertThat(captor60s.getValue().getTimeoutDuration()).isEqualTo(Duration.ofSeconds(30)); + assertThat(captor60s.getValue().getRetrySettings().getTotalTimeoutDuration()) + .isEqualTo(Duration.ofSeconds(30)); + + // 2. 10s global timeout yields 5s chunk budget (not the old 5-minute floor) + clearInvocations(mockChunkCallable); + ResumableUploadCallSettings settings10s = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofSeconds(10)).build(); + callable.futureCall("resource-path", streamOf("hello"), settings10s).get(); + + ArgumentCaptor captor10s = ArgumentCaptor.forClass(ApiCallContext.class); + verify(mockChunkCallable).futureCall(any(), captor10s.capture()); + assertThat(captor10s.getValue().getTimeoutDuration()).isEqualTo(Duration.ofSeconds(5)); + assertThat(captor10s.getValue().getRetrySettings().getTotalTimeoutDuration()) + .isEqualTo(Duration.ofSeconds(5)); + } + + @Test + void testGlobalTimeout_timeoutWhileAttemptInFlight_cancelsInFlightFutureAndDoesNotCorruptBuffer() + throws Exception { + stubStartSession("https://upload.url/in-flight-timeout"); + SettableApiFuture> inFlightFuture = SettableApiFuture.create(); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn(inFlightFuture); + + ResumableUploadCallSettings timeoutSettings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(80)).build(); + + ByteCountingStream stream = new ByteCountingStream("01234567890123456789"); + ResumableUploadFuture future = + callable.futureCall("resource-path", stream, timeoutSettings); + + ExecutionException exception = assertThrows(ExecutionException.class, future::get); + assertThat(exception.getCause()).isInstanceOf(DeadlineExceededException.class); + // In-flight attempt future must be cancelled + assertThat(inFlightFuture.isCancelled()).isTrue(); + + // Stream should have been read only up to the first chunk (chunkSize = 8), not refilled or + // advanced + assertThat(stream.totalBytesRead).isEqualTo(8); + } + + @Test + void testGlobalTimeout_coversStartSessionTimeout() throws Exception { + SettableApiFuture hungStartFuture = SettableApiFuture.create(); + when(mockStartCallable.futureCall(any(), any())).thenReturn(hungStartFuture); + + ResumableUploadCallSettings timeoutSettings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(80)).build(); + + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("hello"), timeoutSettings); + + ExecutionException exception = assertThrows(ExecutionException.class, future::get); + assertThat(exception.getCause()).isInstanceOf(DeadlineExceededException.class); + assertThat(exception.getCause().getMessage()).contains("before session initiation completed"); + assertThat(hungStartFuture.isCancelled()).isTrue(); + } + private static class HttpStatusStatusCode implements StatusCode { private final int httpStatus; private final StatusCode.Code code; @@ -913,6 +1094,7 @@ private static void assertChunk( private static class TrackableStream extends ByteArrayInputStream { boolean closed = false; + int closeCount = 0; int totalBytesRead = 0; TrackableStream(String content) { @@ -931,7 +1113,25 @@ public int read(byte[] b, int off, int len) { @Override public void close() throws IOException { closed = true; + closeCount++; super.close(); } } + + private static class ByteCountingStream extends ByteArrayInputStream { + int totalBytesRead = 0; + + ByteCountingStream(String content) { + super(content.getBytes(StandardCharsets.UTF_8)); + } + + @Override + public int read(byte[] b, int off, int len) { + int read = super.read(b, off, len); + if (read > 0) { + totalBytesRead += read; + } + return read; + } + } }