Add overflow guard in GrpcCacheClient to detect server ignoring `read_offset`.

When retrying a download with `read_offset > 0`, some remote cache servers may ignore `read_offset` and stream from offset 0.

Extract `readOffset` and detect when received bytes exceed the expected digest size, cancel the gRPC request stream, and fail immediately with an explicit `IOException` rather than corrupting the output stream and hash.

Part of https://github.com/bazelbuild/bazel/issues/19005.

PiperOrigin-RevId: 969885342
Change-Id: Ida0841c67fb52146f0f0d2ed0f3fd76c9feb519b

src/main/java/com/google/devtools/build/lib/remote/GrpcCacheClient.java

diff --git a/src/main/java/com/google/devtools/build/lib/remote/GrpcCacheClient.java b/src/main/java/com/google/devtools/build/lib/remote/GrpcCacheClient.java
index 2808f33..bfde6d9 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/GrpcCacheClient.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/GrpcCacheClient.java
@@ -488,11 +488,12 @@
     } catch (IOException e) {
       return Futures.immediateFailedFuture(e);
     }
+    long readOffset = rawOut.getCount();
     bsAsyncStub(context, channel)
         .read(
             ReadRequest.newBuilder()
                 .setResourceName(resourceName)
-                .setReadOffset(rawOut.getCount())
+                .setReadOffset(readOffset)
                 .build(),
             new ClientResponseObserver<ReadRequest, ReadResponse>() {
               private volatile ClientCallStreamObserver<ReadRequest> requestStream;
@@ -512,6 +513,18 @@
               @Override
               public void onNext(ReadResponse readResponse) {
                 ByteString data = readResponse.getData();
+                if (!compressed && rawOut.getCount() + data.size() > digest.getSizeBytes()) {
+                  String msg =
+                      String.format(
+                          "Received more bytes than expected for digest '%s/%d'. "
+                              + "Server may have ignored read_offset.",
+                          digest.getHash(), digest.getSizeBytes());
+                  if (requestStream != null) {
+                    requestStream.cancel(msg, null);
+                  }
+                  future.setException(new IOException(msg));
+                  return;
+                }
                 try {
                   data.writeTo(out);
                 } catch (IOException e) {

src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java

diff --git a/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java b/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java
index 26e8a35..4706e58 100644
--- a/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java
+++ b/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java
@@ -1633,7 +1633,6 @@
 
     assertThat(GrpcCacheClient.isRemoteCacheOptions(options)).isFalse();
   }
-
   @Test
   public void splitBlob_serverCannotSplit_failsWithBlobNotSplittable(
       @TestParameter({"NOT_FOUND", "UNIMPLEMENTED"}) Status.Code code) throws Exception {
@@ -1655,4 +1654,31 @@
         () ->
             getFromFuture(client.splitBlob(context, digest, ChunkingFunction.Value.FAST_CDC_2020)));
   }
+
+  @Test
+  public void testDownloadFailsWhenServerIgnoresReadOffset() throws Exception {
+    Backoff mockBackoff = Mockito.mock(Backoff.class);
+    Mockito.when(mockBackoff.nextDelayMillis(any(Exception.class))).thenReturn(-1L);
+    GrpcCacheClient client = newClient(Options.getDefaults(RemoteOptions.class), () -> mockBackoff);
+    final Digest digest = DIGEST_UTIL.computeAsUtf8("abcdefg");
+    serviceRegistry.addService(
+        new ByteStreamImplBase() {
+          @Override
+          public void read(ReadRequest request, StreamObserver<ReadResponse> responseObserver) {
+            if (request.getReadOffset() == 0) {
+              // First attempt: send partial data and fail
+              responseObserver.onNext(
+                  ReadResponse.newBuilder().setData(ByteString.copyFromUtf8("abcd")).build());
+              responseObserver.onError(Status.DEADLINE_EXCEEDED.asException());
+            } else {
+              // Retry attempt: server ignores read_offset and sends full 7 bytes from offset 0
+              responseObserver.onNext(
+                  ReadResponse.newBuilder().setData(ByteString.copyFromUtf8("abcdefg")).build());
+              responseObserver.onCompleted();
+            }
+          }
+        });
+    IOException e = assertThrows(IOException.class, () -> downloadBlob(context, client, digest));
+    assertThat(e).hasMessageThat().contains("Received more bytes than expected");
+  }
 }