feat(storage): add blobReadSession overload that takes a fast open read

Seventh step of fast open BiDi reads (go/fast-open-bidi-read, b/552263062).

Storage.blobReadSession(BlobId, ReadProjectionConfig, BlobSourceOption...)
sends the read described by the config in the same request that opens the
session. The first readAs with an equal config is given that read, saving the
round trip a separate first read would take. Only asFutureBytes,
asFutureByteString and asChannel configs are accepted, and a null config only
opens the session.

This is additive: blobReadSession(BlobId, BlobSourceOption...) is unchanged.
OtelStorageDecorator passes the config through, so fast open also works with
OpenTelemetry tracing enabled.

Author: Dhriti Chopra

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

@@ -27,6 +27,7 @@
 import static com.google.cloud.storage.Utils.ifNonNull;
 import static com.google.cloud.storage.Utils.projectNameCodec;
 import static com.google.common.base.MoreObjects.firstNonNull;
+import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiFuture;
@@ -55,6 +56,7 @@
 import com.google.cloud.storage.HmacKey.HmacKeyState;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
+import com.google.cloud.storage.ReadProjectionConfig.ProjectionType;
 import com.google.cloud.storage.Retrying.Retrier;
 import com.google.cloud.storage.Retrying.RetrierWithAlg;
 import com.google.cloud.storage.Storage.ComposeRequest.SourceBlob;
@@ -1468,6 +1470,26 @@ public Blob moveBlob(MoveBlobRequest request) {
 
   @Override
   public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption... options) {
+    return blobReadSession(id, ImmutableList.of(), options);
+  }
+
+  @Override
+  public <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id,
+      @Nullable ReadProjectionConfig<Projection> fastOpenConfig,
+      BlobSourceOption... options) {
+    if (fastOpenConfig == null) {
+      return blobReadSession(id, ImmutableList.of(), options);
+    }
+    checkArgument(
+        fastOpenConfig.getType() == ProjectionType.STREAM_READ,
+        "fastOpenConfig must be asFutureBytes(), asFutureByteString() or asChannel(), but was %s",
+        fastOpenConfig);
+    return blobReadSession(id, ImmutableList.of(fastOpenConfig), options);
+  }
+
+  private ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id, List<ReadProjectionConfig<?>> fastOpenConfigs, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options);
     Object object = codecs.blobId().encode(id);
 
@@ -1484,7 +1506,8 @@ public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption...
     BidiReadObjectRequest req = b.build();
 
     GrpcCallContext context = opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
-    ApiFuture<ObjectReadSession> session = storageDataClient.readSession(req, context);
+    ApiFuture<ObjectReadSession> session =
+        storageDataClient.readSession(req, context, fastOpenConfigs);
 
     return BlobReadSessionAdapter.wrap(session);
   }

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

@@ -1475,6 +1475,14 @@ public Blob moveBlob(MoveBlobRequest request) {
 
   @Override
   public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption... options) {
+    return blobReadSession(id, null, options);
+  }
+
+  @Override
+  public <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id,
+      @Nullable ReadProjectionConfig<Projection> fastOpenConfig,
+      BlobSourceOption... options) {
     Span blobReadSessionSpan =
         tracer
             .spanBuilder(BLOB_READ_SESSION)
@@ -1483,7 +1491,8 @@ public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption...
     try (Scope ignore1 = blobReadSessionSpan.makeCurrent()) {
       Context blobReadSessionContext = Context.current();
       Span ready = tracer.spanBuilder(BLOB_READ_SESSION + "/ready").startSpan();
-      ApiFuture<BlobReadSession> blobReadSessionApiFuture = delegate.blobReadSession(id, options);
+      ApiFuture<BlobReadSession> blobReadSessionApiFuture =
+          delegate.blobReadSession(id, fastOpenConfig, options);
       ApiFuture<BlobReadSession> futureDecorated =
           ApiFutures.transform(
               blobReadSessionApiFuture,

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

@@ -5955,6 +5955,44 @@ default ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption..
     return throwGrpcOnly(fmtMethodName("blobReadSession", BlobId.class, BlobSourceOption.class));
   }
 
+  /**
+   * Asynchronously set up a new {@link BlobReadSession} for the specified {@link BlobId} and {@code
+   * options}, sending the read described by {@code fastOpenConfig} in the same request that opens
+   * the session ("fast open"). This saves the round trip a separate first read would take.
+   *
+   * <p>The first {@link BlobReadSession#readAs(ReadProjectionConfig)} with a config equal to {@code
+   * fastOpenConfig} is given that read; any other {@code readAs} sends a new read. If the fast open
+   * read fails, the error is reported by the returned future or by that first {@code readAs}.
+   *
+   * <h4>Example of reading the first 1 KiB of an object with fast open:</h4>
+   *
+   * <pre>{@code
+   * ReadAsFutureBytes first1KiB =
+   *     ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(0, 1024));
+   * try (BlobReadSession session =
+   *     storage.blobReadSession(blobId, first1KiB).get(10, TimeUnit.SECONDS)) {
+   *   byte[] bytes = session.readAs(first1KiB).get(10, TimeUnit.SECONDS);
+   * }
+   * }</pre>
+   *
+   * @param id the blob to read from
+   * @param fastOpenConfig the read to send with the open, or {@code null} to only open the session.
+   *     Must be {@link ReadProjectionConfigs#asFutureBytes()}, {@link
+   *     ReadProjectionConfigs#asFutureByteString()} or {@link ReadProjectionConfigs#asChannel()},
+   *     optionally with a range.
+   * @since 2.77.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  @TransportCompatibility({Transport.GRPC})
+  default <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id,
+      @Nullable ReadProjectionConfig<Projection> fastOpenConfig,
+      BlobSourceOption... options) {
+    return throwGrpcOnly(
+        fmtMethodName(
+            "blobReadSession", BlobId.class, ReadProjectionConfig.class, BlobSourceOption.class));
+  }
+
   /**
    * Create a new {@link BlobAppendableUpload} for the specified {@code blobInfo} and {@code
    * options}.

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

@@ -32,7 +32,9 @@
 import com.google.api.gax.rpc.OutOfRangeException;
 import com.google.cloud.storage.ITObjectReadSessionFakeTest.FakeStorage;
 import com.google.cloud.storage.ReadProjectionConfigs.BaseConfig;
+import com.google.cloud.storage.Storage.BlobSourceOption;
 import com.google.cloud.storage.it.ChecksummedTestContent;
+import com.google.cloud.storage.otel.TestExporter;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -48,6 +50,10 @@
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -69,6 +75,7 @@ public final class ITFastOpenFakeTest {
           .setSize(_2MiB)
           .build();
   private static final GrpcCallContext CTX = GrpcCallContext.createDefault();
+  private static final BlobId BLOB_ID = BlobId.of("b", "o");
 
   private static final BidiReadObjectRequest OPEN =
       BidiReadObjectRequest.newBuilder()
@@ -442,6 +449,133 @@ public void open_whenTheConnectionDropsHalfway_fetchesOnlyTheRestOfTheRange() th
         .inOrder();
   }
 
+  @Test
+  public void blobReadSession_withAFastOpenConfig_doesNotSendAnotherRequest() throws Exception {
+    FakeStorage server =
+        fakeServer(ImmutableMap.of(OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10)));
+
+    try (FakeServer fakeServer = FakeServer.of(server);
+        Storage storage = fakeServer.getGrpcStorageOptions().getService()) {
+      try (BlobReadSession session =
+          storage.blobReadSession(BLOB_ID, READ_10_TO_20).get(5, SECONDS)) {
+        byte[] actual = session.readAs(READ_10_TO_20).get(2, SECONDS);
+        assertThat(xxd(actual)).isEqualTo(xxd(slice(10, 10)));
+      }
+    }
+
+    assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20);
+  }
+
+  @Test
+  public void blobReadSession_withTracing_doesNotSendAnotherRequest_andEndsTheSpans()
+      throws Exception {
+    FakeStorage server =
+        fakeServer(ImmutableMap.of(OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10)));
+    TestExporter exporter = new TestExporter();
+    OpenTelemetrySdk otel =
+        OpenTelemetrySdk.builder()
+            .setTracerProvider(
+                SdkTracerProvider.builder()
+                    .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+                    .build())
+            .build();
+
+    try (FakeServer fakeServer = FakeServer.of(server);
+        Storage storage =
+            fakeServer.getGrpcStorageOptions().toBuilder()
+                .setOpenTelemetry(otel)
+                .build()
+                .getService()) {
+      try (BlobReadSession session =
+          storage.blobReadSession(BLOB_ID, READ_10_TO_20).get(5, SECONDS)) {
+        byte[] actual = session.readAs(READ_10_TO_20).get(2, SECONDS);
+        assertThat(xxd(actual)).isEqualTo(xxd(slice(10, 10)));
+      }
+    }
+
+    assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20);
+    List<String> endedSpans =
+        exporter.getExportedSpans().stream().map(SpanData::getName).collect(toList());
+    assertThat(endedSpans)
+        .containsAtLeast(
+            "com.google.cloud.storage.Storage/blobReadSession",
+            "com.google.cloud.storage.Storage/blobReadSession/readAs");
+  }
+
+  @Test
+  public void blobReadSession_withANullFastOpenConfig_isAPlainOpen() throws Exception {
+    FakeStorage server = fakeServer(ImmutableMap.of(OPEN, metadataOnly()));
+
+    try (FakeServer fakeServer = FakeServer.of(server);
+        Storage storage = fakeServer.getGrpcStorageOptions().getService()) {
+      BlobReadSession session =
+          storage.blobReadSession(BLOB_ID, (ReadAsFutureBytes) null).get(5, SECONDS);
+      session.close();
+    }
+
+    assertThat(requestsReceived).containsExactly(OPEN);
+  }
+
+  @Test
+  public void blobReadSession_withASeekableChannelConfig_isRejected() throws Exception {
+    FakeStorage server = fakeServer(ImmutableMap.of());
+
+    try (FakeServer fakeServer = FakeServer.of(server);
+        Storage storage = fakeServer.getGrpcStorageOptions().getService()) {
+      IllegalArgumentException e =
+          assertThrows(
+              IllegalArgumentException.class,
+              () -> storage.blobReadSession(BLOB_ID, ReadProjectionConfigs.asSeekableChannel()));
+      assertThat(e).hasMessageThat().startsWith("fastOpenConfig must be");
+    }
+
+    assertThat(requestsReceived).isEmpty();
+  }
+
+  @Test
+  public void blobReadSession_whenTheFastOpenReadIsOutOfRange_failsTheOpen() throws Exception {
+    FakeStorage server =
+        FakeStorage.of(
+            ImmutableMap.of(
+                OPEN_WITH_READ_OF_10_TO_20,
+                client -> {
+                  requestsReceived.add(OPEN_WITH_READ_OF_10_TO_20);
+                  client.onError(Status.OUT_OF_RANGE.asRuntimeException());
+                }));
+
+    try (FakeServer fakeServer = FakeServer.of(server);
+        Storage storage = fakeServer.getGrpcStorageOptions().getService()) {
+      ApiFuture<BlobReadSession> open = storage.blobReadSession(BLOB_ID, READ_10_TO_20);
+
+      ExecutionException e = assertThrows(ExecutionException.class, () -> open.get(5, SECONDS));
+      assertThat(e).hasCauseThat().isInstanceOf(StorageException.class);
+      assertThat(((StorageException) e.getCause()).getCode()).isEqualTo(400);
+    }
+
+    assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20);
+  }
+
+  @Test
+  public void blobReadSession_withAFastOpenConfig_stillAppliesBlobSourceOptions() throws Exception {
+    BidiReadObjectRequest openWithGenerationMatchAndRead =
+        OPEN_WITH_READ_OF_10_TO_20.toBuilder()
+            .setReadObjectSpec(OPEN.getReadObjectSpec().toBuilder().setIfGenerationMatch(1))
+            .build();
+    FakeStorage server =
+        fakeServer(ImmutableMap.of(openWithGenerationMatchAndRead, metadataAndBytes(1, 10, 10)));
+
+    try (FakeServer fakeServer = FakeServer.of(server);
+        Storage storage = fakeServer.getGrpcStorageOptions().getService()) {
+      BlobReadSession session =
+          storage
+              .blobReadSession(BLOB_ID, READ_10_TO_20, BlobSourceOption.generationMatch(1))
+              .get(5, SECONDS);
+      session.close();
+    }
+
+    assertThat(requestsReceived).containsExactly(openWithGenerationMatchAndRead);
+  }
+
   /**
    * A fake server that answers each request in {@code responses} and adds it to {@link
    * #requestsReceived}. Any other request is answered with an error.

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/AbstractStorageProxy.java

@@ -41,6 +41,7 @@
 import com.google.cloud.storage.PostPolicyV4;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
+import com.google.cloud.storage.ReadProjectionConfig;
 import com.google.cloud.storage.ServiceAccount;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageBatch;
@@ -498,6 +499,12 @@ public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption...
     return delegate.blobReadSession(id, options);
   }
 
+  @Override
+  public <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id, ReadProjectionConfig<Projection> fastOpenConfig, BlobSourceOption... options) {
+    return delegate.blobReadSession(id, fastOpenConfig, options);
+  }
+
   @Override
   public BlobAppendableUpload blobAppendableUpload(
       BlobInfo blobInfo, BlobAppendableUploadConfig uploadConfig, BlobWriteOption... options) {