feat(storage): log additional bytes received from GCS in read path (#13427)

Fixes bug 475824752. Tracks the number of bytes received from the
transport layer and emits a warning log if GCS sends more bytes than
requested by the user during explicit range requests.

[Generated-by: AI]

---------

Co-authored-by: Dhriti07 <56169283+Dhriti07@users.noreply.github.com>

Author: 2 people

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannel.java

@@ -77,6 +77,7 @@ class ApiaryUnbufferedReadableByteChannel implements UnbufferedReadableByteChann
   private ScatteringByteChannel sbc;
   private boolean open;
   private boolean returnEOF;
+  private long totalBytesReadFromNetwork;
 
   // returned X-Goog-Generation header value
   private Long xGoogGeneration;
@@ -147,6 +148,7 @@ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
           }
         } else {
           totalRead += read;
+          totalBytesReadFromNetwork += read;
         }
         return totalRead;
       } catch (Exception t) {
@@ -182,9 +184,25 @@ public boolean isOpen() {
 
   @Override
   public void close() throws IOException {
-    open = false;
-    if (sbc != null) {
-      sbc.close();
+    try {
+      long requestedLength = apiaryReadRequest.getByteRangeSpec().length();
+      if (requestedLength >= 0
+          && requestedLength < ByteRangeSpec.EFFECTIVE_INFINITY
+          && totalBytesReadFromNetwork > requestedLength) {
+        java.util.logging.Logger.getLogger(ApiaryUnbufferedReadableByteChannel.class.getName())
+            .warning(
+                String.format(
+                    "storage: received %d more bytes than requested from GCS for bucket '%s',"
+                        + " object '%s'",
+                    totalBytesReadFromNetwork - requestedLength,
+                    apiaryReadRequest.getObject().getBucket(),
+                    apiaryReadRequest.getObject().getName()));
+      }
+    } finally {
+      open = false;
+      if (sbc != null) {
+        sbc.close();
+      }
     }
   }
 

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannel.java

@@ -209,37 +209,50 @@ public boolean isOpen() {
 
   @Override
   public void close() throws IOException {
-    open = false;
     try {
-      if (leftovers != null) {
-        leftovers.close();
+      long readLimit = req.getReadLimit();
+      long receivedBytes = fetchOffset.get() - req.getReadOffset();
+      if (readLimit > 0 && receivedBytes > readLimit) {
+        java.util.logging.Logger.getLogger(GapicUnbufferedReadableByteChannel.class.getName())
+            .warning(
+                String.format(
+                    "storage: received %d more bytes than requested from GCS for bucket '%s',"
+                        + " object '%s'",
+                    receivedBytes - readLimit, req.getBucket(), req.getObject()));
       }
-      ReadObjectObserver obs = readObjectObserver;
-      if (obs != null && !obs.cancellation.isDone()) {
-        obs.cancel();
-        drainQueue();
-        try {
-          // make sure our waiting doesn't lockup permanently
-          obs.cancellation.get(1, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-          Thread.currentThread().interrupt();
-          InterruptedIOException ioe = new InterruptedIOException();
-          ioe.initCause(e);
-          ioe.addSuppressed(new AsyncStorageTaskException());
-          throw ioe;
-        } catch (ExecutionException e) {
-          Throwable cause = e;
-          if (e.getCause() != null) {
-            cause = e.getCause();
+    } finally {
+      open = false;
+      try {
+        if (leftovers != null) {
+          leftovers.close();
+        }
+        ReadObjectObserver obs = readObjectObserver;
+        if (obs != null && !obs.cancellation.isDone()) {
+          obs.cancel();
+          drainQueue();
+          try {
+            // make sure our waiting doesn't lockup permanently
+            obs.cancellation.get(1, TimeUnit.SECONDS);
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            InterruptedIOException ioe = new InterruptedIOException();
+            ioe.initCause(e);
+            ioe.addSuppressed(new AsyncStorageTaskException());
+            throw ioe;
+          } catch (ExecutionException e) {
+            Throwable cause = e;
+            if (e.getCause() != null) {
+              cause = e.getCause();
+            }
+            IOException ioException = new IOException(cause);
+            ioException.addSuppressed(new AsyncStorageTaskException());
+            throw ioException;
+          } catch (TimeoutException ignore) {
           }
-          IOException ioException = new IOException(cause);
-          ioException.addSuppressed(new AsyncStorageTaskException());
-          throw ioException;
-        } catch (TimeoutException ignore) {
         }
+      } finally {
+        drainQueue();
       }
-    } finally {
-      drainQueue();
     }
   }
 

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ApiaryUnbufferedReadableByteChannelTest.java

@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
@@ -27,17 +28,25 @@
 import com.google.api.services.storage.Storage;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.ApiaryUnbufferedReadableByteChannel.ApiaryReadRequest;
+import com.google.cloud.storage.Retrying.Retrier;
 import com.google.cloud.storage.Retrying.RetrierWithAlg;
+import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
+import com.google.cloud.storage.spi.v1.AuditingHttpTransport;
 import com.google.common.collect.ImmutableMap;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
+import java.util.logging.Handler;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
 import org.junit.Test;
 
-public class ApiaryUnbufferedReadableByteChannelTest {
+public final class ApiaryUnbufferedReadableByteChannelTest {
 
   private static final byte[] CONTENT_BYTES = "Hello, World!".getBytes();
   private static final String CORRECT_CRC32C_BASE64 = "TVUQaA==";
@@ -218,4 +227,61 @@ public void testRead_partialRangeNoCrc32cValidation() throws IOException {
       assertArrayEquals(CONTENT_BYTES, out.toByteArray());
     }
   }
+
+  @Test
+  public void logsWarning_whenReceivingMoreBytesThanRequested() throws IOException {
+    byte[] content = "0123456789extra_bytes".getBytes();
+    MockLowLevelHttpResponse response =
+        new MockLowLevelHttpResponse()
+            .setContentType("application/octet-stream")
+            .setContent(content)
+            .setStatusCode(200);
+
+    AuditingHttpTransport transport = new AuditingHttpTransport(response);
+    Storage storage =
+        new Storage.Builder(transport, GsonFactory.getDefaultInstance(), null).build();
+
+    StorageObject storageObject = new StorageObject().setBucket("bucket").setName("object");
+    // Explicit range request for 10 bytes
+    ByteRangeSpec byteRangeSpec = ByteRangeSpec.relativeLength(0L, 10L);
+    ApiaryReadRequest apiaryReadRequest =
+        new ApiaryReadRequest(storageObject, ImmutableMap.of(), byteRangeSpec);
+
+    Logger logger = Logger.getLogger(ApiaryUnbufferedReadableByteChannel.class.getName());
+    List<LogRecord> records = new ArrayList<>();
+    Handler handler =
+        new Handler() {
+          @Override
+          public void publish(LogRecord record) {
+            records.add(record);
+          }
+
+          @Override
+          public void flush() {}
+
+          @Override
+          public void close() throws SecurityException {}
+        };
+    logger.addHandler(handler);
+
+    try {
+      try (UnbufferedReadableByteChannel c =
+          new ApiaryUnbufferedReadableByteChannel(
+              apiaryReadRequest,
+              storage,
+              SettableApiFuture.<StorageObject>create(),
+              Retrier.attemptOnce(),
+              Retrying.neverRetry(),
+              Hasher.defaultHasher())) {
+        ByteBuffer dst = ByteBuffer.allocate(25);
+        c.read(dst);
+      }
+
+      boolean warningLogged =
+          records.stream().anyMatch(r -> r.getMessage().contains("more bytes than requested"));
+      assertThat(warningLogged).isTrue();
+    } finally {
+      logger.removeHandler(handler);
+    }
+  }
 }

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedReadableByteChannelTest.java

@@ -330,6 +330,75 @@ public void readObject(
     }
   }
 
+  @Test
+  public void logsWarning_whenReceivingMoreBytesThanRequested()
+      throws IOException, ExecutionException, InterruptedException, TimeoutException {
+    ReadObjectRequest reqWithLimit =
+        ReadObjectRequest.newBuilder()
+            .setObject(objectName)
+            .setReadOffset(0)
+            .setReadLimit(10)
+            .build();
+
+    StorageGrpc.StorageImplBase fakeStorage =
+        new StorageGrpc.StorageImplBase() {
+          @Override
+          public void readObject(
+              ReadObjectRequest request, StreamObserver<ReadObjectResponse> responseObserver) {
+            responseObserver.onNext(resp1); // sends 10 bytes
+            responseObserver.onNext(resp2); // sends another 10 bytes (total 20 > limit 10)
+            responseObserver.onCompleted();
+          }
+        };
+
+    java.util.logging.Logger logger =
+        java.util.logging.Logger.getLogger(GapicUnbufferedReadableByteChannel.class.getName());
+    java.util.List<java.util.logging.LogRecord> records = new java.util.ArrayList<>();
+    java.util.logging.Handler handler =
+        new java.util.logging.Handler() {
+          @Override
+          public void publish(java.util.logging.LogRecord record) {
+            records.add(record);
+          }
+
+          @Override
+          public void flush() {}
+
+          @Override
+          public void close() throws SecurityException {}
+        };
+    logger.addHandler(handler);
+
+    try (FakeServer server = FakeServer.of(fakeStorage);
+        StorageClient storageClient = StorageClient.create(server.storageSettings())) {
+      Retrier retrier = TestUtils.retrierFromStorageOptions(server.getGrpcStorageOptions());
+
+      UnbufferedReadableByteChannelSession<Object> session =
+          new UnbufferedReadSession<>(
+              ApiFutures.immediateFuture(reqWithLimit),
+              (start, resultFuture) ->
+                  new GapicUnbufferedReadableByteChannel(
+                      resultFuture,
+                      new ZeroCopyServerStreamingCallable<>(
+                          storageClient.readObjectCallable(),
+                          ResponseContentLifecycleManager.noop()),
+                      start,
+                      Hasher.noop(),
+                      retrier,
+                      retryOnly(DataLossException.class)));
+      byte[] actualBytes = new byte[15];
+      try (UnbufferedReadableByteChannel c = session.open()) {
+        c.read(ByteBuffer.wrap(actualBytes));
+      }
+
+      boolean warningLogged =
+          records.stream().anyMatch(r -> r.getMessage().contains("more bytes than requested"));
+      assertThat(warningLogged).isTrue();
+    } finally {
+      logger.removeHandler(handler);
+    }
+  }
+
   private static <E extends ApiException> ResultRetryAlgorithm<?> retryOnly(Class<E> c) {
     return new BasicResultRetryAlgorithm<java.lang.Object>() {
       @Override