diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/FeatureTracking.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/FeatureTracking.java new file mode 100644 index 000000000000..0df037b2d800 --- /dev/null +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/FeatureTracking.java @@ -0,0 +1,86 @@ +/* + * 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 com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.primitives.Ints; +import java.util.Base64; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import javax.annotation.concurrent.ThreadSafe; + +/** + * Utilities for reporting {@link TrackedFeature client feature usage} to the service. + * + *

Encoding contract (shared across GCS SDKs): + * + *

+ */ +final class FeatureTracking { + + /** Name of the request header / gRPC metadata key carrying the encoded feature mask. */ + static final String HEADER = "x-goog-storage-java-features"; + + private static final Base64.Encoder ENCODER = Base64.getUrlEncoder(); + + private FeatureTracking() {} + + /** Encodes {@code mask} as a 4 byte big-endian uint32, URL-safe base64 with padding. */ + @VisibleForTesting + static String encode(int mask) { + return ENCODER.encodeToString(Ints.toByteArray(mask)); + } + + /** + * Returns the header map to merge into a request's extra headers: an empty map when {@code mask + * == 0}, otherwise a single entry map for {@link #HEADER}. + */ + private static ImmutableMap> headers(int mask) { + if (mask == 0) { + return ImmutableMap.of(); + } + return ImmutableMap.of(HEADER, ImmutableList.of(encode(mask))); + } + + /** Monotonically growing feature bitmask. Bits can be set but never cleared. */ + @ThreadSafe + static final class Mask { + private final AtomicInteger bits = new AtomicInteger(); + + void set(TrackedFeature feature) { + bits.accumulateAndGet(feature.bitMask(), (a, b) -> a | b); + } + + @VisibleForTesting + int get() { + return bits.get(); + } + + /** Returns the headers to attach to a request, see {@link FeatureTracking#headers(int)}. */ + ImmutableMap> headers() { + return FeatureTracking.headers(bits.get()); + } + } +} diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionState.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionState.java index 86656219a23a..ac3eb2254a03 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionState.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionState.java @@ -56,6 +56,9 @@ final class ObjectReadSessionState { private final AtomicReference<@MonotonicNonNull Object> metadata; private final AtomicLong readIdSeq; + /** Shared between a parent state and all of its children, see {@link #forkChild()}. */ + private final FeatureTracking.Mask featureMask; + @GuardedBy("this.lock") // https://errorprone.info/bugpattern/GuardedBy private final Map> outstandingReads; @@ -69,7 +72,8 @@ final class ObjectReadSessionState { new AtomicLong(1), new AtomicReference<>(), new AtomicReference<>(), - new AtomicReference<>()); + new AtomicReference<>(), + new FeatureTracking.Mask()); } private ObjectReadSessionState( @@ -78,25 +82,37 @@ private ObjectReadSessionState( AtomicLong readIdSeq, AtomicReference<@Nullable BidiReadHandle> bidiReadHandle, AtomicReference<@Nullable String> routingToken, - AtomicReference<@MonotonicNonNull Object> metadata) { + AtomicReference<@MonotonicNonNull Object> metadata, + FeatureTracking.Mask featureMask) { this.baseContext = baseContext; this.openRequest = openRequest; this.bidiReadHandle = bidiReadHandle; this.routingToken = routingToken; this.metadata = metadata; this.readIdSeq = readIdSeq; + this.featureMask = featureMask; this.outstandingReads = new HashMap<>(); this.lock = new ReentrantLock(); } + /** + * Forks a new state for an additional stream of the same read session. + * + *

A fork means the session is now backed by more than one BidiReadObject stream, so this marks + * {@link TrackedFeature#MULTISTREAM_IN_MRD} in the feature mask shared by the parent and all + * children. Any stream opened (or re-opened) by this session afterwards carries the {@link + * FeatureTracking#HEADER} header. + */ ObjectReadSessionState forkChild() { + featureMask.set(TrackedFeature.MULTISTREAM_IN_MRD); return new ObjectReadSessionState( baseContext, openRequest, readIdSeq, new AtomicReference<>(bidiReadHandle.get()), new AtomicReference<>(routingToken.get()), - new AtomicReference<>(metadata.get())); + new AtomicReference<>(metadata.get()), + featureMask); } boolean canHandleNewRead(ObjectReadSessionStreamRead newRead) { @@ -136,14 +152,17 @@ OpenArguments getOpenArguments() { .forEach(b::addReadRanges); ImmutableMap> headers = - ImmutableMap.of( - "x-goog-request-params", - ImmutableList.of( - Stream.of( - "bucket=" + spec.getBucket(), - routingToken != null ? "routing_token=" + routingToken : null) - .filter(Objects::nonNull) - .collect(Collectors.joining("&")))); + ImmutableMap.>builder() + .put( + "x-goog-request-params", + ImmutableList.of( + Stream.of( + "bucket=" + spec.getBucket(), + routingToken != null ? "routing_token=" + routingToken : null) + .filter(Objects::nonNull) + .collect(Collectors.joining("&")))) + .putAll(featureMask.headers()) + .build(); return OpenArguments.of(baseContext.withExtraHeaders(headers), b.build()); } finally { lock.unlock(); diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeature.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeature.java new file mode 100644 index 000000000000..4fefd0f758e1 --- /dev/null +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeature.java @@ -0,0 +1,45 @@ +/* + * 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; + +/** + * A client feature whose usage is reported to the service via the {@link FeatureTracking#HEADER + * x-goog-storage-java-features} request header. + * + *

Each feature is assigned a stable bit position in a 32-bit mask. Bit positions are specific to + * the Java SDK but MUST NOT be renumbered or reused once released. This enum is the single source + * of truth and its constants are parsed by external tooling; the {@code NAME(bit)} shape of each + * constant must stay intact. + */ +enum TrackedFeature { + /** + * A {@link BlobReadSession} opened more than one BidiReadObject stream to serve concurrent reads + * which could not share a single stream. + */ + MULTISTREAM_IN_MRD(0); + + private final int bitPosition; + + TrackedFeature(int bitPosition) { + this.bitPosition = bitPosition; + } + + /** Returns the 32-bit mask with only this feature's bit set. */ + int bitMask() { + return 1 << bitPosition; + } +} diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java index 65c6ab13cb9b..e5cc3fa55dc4 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java @@ -2110,6 +2110,7 @@ static final class Headers extends RpcOptVal> "X-Goog-Encryption-Key-Sha256", "X-Goog-Gcs-Idempotency-Token", "X-Goog-Request-Params", + FeatureTracking.HEADER, "X-Goog-User-Project", "X-HTTP-Method-Override", "X-Upload-Content-Length", diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/FeatureTrackingTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/FeatureTrackingTest.java new file mode 100644 index 000000000000..1062e2a0290b --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/FeatureTrackingTest.java @@ -0,0 +1,190 @@ +/* + * 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.TestUtils.assertAll; +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.google.cloud.storage.Storage.BlobSourceOption; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.Test; + +public final class FeatureTrackingTest { + + private static final int MULTISTREAM_BIT = TrackedFeature.MULTISTREAM_IN_MRD.bitMask(); + + @Test + public void encode_knownVectors() throws Exception { + assertAll( + // the service tolerates an all-zero value, even though the client never sends it + () -> assertThat(FeatureTracking.encode(0)).isEqualTo("AAAAAA=="), + () -> assertThat(FeatureTracking.encode(1)).isEqualTo("AAAAAQ=="), + () -> assertThat(FeatureTracking.encode(2)).isEqualTo("AAAAAg=="), + () -> assertThat(FeatureTracking.encode(3)).isEqualTo("AAAAAw=="), + () -> assertThat(FeatureTracking.encode(4)).isEqualTo("AAAABA=="), + // 0xF8 and 0xFC produce the 62nd and 63rd base64 alphabet characters, which must be the + // URL-safe '-' and '_' rather than the standard '+' and '/' + () -> assertThat(FeatureTracking.encode(0xF8)).isEqualTo("AAAA-A=="), + () -> assertThat(FeatureTracking.encode(0xFC)).isEqualTo("AAAA_A=="), + // high bit set; verify unsigned, big-endian interpretation + () -> assertThat(FeatureTracking.encode(0x80000000)).isEqualTo("gAAAAA=="), + () -> assertThat(FeatureTracking.encode(0xFFFFFFFF)).isEqualTo("_____w==")); + } + + @Test + public void encode_everySingleBit_isUrlSafeWithPadding() { + for (int shift = 0; shift < 32; shift++) { + String encoded = FeatureTracking.encode(1 << shift); + assertThat(encoded).hasLength(8); + assertThat(encoded).doesNotContain("+"); + assertThat(encoded).doesNotContain("/"); + assertThat(encoded).endsWith("=="); + } + } + + @Test + public void mask_headers_emptyWhenNoFeatureSet() { + FeatureTracking.Mask mask = new FeatureTracking.Mask(); + + assertThat(mask.get()).isEqualTo(0); + assertThat(mask.headers()).isEmpty(); + } + + @Test + public void mask_headers_usesWireHeaderNameAndEncodedValue() { + FeatureTracking.Mask mask = new FeatureTracking.Mask(); + + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); + + assertThat(mask.headers()) + .containsExactly("x-goog-storage-java-features", ImmutableList.of("AAAAAQ==")); + } + + @Test + public void mask_set_isIdempotentAndMonotonic() { + FeatureTracking.Mask mask = new FeatureTracking.Mask(); + + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); + + assertThat(mask.get()).isEqualTo(MULTISTREAM_BIT); + assertThat(mask.headers()) + .containsExactly( + FeatureTracking.HEADER, ImmutableList.of(FeatureTracking.encode(MULTISTREAM_BIT))); + } + + /** + * Several threads concurrently set features on a single shared mask; the result must be the OR of + * every feature set, with no lost updates. + */ + @Test + public void mask_set_isSafeUnderConcurrentUpdates() throws Exception { + int threadCount = 8; + int iterationsPerThread = 1_000; + FeatureTracking.Mask mask = new FeatureTracking.Mask(); + CountDownLatch start = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + try { + List> futures = new ArrayList<>(); + for (int i = 0; i < threadCount; i++) { + futures.add( + executor.submit( + () -> { + start.await(); + for (int j = 0; j < iterationsPerThread; j++) { + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); + } + return null; + })); + } + start.countDown(); + for (Future f : futures) { + f.get(30, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + + int expected = 0; + for (TrackedFeature feature : TrackedFeature.values()) { + expected |= feature.bitMask(); + } + assertThat(mask.get()).isEqualTo(expected); + } + + @Test + public void trackedFeature_bitMasksAreDistinctSingleBits() { + Set seen = new HashSet<>(); + for (TrackedFeature feature : TrackedFeature.values()) { + assertThat(Integer.bitCount(feature.bitMask())).isEqualTo(1); + seen.add(feature.bitMask()); + } + assertThat(seen).hasSize(TrackedFeature.values().length); + } + + /** The wire value of each feature bit is a cross-SDK contract; pin it. */ + @Test + public void trackedFeature_multistreamInMrd_isBitZero() { + assertThat(TrackedFeature.MULTISTREAM_IN_MRD.bitMask()).isEqualTo(1); + } + + @Test + public void extraHeaders_featureTrackingHeader_throws() { + ImmutableMap headers = ImmutableMap.of(FeatureTracking.HEADER, "AAAAAQ=="); + + IllegalArgumentException thrown = + assertThrows(IllegalArgumentException.class, () -> UnifiedOpts.extraHeaders(headers)); + + assertThat(thrown).hasMessageThat().contains("Disallowed headers"); + assertThat(thrown).hasMessageThat().contains("x-goog-storage-java-features"); + } + + @Test + public void extraHeaders_featureTrackingHeaderMixedCase_throws() { + ImmutableMap headers = ImmutableMap.of("X-Goog-Storage-Java-Features", "a"); + + IllegalArgumentException thrown = + assertThrows(IllegalArgumentException.class, () -> UnifiedOpts.extraHeaders(headers)); + + assertThat(thrown).hasMessageThat().contains("x-goog-storage-java-features"); + } + + /** + * The public option factories are transport agnostic and validate eagerly, so a blocked header is + * rejected before it can reach either the gRPC or the HTTP/JSON transport. + */ + @Test + public void blobSourceOption_extraHeaders_featureTrackingHeader_throws() { + ImmutableMap headers = ImmutableMap.of(FeatureTracking.HEADER, "AAAAAQ=="); + + IllegalArgumentException thrown = + assertThrows(IllegalArgumentException.class, () -> BlobSourceOption.extraHeaders(headers)); + + assertThat(thrown).hasMessageThat().contains("x-goog-storage-java-features"); + } +} diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java index 1cb32704982b..5d77cf30646a 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java @@ -43,6 +43,7 @@ import com.google.cloud.storage.Storage.BlobSourceOption; import com.google.cloud.storage.StorageDataClient.FastOpenObjectReadSession; import com.google.cloud.storage.ZeroCopySupport.DisposableByteString; +import com.google.cloud.storage.it.AssertRequestHeaders.FilteringPolicy; import com.google.cloud.storage.it.ChecksummedTestContent; import com.google.cloud.storage.it.GrpcPlainRequestLoggingInterceptor; import com.google.cloud.storage.it.GrpcRequestAuditing; @@ -133,6 +134,12 @@ public final class ITObjectReadSessionFakeTest { Metadata.Key.of("x-goog-gcs-idempotency-token", Metadata.ASCII_STRING_MARSHALLER); private static final Metadata.Key X_GOOG_USER_PROJECT = Metadata.Key.of("x-goog-user-project", Metadata.ASCII_STRING_MARSHALLER); + private static final Metadata.Key X_GOOG_STORAGE_JAVA_FEATURES = + Metadata.Key.of(FeatureTracking.HEADER, Metadata.ASCII_STRING_MARSHALLER); + + /** URL-safe base64 of a mask with only {@link TrackedFeature#MULTISTREAM_IN_MRD} set. */ + private static final String MULTISTREAM_IN_MRD_HEADER_VALUE = "AAAAAQ=="; + private static final Correspondence IS_UUID = Correspondence.transforming(UUID::fromString, "is a UUID"); @@ -1309,6 +1316,371 @@ public void onCompleted() { } } + /** + * Validates that a session served by a single stream never reports {@code MULTISTREAM_IN_MRD}. + * + *

    + *
  1. Open blob descriptor (stream 1) + *
  2. read bytes 10-20 as a future, then bytes 20-30 as a future; both can share stream 1 + *
  3. expect exactly one stream, and no request carried {@code x-goog-storage-java-features} + *
+ */ + @Test + public void singleStreamSessionNeverSendsFeatureTrackingHeader() throws Exception { + ChecksummedTestContent content1 = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 10); + ChecksummedTestContent content2 = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 20, 10); + BidiReadObjectRequest req2 = read(1, 10, 10); + BidiReadObjectResponse res2 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content1.asChecksummedData()) + .setReadRange(getReadRange(1, 10, content1)) + .setRangeEnd(true) + .build()) + .build(); + BidiReadObjectRequest req3 = read(2, 20, 10); + BidiReadObjectResponse res3 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content2.asChecksummedData()) + .setReadRange(getReadRange(2, 20, content2)) + .setRangeEnd(true) + .build()) + .build(); + + AtomicInteger bidiReadObjectCount = new AtomicInteger(); + StorageImplBase fakeStorage = + new StorageImplBase() { + @Override + public StreamObserver bidiReadObject( + StreamObserver respond) { + bidiReadObjectCount.getAndIncrement(); + return new AbstractObserver(respond) { + @Override + public void onNext(BidiReadObjectRequest request) { + if (request.equals(REQ_OPEN)) { + respond.onNext(RES_OPEN); + } else if (request.equals(req2)) { + respond.onNext(res2); + } else if (request.equals(req3)) { + respond.onNext(res3); + } else { + respond.onError(TestUtils.apiException(Code.UNIMPLEMENTED, "Unexpected request")); + } + } + }; + } + }; + + GrpcRequestAuditing requestAuditing = new GrpcRequestAuditing(); + try (FakeServer fakeServer = FakeServer.of(fakeStorage); + Storage storage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(1).build()) + .setGrpcInterceptorProvider(requestAuditing) + .build() + .getService()) { + + BlobId id = BlobId.of("b", "o"); + ApiFuture futureObjectDescriptor = storage.blobReadSession(id); + + byte[] bytes1; + byte[] bytes2; + try (BlobReadSession bd = futureObjectDescriptor.get(5, TimeUnit.SECONDS)) { + bytes1 = + bd.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10L, 10L))) + .get(3, TimeUnit.SECONDS); + bytes2 = + bd.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(20L, 10L))) + .get(3, TimeUnit.SECONDS); + } + + assertAll( + () -> assertThat(bidiReadObjectCount.get()).isEqualTo(1), + () -> requestAuditing.assertRequestHeader(X_GOOG_STORAGE_JAVA_FEATURES).isEmpty(), + () -> requestAuditing.assertRequestHeader(X_GOOG_REQUEST_PARAMS).isNotEmpty(), + () -> assertThat(xxd(bytes1)).isEqualTo(xxd(content1.getBytes())), + () -> assertThat(xxd(bytes2)).isEqualTo(xxd(content2.getBytes()))); + } + } + + /** + * Validates that only a transparently forked stream carries the feature tracking header. + * + *
    + *
  1. Open blob descriptor (stream 1, no feature header) + *
  2. start a channel read of bytes 10-20 (stream 1) + *
  3. while the channel read is outstanding, request bytes 10-20 as a future, which can not + * share the stream and therefore forks a second stream (stream 2) + *
  4. expect stream 2 to carry {@code x-goog-storage-java-features} with the {@code + * MULTISTREAM_IN_MRD} bit set, and stream 1 to not have carried it + *
+ */ + @Test + public void forkedStreamCarriesFeatureTrackingHeader() throws Exception { + ChecksummedTestContent content = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 20); + ChecksummedTestContent content1 = ChecksummedTestContent.of(content.getBytes(), 0, 10); + ChecksummedTestContent content2 = ChecksummedTestContent.of(content.getBytes(), 10, 10); + BidiReadObjectRequest req2 = read(1, 10, 20); + BidiReadObjectResponse res2_1 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content1.asChecksummedData()) + .setReadRange(getReadRange(1, 10, content1)) + .build()) + .build(); + BidiReadObjectResponse res2_2 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content2.asChecksummedData()) + .setReadRange(getReadRange(1, 20, content2)) + .setRangeEnd(true) + .build()) + .build(); + + BidiReadObjectRequest req3 = + read(2, 10, 20).toBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(METADATA.getGeneration()) + .build()) + .build(); + BidiReadObjectResponse res3 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content.asChecksummedData()) + .setReadRange(getReadRange(2, 10, content)) + .setRangeEnd(true) + .build()) + .build(); + + AtomicInteger bidiReadObjectCount = new AtomicInteger(); + CountDownLatch cdl = new CountDownLatch(1); + + StorageImplBase fakeStorage = + new StorageImplBase() { + @Override + public StreamObserver bidiReadObject( + StreamObserver respond) { + bidiReadObjectCount.getAndIncrement(); + return new AbstractObserver(respond) { + @Override + public void onNext(BidiReadObjectRequest request) { + if (request.equals(REQ_OPEN)) { + respond.onNext(RES_OPEN); + } else if (request.equals(req2)) { + respond.onNext(res2_1); + try { + cdl.await(); + } catch (InterruptedException e) { + respond.onError(TestUtils.apiException(Code.UNIMPLEMENTED, e.getMessage())); + } + respond.onNext(res2_2); + respond.onCompleted(); + } else if (request.equals(req3)) { + respond.onNext(res3); + respond.onCompleted(); + cdl.countDown(); + } else { + respond.onError(TestUtils.apiException(Code.UNIMPLEMENTED, "Unexpected request")); + } + } + }; + } + }; + + GrpcRequestAuditing requestAuditing = new GrpcRequestAuditing(); + try (FakeServer fakeServer = FakeServer.of(fakeStorage); + Storage storage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(1).build()) + .setGrpcInterceptorProvider(requestAuditing) + .build() + .getService()) { + + BlobId id = BlobId.of("b", "o"); + ApiFuture futureObjectDescriptor = storage.blobReadSession(id); + + ByteBuffer buf = ByteBuffer.allocate(50); + byte[] bytes; + try (BlobReadSession bd = futureObjectDescriptor.get(5, TimeUnit.SECONDS)) { + // only the initial stream has been opened so far, no feature header expected + requestAuditing.assertRequestHeader(X_GOOG_STORAGE_JAVA_FEATURES).isEmpty(); + try (ScatteringByteChannel c = + bd.readAs(ReadProjectionConfigs.asChannel().withRangeSpec(RangeSpec.of(10L, 20L)))) { + buf.limit(5); + Buffers.fillFrom(buf, c); + buf.limit(buf.capacity()); + ApiFuture future = + bd.readAs( + ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10L, 20L))); + bytes = future.get(3, TimeUnit.SECONDS); + Buffers.fillFrom(buf, c); + } + } + + byte[] futureBytes = bytes; + assertAll( + () -> assertThat(bidiReadObjectCount.get()).isEqualTo(2), + () -> + // exactly one of the two streams (the forked one) carried the header + requestAuditing + .assertRequestHeader(FeatureTracking.HEADER, FilteringPolicy.NO_FILTER) + .containsExactly(MULTISTREAM_IN_MRD_HEADER_VALUE), + () -> assertThat(xxd(buf)).isEqualTo(xxd(content.getBytes())), + () -> assertThat(xxd(futureBytes)).isEqualTo(xxd(content.getBytes()))); + } + } + + /** + * Validates that once a session has forked, a retry re-open of the original stream also + * carries the feature tracking header. + * + *
    + *
  1. Open blob descriptor (stream 1, no feature header) + *
  2. start a channel read of bytes 10-30 (stream 1); the server delivers bytes 10-20 and then + * holds the stream open + *
  3. while the channel read is outstanding, request bytes 10-30 as a future, which can not + * share the stream and therefore forks a second stream (stream 2, with feature header) + *
  4. once stream 2 has completed, the server fails stream 1 with a retryable UNAVAILABLE + *
  5. expect the client to re-open stream 1 (stream 3) for the remaining bytes 20-30, carrying + * the feature header, and the channel read to complete successfully + *
+ */ + @Test + public void streamReopenedAfterForkCarriesFeatureTrackingHeader() throws Exception { + ChecksummedTestContent content = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 20); + ChecksummedTestContent content1 = ChecksummedTestContent.of(content.getBytes(), 0, 10); + ChecksummedTestContent content2 = ChecksummedTestContent.of(content.getBytes(), 10, 10); + BidiReadObjectRequest req2 = read(1, 10, 20); + BidiReadObjectResponse res2 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content1.asChecksummedData()) + .setReadRange(getReadRange(1, 10, content1)) + .build()) + .build(); + + BidiReadObjectSpec resolvedSpec = + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(METADATA.getGeneration()) + .build(); + // forked stream: open + read id 2 in one request + BidiReadObjectRequest req3 = + read(2, 10, 20).toBuilder().setReadObjectSpec(resolvedSpec).build(); + BidiReadObjectResponse res3 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content.asChecksummedData()) + .setReadRange(getReadRange(2, 10, content)) + .setRangeEnd(true) + .build()) + .build(); + // re-open of the original stream: open + resume read id 1 from the current offset + BidiReadObjectRequest req4 = + read(1, 20, 10).toBuilder().setReadObjectSpec(resolvedSpec).build(); + BidiReadObjectResponse res4 = + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content2.asChecksummedData()) + .setReadRange(getReadRange(1, 20, content2)) + .setRangeEnd(true) + .build()) + .build(); + + AtomicInteger bidiReadObjectCount = new AtomicInteger(); + CountDownLatch forkedStreamDone = new CountDownLatch(1); + + StorageImplBase fakeStorage = + new StorageImplBase() { + @Override + public StreamObserver bidiReadObject( + StreamObserver respond) { + bidiReadObjectCount.getAndIncrement(); + return new AbstractObserver(respond) { + @Override + public void onNext(BidiReadObjectRequest request) { + if (request.equals(REQ_OPEN)) { + respond.onNext(RES_OPEN); + } else if (request.equals(req2)) { + respond.onNext(res2); + try { + forkedStreamDone.await(); + } catch (InterruptedException e) { + respond.onError(TestUtils.apiException(Code.UNIMPLEMENTED, e.getMessage())); + return; + } + respond.onError(Status.UNAVAILABLE.asRuntimeException()); + } else if (request.equals(req3)) { + respond.onNext(res3); + respond.onCompleted(); + forkedStreamDone.countDown(); + } else if (request.equals(req4)) { + respond.onNext(res4); + respond.onCompleted(); + } else { + respond.onError(TestUtils.apiException(Code.UNIMPLEMENTED, "Unexpected request")); + } + } + }; + } + }; + + GrpcRequestAuditing requestAuditing = new GrpcRequestAuditing(); + try (FakeServer fakeServer = FakeServer.of(fakeStorage); + Storage storage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(3).build()) + .setGrpcInterceptorProvider(requestAuditing) + .build() + .getService()) { + + BlobId id = BlobId.of("b", "o"); + ApiFuture futureObjectDescriptor = storage.blobReadSession(id); + + ByteBuffer buf = ByteBuffer.allocate(50); + byte[] bytes; + try (BlobReadSession bd = futureObjectDescriptor.get(5, TimeUnit.SECONDS)) { + try (ScatteringByteChannel c = + bd.readAs(ReadProjectionConfigs.asChannel().withRangeSpec(RangeSpec.of(10L, 20L)))) { + buf.limit(5); + Buffers.fillFrom(buf, c); + buf.limit(buf.capacity()); + ApiFuture future = + bd.readAs( + ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10L, 20L))); + bytes = future.get(3, TimeUnit.SECONDS); + Buffers.fillFrom(buf, c); + } + } + + byte[] futureBytes = bytes; + assertAll( + () -> assertThat(bidiReadObjectCount.get()).isEqualTo(3), + () -> + // the forked stream and the re-opened original stream carried the header, the + // initial open did not + requestAuditing + .assertRequestHeader(FeatureTracking.HEADER, FilteringPolicy.NO_FILTER) + .containsExactly( + MULTISTREAM_IN_MRD_HEADER_VALUE, MULTISTREAM_IN_MRD_HEADER_VALUE), + () -> assertThat(xxd(buf)).isEqualTo(xxd(content.getBytes())), + () -> assertThat(xxd(futureBytes)).isEqualTo(xxd(content.getBytes()))); + } + } + @Test public void gettingSessionFromFastOpenKeepsTheSessionOpenUntilClosed() throws Exception { ChecksummedTestContent expected = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 30); diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStateTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStateTest.java index 2714d177c82d..5e8aea9cab7e 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStateTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ObjectReadSessionStateTest.java @@ -41,6 +41,18 @@ public final class ObjectReadSessionStateTest { + private static final BidiReadObjectRequest BASE_REQUEST = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket("projects/_/buckets/my-bucket") + .setObject("my-object")) + .build(); + private static final ImmutableList MULTISTREAM_HEADER_VALUE = + ImmutableList.of("AAAAAQ=="); + private static final ImmutableList REQUEST_PARAMS_HEADER_VALUE = + ImmutableList.of("bucket=projects/_/buckets/my-bucket"); + @Test public void getOpenArguments_includesAllRelevantModifications() throws Exception { BidiReadObjectRequest base = @@ -234,4 +246,91 @@ public void canHandleNewRead() throws Exception { () -> assertThat(state2.canHandleNewRead(streaming2)).isFalse()); } } + + @Test + public void featureTrackingHeader_absentBeforeFork() { + ObjectReadSessionState state = + new ObjectReadSessionState(GrpcCallContext.createDefault(), BASE_REQUEST); + + Map> headers = state.getOpenArguments().getCtx().getExtraHeaders(); + + assertThat(headers).containsExactly("x-goog-request-params", REQUEST_PARAMS_HEADER_VALUE); + } + + @Test + public void featureTrackingHeader_presentOnChildAfterFork() { + ObjectReadSessionState parent = + new ObjectReadSessionState(GrpcCallContext.createDefault(), BASE_REQUEST); + + ObjectReadSessionState child = parent.forkChild(); + + Map> headers = child.getOpenArguments().getCtx().getExtraHeaders(); + assertThat(headers) + .containsExactly( + "x-goog-request-params", REQUEST_PARAMS_HEADER_VALUE, + "x-goog-storage-java-features", MULTISTREAM_HEADER_VALUE); + } + + /** Any subsequent (re)open of the parent stream must also report the feature. */ + @Test + public void featureTrackingHeader_presentOnParentAfterFork() { + ObjectReadSessionState parent = + new ObjectReadSessionState(GrpcCallContext.createDefault(), BASE_REQUEST); + + ObjectReadSessionState unused = parent.forkChild(); + + Map> headers = parent.getOpenArguments().getCtx().getExtraHeaders(); + assertThat(headers) + .containsExactly( + "x-goog-request-params", REQUEST_PARAMS_HEADER_VALUE, + "x-goog-storage-java-features", MULTISTREAM_HEADER_VALUE); + } + + @Test + public void featureTrackingHeader_secondForkDoesNotDuplicateOrAlterValue() throws Exception { + ObjectReadSessionState parent = + new ObjectReadSessionState(GrpcCallContext.createDefault(), BASE_REQUEST); + + ObjectReadSessionState child1 = parent.forkChild(); + ObjectReadSessionState child2 = parent.forkChild(); + ObjectReadSessionState grandChild = child1.forkChild(); + + assertAll( + () -> + assertThat(parent.getOpenArguments().getCtx().getExtraHeaders()) + .containsEntry("x-goog-storage-java-features", MULTISTREAM_HEADER_VALUE), + () -> + assertThat(child1.getOpenArguments().getCtx().getExtraHeaders()) + .containsEntry("x-goog-storage-java-features", MULTISTREAM_HEADER_VALUE), + () -> + assertThat(child2.getOpenArguments().getCtx().getExtraHeaders()) + .containsEntry("x-goog-storage-java-features", MULTISTREAM_HEADER_VALUE), + () -> + assertThat(grandChild.getOpenArguments().getCtx().getExtraHeaders()) + .containsEntry("x-goog-storage-java-features", MULTISTREAM_HEADER_VALUE)); + } + + /** + * User supplied extra headers live on the base context; the per-open headers must be merged with + * them rather than replace them, and the routing token must still be reflected in {@code + * x-goog-request-params}. + */ + @Test + public void featureTrackingHeader_coexistsWithUserExtraHeadersAndRequestParams() { + GrpcCallContext baseContext = + GrpcCallContext.createDefault() + .withExtraHeaders(ImmutableMap.of("x-user-header", ImmutableList.of("user-value"))); + ObjectReadSessionState parent = new ObjectReadSessionState(baseContext, BASE_REQUEST); + parent.setRoutingToken("token-1"); + + ObjectReadSessionState child = parent.forkChild(); + + Map> headers = child.getOpenArguments().getCtx().getExtraHeaders(); + assertThat(headers) + .containsExactly( + "x-user-header", ImmutableList.of("user-value"), + "x-goog-request-params", + ImmutableList.of("bucket=projects/_/buckets/my-bucket&routing_token=token-1"), + "x-goog-storage-java-features", MULTISTREAM_HEADER_VALUE); + } } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java index 38a40fd0546b..9c96b55a7ff6 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/UnifiedOptsGrpcTest.java @@ -1388,6 +1388,7 @@ public void headersOnBlocklistResultInIllegalArgumentException() { .put("X-Goog-Encryption-Key-Sha256", "a") .put("X-Goog-Meta-A", "a") .put("X-Goog-Request-Params", "a") + .put("X-Goog-Storage-Java-Features", "a") .put("X-Goog-User-Project", "a") .put("X-HTTP-Method-Override", "a") .put("X-Upload-Content-Length", "a") @@ -1404,8 +1405,8 @@ public void headersOnBlocklistResultInIllegalArgumentException() { + " x-goog-copy-source-encryption-algorithm, x-goog-copy-source-encryption-key," + " x-goog-copy-source-encryption-key-sha256, x-goog-encryption-algorithm," + " x-goog-encryption-key, x-goog-encryption-key-sha256, x-goog-meta-a," - + " x-goog-request-params, x-goog-user-project, x-http-method-override," - + " x-upload-content-length, x-upload-content-type]"); + + " x-goog-request-params, x-goog-storage-java-features, x-goog-user-project," + + " x-http-method-override, x-upload-content-length, x-upload-content-type]"); } }