From 50ffe267b2d7891173db9b8c02b5fe6a5733f17e Mon Sep 17 00:00:00 2001 From: cpriti-os Date: Sat, 24 Oct 2026 13:29:34 +0000 Subject: [PATCH 1/4] feat(storage): add client feature tracking header Add a package-private feature tracking mechanism so the service can observe SDK-side features it otherwise cannot see. The first tracked feature is MULTISTREAM_IN_MRD: a BlobReadSession transparently opening more than one BidiReadObject stream, which has no user-facing option. * TrackedFeatures: single source of truth enum, feature -> bit position. * FeatureTracking: header name `x-goog-storage-java-features`, encoding (4-byte big-endian uint32 -> URL-safe base64 with padding), and a thread-safe Mask. Header is omitted when the mask is 0. * ObjectReadSessionState: a Mask shared by a session's root state and all forkChild() states; forkChild() sets MULTISTREAM_IN_MRD and getOpenArguments() attaches the header on subsequent stream opens. * UnifiedOpts.Headers.BLOCKLIST: reject user-supplied x-goog-storage-java-features. Parity with google-cloud-go `x-goog-storage-go-features` (googleapis/google-cloud-go#20684). --- .../google/cloud/storage/FeatureTracking.java | 93 ++++++++++++ .../cloud/storage/ObjectReadSessionState.java | 40 +++-- .../google/cloud/storage/TrackedFeatures.java | 43 ++++++ .../com/google/cloud/storage/UnifiedOpts.java | 1 + .../cloud/storage/FeatureTrackingTest.java | 112 ++++++++++++++ .../storage/ITObjectReadSessionFakeTest.java | 139 ++++++++++++++++++ .../storage/ObjectReadSessionStateTest.java | 32 ++++ .../cloud/storage/UnifiedOptsGrpcTest.java | 5 +- 8 files changed, 452 insertions(+), 13 deletions(-) create mode 100644 java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/FeatureTracking.java create mode 100644 java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java create mode 100644 java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/FeatureTrackingTest.java 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..9a2afbae44b0 --- /dev/null +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/FeatureTracking.java @@ -0,0 +1,93 @@ +/* + * 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.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.nio.ByteBuffer; +import java.util.Base64; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Utilities for reporting {@link TrackedFeatures 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() {} + + static int bit(TrackedFeatures feature) { + return 1 << feature.getBitPosition(); + } + + /** Encode {@code mask} as a 4 byte big-endian uint32, URL-safe base64 with padding. */ + static String encode(int mask) { + ByteBuffer buf = ByteBuffer.allocate(4); + buf.putInt(mask); // ByteBuffer default byte order is BIG_ENDIAN + return ENCODER.encodeToString(buf.array()); + } + + /** + * Produce the header map to merge into a request's extra headers. + * + * @return an empty map when {@code mask == 0}, otherwise a single entry map for {@link #HEADER} + */ + static ImmutableMap> headers(int mask) { + if (mask == 0) { + return ImmutableMap.of(); + } + return ImmutableMap.of(HEADER, ImmutableList.of(encode(mask))); + } + + /** Thread-safe, monotonically growing feature bitmask. */ + static final class Mask { + private final AtomicInteger bits = new AtomicInteger(); + + Mask() {} + + void set(TrackedFeatures feature) { + bits.updateAndGet(current -> current | bit(feature)); + } + + boolean isSet(TrackedFeatures feature) { + return (bits.get() & bit(feature)) != 0; + } + + int get() { + return bits.get(); + } + + ImmutableMap> headers() { + return FeatureTracking.headers(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..078efe9c24c5 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 @@ -55,6 +55,8 @@ final class ObjectReadSessionState { private final AtomicReference<@Nullable String> routingToken; private final AtomicReference<@MonotonicNonNull Object> metadata; private final AtomicLong readIdSeq; + // shared between a parent state and all of its children, see #forkChild + private final FeatureTracking.Mask featureMask; @GuardedBy("this.lock") // https://errorprone.info/bugpattern/GuardedBy private final Map> outstandingReads; @@ -69,7 +71,8 @@ final class ObjectReadSessionState { new AtomicLong(1), new AtomicReference<>(), new AtomicReference<>(), - new AtomicReference<>()); + new AtomicReference<>(), + new FeatureTracking.Mask()); } private ObjectReadSessionState( @@ -78,25 +81,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(); } + /** + * Fork 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 TrackedFeatures#MULTISTREAM_IN_MRD} in the feature mask shared by the parent and all + * children. Any stream opened (or re-opened) by this session afterwards will carry the {@link + * FeatureTracking#HEADER} header. + */ ObjectReadSessionState forkChild() { + featureMask.set(TrackedFeatures.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 +151,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/TrackedFeatures.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java new file mode 100644 index 000000000000..82dfe6772065 --- /dev/null +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java @@ -0,0 +1,43 @@ +/* + * 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; + +/** + * Client features 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 is parsed by external tooling; keep the {@code NAME(bit)} shape intact. + */ +enum TrackedFeatures { + /** + * 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; + + TrackedFeatures(int bitPosition) { + this.bitPosition = bitPosition; + } + + int getBitPosition() { + return 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..e1519c436138 --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/FeatureTrackingTest.java @@ -0,0 +1,112 @@ +/* + * 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.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.google.cloud.storage.UnifiedOpts.Opts; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.util.HashSet; +import java.util.Set; +import org.junit.Test; + +public final class FeatureTrackingTest { + + @Test + public void headerName() { + assertThat(FeatureTracking.HEADER).isEqualTo("x-goog-storage-java-features"); + } + + @Test + public void encode_knownVectors() { + assertThat(FeatureTracking.encode(0)).isEqualTo("AAAAAA=="); + assertThat(FeatureTracking.encode(1)).isEqualTo("AAAAAQ=="); + assertThat(FeatureTracking.encode(4)).isEqualTo("AAAABA=="); + assertThat(FeatureTracking.encode(0xF8)).isEqualTo("AAAA-A=="); + } + + @Test + public void encode_isUrlSafeWithPadding() { + for (int shift = 0; shift < 32; shift++) { + int mask = 1 << shift; + String encoded = FeatureTracking.encode(mask); + assertThat(encoded).hasLength(8); + assertThat(encoded).doesNotContain("+"); + assertThat(encoded).doesNotContain("/"); + assertThat(encoded).endsWith("=="); + } + // high bit set; verify unsigned/big-endian interpretation + assertThat(FeatureTracking.encode(0xFFFFFFFF)).isEqualTo("_____w=="); + assertThat(FeatureTracking.encode(0x80000000)).isEqualTo("gAAAAA=="); + } + + @Test + public void headers_omittedWhenMaskIsZero() { + assertThat(FeatureTracking.headers(0)).isEmpty(); + assertThat(new FeatureTracking.Mask().headers()).isEmpty(); + } + + @Test + public void headers_presentWhenMaskIsNonZero() { + assertThat(FeatureTracking.headers(1)) + .containsExactly(FeatureTracking.HEADER, ImmutableList.of("AAAAAQ==")); + } + + @Test + public void mask_setIsIdempotentAndMonotonic() { + FeatureTracking.Mask mask = new FeatureTracking.Mask(); + assertThat(mask.isSet(TrackedFeatures.MULTISTREAM_IN_MRD)).isFalse(); + mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); + mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); + assertThat(mask.isSet(TrackedFeatures.MULTISTREAM_IN_MRD)).isTrue(); + assertThat(mask.get()).isEqualTo(1); + assertThat(mask.headers()) + .containsExactly(FeatureTracking.HEADER, ImmutableList.of("AAAAAQ==")); + } + + @Test + public void trackedFeatures_bitPositionsAreUniqueAndInRange() { + Set seen = new HashSet<>(); + for (TrackedFeatures f : TrackedFeatures.values()) { + assertThat(f.getBitPosition()).isAtLeast(0); + assertThat(f.getBitPosition()).isLessThan(32); + assertThat(seen.add(f.getBitPosition())).isTrue(); + } + assertThat(TrackedFeatures.MULTISTREAM_IN_MRD.getBitPosition()).isEqualTo(0); + } + + @Test + public void extraHeaders_rejectsFeatureTrackingHeader() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> + Opts.from( + UnifiedOpts.extraHeaders(ImmutableMap.of(FeatureTracking.HEADER, "AAAAAQ==")))); + assertThat(e).hasMessageThat().contains(FeatureTracking.HEADER); + + // case-insensitive + assertThrows( + IllegalArgumentException.class, + () -> + Opts.from( + UnifiedOpts.extraHeaders( + ImmutableMap.of("X-Goog-Storage-Java-Features", "AAAAAQ==")))); + } +} 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..f123a1418161 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,8 @@ 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); private static final Correspondence IS_UUID = Correspondence.transforming(UUID::fromString, "is a UUID"); @@ -1166,6 +1169,142 @@ public void onCompleteWithoutAValue() throws Exception { *

validate that two streams are opened and that getting the accumulated byte array can succeed * while the channel hasn't been fully consumed. */ + /** + * + * + *

    + *
  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(); + String expectedHeaderValue = + FeatureTracking.encode(FeatureTracking.bit(TrackedFeatures.MULTISTREAM_IN_MRD)); + 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); + } + } + + final byte[] finalBytes = 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(expectedHeaderValue), + () -> assertThat(expectedHeaderValue).isEqualTo("AAAAAQ=="), + () -> assertThat(xxd(buf)).isEqualTo(xxd(content.getBytes())), + () -> assertThat(xxd(finalBytes)).isEqualTo(xxd(content.getBytes()))); + } + } + @Test public void blobDescriptorTransparentlyForksStreamIfNeeded() throws Exception { ChecksummedTestContent content = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 20); 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..4003e3bcf618 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 @@ -234,4 +234,36 @@ public void canHandleNewRead() throws Exception { () -> assertThat(state2.canHandleNewRead(streaming2)).isFalse()); } } + + @Test + public void featureTrackingHeader_absentUntilFork_presentOnParentAndChildAfterFork() + throws Exception { + BidiReadObjectRequest base = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket("projects/_/buckets/my-bucket") + .setObject("my-object")) + .build(); + + ObjectReadSessionState parent = + new ObjectReadSessionState(GrpcCallContext.createDefault(), base); + + Map> before = parent.getOpenArguments().getCtx().getExtraHeaders(); + assertThat(before).doesNotContainKey(FeatureTracking.HEADER); + + ObjectReadSessionState child = parent.forkChild(); + + List expected = + ImmutableList.of( + FeatureTracking.encode(FeatureTracking.bit(TrackedFeatures.MULTISTREAM_IN_MRD))); + Map> childHeaders = child.getOpenArguments().getCtx().getExtraHeaders(); + // any subsequent (re)open of the parent stream must also report the feature + Map> parentHeaders = parent.getOpenArguments().getCtx().getExtraHeaders(); + assertAll( + () -> assertThat(childHeaders).containsEntry(FeatureTracking.HEADER, expected), + () -> assertThat(parentHeaders).containsEntry(FeatureTracking.HEADER, expected), + () -> assertThat(childHeaders).containsKey("x-goog-request-params"), + () -> assertThat(expected).containsExactly("AAAAAQ==")); + } } 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]"); } } From 01e52f522ac67893ad9327e40c4e7ba3c58380d1 Mon Sep 17 00:00:00 2001 From: cpriti-os Date: Sun, 25 Oct 2026 03:50:58 +0000 Subject: [PATCH 2/4] fix(storage): tighten FeatureTracking visibility and readability * Move the bit-mask derivation onto the enum (TrackedFeatures#bitMask) and drop the static FeatureTracking#bit helper. * Make FeatureTracking#headers(int) private; production code only goes through Mask#headers(). Remove the unused Mask#isSet, mark the test-only accessors @VisibleForTesting, and drop the redundant explicit Mask constructor. * Encode with Guava Ints.toByteArray (big-endian by contract) instead of a ByteBuffer and a byte-order comment. * Annotate Mask @ThreadSafe (consistent with ByteRangeSpec) and fix Javadoc summary fragments to the declarative form. No behavior change; all additions remain package-private. --- .../google/cloud/storage/FeatureTracking.java | 37 ++++++++----------- .../cloud/storage/ObjectReadSessionState.java | 7 ++-- .../google/cloud/storage/TrackedFeatures.java | 7 +++- 3 files changed, 25 insertions(+), 26 deletions(-) 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 index 9a2afbae44b0..18aa1bf341fb 100644 --- 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 @@ -16,12 +16,14 @@ package com.google.cloud.storage; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; -import java.nio.ByteBuffer; +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 TrackedFeatures client feature usage} to the service. @@ -45,49 +47,40 @@ final class FeatureTracking { private FeatureTracking() {} - static int bit(TrackedFeatures feature) { - return 1 << feature.getBitPosition(); - } - - /** Encode {@code mask} as a 4 byte big-endian uint32, URL-safe base64 with padding. */ + /** Encodes {@code mask} as a 4 byte big-endian uint32, URL-safe base64 with padding. */ + @VisibleForTesting static String encode(int mask) { - ByteBuffer buf = ByteBuffer.allocate(4); - buf.putInt(mask); // ByteBuffer default byte order is BIG_ENDIAN - return ENCODER.encodeToString(buf.array()); + return ENCODER.encodeToString(Ints.toByteArray(mask)); } /** - * Produce the header map to merge into a request's extra headers. - * - * @return an empty map when {@code mask == 0}, otherwise a single entry map for {@link #HEADER} + * 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}. */ - static ImmutableMap> headers(int mask) { + private static ImmutableMap> headers(int mask) { if (mask == 0) { return ImmutableMap.of(); } return ImmutableMap.of(HEADER, ImmutableList.of(encode(mask))); } - /** Thread-safe, monotonically growing feature bitmask. */ + /** Monotonically growing feature bitmask. Bits can be set but never cleared. */ + @ThreadSafe static final class Mask { private final AtomicInteger bits = new AtomicInteger(); - Mask() {} - void set(TrackedFeatures feature) { - bits.updateAndGet(current -> current | bit(feature)); - } - - boolean isSet(TrackedFeatures feature) { - return (bits.get() & bit(feature)) != 0; + bits.updateAndGet(current -> current | feature.bitMask()); } + @VisibleForTesting int get() { return bits.get(); } + /** Returns the headers to attach to a request, see {@link FeatureTracking#headers(int)}. */ ImmutableMap> headers() { - return FeatureTracking.headers(get()); + 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 078efe9c24c5..75f51bb35736 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 @@ -55,7 +55,8 @@ final class ObjectReadSessionState { private final AtomicReference<@Nullable String> routingToken; private final AtomicReference<@MonotonicNonNull Object> metadata; private final AtomicLong readIdSeq; - // shared between a parent state and all of its children, see #forkChild + + /** 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 @@ -95,11 +96,11 @@ private ObjectReadSessionState( } /** - * Fork a new state for an additional stream of the same read session. + * 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 TrackedFeatures#MULTISTREAM_IN_MRD} in the feature mask shared by the parent and all - * children. Any stream opened (or re-opened) by this session afterwards will carry the {@link + * children. Any stream opened (or re-opened) by this session afterwards carries the {@link * FeatureTracking#HEADER} header. */ ObjectReadSessionState forkChild() { diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java index 82dfe6772065..663ccc8d14a4 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java @@ -17,7 +17,7 @@ package com.google.cloud.storage; /** - * Client features whose usage is reported to the service via the {@link FeatureTracking#HEADER + * 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 @@ -40,4 +40,9 @@ enum TrackedFeatures { int getBitPosition() { return bitPosition; } + + /** Returns the 32-bit mask with only this feature's bit set. */ + int bitMask() { + return 1 << bitPosition; + } } From 09d500498903eeea678fc8a9de067276074d95d6 Mon Sep 17 00:00:00 2001 From: cpriti-os Date: Sun, 25 Oct 2026 03:51:04 +0000 Subject: [PATCH 3/4] test(storage): broaden feature tracking header coverage FeatureTrackingTest * Encoding vectors for 0, 1, 2, 3, 4, 0xF8 ('-'), 0xFC ('_'), 0x80000000 and 0xFFFFFFFF; every single bit is URL-safe and padded. * Mask: empty headers when nothing is set, literal wire header name and value, idempotent set, and a deterministic multi-threaded test that concurrent set() calls lose no bits. * Bit positions unique and < 32; MULTISTREAM_IN_MRD pinned to bit 0. * Blocklist: exact and mixed-case header rejected by UnifiedOpts.extraHeaders and by the public BlobSourceOption factory (transport agnostic, so both gRPC and HTTP/JSON paths are covered). ObjectReadSessionStateTest * Header absent before fork; present on child and on parent after fork; a second fork / grandchild fork does not duplicate or alter the value; x-goog-request-params (with routing token) and user supplied extra headers on the base context coexist with the feature header. ITObjectReadSessionFakeTest * singleStreamSessionNeverSendsFeatureTrackingHeader: two sequential reads sharing one stream never send the header. * forkedStreamCarriesFeatureTrackingHeader: moved below the test whose Javadoc it had been inserted in front of; redundant assertions removed. * streamReopenedAfterForkCarriesFeatureTrackingHeader: after a fork, the original stream is failed with UNAVAILABLE; the retry re-open of that stream also carries the header (3 streams, header on exactly 2). --- .../cloud/storage/FeatureTrackingTest.java | 174 +++++--- .../storage/ITObjectReadSessionFakeTest.java | 373 ++++++++++++++---- .../storage/ObjectReadSessionStateTest.java | 113 ++++-- 3 files changed, 520 insertions(+), 140 deletions(-) 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 index e1519c436138..c4406a5ffd37 100644 --- 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 @@ -16,97 +16,177 @@ 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.UnifiedOpts.Opts; +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 { - @Test - public void headerName() { - assertThat(FeatureTracking.HEADER).isEqualTo("x-goog-storage-java-features"); - } + private static final int MULTISTREAM_BIT = TrackedFeatures.MULTISTREAM_IN_MRD.bitMask(); @Test - public void encode_knownVectors() { - assertThat(FeatureTracking.encode(0)).isEqualTo("AAAAAA=="); - assertThat(FeatureTracking.encode(1)).isEqualTo("AAAAAQ=="); - assertThat(FeatureTracking.encode(4)).isEqualTo("AAAABA=="); - assertThat(FeatureTracking.encode(0xF8)).isEqualTo("AAAA-A=="); + 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_isUrlSafeWithPadding() { + public void encode_everySingleBit_isUrlSafeWithPadding() { for (int shift = 0; shift < 32; shift++) { - int mask = 1 << shift; - String encoded = FeatureTracking.encode(mask); + String encoded = FeatureTracking.encode(1 << shift); assertThat(encoded).hasLength(8); assertThat(encoded).doesNotContain("+"); assertThat(encoded).doesNotContain("/"); assertThat(encoded).endsWith("=="); } - // high bit set; verify unsigned/big-endian interpretation - assertThat(FeatureTracking.encode(0xFFFFFFFF)).isEqualTo("_____w=="); - assertThat(FeatureTracking.encode(0x80000000)).isEqualTo("gAAAAA=="); } @Test - public void headers_omittedWhenMaskIsZero() { - assertThat(FeatureTracking.headers(0)).isEmpty(); - assertThat(new FeatureTracking.Mask().headers()).isEmpty(); + public void mask_headers_emptyWhenNoFeatureSet() { + FeatureTracking.Mask mask = new FeatureTracking.Mask(); + + assertThat(mask.get()).isEqualTo(0); + assertThat(mask.headers()).isEmpty(); } @Test - public void headers_presentWhenMaskIsNonZero() { - assertThat(FeatureTracking.headers(1)) - .containsExactly(FeatureTracking.HEADER, ImmutableList.of("AAAAAQ==")); + public void mask_headers_usesWireHeaderNameAndEncodedValue() { + FeatureTracking.Mask mask = new FeatureTracking.Mask(); + + mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); + + assertThat(mask.headers()) + .containsExactly("x-goog-storage-java-features", ImmutableList.of("AAAAAQ==")); } @Test - public void mask_setIsIdempotentAndMonotonic() { + public void mask_set_isIdempotentAndMonotonic() { FeatureTracking.Mask mask = new FeatureTracking.Mask(); - assertThat(mask.isSet(TrackedFeatures.MULTISTREAM_IN_MRD)).isFalse(); + mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); - assertThat(mask.isSet(TrackedFeatures.MULTISTREAM_IN_MRD)).isTrue(); - assertThat(mask.get()).isEqualTo(1); + + assertThat(mask.get()).isEqualTo(MULTISTREAM_BIT); assertThat(mask.headers()) - .containsExactly(FeatureTracking.HEADER, ImmutableList.of("AAAAAQ==")); + .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(TrackedFeatures.MULTISTREAM_IN_MRD); + } + return null; + })); + } + start.countDown(); + for (Future f : futures) { + f.get(30, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + + int expected = 0; + for (TrackedFeatures feature : TrackedFeatures.values()) { + expected |= feature.bitMask(); + } + assertThat(mask.get()).isEqualTo(expected); } @Test public void trackedFeatures_bitPositionsAreUniqueAndInRange() { Set seen = new HashSet<>(); - for (TrackedFeatures f : TrackedFeatures.values()) { - assertThat(f.getBitPosition()).isAtLeast(0); - assertThat(f.getBitPosition()).isLessThan(32); - assertThat(seen.add(f.getBitPosition())).isTrue(); + for (TrackedFeatures feature : TrackedFeatures.values()) { + assertThat(feature.getBitPosition()).isAtLeast(0); + assertThat(feature.getBitPosition()).isLessThan(32); + assertThat(feature.bitMask()).isEqualTo(1 << feature.getBitPosition()); + seen.add(feature.getBitPosition()); } + assertThat(seen).hasSize(TrackedFeatures.values().length); + } + + /** The wire value of each bit position is a cross-SDK contract; pin it. */ + @Test + public void trackedFeatures_multistreamInMrd_isBitZero() { assertThat(TrackedFeatures.MULTISTREAM_IN_MRD.getBitPosition()).isEqualTo(0); } @Test - public void extraHeaders_rejectsFeatureTrackingHeader() { - IllegalArgumentException e = - assertThrows( - IllegalArgumentException.class, - () -> - Opts.from( - UnifiedOpts.extraHeaders(ImmutableMap.of(FeatureTracking.HEADER, "AAAAAQ==")))); - assertThat(e).hasMessageThat().contains(FeatureTracking.HEADER); - - // case-insensitive - assertThrows( - IllegalArgumentException.class, - () -> - Opts.from( - UnifiedOpts.extraHeaders( - ImmutableMap.of("X-Goog-Storage-Java-Features", "AAAAAQ==")))); + 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 f123a1418161..f683d9c1f0ef 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 @@ -136,6 +136,10 @@ public final class ITObjectReadSessionFakeTest { 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 TrackedFeatures#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"); @@ -1169,20 +1173,8 @@ public void onCompleteWithoutAValue() throws Exception { *

validate that two streams are opened and that getting the accumulated byte array can succeed * while the channel hasn't been fully consumed. */ - /** - * - * - *

    - *
  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 { + public void blobDescriptorTransparentlyForksStreamIfNeeded() 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); @@ -1233,12 +1225,14 @@ public void forkedStreamCarriesFeatureTrackingHeader() throws Exception { public StreamObserver bidiReadObject( StreamObserver respond) { bidiReadObjectCount.getAndIncrement(); - return new AbstractObserver(respond) { + return new StreamObserver() { @Override public void onNext(BidiReadObjectRequest request) { if (request.equals(REQ_OPEN)) { respond.onNext(RES_OPEN); } else if (request.equals(req2)) { + // respond with the first half of the bytes, then wait for the second request to + // be received before sending the second half. respond.onNext(res2_1); try { cdl.await(); @@ -1250,23 +1244,31 @@ public void onNext(BidiReadObjectRequest request) { } else if (request.equals(req3)) { respond.onNext(res3); respond.onCompleted(); + // signal the second request was received cdl.countDown(); } else { respond.onError(TestUtils.apiException(Code.UNIMPLEMENTED, "Unexpected request")); } } + + @Override + public void onError(Throwable t) { + System.out.println("ITObjectReadSessionFakeTest.onError"); + respond.onError(t); + } + + @Override + public void onCompleted() { + System.out.println("ITObjectReadSessionFakeTest.onCompleted"); + respond.onCompleted(); + } }; } }; - - GrpcRequestAuditing requestAuditing = new GrpcRequestAuditing(); - String expectedHeaderValue = - FeatureTracking.encode(FeatureTracking.bit(TrackedFeatures.MULTISTREAM_IN_MRD)); try (FakeServer fakeServer = FakeServer.of(fakeStorage); Storage storage = fakeServer.getGrpcStorageOptions().toBuilder() .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(1).build()) - .setGrpcInterceptorProvider(requestAuditing) .build() .getService()) { @@ -1274,10 +1276,9 @@ public void onNext(BidiReadObjectRequest request) { ApiFuture futureObjectDescriptor = storage.blobReadSession(id); ByteBuffer buf = ByteBuffer.allocate(50); - byte[] bytes; + byte[] bytes = new byte[0]; + Exception caught = null; 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); @@ -1289,24 +1290,136 @@ public void onNext(BidiReadObjectRequest request) { bytes = future.get(3, TimeUnit.SECONDS); Buffers.fillFrom(buf, c); } + } catch (Exception e) { + // stash off any runtime failure so we can still do our assertions to help determine + // the true failure + caught = e; + } finally { + final byte[] finalBytes = bytes; + final Exception finalCaught = caught; + assertAll( + () -> assertThat(bidiReadObjectCount.get()).isEqualTo(2), + () -> + assertWithMessage("Channel bytes missmatch") + .that(xxd(buf)) + .isEqualTo(xxd(content.getBytes())), + () -> + assertWithMessage("Future bytes missmatch") + .that(xxd(finalBytes)) + .isEqualTo(xxd(content.getBytes())), + () -> { + if (finalCaught != null) { + throw new Exception("exception during test", finalCaught); + } + }); + } + } + } + + /** + * 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); } - final byte[] finalBytes = 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(expectedHeaderValue), - () -> assertThat(expectedHeaderValue).isEqualTo("AAAAAQ=="), - () -> assertThat(xxd(buf)).isEqualTo(xxd(content.getBytes())), - () -> assertThat(xxd(finalBytes)).isEqualTo(xxd(content.getBytes()))); + () -> 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 blobDescriptorTransparentlyForksStreamIfNeeded() throws Exception { + 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); @@ -1357,14 +1470,12 @@ public void blobDescriptorTransparentlyForksStreamIfNeeded() throws Exception { public StreamObserver bidiReadObject( StreamObserver respond) { bidiReadObjectCount.getAndIncrement(); - return new StreamObserver() { + 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 with the first half of the bytes, then wait for the second request to - // be received before sending the second half. respond.onNext(res2_1); try { cdl.await(); @@ -1376,31 +1487,163 @@ public void onNext(BidiReadObjectRequest request) { } else if (request.equals(req3)) { respond.onNext(res3); respond.onCompleted(); - // signal the second request was received cdl.countDown(); } else { respond.onError(TestUtils.apiException(Code.UNIMPLEMENTED, "Unexpected request")); } } + }; + } + }; - @Override - public void onError(Throwable t) { - System.out.println("ITObjectReadSessionFakeTest.onError"); - respond.onError(t); - } + 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 onCompleted() { - System.out.println("ITObjectReadSessionFakeTest.onCompleted"); - respond.onCompleted(); + 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(1).build()) + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(3).build()) + .setGrpcInterceptorProvider(requestAuditing) .build() .getService()) { @@ -1408,8 +1651,7 @@ public void onCompleted() { ApiFuture futureObjectDescriptor = storage.blobReadSession(id); ByteBuffer buf = ByteBuffer.allocate(50); - byte[] bytes = new byte[0]; - Exception caught = null; + byte[] bytes; try (BlobReadSession bd = futureObjectDescriptor.get(5, TimeUnit.SECONDS)) { try (ScatteringByteChannel c = bd.readAs(ReadProjectionConfigs.asChannel().withRangeSpec(RangeSpec.of(10L, 20L)))) { @@ -1422,29 +1664,20 @@ public void onCompleted() { bytes = future.get(3, TimeUnit.SECONDS); Buffers.fillFrom(buf, c); } - } catch (Exception e) { - // stash off any runtime failure so we can still do our assertions to help determine - // the true failure - caught = e; - } finally { - final byte[] finalBytes = bytes; - final Exception finalCaught = caught; - assertAll( - () -> assertThat(bidiReadObjectCount.get()).isEqualTo(2), - () -> - assertWithMessage("Channel bytes missmatch") - .that(xxd(buf)) - .isEqualTo(xxd(content.getBytes())), - () -> - assertWithMessage("Future bytes missmatch") - .that(xxd(finalBytes)) - .isEqualTo(xxd(content.getBytes())), - () -> { - if (finalCaught != null) { - throw new Exception("exception during test", finalCaught); - } - }); } + + 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()))); } } 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 4003e3bcf618..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 = @@ -236,34 +248,89 @@ public void canHandleNewRead() throws Exception { } @Test - public void featureTrackingHeader_absentUntilFork_presentOnParentAndChildAfterFork() - throws Exception { - BidiReadObjectRequest base = - BidiReadObjectRequest.newBuilder() - .setReadObjectSpec( - BidiReadObjectSpec.newBuilder() - .setBucket("projects/_/buckets/my-bucket") - .setObject("my-object")) - .build(); + public void featureTrackingHeader_absentBeforeFork() { + ObjectReadSessionState state = + new ObjectReadSessionState(GrpcCallContext.createDefault(), BASE_REQUEST); - ObjectReadSessionState parent = - new ObjectReadSessionState(GrpcCallContext.createDefault(), base); + Map> headers = state.getOpenArguments().getCtx().getExtraHeaders(); + + assertThat(headers).containsExactly("x-goog-request-params", REQUEST_PARAMS_HEADER_VALUE); + } - Map> before = parent.getOpenArguments().getCtx().getExtraHeaders(); - assertThat(before).doesNotContainKey(FeatureTracking.HEADER); + @Test + public void featureTrackingHeader_presentOnChildAfterFork() { + ObjectReadSessionState parent = + new ObjectReadSessionState(GrpcCallContext.createDefault(), BASE_REQUEST); ObjectReadSessionState child = parent.forkChild(); - List expected = - ImmutableList.of( - FeatureTracking.encode(FeatureTracking.bit(TrackedFeatures.MULTISTREAM_IN_MRD))); - Map> childHeaders = child.getOpenArguments().getCtx().getExtraHeaders(); - // any subsequent (re)open of the parent stream must also report the feature - Map> parentHeaders = parent.getOpenArguments().getCtx().getExtraHeaders(); + 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(childHeaders).containsEntry(FeatureTracking.HEADER, expected), - () -> assertThat(parentHeaders).containsEntry(FeatureTracking.HEADER, expected), - () -> assertThat(childHeaders).containsKey("x-goog-request-params"), - () -> assertThat(expected).containsExactly("AAAAAQ==")); + () -> + 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); } } From e05316c368284242d4ab67f9dfc63a3f5a135fb1 Mon Sep 17 00:00:00 2001 From: cpriti-os Date: Mon, 26 Oct 2026 07:09:14 +0000 Subject: [PATCH 4/4] refactor(storage): rename TrackedFeature and drop test-only accessor * Rename enum TrackedFeatures -> TrackedFeature (singular, matching the C++ TrackedFeature / Go trackedFeature naming). * Remove getBitPosition(); it was only used by tests. bitMask() remains the single accessor. Tests now assert MULTISTREAM_IN_MRD.bitMask() == 1 and that every bitMask() is a distinct single bit. * Mask#set uses accumulateAndGet(bitMask, (a, b) -> a | b). * Clarify the enum Javadoc: external tooling parses the NAME(bit) constants, the type name is irrelevant to it. --- .../google/cloud/storage/FeatureTracking.java | 10 +++---- .../cloud/storage/ObjectReadSessionState.java | 4 +-- ...ackedFeatures.java => TrackedFeature.java} | 11 +++---- .../cloud/storage/FeatureTrackingTest.java | 30 +++++++++---------- .../storage/ITObjectReadSessionFakeTest.java | 2 +- 5 files changed, 26 insertions(+), 31 deletions(-) rename java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/{TrackedFeatures.java => TrackedFeature.java} (87%) 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 index 18aa1bf341fb..0df037b2d800 100644 --- 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 @@ -26,13 +26,13 @@ import javax.annotation.concurrent.ThreadSafe; /** - * Utilities for reporting {@link TrackedFeatures client feature usage} to the service. + * Utilities for reporting {@link TrackedFeature client feature usage} to the service. * *

Encoding contract (shared across GCS SDKs): * *

    - *
  • The set of used features is a 32-bit mask where bit {@code n} corresponds to the feature - * with {@link TrackedFeatures#getBitPosition()} {@code == n}. + *
  • The set of used features is a 32-bit mask, the bitwise OR of {@link + * TrackedFeature#bitMask()} for every feature used. *
  • The mask is serialized as 4 bytes big-endian, then encoded with URL-safe base64 (RFC 4648 * §5) with padding. The service rejects {@code +} and {@code /}. *
  • When the mask is {@code 0} the header is omitted entirely. @@ -69,8 +69,8 @@ private static ImmutableMap> headers(int mask) { static final class Mask { private final AtomicInteger bits = new AtomicInteger(); - void set(TrackedFeatures feature) { - bits.updateAndGet(current -> current | feature.bitMask()); + void set(TrackedFeature feature) { + bits.accumulateAndGet(feature.bitMask(), (a, b) -> a | b); } @VisibleForTesting 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 75f51bb35736..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 @@ -99,12 +99,12 @@ private ObjectReadSessionState( * 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 TrackedFeatures#MULTISTREAM_IN_MRD} in the feature mask shared by the parent and all + * {@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(TrackedFeatures.MULTISTREAM_IN_MRD); + featureMask.set(TrackedFeature.MULTISTREAM_IN_MRD); return new ObjectReadSessionState( baseContext, openRequest, diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeature.java similarity index 87% rename from java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java rename to java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeature.java index 663ccc8d14a4..4fefd0f758e1 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeatures.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/TrackedFeature.java @@ -22,9 +22,10 @@ * *

    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 is parsed by external tooling; keep the {@code NAME(bit)} shape intact. + * of truth and its constants are parsed by external tooling; the {@code NAME(bit)} shape of each + * constant must stay intact. */ -enum TrackedFeatures { +enum TrackedFeature { /** * A {@link BlobReadSession} opened more than one BidiReadObject stream to serve concurrent reads * which could not share a single stream. @@ -33,14 +34,10 @@ enum TrackedFeatures { private final int bitPosition; - TrackedFeatures(int bitPosition) { + TrackedFeature(int bitPosition) { this.bitPosition = bitPosition; } - int getBitPosition() { - return 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/test/java/com/google/cloud/storage/FeatureTrackingTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/FeatureTrackingTest.java index c4406a5ffd37..1062e2a0290b 100644 --- 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 @@ -36,7 +36,7 @@ public final class FeatureTrackingTest { - private static final int MULTISTREAM_BIT = TrackedFeatures.MULTISTREAM_IN_MRD.bitMask(); + private static final int MULTISTREAM_BIT = TrackedFeature.MULTISTREAM_IN_MRD.bitMask(); @Test public void encode_knownVectors() throws Exception { @@ -79,7 +79,7 @@ public void mask_headers_emptyWhenNoFeatureSet() { public void mask_headers_usesWireHeaderNameAndEncodedValue() { FeatureTracking.Mask mask = new FeatureTracking.Mask(); - mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); assertThat(mask.headers()) .containsExactly("x-goog-storage-java-features", ImmutableList.of("AAAAAQ==")); @@ -89,8 +89,8 @@ public void mask_headers_usesWireHeaderNameAndEncodedValue() { public void mask_set_isIdempotentAndMonotonic() { FeatureTracking.Mask mask = new FeatureTracking.Mask(); - mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); - mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); assertThat(mask.get()).isEqualTo(MULTISTREAM_BIT); assertThat(mask.headers()) @@ -117,7 +117,7 @@ public void mask_set_isSafeUnderConcurrentUpdates() throws Exception { () -> { start.await(); for (int j = 0; j < iterationsPerThread; j++) { - mask.set(TrackedFeatures.MULTISTREAM_IN_MRD); + mask.set(TrackedFeature.MULTISTREAM_IN_MRD); } return null; })); @@ -131,28 +131,26 @@ public void mask_set_isSafeUnderConcurrentUpdates() throws Exception { } int expected = 0; - for (TrackedFeatures feature : TrackedFeatures.values()) { + for (TrackedFeature feature : TrackedFeature.values()) { expected |= feature.bitMask(); } assertThat(mask.get()).isEqualTo(expected); } @Test - public void trackedFeatures_bitPositionsAreUniqueAndInRange() { + public void trackedFeature_bitMasksAreDistinctSingleBits() { Set seen = new HashSet<>(); - for (TrackedFeatures feature : TrackedFeatures.values()) { - assertThat(feature.getBitPosition()).isAtLeast(0); - assertThat(feature.getBitPosition()).isLessThan(32); - assertThat(feature.bitMask()).isEqualTo(1 << feature.getBitPosition()); - seen.add(feature.getBitPosition()); + for (TrackedFeature feature : TrackedFeature.values()) { + assertThat(Integer.bitCount(feature.bitMask())).isEqualTo(1); + seen.add(feature.bitMask()); } - assertThat(seen).hasSize(TrackedFeatures.values().length); + assertThat(seen).hasSize(TrackedFeature.values().length); } - /** The wire value of each bit position is a cross-SDK contract; pin it. */ + /** The wire value of each feature bit is a cross-SDK contract; pin it. */ @Test - public void trackedFeatures_multistreamInMrd_isBitZero() { - assertThat(TrackedFeatures.MULTISTREAM_IN_MRD.getBitPosition()).isEqualTo(0); + public void trackedFeature_multistreamInMrd_isBitZero() { + assertThat(TrackedFeature.MULTISTREAM_IN_MRD.bitMask()).isEqualTo(1); } @Test 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 f683d9c1f0ef..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 @@ -137,7 +137,7 @@ public final class ITObjectReadSessionFakeTest { 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 TrackedFeatures#MULTISTREAM_IN_MRD} set. */ + /** 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 =