Skip to content

Commit f548335

Browse files
BenWhiteheadBenWhitehead
authored
fix: update ApiaryUnbufferedWritableByteChannel to be graceful of non-quantum aligned write calls (#2493)
Despite GCS only allowing incremental updates to a resumable session at 256KiB byte boundaries, we have observed an extremely rare case of an incremental write being non-quantum aligned. The change in this PR makes ApiaryUnbufferedWritableByteChannel graceful to this possibility, and will only set the finalization header when close is invoked. If a write is not quantum aligned, it will either: 1) not be consumed at all, at which point write can be called again with the still enqueued bytes 2) partially consumed, with the position of the provided ByteBuffers updated to reflect how much of their bytes were consumed, matching up with the number of bytes actually consumed returned from `write()` Add new integration test to intentionally perform non-quantum aligned `write()` calls to ApiaryUnbufferedWritableByteChannel. Separate change to gRPC affected code path to come in a later PR. b/330550326
1 parent 3a0b829 commit f548335

6 files changed

Lines changed: 131 additions & 59 deletions

File tree

‎google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedWritableByteChannel.java‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -96,30 +96,32 @@ private long internalWrite(ByteBuffer[] srcs, int offset, int length, boolean fi
9696
}
9797
RewindableContent content = RewindableContent.of(Utils.subArray(srcs, offset, length));
9898
long available = content.getLength();
99+
// as long as request has at least 256KiB GCS will accept bytes in 256KiB increments,
100+
// however if a request is smaller than 256KiB it MUST be the finalization request.
101+
if (!finalize && available < ByteSizeConstants._256KiB) {
102+
return 0;
103+
}
99104
long newFinalByteOffset = cumulativeByteCount + available;
100105
final HttpContentRange header;
101106
ByteRangeSpec rangeSpec = ByteRangeSpec.explicit(cumulativeByteCount, newFinalByteOffset);
102-
boolean quantumAligned = available % ByteSizeConstants._256KiB == 0;
103-
if (quantumAligned && finalize) {
107+
if (finalize) {
104108
header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
105109
finished = true;
106-
} else if (quantumAligned) {
110+
} else {
107111
header = HttpContentRange.of(rangeSpec);
108-
} else { // not quantum aligned, have to finalize
109-
header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
110-
finished = true;
111112
}
112113
try {
113114
ResumableOperationResult<@Nullable StorageObject> operationResult =
114115
session.put(content, header);
115116
long persistedSize = operationResult.getPersistedSize();
116117
committedBytesCallback.accept(persistedSize);
118+
long written = persistedSize - cumulativeByteCount;
117119
this.cumulativeByteCount = persistedSize;
118120
if (finished) {
119121
StorageObject storageObject = operationResult.getObject();
120122
result.set(storageObject);
121123
}
122-
return available;
124+
return written;
123125
} catch (Exception e) {
124126
result.setException(e);
125127
throw StorageException.coalesce(e);

‎google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,8 +64,7 @@ enum JsonResumableSessionFailureScenario {
6464
SCENARIO_7(
6565
BaseServiceException.UNKNOWN_CODE,
6666
"dataLoss",
67-
"Client side data loss detected. Bytes acked is more than client sent."),
68-
SCENARIO_9(503, "backendNotConnected", "Ack less than bytes sent");
67+
"Client side data loss detected. Bytes acked is more than client sent.");
6968

7069
private static final String PREFIX_I = "\t|< ";
7170
private static final String PREFIX_O = "\t|> ";

‎google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -109,10 +109,9 @@ public void rewindTo(long offset) {
109109
success = true;
110110
return ResumableOperationResult.incremental(ackRange.endOffset());
111111
} else if (ackRange.endOffset() < effectiveEnd) {
112-
StorageException se =
113-
JsonResumableSessionFailureScenario.SCENARIO_9.toStorageException(uploadId, response);
114-
span.setStatus(Status.UNKNOWN.withDescription(se.getMessage()));
115-
throw se;
112+
rewindTo(ackRange.endOffset());
113+
success = true;
114+
return ResumableOperationResult.incremental(ackRange.endOffset());
116115
} else {
117116
StorageException se =
118117
JsonResumableSessionFailureScenario.SCENARIO_7.toStorageException(uploadId, response);

‎google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java‎

Lines changed: 0 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -110,51 +110,6 @@ public void emptyObjectHappyPath() throws Exception {
110110
}
111111
}
112112

113-
/**
114-
*
115-
*
116-
* <h4>S.9</h4>
117-
*
118-
* Partial successful append to session
119-
*
120-
* <p>The client has sent N bytes, the server confirmed N bytes as committed. The client sends K
121-
* bytes starting at offset N. The server responds with only N + L with 0 &lt;= L &lt; K bytes as
122-
* committed.
123-
*/
124-
@Test
125-
public void scenario9() throws Exception {
126-
127-
HttpRequestHandler handler =
128-
req -> {
129-
String contentRangeString = req.headers().get(CONTENT_RANGE);
130-
HttpContentRange parse = HttpContentRange.parse(contentRangeString);
131-
long endInclusive = ((HttpContentRange.HasRange<?>) parse).range().endOffsetInclusive();
132-
FullHttpResponse resp =
133-
new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
134-
ByteRangeSpec range = ByteRangeSpec.explicitClosed(0L, endInclusive - 1);
135-
resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader());
136-
return resp;
137-
};
138-
139-
try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) {
140-
URI endpoint = fakeHttpServer.getEndpoint();
141-
String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
142-
143-
AtomicLong confirmedBytes = new AtomicLong(-1L);
144-
145-
JsonResumableSessionPutTask task =
146-
new JsonResumableSessionPutTask(
147-
httpClientContext,
148-
uploadUrl,
149-
RewindableContent.empty(),
150-
HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L)));
151-
152-
StorageException se = assertThrows(StorageException.class, task::call);
153-
assertThat(se.getCode()).isEqualTo(503);
154-
assertThat(confirmedBytes.get()).isEqualTo(-1L);
155-
}
156-
}
157-
158113
/**
159114
*
160115
*

‎google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ public void rewindWillQueryStatusOnlyWhenDirty() throws Exception {
9898
DefaultFullHttpResponse resp =
9999
new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE);
100100
if (range1.getHeaderValue().equals(contentRange)) {
101-
resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
101+
return new DefaultFullHttpResponse(req.protocolVersion(), SERVICE_UNAVAILABLE);
102102
} else if (range2.getHeaderValue().equals(contentRange)) {
103103
resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader());
104104
} else {

‎google-cloud-storage/src/test/java/com/google/cloud/storage/ITUnbufferedResumableUploadTest.java‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
/*
2+
* Copyright 2024 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.storage;
18+
19+
import static com.google.common.truth.Truth.assertThat;
20+
21+
import com.google.api.core.ApiFutures;
22+
import com.google.api.services.storage.model.StorageObject;
23+
import com.google.cloud.storage.ITUnbufferedResumableUploadTest.ObjectSizes;
24+
import com.google.cloud.storage.TransportCompatibility.Transport;
25+
import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
26+
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
27+
import com.google.cloud.storage.UnifiedOpts.Opts;
28+
import com.google.cloud.storage.it.runner.StorageITRunner;
29+
import com.google.cloud.storage.it.runner.annotations.Backend;
30+
import com.google.cloud.storage.it.runner.annotations.CrossRun;
31+
import com.google.cloud.storage.it.runner.annotations.CrossRun.Exclude;
32+
import com.google.cloud.storage.it.runner.annotations.Inject;
33+
import com.google.cloud.storage.it.runner.annotations.Parameterized;
34+
import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter;
35+
import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersProvider;
36+
import com.google.cloud.storage.it.runner.registry.Generator;
37+
import com.google.cloud.storage.spi.v1.StorageRpc;
38+
import com.google.common.collect.ImmutableList;
39+
import java.io.IOException;
40+
import java.math.BigInteger;
41+
import java.nio.ByteBuffer;
42+
import java.util.Map;
43+
import java.util.concurrent.ExecutionException;
44+
import java.util.concurrent.TimeUnit;
45+
import java.util.concurrent.TimeoutException;
46+
import java.util.function.Supplier;
47+
import org.junit.Test;
48+
import org.junit.runner.RunWith;
49+
50+
@RunWith(StorageITRunner.class)
51+
@CrossRun(
52+
backends = {Backend.PROD},
53+
transports = {Transport.HTTP, Transport.GRPC})
54+
@Parameterized(ObjectSizes.class)
55+
public final class ITUnbufferedResumableUploadTest {
56+
57+
@Inject public Storage storage;
58+
@Inject public BucketInfo bucket;
59+
@Inject public Generator generator;
60+
61+
@Parameter public int objectSize;
62+
63+
public static final class ObjectSizes implements ParametersProvider {
64+
65+
@Override
66+
public ImmutableList<Integer> parameters() {
67+
return ImmutableList.of(256 * 1024, 2 * 1024 * 1024);
68+
}
69+
}
70+
71+
@Test
72+
@Exclude(transports = Transport.GRPC)
73+
public void json()
74+
throws IOException, ExecutionException, InterruptedException, TimeoutException {
75+
BlobInfo blobInfo = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
76+
Opts<ObjectTargetOpt> opts = Opts.empty();
77+
final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
78+
BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
79+
BlobInfo updated = opts.blobInfoMapper().apply(builder).build();
80+
81+
StorageObject encode = Conversions.json().blobInfo().encode(updated);
82+
HttpStorageOptions options = (HttpStorageOptions) storage.getOptions();
83+
Supplier<String> uploadIdSupplier =
84+
ResumableMedia.startUploadForBlobInfo(
85+
options,
86+
updated,
87+
optionsMap,
88+
StorageRetryStrategy.getUniformStorageRetryStrategy().getIdempotentHandler());
89+
JsonResumableWrite jsonResumableWrite =
90+
JsonResumableWrite.of(encode, optionsMap, uploadIdSupplier.get(), 0);
91+
92+
UnbufferedWritableByteChannelSession<StorageObject> session =
93+
ResumableMedia.http()
94+
.write()
95+
.byteChannel(HttpClientContext.from(options.getStorageRpcV1()))
96+
.resumable()
97+
.unbuffered()
98+
.setStartAsync(ApiFutures.immediateFuture(jsonResumableWrite))
99+
.build();
100+
101+
int additional = 13;
102+
long size = objectSize + additional;
103+
ByteBuffer b = DataGenerator.base64Characters().genByteBuffer(size);
104+
105+
UnbufferedWritableByteChannel open = session.open();
106+
int written = open.write(b);
107+
assertThat(written).isEqualTo(objectSize);
108+
assertThat(b.remaining()).isEqualTo(additional);
109+
110+
int writtenAndClose = open.writeAndClose(b);
111+
assertThat(writtenAndClose).isEqualTo(additional);
112+
open.close();
113+
114+
StorageObject storageObject = session.getResult().get(2, TimeUnit.SECONDS);
115+
assertThat(storageObject.getSize()).isEqualTo(BigInteger.valueOf(size));
116+
}
117+
}

0 commit comments

Comments
 (0)