Is `--experimental_remote_cache_compression` being used as well? Actually the reason for this error seems to be `--experimental_remote_cache_compression`. From my early tests locally that flag reduced build times by 20-25%, but I see `BulkTransferException` in every single CI build. I've removed `--experimental_remote_cache_async` and only enabled `--experimental_remote_cache_compression` and the problem persists. This is where those errors are thrown:
https://github.com/bazelbuild/bazel/blob/72d525575a7a6c22abee5e7791f1072bb334340e/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java#L466-L476 Highlighting the relevant parts of the error:

```
	Suppressed: java.io.IOException: write incomplete: committed_size 721 for 374 total
	Suppressed: java.io.IOException: write incomplete: committed_size 144 for 125 total
	Suppressed: java.io.IOException: write incomplete: committed_size 2 for 14 total
```

I am not super familiar with the byte stream uploader (attempting to grok it now), but is it possible that the `!chunker.hasNext()` condition does not necessarily imply that the uploader was the one who completed the upload, as the comment states? (Could it be that `chunker.hasNext()` returns false in some cases when there was only 1 chunk to flush, and the server returned that the chunk already exists, returning the blob's uncompressed size as the committed size?)

Asking because those error messages look as though the server is returning uncompressed blob sizes as the committed size (e.g. a 721 byte chunk compressing to 374 bytes looks about right, and a 2 byte chunk compressing to 14 bytes looks accurate as well since the compression overhead probably requires more space than the bytes themselves). CC @coeuvre , @benjaminp , @buchgr , @Wyverald  CC @bazelbuild/remote-execution  I started a draft PR which has a failing test illustrating the issue, and I am fairly certain that the fake ByteStream impl in the test adheres to protocol, but please correct me if I'm wrong: https://github.com/bazelbuild/bazel/pull/14655

I think there is an ambiguity with the current way that the `ByteStream.Write` protocol is specified. The exact spec is here: https://github.com/bazelbuild/remote-apis/blob/636121a32fa7b9114311374e4786597d8e7a69f3/build/bazel/remote/execution/v2/remote_execution.proto#L256-L264

```
// When attempting an upload, if another client has already completed the upload
// (which may occur in the middle of a single upload if another client uploads
// the same blob concurrently), the request will terminate immediately with
// a response whose `committed_size` is the full size of the uploaded file
// (regardless of how much data was transmitted by the client). If the client
// completes the upload but the
// [Digest][build.bazel.remote.execution.v2.Digest] does not match, an
// `INVALID_ARGUMENT` error will be returned. In either case, the client should
// not attempt to retry the upload.
```

In summary, one of two things can happen in response to clients sending their final chunk:

1. The server can respond with `committed_size = client_uploaded_size` which means that the client successfully performed the upload, or
2. the server can respond with `committed_size = "full size of the uploaded file"` 

But, what does "full size of the uploaded file" mean?
* If it means "compressed size of the file, as uploaded by the other client," then the "full size" could be pretty arbitrary since there is more than one valid way to compress a file, and it's useless to return this to the client.
* But if it means "uncompressed size of the file" (i.e. `Digest.size_bytes`), then the client needs to check that, once they have received a `WriteResponse` and they have already uploaded all local chunks, then _either_ the `WriteResponse.committed_size` equals `Digest.size_bytes`, _or_ it equals the total number of bytes written by the local client (`chunker.getOffset()`) -- because the protocol gives us no way of knowing whether the successful upload of the last chunk was due to the local client's chunk being accepted, or if it was due to some other client completing their upload.

Given that the former interpretation would be useless to clients, I imagine the latter interpretation is what was originally intended. If so, I think it implies that the current implementation of `ByteStreamUploader` has a bug, because it does not check whether the `committed_size` could equal `Digest.size_bytes`. The current Bazel implementation seems to have used the first interpretation, based on [this comment](https://github.com/bazelbuild/bazel/blob/90abed5a31ae75761252c7e082572babafeb0ae6/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java#L466-L468):

```
              // Only check for matching committed size if we have completed the upload.
              // If another client did, they might have used a different compression
              // level/algorithm, so we cannot know the expected committed offset
```

CC @mostynb who may be able to help clarify the spec You're right that `WriteResponse.committed_size` is kind of useless for compressed blobs, but I don't think clients need to check that value- the server should return an error status code if and only if the write failed, and the client can check that instead of the `committed_size` field. @mostynb Thanks for your input!

I would be happy to send a PR so that Bazel stops checking the `committed_size` in the `WriteResponse`, so long as the folks familiar with the ByteStreamUploader don't see any problems with removing the check. I am not 100% sure whether this would a breaking change, but it seems like it wouldn't be. If anything, it might break some expectations on the remote cache side, so it might help to have input from some remote cache impl maintainers as well.

Separately, I am trying to figure out whether the `compressed_size` check on the Bazel side is possible for a remote cache implementation to work around in the meantime. Specifically, I'm thinking about the following edge case and not sure how to handle it:

```
1. Bazel sends chunk #1, FinishWrite=false
2. Bazel sends chunk #2, FinishWrite=true
3. Server receives chunk #1, parses ResourceName, sees that the digest already exists in the cache
   Server immediately returns `WriteResponse{committed_size = digest_size}`
4. Bazel receives WriteResponse from the server. Since it has already sent all chunks, it thinks it has completed
   the upload, and then asserts `committed_size == chunker.offset`. This fails.
```

Firstly, is this scenario is even possible? (not sure if client-streaming gRPC works the same way in Java as it does in Go, but IIUC the client does not need to wait for the server to receive and respond to each message before proceeding to send the next message)

Secondly, is the only workaround on the cache side just to wait for Bazel to upload the entire stream, regardless of whether the digest already exists in cache or not, and then return the total length of the compressed stream so that it satisfies Bazel's committed_size expectation? @mostynb FYI, I think this issue might affect `bazel-remote` as well:

https://github.com/buchgr/bazel-remote/blob/a3c6189b64cff12065750073b11a116c8c1ed00a/server/grpc_bytestream.go#L415-L421

It looks like on the first WriteRequest, if the digest already exists in cache, a `WriteResponse` with `committed_size=uncompressed_size` is effectively returned. Maybe the best we can do is update the REAPI spec to advise clients to ignore `committed_size` for compressed writes and to rely on the error status instead in that case?

I'm not sure how the early-exit mechanism is useful in practice actually. As you mentioned the client calls `Send` until it thinks it has sent all the data, and only then calls `CloseAndRecv` to get the `WriteResponse` (at least in the go bindings). At this point the client has sent all the data even if the server decided to return early. So instead of returning early the server could have discarded all the received data and just counted how much compressed data was sent and returned that number. So maybe we should instead update the REAPI spec to advise servers to do that for compressed-blobs writes instead of returning early? > instead of returning early the server could have discarded all the received data and just counted how much compressed data was sent and returned that number. 

This is exactly the workaround that I'm currently implementing :+1: 

> advise servers to do that for compressed-blobs writes instead of returning early

I think it's good advice for servers that want to be compatible with Bazel's current behavior, so it's probably worth mentioning this approach in the REAPI spec as a notable client quirk. I'm not sure whether it'd be the best advice though if/when Bazel removes the `committed_size` check, since I don't have a great intuition (nor any data) as to whether the early-exit mechanism provides significant savings in practice. Let's open an REAPI issue to get more input- can you do that or would you like me to? I'll go ahead and open one @bazel-io fork 5.1 Still seeing this, removing `--experimental_remote_cache_compression` works around it @keith: your remote cache might require an update- is it something opensource? If so which one and which version? Using Google's 🙃  cc @bergsieker This will be fixed in Bazel 5.3 😍.
`--experimental_remote_cache_compression` causes incomplete writes
### Description of the problem / feature request:

We upgraded to Bazel 5.0.0 and are currently trying out the `--experimental_remote_cache_async`. During our build, I see over 1000 messages printed out to stdout that look like this possibly caused by `BulkTransferException`s:
```
WARNING: Remote Cache: 3 errors during bulk transfer
com.google.devtools.build.lib.remote.common.BulkTransferException: 3 errors during bulk transfer
	at com.google.devtools.build.lib.remote.util.RxUtils$BulkTransferExceptionCollector.onResult(RxUtils.java:91)
	at io.reactivex.rxjava3.internal.operators.flowable.FlowableCollectSingle$CollectSubscriber.onNext(FlowableCollectSingle.java:94)
	at io.reactivex.rxjava3.internal.operators.flowable.FlowableFlatMapSingle$FlatMapSingleSubscriber.innerSuccess(FlowableFlatMapSingle.java:173)
	at io.reactivex.rxjava3.internal.operators.flowable.FlowableFlatMapSingle$FlatMapSingleSubscriber$InnerObserver.onSuccess(FlowableFlatMapSingle.java:342)
	at io.reactivex.rxjava3.internal.operators.single.SingleDoFinally$DoFinallyObserver.onSuccess(SingleDoFinally.java:73)
	at io.reactivex.rxjava3.internal.observers.ResumeSingleObserver.onSuccess(ResumeSingleObserver.java:46)
	at io.reactivex.rxjava3.internal.operators.single.SingleJust.subscribeActual(SingleJust.java:30)
	at io.reactivex.rxjava3.core.Single.subscribe(Single.java:4855)
	at io.reactivex.rxjava3.internal.operators.single.SingleResumeNext$ResumeMainSingleObserver.onError(SingleResumeNext.java:80)
	at io.reactivex.rxjava3.internal.operators.completable.CompletableToSingle$ToSingle.onError(CompletableToSingle.java:73)
	at io.reactivex.rxjava3.internal.operators.completable.CompletableCreate$Emitter.tryOnError(CompletableCreate.java:91)
	at io.reactivex.rxjava3.internal.operators.completable.CompletableCreate$Emitter.onError(CompletableCreate.java:77)
	at com.google.devtools.build.lib.remote.util.RxFutures$OnceCompletableOnSubscribe$1.onFailure(RxFutures.java:102)
	at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1074)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:771)
	at com.google.devtools.build.lib.remote.util.RxFutures$CompletableFuture.setException(RxFutures.java:276)
	at com.google.devtools.build.lib.remote.util.RxFutures$1.onError(RxFutures.java:210)
	at io.reactivex.rxjava3.internal.operators.completable.CompletableFromSingle$CompletableFromSingleObserver.onError(CompletableFromSingle.java:41)
	at io.reactivex.rxjava3.internal.operators.single.SingleCreate$Emitter.tryOnError(SingleCreate.java:95)
	at io.reactivex.rxjava3.internal.operators.single.SingleCreate$Emitter.onError(SingleCreate.java:81)
	at com.google.devtools.build.lib.remote.util.AsyncTaskCache$1.onError(AsyncTaskCache.java:306)
	at com.google.devtools.build.lib.remote.util.AsyncTaskCache$Execution.onError(AsyncTaskCache.java:197)
	at io.reactivex.rxjava3.internal.operators.completable.CompletableToSingle$ToSingle.onError(CompletableToSingle.java:73)
	at io.reactivex.rxjava3.internal.operators.completable.CompletableCreate$Emitter.tryOnError(CompletableCreate.java:91)
	at io.reactivex.rxjava3.internal.operators.completable.CompletableCreate$Emitter.onError(CompletableCreate.java:77)
	at com.google.devtools.build.lib.remote.util.RxFutures$OnceCompletableOnSubscribe$1.onFailure(RxFutures.java:102)
	at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1066)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.setFuture(AbstractFuture.java:814)
	at com.google.common.util.concurrent.AbstractCatchingFuture.run(AbstractCatchingFuture.java:115)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:771)
	at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:100)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.setFuture(AbstractFuture.java:814)
	at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.setResult(AbstractTransformFuture.java:224)
	at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.setResult(AbstractTransformFuture.java:202)
	at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:163)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:746)
	at com.google.common.util.concurrent.AbstractCatchingFuture.run(AbstractCatchingFuture.java:110)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:746)
	at com.google.common.util.concurrent.AbstractCatchingFuture.run(AbstractCatchingFuture.java:110)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:746)
	at com.google.common.util.concurrent.AbstractCatchingFuture.run(AbstractCatchingFuture.java:110)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:746)
	at com.google.common.util.concurrent.AbstractTransformFuture$TransformFuture.setResult(AbstractTransformFuture.java:247)
	at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:163)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:746)
	at com.google.devtools.build.lib.remote.util.RxFutures$CompletableFuture.set(RxFutures.java:270)
	at com.google.devtools.build.lib.remote.util.RxFutures$2.onSuccess(RxFutures.java:233)
	at io.reactivex.rxjava3.internal.operators.single.SingleFlatMap$SingleFlatMapCallback$FlatMapSingleObserver.onSuccess(SingleFlatMap.java:112)
	at io.reactivex.rxjava3.internal.operators.single.SingleUsing$UsingSingleObserver.onSuccess(SingleUsing.java:154)
	at io.reactivex.rxjava3.internal.operators.single.SingleCreate$Emitter.onSuccess(SingleCreate.java:68)
	at com.google.devtools.build.lib.remote.util.RxFutures$OnceSingleOnSubscribe$1.onSuccess(RxFutures.java:155)
	at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1080)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1213)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
	at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:746)
	at com.google.common.util.concurrent.SettableFuture.set(SettableFuture.java:47)
	at com.google.devtools.build.lib.remote.ByteStreamUploader$AsyncUpload$1.onClose(ByteStreamUploader.java:579)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at com.google.devtools.build.lib.remote.logging.LoggingInterceptor$LoggingForwardingCall$1.onClose(LoggingInterceptor.java:157)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:557)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:69)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:738)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:717)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
	Suppressed: java.io.IOException: write incomplete: committed_size 721 for 374 total
		at com.google.devtools.build.lib.remote.ByteStreamUploader$AsyncUpload.lambda$start$2(ByteStreamUploader.java:448)
		at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:213)
		at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:202)
		at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:118)
		... 51 more
	Suppressed: java.io.IOException: write incomplete: committed_size 144 for 125 total
		at com.google.devtools.build.lib.remote.ByteStreamUploader$AsyncUpload.lambda$start$2(ByteStreamUploader.java:448)
		at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:213)
		at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:202)
		at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:118)
		... 51 more
	Suppressed: java.io.IOException: write incomplete: committed_size 2 for 14 total
		at com.google.devtools.build.lib.remote.ByteStreamUploader$AsyncUpload.lambda$start$2(ByteStreamUploader.java:448)
		at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:213)
		at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:202)
		at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:118)
		... 51 more
```

### Feature requests: what underlying problem are you trying to solve with this feature?

I'm testing the new `--experimental_remote_cache_compression` flag to speed up artifacts download and upload.

### Bugs: what's the simplest, easiest way to reproduce this bug? Please provide a minimal example if possible.

I couldn't reproduce it yet. This is a pretty big project with a lot of flags. Maybe this flag is conflicting with one of these as well:
```
build --remote_timeout=600s
build --remote_upload_local_results
build --remote_download_toplevel
```

macOS 12.1

### What's the output of `bazel info release`?

`release 5.0.0`

CONTRIBUTORS src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java
CONTRIBUTORS src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java
diff --git a/CONTRIBUTORS b/CONTRIBUTORS
index a95cecd8def926..6936c814fe7bd0 100644
--- a/CONTRIBUTORS
+++ b/CONTRIBUTORS
@@ -20,6 +20,7 @@ Shreya Bhattarai <shreyax@google.com>
 Kevin Bierhoff <kmb@google.com>
 Klaas Boesche <klaasb@google.com>
 Phil Bordelon <sunfall@google.com>
+Mostyn Bramley-Moore <mostyn@antipode.se>
 Jon Brandvein <brandjon@google.com>
 Volker Braun <vbraun.name@gmail.com>
 Thomas Broyer <t.broyer@ltgt.net>
diff --git a/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java b/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java
index d4aa19061c3c34..220a3046161f7c 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java
@@ -440,14 +440,33 @@ ListenableFuture<Void> start() {
               // level/algorithm, so we cannot know the expected committed offset
               long committedSize = committedOffset.get();
               long expected = chunker.getOffset();
-              if (!chunker.hasNext() && committedSize != expected) {
+
+              if (committedSize == expected) {
+                // Both compressed and uncompressed uploads can succeed
+                // with this result.
+                return immediateVoidFuture();
+              }
+
+              if (chunker.isCompressed()) {
+                if (committedSize == -1) {
+                  // Returned early, blob already available.
+                  return immediateVoidFuture();
+                }
+
                 String message =
                     format(
-                        "write incomplete: committed_size %d for %d total",
+                        "compressed write incomplete: committed_size %d is neither -1 nor total %d",
                         committedSize, expected);
                 return Futures.immediateFailedFuture(new IOException(message));
               }
+
+              // Uncompressed upload failed.
+              String message =
+                  format(
+                      "write incomplete: committed_size %d for %d total", committedSize, expected);
+              return Futures.immediateFailedFuture(new IOException(message));
             }
+
             return immediateVoidFuture();
           },
           MoreExecutors.directExecutor());
