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 e836f2b1aacd..dd3581e00a87 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,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 @@ 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 = @@ -94,6 +121,8 @@ class ChunkAttemptCallable implements Callable> retryingFuture) { @@ -129,9 +158,36 @@ private void prepareAttempt( SettableApiFuture> attemptFuture, ApiCallContext attemptContext, RetryingFuture> 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> 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 a74b6964cfb7..882c3f8e81d9 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 @@ -52,6 +52,9 @@ import java.io.InputStream; import java.time.Duration; import java.util.concurrent.CancellationException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import org.jspecify.annotations.NullMarked; @@ -65,16 +68,13 @@ @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. static final RetrySettings DEFAULT_CHUNK_RETRY_SETTINGS = RetrySettings.newBuilder() .setInitialRetryDelayDuration(Duration.ofMillis(100)) .setRetryDelayMultiplier(1.3) .setMaxRetryDelayDuration(Duration.ofMinutes(1)) - .setInitialRpcTimeoutDuration(Duration.ofSeconds(30)) - .setRpcTimeoutMultiplier(1.0) - .setMaxRpcTimeoutDuration(Duration.ofSeconds(30)) - .setTotalTimeoutDuration(Duration.ofMinutes(5)) - .setMaxAttempts(5) .build(); private final Object lock = new Object(); @@ -87,15 +87,17 @@ final class ResumableUploadChunkCoordinator { uploadChunkCallable; private final UnaryCallable> queryStatusCallable; + private final ScheduledExecutorService executor; private final ScheduledRetryingExecutor> retryingExecutor; private final InputStream payload; + private final ResumableUploadCallSettings settings; private final int chunkSize; private final ApiCallContext callContext; private final ClientContext clientContext; - private final RetrySettings chunkRetrySettings; private volatile @Nullable String uploadSessionUrl; private volatile @Nullable RewindableStreamBuffer buffer; + private volatile long deadlineNanos; @GuardedBy("lock") private boolean done; @@ -106,6 +108,9 @@ final class ResumableUploadChunkCoordinator { @GuardedBy("lock") private @Nullable ApiFuture inFlightFuture; + @GuardedBy("lock") + private @Nullable ScheduledFuture timeoutFuture; + ResumableUploadChunkCoordinator( SettableApiFuture result, ApiFuture startFuture, @@ -122,19 +127,18 @@ final class ResumableUploadChunkCoordinator { this.queryStatusCallable = checkNotNull(queryStatusCallable, "queryStatusCallable must not be null"); this.payload = checkNotNull(payload, "payload must not be null"); - checkNotNull(settings, "settings must not be null"); + this.settings = checkNotNull(settings, "settings must not be null"); checkArgument(settings.getChunkSize() > 0, "chunkSize must be > 0"); this.chunkSize = settings.getChunkSize(); this.callContext = checkNotNull(callContext, "callContext must not be null"); this.clientContext = checkNotNull(clientContext, "clientContext must not be null"); - this.chunkRetrySettings = DEFAULT_CHUNK_RETRY_SETTINGS; + this.executor = checkNotNull(clientContext.getExecutor(), "executor must not be null"); RetryAlgorithm> retryAlgorithm = new RetryAlgorithm<>( new ResumableUploadResultRetryAlgorithm<>(ResumableUploadCommand.UPLOAD), - new ExponentialRetryAlgorithm(chunkRetrySettings, clientContext.getClock())); - this.retryingExecutor = - new ScheduledRetryingExecutor<>(retryAlgorithm, clientContext.getExecutor()); + new ExponentialRetryAlgorithm(DEFAULT_CHUNK_RETRY_SETTINGS, clientContext.getClock())); + this.retryingExecutor = new ScheduledRetryingExecutor<>(retryAlgorithm, this.executor); synchronized (lock) { this.inFlightFuture = startFuture; @@ -142,6 +146,15 @@ final class ResumableUploadChunkCoordinator { } void start() { + Duration timeout = settings.getGlobalTimeout(); + this.deadlineNanos = clientContext.getClock().nanoTime() + timeout.toNanos(); + synchronized (lock) { + if (!done) { + this.timeoutFuture = + executor.schedule(this::onTimeout, timeout.toMillis(), TimeUnit.MILLISECONDS); + } + } + ApiFutures.addCallback( startFuture, new ApiFutureCallback() { @@ -168,6 +181,21 @@ public void onFailure(Throwable t) { MoreExecutors.directExecutor()); } + private void onTimeout() { + synchronized (lock) { + if (done) { + return; + } + } + String message = + uploadSessionUrl != null + ? "Resumable upload timed out for session: " + uploadSessionUrl + : "Resumable upload timed out before session initiation completed"; + finish( + null, + new DeadlineExceededException(message, null, UploadErrors.TIMEOUT_STATUS_CODE, false)); + } + @Nullable String getUploadSessionUrl() { return uploadSessionUrl; } @@ -176,7 +204,7 @@ void setInFlightFuture(ApiFuture future) { boolean shouldCancel = false; synchronized (lock) { if (done) { - shouldCancel = result.isCancelled(); + shouldCancel = true; } else { this.inFlightFuture = future; } @@ -187,6 +215,7 @@ void setInFlightFuture(ApiFuture future) { } void cancel(boolean mayInterruptIfRunning) { + ScheduledFuture timeout; ApiFuture inFlight; synchronized (lock) { if (done) { @@ -195,6 +224,11 @@ void cancel(boolean mayInterruptIfRunning) { done = true; inFlight = this.inFlightFuture; this.inFlightFuture = null; + timeout = this.timeoutFuture; + this.timeoutFuture = null; + } + if (timeout != null) { + timeout.cancel(false); } if (inFlight != null) { inFlight.cancel(mayInterruptIfRunning); @@ -203,12 +237,23 @@ void cancel(boolean mayInterruptIfRunning) { } private void finish(@Nullable ResponseT response, @Nullable Throwable error) { + ScheduledFuture timeout; + ApiFuture inFlight; synchronized (lock) { if (done) { return; } done = true; - inFlightFuture = null; + inFlight = this.inFlightFuture; + this.inFlightFuture = null; + timeout = this.timeoutFuture; + this.timeoutFuture = null; + } + if (timeout != null) { + timeout.cancel(false); + } + if (inFlight != null && error != null) { + inFlight.cancel(true); } IOException closeError = closePayload(); if (error == null) { @@ -298,6 +343,21 @@ private void transmitSingleChunk(long currentOffset) { .setFinal(streamBuffer.isFinal()) .build(); + // 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 = + DEFAULT_CHUNK_RETRY_SETTINGS.toBuilder() + .setTotalTimeoutDuration(chunkDeadline) + .setInitialRpcTimeoutDuration(chunkDeadline) + .setRpcTimeoutMultiplier(1.0) + .setMaxRpcTimeoutDuration(chunkDeadline) + .build(); + ApiCallContext chunkCallContext = callContext.withRetrySettings(derivedChunkRetrySettings); + ChunkAttemptCallable attemptCallable = new ChunkAttemptCallable<>( uploadChunkCallable, @@ -305,11 +365,13 @@ private void transmitSingleChunk(long currentOffset) { streamBuffer, url, chunkRequest, - callContext, - command); + chunkCallContext, + command, + deadlineNanos, + clientContext.getClock()); RetryingFuture> retryingFuture = - retryingExecutor.createFuture(attemptCallable, callContext); + retryingExecutor.createFuture(attemptCallable, chunkCallContext); attemptCallable.setRetryingFuture(retryingFuture); setInFlightFuture(retryingFuture); 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 17c420635553..7de11039fcc7 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; @@ -53,6 +56,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; @@ -60,6 +64,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; @@ -80,6 +85,7 @@ class ResumableUploadCallableImplTest { private ResumableUploadCallSettings defaultSettings; private FakeCallContext callContext; private ScheduledExecutorService executor; + private ClientContext clientContext; private ResumableUploadCallableImpl callable; @BeforeEach @@ -97,7 +103,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); } @@ -270,7 +276,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 @@ -282,7 +288,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 @@ -449,16 +455,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 @@ -487,7 +495,7 @@ void testChunkRetry_cancelSession_cancelsInFlightHttpFuture() throws Exception { // The in-flight HTTP chunk future must have been cancelled assertThat(inFlightChunkFuture.isCancelled()).isTrue(); // The payload stream must be closed - assertThat(stream.closed).isTrue(); + assertThat(stream.closeCount).isEqualTo(1); } @Test @@ -861,6 +869,204 @@ void testBufferWindow_noArrayCopyForPartialChunk_backingArrayIdentityPreserved() assertThat(chunk.getPayload().length).isEqualTo(8); } + @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(true, "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(true, "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 testUploadCallable_failureOutcome_attachesCloseExceptionViaAddSuppressed() { + when(mockStartCallable.futureCall(any(), any())) + .thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("upload failed"))); + + InputStream failingStream = + new InputStream() { + @Override + public int read() { + return -1; + } + + @Override + public void close() throws IOException { + throw new IOException("stream close error"); + } + }; + + ResumableUploadFuture future = + callable.futureCall("resource-path", failingStream, null); + ExecutionException exception = assertThrows(ExecutionException.class, future::get); + assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(exception.getCause().getSuppressed()).asList().hasSize(1); + assertThat(exception.getCause().getSuppressed()[0]).isInstanceOf(IOException.class); + assertThat(exception.getCause().getSuppressed()[0]) + .hasMessageThat() + .contains("stream close error"); + } + + @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; @@ -932,7 +1138,7 @@ private static void assertChunk( } private static class TrackableStream extends ByteArrayInputStream { - boolean closed = false; + int closeCount = 0; TrackableStream(String content) { super(content.getBytes(StandardCharsets.UTF_8)); @@ -940,7 +1146,7 @@ private static class TrackableStream extends ByteArrayInputStream { @Override public void close() throws IOException { - closed = true; + closeCount++; super.close(); } }