Don't report cancelled remote transfers as errors

src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java

@@ -314,7 +314,7 @@ public Single<ValueT> 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);
                   }
                 });
           }

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;
   }
-
 }

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<TransferResult> 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);