From a227ccfa41b5c205f24b0dcad638544bdbd9efc1 Mon Sep 17 00:00:00 2001 From: Fabian Meumertzheim Date: Mon, 10 Aug 2026 14:29:35 +0200 Subject: [PATCH 1/2] Don't report cancelled remote transfers as errors When a bulk transfer to or from the remote cache is cancelled, e.g. because the action owning it is interrupted, every one of its in-flight transfers fails with a `CancellationException`. `mergeBulkTransfer` subscribes to all of them at once via `flatMapSingle`, which terminates on the first error and hands all subsequent ones to RxJava's global error handler. `RemoteModule` installs a handler that reports those as error events, so a single cancelled bulk transfer printed a `java.util.concurrent.CancellationException: Task was cancelled` stack trace for every transfer but the first. A transfer is only ever cancelled as part of tearing down the operation it belongs to, so `toTransferResult` now maps `CancellationException` to the same result as `InterruptedException` instead of propagating it. Also replaces two `isDisposed()`-then-`onError()` sequences with `tryOnError`: the emitter can be disposed in between, in which case `onError` hands the error to the global error handler as well. --- .../build/lib/remote/util/AsyncTaskCache.java | 9 +++++---- .../build/lib/remote/util/RxFutures.java | 20 +++++++++++++------ .../build/lib/remote/util/RxUtils.java | 9 ++++++++- 3 files changed, 27 insertions(+), 11 deletions(-) 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..ab833a2ba91194 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,10 @@ public void onSuccess(@NonNull ValueT valueT) { @Override public void onError(@NonNull Throwable e) { - if (!emitter.isDisposed()) { - emitter.onError(e); - } + // The emitter can be disposed at any time, in particular right after an + // isDisposed() check, so use tryOnError: onError would hand the error to + // RxJavaPlugins' global error handler if it has been disposed in the meantime. + 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..46f1f539839e49 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,14 @@ 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. Since the Completable can be disposed at any time, in particular + * after an isDisposed() check, this is done with tryOnError: onError would hand + * the exception to RxJavaPlugins' global error handler if the Completable has + * been disposed in the meantime. */ - if (throwable instanceof CancellationException && emitter.isDisposed()) { + if (throwable instanceof CancellationException) { + emitter.tryOnError(throwable); return; } @@ -162,10 +166,14 @@ 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. Since the Single can be disposed at any time, in particular after an + * isDisposed() check, this is done with tryOnError: onError would hand the + * exception to RxJavaPlugins' global error handler if the Single has been disposed + * in the meantime. */ - if (throwable instanceof CancellationException && emitter.isDisposed()) { + if (throwable instanceof CancellationException) { + emitter.tryOnError(throwable); return; } 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..5c22fe86f68fab 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,13 @@ 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 cancelled as part of tearing down the operation it + // belongs to, so this is reported as an interruption rather than propagated + // downstream: mergeBulkTransfer subscribes to all transfers of a bulk operation at + // once and RxJava hands every error but the first one to RxJavaPlugins' global + // error handler, which Bazel reports as an error event. return Single.just(TransferResult.interrupted()); } else { return Single.error(error); From d02adb8b2be81cdcf7c6e966456289fe230eccbd Mon Sep 17 00:00:00 2001 From: Fabian Meumertzheim Date: Mon, 10 Aug 2026 14:34:38 +0200 Subject: [PATCH 2/2] Streamline comments --- .../build/lib/remote/util/AsyncTaskCache.java | 5 ++--- .../devtools/build/lib/remote/util/RxFutures.java | 11 ++--------- .../devtools/build/lib/remote/util/RxUtils.java | 6 ++---- 3 files changed, 6 insertions(+), 16 deletions(-) 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 ab833a2ba91194..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 @@ -357,9 +357,8 @@ public void onSuccess(@NonNull ValueT valueT) { @Override public void onError(@NonNull Throwable e) { - // The emitter can be disposed at any time, in particular right after an - // isDisposed() check, so use tryOnError: onError would hand the error to - // RxJavaPlugins' global error handler if it has been disposed in the meantime. + // 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 46f1f539839e49..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 @@ -93,10 +93,7 @@ public void onFailure(Throwable throwable) { * 2. Completable is disposed by downstream. * * In the second case, the CancellationException must not be propagated to - * downstream. Since the Completable can be disposed at any time, in particular - * after an isDisposed() check, this is done with tryOnError: onError would hand - * the exception to RxJavaPlugins' global error handler if the Completable has - * been disposed in the meantime. + * downstream. */ if (throwable instanceof CancellationException) { emitter.tryOnError(throwable); @@ -167,10 +164,7 @@ public void onFailure(Throwable throwable) { * 2. Single is disposed by downstream. * * In the second case, the CancellationException must not be propagated to - * downstream. Since the Single can be disposed at any time, in particular after an - * isDisposed() check, this is done with tryOnError: onError would hand the - * exception to RxJavaPlugins' global error handler if the Single has been disposed - * in the meantime. + * downstream. */ if (throwable instanceof CancellationException) { emitter.tryOnError(throwable); @@ -270,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 5c22fe86f68fab..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 @@ -88,11 +88,9 @@ public static Single toTransferResult(Completable completable) { return Single.just(TransferResult.error(ioException)); } else if (error instanceof InterruptedException || error instanceof CancellationException) { - // A transfer is only ever cancelled as part of tearing down the operation it + // 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: mergeBulkTransfer subscribes to all transfers of a bulk operation at - // once and RxJava hands every error but the first one to RxJavaPlugins' global - // error handler, which Bazel reports as an error event. + // downstream. return Single.just(TransferResult.interrupted()); } else { return Single.error(error);