From 19a09bd31136d1bc6694223938dcb22df8a67ebb Mon Sep 17 00:00:00 2001 From: Dhriti Chopra Date: Tue, 27 Oct 2026 16:14:36 +0000 Subject: [PATCH] feat(storage): readSession can send fast open reads in the open request First step of fast open BiDi reads (go/fast-open-bidi-read, b/552263062). Today a read takes two round trips: one to open the session, one to send the first range. StorageDataClient.readSession(req, ctx, fastOpenConfigs) creates a read for each config and registers it in the session state before the stream is opened. The open request is now built from the session state (ObjectReadSessionState.getOpenArguments(), already used when the stream restarts), which adds the range of every registered read, so the fast open reads are part of the open request and the bytes the server returns with the open are delivered to them. The existing two-argument method opens with no fast open reads and sends exactly the same request as before. The session does not hand these reads out yet; that is the next change, which also replaces the single-range prototype fastOpenReadSession. --- .../cloud/storage/StorageDataClient.java | 28 +++- .../cloud/storage/ITFastOpenFakeTest.java | 141 ++++++++++++++++++ 2 files changed, 168 insertions(+), 1 deletion(-) create mode 100644 java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java index 43fd503a3679..0ef051bb8c4a 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java @@ -26,10 +26,12 @@ import com.google.cloud.storage.ObjectReadSessionState.OpenArguments; import com.google.cloud.storage.ReadProjectionConfig.ProjectionType; import com.google.cloud.storage.RetryContext.RetryContextProvider; +import com.google.common.collect.ImmutableList; import com.google.storage.v2.BidiReadObjectRequest; import com.google.storage.v2.BidiReadObjectResponse; import java.io.IOException; import java.time.Duration; +import java.util.List; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -57,6 +59,21 @@ private StorageDataClient( } ApiFuture readSession(BidiReadObjectRequest req, GrpcCallContext ctx) { + return readSession(req, ctx, ImmutableList.of()); + } + + /** + * Open a new {@link ObjectReadSession} using fast open: the ranges in {@code fastOpenConfigs} are + * sent in the same request that opens the session, saving a round trip before the first bytes. + * + * @param fastOpenConfigs the reads to send with the open, for example {@code + * ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(0, 1024))}; empty to only + * open the session + */ + ApiFuture readSession( + BidiReadObjectRequest req, + GrpcCallContext ctx, + List> fastOpenConfigs) { checkArgument( req.getReadRangesList().isEmpty(), "ranges included in the initial request are not supported"); @@ -67,13 +84,22 @@ ApiFuture readSession(BidiReadObjectRequest req, GrpcCallCont ObjectReadSessionStream stream = ObjectReadSessionStream.create(executor, callable, state, retryContextProvider.create()); + // Add each fast open read to the session state. The open request below is built from the + // state, so these reads are sent as part of the open. + for (ReadProjectionConfig config : fastOpenConfigs) { + long readId = state.newReadId(); + ObjectReadSessionStreamRead read = + config.cast().newRead(readId, retryContextProvider.create()); + state.putOutstandingRead(readId, read); + } + ApiFuture objectReadSessionFuture = ApiFutures.transform( stream, nowOpen -> new ObjectReadSessionImpl(executor, callable, stream, state, retryContextProvider), executor); - stream.send(req); + stream.send(state.getOpenArguments().getReq()); return objectReadSessionFuture; } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java new file mode 100644 index 000000000000..ebc753333ae5 --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java @@ -0,0 +1,141 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.ByteSizeConstants._2MiB; +import static com.google.cloud.storage.ITObjectReadSessionFakeTest.getReadRange; +import static com.google.cloud.storage.PackagePrivateMethodWorkarounds.maybeGetStorageDataClient; +import static com.google.common.truth.Truth.assertThat; +import static java.util.concurrent.TimeUnit.SECONDS; + +import com.google.api.gax.grpc.GrpcCallContext; +import com.google.cloud.storage.ITObjectReadSessionFakeTest.FakeStorage; +import com.google.cloud.storage.it.ChecksummedTestContent; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.storage.v2.BidiReadObjectRequest; +import com.google.storage.v2.BidiReadObjectResponse; +import com.google.storage.v2.BidiReadObjectSpec; +import com.google.storage.v2.BucketName; +import com.google.storage.v2.Object; +import com.google.storage.v2.ObjectRangeData; +import io.grpc.stub.StreamObserver; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import org.junit.Test; + +public final class ITFastOpenFakeTest { + + private static final byte[] ALL_OBJECT_BYTES = DataGenerator.base64Characters().genBytes(64); + private static final Object METADATA = + Object.newBuilder() + .setBucket(BucketName.format("_", "b")) + .setName("o") + .setGeneration(1) + .setSize(_2MiB) + .build(); + private static final GrpcCallContext CTX = GrpcCallContext.createDefault(); + + private static final BidiReadObjectRequest OPEN = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .build()) + .build(); + private static final BidiReadObjectRequest OPEN_WITH_READ_OF_10_TO_20 = + OPEN.toBuilder().addReadRanges(getReadRange(1, 10, 10)).build(); + + private static final ReadAsFutureBytes READ_10_TO_20 = + ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10, 10)); + + /** Every request the fake server received, in order. */ + private final List requestsReceived = + Collections.synchronizedList(new ArrayList<>()); + + @Test + public void openRequest_includesTheFastOpenRead() 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()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of(READ_10_TO_20)).get(5, SECONDS); + session.close(); + } + + assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20); + } + + @Test + public void openRequest_withoutFastOpenReads_isUnchanged() throws Exception { + FakeStorage server = fakeServer(ImmutableMap.of(OPEN, metadataOnly())); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of()).get(5, SECONDS); + session.close(); + } + + assertThat(requestsReceived).containsExactly(OPEN); + } + + /** + * A fake server that answers each request in {@code responses} and adds it to {@link + * #requestsReceived}. Any other request is answered with an error. + */ + private FakeStorage fakeServer(Map responses) { + ImmutableMap.Builder>> + behaviour = ImmutableMap.builder(); + responses.forEach( + (request, response) -> + behaviour.put( + request, + client -> { + requestsReceived.add(request); + client.onNext(response); + })); + return FakeStorage.of(behaviour.buildOrThrow()); + } + + private static BidiReadObjectResponse metadataOnly() { + return BidiReadObjectResponse.newBuilder().setMetadata(METADATA).build(); + } + + /** The response to an open request that has a read of {@code [offset, offset + length)}. */ + private static BidiReadObjectResponse metadataAndBytes(int readId, int offset, int length) { + return BidiReadObjectResponse.newBuilder() + .setMetadata(METADATA) + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData( + ChecksummedTestContent.of(ALL_OBJECT_BYTES, offset, length).asChecksummedData()) + .setReadRange(getReadRange(readId, offset, length)) + .setRangeEnd(true)) + .build(); + } +}