diff --git a/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java b/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java index 7da0022e0805e8..51df61a6b450a4 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java +++ b/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java @@ -314,7 +314,7 @@ public Single execute( emitter -> { synchronized (lock) { if (state != STATE_ACTIVE) { - emitter.onError(new CancellationException("already shutdown")); + emitter.tryOnError(new CancellationException("already shutdown")); return; } @@ -357,9 +357,9 @@ public void onSuccess(@NonNull ValueT valueT) { @Override public void onError(@NonNull Throwable e) { - if (!emitter.isDisposed()) { - emitter.onError(e); - } + // Don't report via RxJava's global error handler if the emitter has been + // disposed. + emitter.tryOnError(e); } }); } diff --git a/src/main/java/com/google/devtools/build/lib/remote/util/RxFutures.java b/src/main/java/com/google/devtools/build/lib/remote/util/RxFutures.java index 43e28baed00ef4..41e94ef67ceafa 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/util/RxFutures.java +++ b/src/main/java/com/google/devtools/build/lib/remote/util/RxFutures.java @@ -92,10 +92,11 @@ public void onFailure(Throwable throwable) { * 1. The ListenableFuture itself is cancelled. * 2. Completable is disposed by downstream. * - * This check is used to prevent propagating CancellationException to downstream - * when it has already disposed the Completable. + * In the second case, the CancellationException must not be propagated to + * downstream. */ - if (throwable instanceof CancellationException && emitter.isDisposed()) { + if (throwable instanceof CancellationException) { + emitter.tryOnError(throwable); return; } @@ -162,10 +163,11 @@ public void onFailure(Throwable throwable) { * 1. The ListenableFuture itself is cancelled. * 2. Single is disposed by downstream. * - * This check is used to prevent propagating CancellationException to downstream - * when it has already disposed the Single. + * In the second case, the CancellationException must not be propagated to + * downstream. */ - if (throwable instanceof CancellationException && emitter.isDisposed()) { + if (throwable instanceof CancellationException) { + emitter.tryOnError(throwable); return; } @@ -262,5 +264,4 @@ public void onError(Throwable e) { }); return future; } - } diff --git a/src/main/java/com/google/devtools/build/lib/remote/util/RxUtils.java b/src/main/java/com/google/devtools/build/lib/remote/util/RxUtils.java index 6d5260c3d717c0..8d6f22d9571b5b 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/util/RxUtils.java +++ b/src/main/java/com/google/devtools/build/lib/remote/util/RxUtils.java @@ -20,6 +20,7 @@ import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Single; import java.io.IOException; +import java.util.concurrent.CancellationException; import javax.annotation.Nullable; /** Utility methods for the Rx. * */ @@ -85,7 +86,11 @@ public static Single toTransferResult(Completable completable) { error -> { if (error instanceof IOException ioException) { return Single.just(TransferResult.error(ioException)); - } else if (error instanceof InterruptedException) { + } else if (error instanceof InterruptedException + || error instanceof CancellationException) { + // A transfer is only ever canceled as part of tearing down the operation it + // belongs to, so this is reported as an interruption rather than propagated + // downstream. return Single.just(TransferResult.interrupted()); } else { return Single.error(error);