From 05d88ebcd7b8137dcb765b252f055beaf9ba4adb Mon Sep 17 00:00:00 2001 From: whowes Date: Fri, 6 Nov 2026 18:07:03 +0000 Subject: [PATCH] feat(gax): enable stall control for resumable uploads --- .../rpc/ResumableUploadChunkCoordinator.java | 62 ++++- .../gax/rpc/ResumableUploadFutureImpl.java | 51 ++++- .../api/gax/rpc/ResumableUploadOptions.java | 64 ++++++ .../rpc/ResumableUploadCallableImplTest.java | 213 ++++++++++++++++++ .../gax/rpc/ResumableUploadOptionsTest.java | 85 +++++++ 5 files changed, 460 insertions(+), 15 deletions(-) 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 a5eb146fd2a5..5aae15937746 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 @@ -42,11 +42,13 @@ import com.google.api.gax.resumable.QueryStatusResponse; import com.google.api.gax.resumable.ResumableUploadStatus; import com.google.api.gax.retrying.ExponentialRetryAlgorithm; +import com.google.api.gax.retrying.RetrySettings; import com.google.api.gax.retrying.TimedAttemptSettings; import com.google.api.gax.rpc.ResumableUploadErrorClassifier.Category; import com.google.common.util.concurrent.MoreExecutors; import java.io.IOException; import java.io.InputStream; +import java.time.Duration; import java.util.Optional; import java.util.concurrent.CancellationException; import java.util.concurrent.Future; @@ -75,6 +77,7 @@ final class ResumableUploadChunkCoordinator { private final ExponentialRetryAlgorithm recoveryAlgorithm; private final ScheduledExecutorService executor; private final ResumableUploadProgressTracker progressTracker; + private final @Nullable ResumableUploadStallController stallController; private final SettableApiFuture uploadResultFuture = SettableApiFuture.create(); private volatile @Nullable Future inFlightFuture; private TimedAttemptSettings recoverySettings; @@ -87,7 +90,8 @@ final class ResumableUploadChunkCoordinator { InputStream payload, int chunkSize, ClientContext clientContext, - ResumableUploadProgressTracker progressTracker) { + ResumableUploadProgressTracker progressTracker, + @Nullable ResumableUploadStallController stallController) { this.uploadChunkCallable = checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null"); this.queryStatusCallable = @@ -101,11 +105,23 @@ final class ResumableUploadChunkCoordinator { new ExponentialRetryAlgorithm( ResumableUploadCallableImpl.RETRY_SETTINGS, clientContext.getClock()); this.executor = checkNotNull(clientContext.getExecutor(), "executor must not be null"); - this.recoverySettings = recoveryAlgorithm.createFirstAttempt(); this.progressTracker = checkNotNull(progressTracker, "progressTracker must not be null"); + this.stallController = stallController; + this.recoverySettings = createFirstRecoveryAttempt(); this.buffer = new RewindableStreamBuffer(payload, chunkSize); } + private TimedAttemptSettings createFirstRecoveryAttempt() { + if (stallController != null) { + return recoveryAlgorithm.createFirstAttempt( + callContext.withRetrySettings( + ResumableUploadCallableImpl.RETRY_SETTINGS.toBuilder() + .setTotalTimeoutDuration(Duration.ZERO) + .build())); + } + return recoveryAlgorithm.createFirstAttempt(); + } + ApiFuture getFuture() { return uploadResultFuture; } @@ -149,10 +165,16 @@ private void transmitChunk() { } ChunkUploadRequest chunkRequest = buildCurrentChunkRequest(); + ApiCallContext chunkCallContext = callContext; + if (stallController != null) { + chunkCallContext = + callContext.withRetrySettings( + stallController.onStartChunk(chunkRequest.getPayload().length)); + } // Dispatch the chunk upload call and register the in-flight future for cancellation. ApiFuture> chunkFuture = - uploadChunkCallable.futureCall(chunkRequest, callContext); + uploadChunkCallable.futureCall(chunkRequest, chunkCallContext); if (!tryRegisterInFlightFuture(chunkFuture)) { return; } @@ -189,6 +211,9 @@ public void onFailure(Throwable t) { } else { // Category.TRANSIENT errors reaching here have already exhausted their retry budget // in the underlying RetryingCallable and become fatal per protocol specification. + if (stallController != null) { + t = stallController.translateException(t, ResumableUploadCommand.UPLOAD); + } uploadResultFuture.setException(t); } } @@ -203,8 +228,7 @@ private void recover(Throwable cause) { progressTracker.onRecovering(); try { if (madeProgressSinceRecovery) { - recoverySettings = - recoveryAlgorithm.createNextAttempt(recoveryAlgorithm.createFirstAttempt()); + recoverySettings = recoveryAlgorithm.createNextAttempt(createFirstRecoveryAttempt()); } else { recoverySettings = recoveryAlgorithm.createNextAttempt(recoverySettings); } @@ -213,24 +237,34 @@ private void recover(Throwable cause) { uploadResultFuture.setException(cause); return; } + Duration backoffDelay = recoverySettings.getRandomizedRetryDelayDuration(); + RetrySettings queryRetrySettings = null; + if (stallController != null) { + queryRetrySettings = stallController.onStartQuery(backoffDelay); + } + RetrySettings finalQueryRetrySettings = queryRetrySettings; tryRegisterInFlightFuture( executor.schedule( - this::queryStatus, - recoverySettings.getRandomizedRetryDelayDuration().toNanos(), + () -> queryStatus(finalQueryRetrySettings), + backoffDelay.toNanos(), TimeUnit.NANOSECONDS)); } catch (Throwable t) { uploadResultFuture.setException(t); } } - private void queryStatus() { + private void queryStatus(@Nullable RetrySettings queryRetrySettings) { try { if (uploadResultFuture.isDone()) { return; } + ApiCallContext queryCallContext = callContext; + if (queryRetrySettings != null) { + queryCallContext = callContext.withRetrySettings(queryRetrySettings); + } // Dispatch the query status call and register the in-flight future for cancellation. ApiFuture> queryFuture = - queryStatusCallable.futureCall(QueryStatusRequest.create(uploadUrl), callContext); + queryStatusCallable.futureCall(QueryStatusRequest.create(uploadUrl), queryCallContext); if (!tryRegisterInFlightFuture(queryFuture)) { return; } @@ -254,6 +288,9 @@ public void onFailure(Throwable t) { if (t instanceof CancellationException || uploadResultFuture.isDone()) { return; } + if (stallController != null) { + t = stallController.translateException(t, ResumableUploadCommand.QUERY); + } uploadResultFuture.setException(t); } }, @@ -294,6 +331,10 @@ private void handleQueryResponse(QueryStatusResponse queryResponse) throw new IllegalStateException( "Incomplete query status response did not include a committed offset"); } + long previousBaseOffset = buffer.getBufferBaseOffset(); + if (stallController != null) { + stallController.onRecoveryOffsetReceived(committedOffset - previousBaseOffset); + } progressTracker.onOffsetReceived(committedOffset, chunkSize); buffer.realignTo(committedOffset); transmitChunk(); @@ -309,6 +350,9 @@ private void handleChunkResponse(ChunkUploadResponse response) throws "Upload stream ended and final chunk was transmitted, but server returned" + " incomplete status")); } else { + if (stallController != null) { + stallController.onChunkSuccess(buffer.getPayload().length); + } madeProgressSinceRecovery = true; buffer.fill(); progressTracker.onChunkUploaded(buffer.getBufferBaseOffset()); 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 57bc530ae39f..c82a9aa27914 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 @@ -78,7 +78,9 @@ final class ResumableUploadFutureImpl implements ResumableUploadFutur private final ResumableUploadOptions options; private final ClientContext clientContext; private final ScheduledExecutorService executor; + private final boolean isResume; private final ResumableUploadProgressTracker progressTracker; + private final @Nullable ResumableUploadStallController stallController; private final SettableApiFuture resultFuture = SettableApiFuture.create(); private volatile @Nullable InputStream payload; @@ -157,9 +159,13 @@ static ResumableUploadFutureImpl createForResume( sessionUrl, uploadChunkCallable, queryStatusCallable, options, clientContext); ApiFuture> queryFuture; try { + ApiCallContext queryCallContext = clientContext.getDefaultCallContext(); + if (future.stallController != null) { + queryCallContext = + queryCallContext.withRetrySettings(future.stallController.onStartQuery(Duration.ZERO)); + } queryFuture = - queryStatusCallable.futureCall( - QueryStatusRequest.create(sessionUrl), clientContext.getDefaultCallContext()); + queryStatusCallable.futureCall(QueryStatusRequest.create(sessionUrl), queryCallContext); } catch (Throwable t) { queryFuture = ApiFutures.immediateFailedFuture(t); } @@ -188,6 +194,7 @@ private ResumableUploadFutureImpl( ResumableUploadOptions options, ClientContext clientContext) { this.uploadSessionUrl = sessionUrl; + this.isResume = sessionUrl != null; this.uploadChunkCallable = checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null"); this.queryStatusCallable = @@ -196,8 +203,10 @@ private ResumableUploadFutureImpl( checkArgument(options.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"); - if (sessionUrl != null) { - this.progressTracker = ResumableUploadProgressTracker.createForResume(sessionUrl); + this.stallController = createStallController(options, clientContext); + if (this.isResume) { + this.progressTracker = + ResumableUploadProgressTracker.createForResume(checkNotNull(sessionUrl)); } else { this.progressTracker = ResumableUploadProgressTracker.create(); } @@ -207,6 +216,24 @@ private ResumableUploadFutureImpl( } } + private static @Nullable ResumableUploadStallController createStallController( + ResumableUploadOptions options, ClientContext clientContext) { + if (!options.isStallControlEnabled()) { + return null; + } + Duration globalDeadline = null; + if (options.getGlobalTimeout() != null) { + globalDeadline = + Duration.ofMillis(clientContext.getClock().millisTime()).plus(options.getGlobalTimeout()); + } + return new ResumableUploadStallController( + clientContext.getClock(), + checkNotNull(options.getTransferStallMinimumRate()), + checkNotNull(options.getTransferStallTimeout()), + globalDeadline, + ResumableUploadCallableImpl.RETRY_SETTINGS); + } + private void initiate( ApiFuture initiationFuture, InputStreamSupplier payloadSupplier, @@ -235,6 +262,9 @@ public void onFailure(Throwable t) { if (t instanceof CancellationException || resultFuture.isDone()) { return; } + if (isResume && stallController != null) { + t = stallController.translateException(t, ResumableUploadCommand.QUERY); + } fail(t); } }, @@ -242,7 +272,15 @@ public void onFailure(Throwable t) { } private void setUpGlobalTimeout() { - Duration timeout = firstNonNull(options.getGlobalTimeout(), DEFAULT_GLOBAL_TIMEOUT); + Duration timeout; + if (options.isStallControlEnabled()) { + timeout = options.getGlobalTimeout(); + } else { + timeout = firstNonNull(options.getGlobalTimeout(), DEFAULT_GLOBAL_TIMEOUT); + } + if (timeout == null) { + return; + } ScheduledFuture timeoutFuture = executor.schedule(this::onTimeout, timeout.toMillis(), TimeUnit.MILLISECONDS); resultFuture.addListener(() -> timeoutFuture.cancel(false), MoreExecutors.directExecutor()); @@ -259,7 +297,8 @@ private void setUpGlobalTimeout() { checkNotNull(payload), chunkSize, clientContext, - progressTracker); + progressTracker, + stallController); ApiFuture uploadFuture = coordinator.getFuture(); synchronized (lock) { if (inFlightFuture == null) { diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadOptions.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadOptions.java index 4c1c6385c6c9..7d9bf118a46f 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadOptions.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadOptions.java @@ -61,6 +61,24 @@ public abstract class ResumableUploadOptions { */ public abstract @Nullable ResumableUploadProgressListener getProgressListener(); + /** + * Returns the minimum acceptable transfer rate in MiB/s (1 MiB = 1,048,576 bytes) for stall + * detection, or {@code null} if unset. Stall control is only active when both {@link + * #getTransferStallMinimumRate()} and {@link #getTransferStallTimeout()} are set. + */ + public abstract @Nullable Double getTransferStallMinimumRate(); + + /** + * Returns the maximum duration the upload may remain below {@link #getTransferStallMinimumRate()} + * before being aborted as stalled, or {@code null} if unset. Stall control is only active when + * both {@link #getTransferStallMinimumRate()} and {@link #getTransferStallTimeout()} are set. + */ + public abstract @Nullable Duration getTransferStallTimeout(); + + boolean isStallControlEnabled() { + return getTransferStallMinimumRate() != null && getTransferStallTimeout() != null; + } + /** * Merges another {@code ResumableUploadOptions} instance with this one. Fields set in {@code * other} override fields in this instance. @@ -76,8 +94,17 @@ public ResumableUploadOptions merge(@Nullable ResumableUploadOptions other) { if (other.getChunkSize() > 0) { builder.setChunkSize(other.getChunkSize()); } + if (other.getTransferStallMinimumRate() != null) { + builder.setTransferStallMinimumRate(other.getTransferStallMinimumRate()); + } + if (other.getTransferStallTimeout() != null) { + builder.setTransferStallTimeout(other.getTransferStallTimeout()); + } if (other.getGlobalTimeout() != null) { builder.setGlobalTimeout(other.getGlobalTimeout()); + } else if (builder.getTransferStallMinimumRate() != null + && builder.getTransferStallTimeout() != null) { + builder.setGlobalTimeout(null); } ResumableUploadProgressListener otherProgressListener = other.getProgressListener(); if (otherProgressListener != null) { @@ -128,6 +155,33 @@ public abstract static class Builder { public abstract Builder setProgressListener( @Nullable ResumableUploadProgressListener progressListener); + /** + * Sets the minimum acceptable transfer rate in MiB/s (1 MiB = 1,048,576 bytes) for stall + * detection. Must be finite and positive. Stall control is only active when both {@code + * transferStallMinimumRate} and {@code transferStallTimeout} are set; enabling stall control + * without an explicit {@link #setGlobalTimeout(Duration)} disables the default global timeout. + */ + public abstract Builder setTransferStallMinimumRate(@Nullable Double transferStallMinimumRate); + + /** + * Returns the minimum acceptable transfer rate in MiB/s, or {@code null} if unset. See {@link + * #setTransferStallMinimumRate(Double)}. + */ + public abstract @Nullable Double getTransferStallMinimumRate(); + + /** + * Sets the maximum duration the upload may remain below {@link #getTransferStallMinimumRate()} + * before being aborted as stalled. Must be positive. Stall control is only active when both + * {@code transferStallMinimumRate} and {@code transferStallTimeout} are set. + */ + public abstract Builder setTransferStallTimeout(@Nullable Duration transferStallTimeout); + + /** + * Returns the stall timeout duration, or {@code null} if unset. See {@link + * #setTransferStallTimeout(Duration)}. + */ + public abstract @Nullable Duration getTransferStallTimeout(); + abstract ResumableUploadOptions autoBuild(); public ResumableUploadOptions build() { @@ -137,6 +191,16 @@ public ResumableUploadOptions build() { !getGlobalTimeout().isNegative() && !getGlobalTimeout().isZero(), "globalTimeout must be positive"); } + if (getTransferStallMinimumRate() != null) { + Preconditions.checkArgument( + Double.isFinite(getTransferStallMinimumRate()) && getTransferStallMinimumRate() > 0.0, + "transferStallMinimumRate must be finite and positive"); + } + if (getTransferStallTimeout() != null) { + Preconditions.checkArgument( + !getTransferStallTimeout().isNegative() && !getTransferStallTimeout().isZero(), + "transferStallTimeout must be positive"); + } return autoBuild(); } } 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 ad2eca4f1cec..05b577a46c9a 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 @@ -48,6 +48,8 @@ import com.google.api.core.ApiFutures; import com.google.api.core.ForwardingApiFuture; import com.google.api.core.SettableApiFuture; +import com.google.api.gax.core.FakeApiClock; +import com.google.api.gax.core.RecordingScheduler; import com.google.api.gax.resumable.ChunkUploadRequest; import com.google.api.gax.resumable.ChunkUploadResponse; import com.google.api.gax.resumable.QueryStatusRequest; @@ -71,11 +73,14 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.jspecify.annotations.Nullable; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.ArgumentCaptor; import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.stubbing.Answer; @ExtendWith(MockitoExtension.class) class ResumableUploadCallableImplTest { @@ -1544,6 +1549,214 @@ void testResumeCall_failure_includesSessionUrlAndEffectiveChunkSize() { assertThat(shortStream.closed).isTrue(); } + @Nested + class StallControlTest { + private FakeApiClock clock; + private RecordingScheduler scheduler; + private ResumableUploadCallableImpl stallCallable; + + @BeforeEach + void setUpStall() { + clock = new FakeApiClock(0L); + scheduler = RecordingScheduler.create(clock); + stallCallable = createStallCallable(scheduler, defaultStallOptions()); + } + + @AfterEach + void tearDownStall() { + scheduler.shutdownNow(); + } + + @Test + void testSlowNonFinalChunksStall() { + stubStartSession("https://upload.url/stall-non-final"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenAnswer( + respondsAfter(5, ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null))) + .thenAnswer( + respondsAfter(15, ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null))); + + ExecutionException stalled = + assertThrows( + ExecutionException.class, + () -> + stallCallable + .futureCall("resource-path", payloadOf("0123456789abcdefghij"), null) + .get()); + assertThat(stalled.getCause()).isInstanceOf(DeadlineExceededException.class); + assertThat(stalled.getCause().getMessage()).contains("Resumable upload stalled"); + } + + @Test + void testSlowFinalChunkSucceeds() throws Exception { + stubStartSession("https://upload.url/stall-final-slow"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenAnswer( + respondsAfter(5, ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null))) + .thenAnswer( + respondsAfter( + 30, ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "final-ok"))); + + // 12-byte payload with chunkSize=8 makes chunk 2 (4 bytes) the genuine final chunk. + assertThat(stallCallable.futureCall("resource-path", payloadOf("0123456789ab"), null).get()) + .isEqualTo("final-ok"); + // Confirms setUpGlobalTimeout() did not schedule the 15-minute DEFAULT_GLOBAL_TIMEOUT. + assertThat(scheduler.getIterationsCount()).isEqualTo(0); + } + + @Test + void testRecoveryCreditsPartialProgressAndStallsWhenTooSlow() { + stubStartSession("https://upload.url/stall-recovery"); + // Rate = 8 bytes / 5s (expected = 5s per 8-byte chunk). Chunk 1 (6s) incurs 1s lag and sets + // stallStartedAt=6s. Chunk 2 gets a 14s deadline (5s - 1s + 10s, until t=20s) and fails with + // 400 at t=12s; queryStatus (deadline ~7s) responds in 5s (t=17s..18s) with 4 committed + // bytes, exceeding stallStartedAt + 10s (t=16s) when crediting partial recovery progress. + ResumableUploadOptions rateOptions = + defaultStallOptions().toBuilder() + .setTransferStallMinimumRate(8.0 / (5.0 * 1024 * 1024)) + .build(); + ResumableUploadCallableImpl callable = + createStallCallable(scheduler, rateOptions); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenAnswer( + respondsAfter(6, ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null))) + .thenAnswer(failsAfter(6, createApiException(400, StatusCode.Code.INVALID_ARGUMENT))); + when(mockQueryCallable.futureCall(any(), any())) + .thenAnswer( + respondsAfter(5, createQueryResponse(12L, null, ResumableUploadStatus.ACTIVE))); + + ExecutionException ex = + assertThrows( + ExecutionException.class, + () -> + callable + .futureCall("resource-path", payloadOf("0123456789abcdefghij"), null) + .get()); + assertThat(ex.getCause()).isInstanceOf(DeadlineExceededException.class); + assertThat(ex.getCause().getMessage()).contains("Resumable upload stalled"); + verify(mockQueryCallable, times(1)).futureCall(any(), any()); + } + + @Test + void testResumeCallDoesNotTreatPreExistingOffsetAsStallLag() throws Exception { + // At t=100s, resuming with 8 pre-existing bytes keeps lag=0s so chunk 1 gets a fresh 10s + // deadline (allowing a 2s 503 attempt to retry and succeed instead of clamping the deadline + // to 1s). + clock.incrementNanoTime(TimeUnit.SECONDS.toNanos(100)); + String sessionUrl = "https://upload.url/stall-resume"; + when(mockQueryCallable.futureCall(eq(QueryStatusRequest.create(sessionUrl)), any())) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(8L, null, ResumableUploadStatus.ACTIVE))); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenAnswer(failsAfter(2, createApiException(503, StatusCode.Code.UNAVAILABLE))) + .thenAnswer( + respondsAfter( + 2, ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "resumed-ok"))); + + assertThat(stallCallable.resumeCall(sessionUrl, payloadOf("0123456789ab"), null).get()) + .isEqualTo("resumed-ok"); + } + + @Test + void testExhaustedTransientRetriesReportStalled() { + stubStartSession("https://upload.url/stall-503"); + ApiException unavailable503 = createApiException(503, StatusCode.Code.UNAVAILABLE); + // Each 503 takes 6s against a 10s stallTimeout: attempt 1 (6s + <=1s backoff) fits in 10s, + // and attempt 2 (12s) exceeds the 10s chunk deadline (2 attempts vs 5 under default + // settings). + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenAnswer(failsAfter(6, unavailable503)); + + ExecutionException stalled = + assertThrows( + ExecutionException.class, + () -> stallCallable.futureCall("resource-path", payloadOf("hello"), null).get()); + assertThat(stalled.getCause()).isInstanceOf(DeadlineExceededException.class); + assertThat(stalled.getCause().getMessage()).contains("Resumable upload stalled"); + assertThat(stalled.getCause().getCause()).isSameInstanceAs(unavailable503); + verify(mockChunkCallable, times(2)).futureCall(any(), any()); + } + + @Test + void testGlobalTimeoutPrecedenceReportsTimedOut() { + stubStartSession("https://upload.url/stall-global-precedence"); + // Mock executor prevents RecordingScheduler from firing the global timeout timer immediately. + ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class); + when(mockExecutor.schedule(any(Runnable.class), anyLong(), any())) + .thenReturn(mock(ScheduledFuture.class)); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenAnswer(failsAfter(4, createApiException(503, StatusCode.Code.UNAVAILABLE))); + + ResumableUploadOptions tightGlobal = + defaultStallOptions().toBuilder().setGlobalTimeout(Duration.ofSeconds(4)).build(); + ExecutionException timedOut = + assertThrows( + ExecutionException.class, + () -> + createStallCallable(mockExecutor, tightGlobal) + .futureCall("resource-path", payloadOf("hello"), null) + .get()); + assertThat(timedOut.getCause()).isInstanceOf(DeadlineExceededException.class); + assertThat(timedOut.getCause().getMessage()).contains("Resumable upload timed out"); + } + + @Test + void testRecoveryMaxAttemptsExhaustedSurfacesOriginalErrorBeyondTwoMinutes() { + stubStartSession("https://upload.url/stall-recovery-max-attempts"); + // Each chunk attempt takes 40s (4 recovery rounds span > 160s, exceeding the default 2m + // RETRY_SETTINGS totalTimeoutDuration) while remaining within a 600s stallTimeout. + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenAnswer(failsAfter(40, createApiException(400, StatusCode.Code.INVALID_ARGUMENT))); + when(mockQueryCallable.futureCall(any(), any())) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(0L, null, ResumableUploadStatus.ACTIVE))); + + ResumableUploadOptions generousStallOptions = + defaultStallOptions().toBuilder() + .setTransferStallTimeout(Duration.ofSeconds(600)) + .build(); + ResumableUploadCallableImpl generousCallable = + createStallCallable(scheduler, generousStallOptions); + + ExecutionException ex = + assertThrows( + ExecutionException.class, + () -> generousCallable.futureCall("resource-path", payloadOf("hello"), null).get()); + assertThat(ex.getCause()).isInstanceOf(InvalidArgumentException.class); + verify(mockQueryCallable, times(4)).futureCall(any(), any()); + verify(mockChunkCallable, times(5)).futureCall(any(), any()); + } + + private ResumableUploadCallableImpl createStallCallable( + ScheduledExecutorService executor, ResumableUploadOptions options) { + ClientContext ctx = clientContext.toBuilder().setClock(clock).setExecutor(executor).build(); + return new ResumableUploadCallableImpl<>(mockClient, options, ctx); + } + + private ResumableUploadOptions defaultStallOptions() { + return defaultSettings.toBuilder() + .setTransferStallMinimumRate(1.0) + .setTransferStallTimeout(Duration.ofSeconds(10)) + .build(); + } + + private Answer> respondsAfter(long seconds, T response) { + return inv -> { + clock.incrementNanoTime(TimeUnit.SECONDS.toNanos(seconds)); + return ApiFutures.immediateFuture(response); + }; + } + + private Answer> failsAfter(long seconds, ApiException error) { + return inv -> { + clock.incrementNanoTime(TimeUnit.SECONDS.toNanos(seconds)); + return ApiFutures.immediateFailedFuture(error); + }; + } + } + private static class HttpStatusStatusCode implements StatusCode { private final int httpStatus; private final StatusCode.Code code; diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadOptionsTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadOptionsTest.java index a0f96b17f19b..54f33e3bb059 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadOptionsTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadOptionsTest.java @@ -30,27 +30,81 @@ package com.google.api.gax.rpc; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.time.Duration; import org.junit.jupiter.api.Test; public class ResumableUploadOptionsTest { + @Test + public void testDefaultOptions() { + ResumableUploadOptions options = ResumableUploadOptions.newBuilder().build(); + + assertEquals(8 * 1024 * 1024, options.getChunkSize()); + assertNull(options.getGlobalTimeout()); + assertNull(options.getProgressListener()); + assertNull(options.getTransferStallMinimumRate()); + assertNull(options.getTransferStallTimeout()); + assertFalse(options.isStallControlEnabled()); + } + @Test public void testCustomOptionsAndToBuilder() { ResumableUploadOptions options = ResumableUploadOptions.newBuilder() .setChunkSize(16 * 1024 * 1024) .setGlobalTimeout(Duration.ofMinutes(15)) + .setTransferStallMinimumRate(2.5) + .setTransferStallTimeout(Duration.ofSeconds(45)) .build(); assertEquals(16 * 1024 * 1024, options.getChunkSize()); assertEquals(Duration.ofMinutes(15), options.getGlobalTimeout()); + assertEquals(2.5, options.getTransferStallMinimumRate()); + assertEquals(Duration.ofSeconds(45), options.getTransferStallTimeout()); + assertTrue(options.isStallControlEnabled()); assertEquals(options, options.toBuilder().build()); } + @Test + public void testPartialStallPair_stallControlDisabled() { + assertFalse( + ResumableUploadOptions.newBuilder() + .setTransferStallMinimumRate(10.0) + .build() + .isStallControlEnabled()); + assertFalse( + ResumableUploadOptions.newBuilder() + .setTransferStallTimeout(Duration.ofSeconds(30)) + .build() + .isStallControlEnabled()); + } + + @Test + public void testInvalidTransferStallOptions_throwsIllegalArgumentException() { + for (double invalidRate : + new double[] {0.0, -1.0, Double.NaN, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY}) { + assertThrows( + IllegalArgumentException.class, + () -> + ResumableUploadOptions.newBuilder().setTransferStallMinimumRate(invalidRate).build()); + } + assertThrows( + IllegalArgumentException.class, + () -> ResumableUploadOptions.newBuilder().setTransferStallTimeout(Duration.ZERO).build()); + assertThrows( + IllegalArgumentException.class, + () -> + ResumableUploadOptions.newBuilder() + .setTransferStallTimeout(Duration.ofSeconds(-5)) + .build()); + } + @Test public void testInvalidChunkSize_throwsIllegalArgumentException() { assertThrows( @@ -138,4 +192,35 @@ public void testMerge_nullProgressListenerDoesNotOverride() { assertSame(defaultListener, merged.getProgressListener()); } + + @Test + public void testMerge_stallControlClearsDefaultGlobalTimeoutUnlessExplicitOnOverride() { + ResumableUploadOptions defaults = + ResumableUploadOptions.newBuilder().setGlobalTimeout(Duration.ofSeconds(5)).build(); + ResumableUploadOptions stallPair = + ResumableUploadOptions.newBuilder() + .setTransferStallMinimumRate(5.0) + .setTransferStallTimeout(Duration.ofSeconds(60)) + .build(); + + // 1. Enabling stall control per-call clears the default globalTimeout. + ResumableUploadOptions mergedPerCall = defaults.merge(stallPair); + assertTrue(mergedPerCall.isStallControlEnabled()); + assertNull(mergedPerCall.getGlobalTimeout()); + + // 2. Completing the stall pair across default + per-call also clears the default globalTimeout. + ResumableUploadOptions defaultWithRate = + defaults.toBuilder().setTransferStallMinimumRate(5.0).build(); + ResumableUploadOptions perCallWithTimeout = + ResumableUploadOptions.newBuilder().setTransferStallTimeout(Duration.ofSeconds(60)).build(); + ResumableUploadOptions mergedSplit = defaultWithRate.merge(perCallWithTimeout); + assertTrue(mergedSplit.isStallControlEnabled()); + assertNull(mergedSplit.getGlobalTimeout()); + + // 3. Explicit per-call globalTimeout is preserved alongside stall control. + ResumableUploadOptions stallWithExplicitGlobal = + stallPair.toBuilder().setGlobalTimeout(Duration.ofMinutes(30)).build(); + assertEquals( + Duration.ofMinutes(30), defaults.merge(stallWithExplicitGlobal).getGlobalTimeout()); + } }