Skip to content

Commit 485be18

Browse files
BenWhiteheadBenWhitehead
authored
fix: fix appendable upload finalization race condition (#3295)
When finalizing an appendable upload depending on how quickly gcs is acking bytes, we could run into a case where finish_write: true was sent before all bytes had been enqueued. Regression introduced in 2.57.0
1 parent b0b0581 commit 485be18

File tree

5 files changed

+211
-79
lines changed

5 files changed

+211
-79
lines changed

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -145,6 +145,7 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th
145145
// those bytes to implicitly flag as dirty.
146146
rewindableContent.flagDirty();
147147

148+
long remainingAfterPacking = Buffers.totalRemaining(srcs, srcsOffset, srcsLength);
148149
long bytesConsumed = 0;
149150
for (int i = 0, len = data.length, lastIdx = len - 1; i < len; i++) {
150151
ChunkSegment datum = data[i];
@@ -153,7 +154,7 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th
153154
boolean appended;
154155
if (i < lastIdx && !shouldFlush) {
155156
appended = stream.append(datum);
156-
} else if (i == lastIdx && nextWriteShouldFinalize) {
157+
} else if (i == lastIdx && remainingAfterPacking == 0 && nextWriteShouldFinalize) {
157158
appended = stream.appendAndFinalize(datum);
158159
} else {
159160
appended = stream.appendAndFlush(datum);

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

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -416,10 +416,10 @@ final boolean offer(ChunkSegmenter.@NonNull ChunkSegment datum) {
416416
try {
417417
requireNonNull(datum, "data must be non null");
418418
validateCurrentStateIsOneOf(State.allNonTerminal);
419-
checkNotFinalizing();
420419
ByteString b = datum.getB();
421-
long availableCapacity = availableCapacity();
422420
int size = b.size();
421+
checkNotFinalizing(size);
422+
long availableCapacity = availableCapacity();
423423
if (size <= availableCapacity) {
424424
Crc32cLengthKnown crc32c = datum.getCrc32c();
425425
ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
@@ -449,7 +449,7 @@ public boolean finalFlush(long totalLength) {
449449
lock.lock();
450450
try {
451451
validateCurrentStateIsOneOf(State.allNonTerminal);
452-
checkNotFinalizing();
452+
checkNotFinalizing(0);
453453
checkArgument(
454454
totalLength == totalSentBytes,
455455
"(totalLength == totalSentBytes) (%s == %s)",
@@ -490,7 +490,7 @@ final boolean offer(@NonNull BidiWriteObjectRequest e) {
490490
requireNonNull(e, "e must be non null");
491491
validateCurrentStateIsOneOf(State.allNonTerminal);
492492
if (e.hasChecksummedData()) {
493-
checkNotFinalizing();
493+
checkNotFinalizing(e.getChecksummedData().getContent().size());
494494
}
495495
int size = e.getChecksummedData().getContent().size();
496496
long availableCapacity = availableCapacity();
@@ -827,10 +827,15 @@ protected final void validateCurrentStateIsOneOf(State... allowed) {
827827
state);
828828
}
829829

830-
private void checkNotFinalizing() {
830+
private void checkNotFinalizing(int size) {
831831
checkState(
832832
finishWriteOffset == -1,
833-
"Attempting to append bytes even though finalization has previously been signaled.");
833+
"Attempting to append bytes even though finalization has previously been signaled."
834+
+ " (finishWriteOffset: %s, totalSentBytes: %s, confirmedBytes: %s, size: %s)",
835+
finishWriteOffset,
836+
totalSentBytes,
837+
confirmedBytes,
838+
size);
834839
}
835840

836841
protected final boolean internalOffer(BidiWriteObjectRequest e) {

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

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
/*
2+
* Copyright 2025 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.SettableApiFuture;
22+
import com.google.api.gax.grpc.GrpcCallContext;
23+
import com.google.cloud.storage.BidiUploadState.AppendableUploadState;
24+
import com.google.cloud.storage.ITAppendableUploadFakeTest.FakeStorage;
25+
import com.google.cloud.storage.it.ChecksummedTestContent;
26+
import com.google.common.collect.ImmutableList;
27+
import com.google.storage.v2.BidiWriteObjectResponse;
28+
import java.io.IOException;
29+
import java.nio.ByteBuffer;
30+
import java.time.OffsetDateTime;
31+
import java.util.concurrent.Executors;
32+
import java.util.concurrent.ScheduledExecutorService;
33+
import java.util.concurrent.atomic.AtomicLong;
34+
import org.junit.Rule;
35+
import org.junit.Test;
36+
import org.junit.rules.TestName;
37+
38+
public final class BidiAppendableUnbufferedWritableByteChannelTest {
39+
@Rule public final TestName testName = new TestName();
40+
41+
@Test
42+
public void appendAndFinalizeOnlyPerformedIfAllBytesConsumed() throws IOException {
43+
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
44+
ChecksummedTestContent ctc = ChecksummedTestContent.gen(27);
45+
AppendableUploadState state =
46+
BidiUploadState.appendableNew(
47+
BidiUploadTest.appendRequestNew,
48+
GrpcCallContext::createDefault,
49+
16,
50+
SettableApiFuture.create(),
51+
Crc32cValue.zero());
52+
AtomicLong finishWriteOffset = new AtomicLong(-1);
53+
BidiUploadStreamingStream stream =
54+
new BidiUploadStreamingStream(
55+
state,
56+
executor,
57+
BidiUploadTestUtils.adaptOnlySend(
58+
respond ->
59+
request -> {
60+
if (request.getFinishWrite()) {
61+
finishWriteOffset.set(
62+
request.getWriteOffset()
63+
+ request.getChecksummedData().getContent().size());
64+
}
65+
executor.submit(
66+
() -> {
67+
switch ((int) request.getWriteOffset()) {
68+
case 0:
69+
respond.onResponse(BidiUploadTest.resourceWithSize(0));
70+
break;
71+
case 4:
72+
case 8:
73+
// do not ack any bytes until we receive 16, this simulates
74+
// latency on the bytes being ack'd.
75+
break;
76+
case 12:
77+
respond.onResponse(BidiUploadTestUtils.incremental(8));
78+
break;
79+
case 16:
80+
respond.onResponse(BidiUploadTestUtils.incremental(12));
81+
break;
82+
case 20:
83+
respond.onResponse(BidiUploadTestUtils.incremental(16));
84+
break;
85+
case 24:
86+
BidiWriteObjectResponse.Builder b =
87+
BidiUploadTest.resourceFor(ctc).toBuilder();
88+
b.getResourceBuilder()
89+
.setFinalizeTime(
90+
Conversions.grpc()
91+
.timestampCodec
92+
.encode(OffsetDateTime.now()));
93+
respond.onResponse(b.build());
94+
break;
95+
default:
96+
respond.onError(
97+
FakeStorage.unexpectedRequest(request, ImmutableList.of()));
98+
break;
99+
}
100+
});
101+
}),
102+
3,
103+
RetryContext.neverRetry());
104+
ChunkSegmenter chunkSegmenter =
105+
new ChunkSegmenter(Hasher.enabled(), ByteStringStrategy.copy(), 4, 2);
106+
BidiAppendableUnbufferedWritableByteChannel channel =
107+
new BidiAppendableUnbufferedWritableByteChannel(stream, chunkSegmenter, 4, 0);
108+
109+
ByteBuffer buf = ctc.asByteBuffer();
110+
int written1 = channel.write(buf);
111+
// fill up the outbound queue
112+
assertThat(written1).isEqualTo(16);
113+
114+
// asynchronously bytes will be ack'd 4 at a time, eventually there will be enough space in the
115+
// outbound queue to allow writeAndClose to start consuming bytes.
116+
channel.nextWriteShouldFinalize();
117+
int written2 = channel.writeAndClose(buf);
118+
assertThat(written2).isEqualTo(11);
119+
assertThat(finishWriteOffset.get()).isEqualTo(ctc.length());
120+
}
121+
}

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

Lines changed: 2 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717
package com.google.cloud.storage;
1818

1919
import static com.google.cloud.storage.BidiUploadState.appendableNew;
20+
import static com.google.cloud.storage.BidiUploadTestUtils.adaptOnlySend;
21+
import static com.google.cloud.storage.BidiUploadTestUtils.alwaysErrorBidiStreamingCallable;
2022
import static com.google.cloud.storage.BidiUploadTestUtils.createSegment;
2123
import static com.google.cloud.storage.BidiUploadTestUtils.finishAt;
2224
import static com.google.cloud.storage.BidiUploadTestUtils.incremental;
@@ -39,11 +41,7 @@
3941
import com.google.api.gax.grpc.GrpcCallContext;
4042
import com.google.api.gax.grpc.GrpcStatusCode;
4143
import com.google.api.gax.rpc.AbortedException;
42-
import com.google.api.gax.rpc.ApiCallContext;
4344
import com.google.api.gax.rpc.ApiExceptionFactory;
44-
import com.google.api.gax.rpc.BidiStreamingCallable;
45-
import com.google.api.gax.rpc.ClientStream;
46-
import com.google.api.gax.rpc.ClientStreamReadyObserver;
4745
import com.google.api.gax.rpc.ErrorDetails;
4846
import com.google.api.gax.rpc.ResponseObserver;
4947
import com.google.api.gax.rpc.StreamController;
@@ -64,7 +62,6 @@
6462
import com.google.common.collect.Range;
6563
import com.google.protobuf.Any;
6664
import com.google.protobuf.ByteString;
67-
import com.google.protobuf.Message;
6865
import com.google.protobuf.TextFormat;
6966
import com.google.rpc.Code;
7067
import com.google.storage.v2.AppendObjectSpec;
@@ -1840,59 +1837,6 @@ public <T extends Throwable> void recordError(
18401837
exec1.shutdownNow();
18411838
}
18421839
}
1843-
1844-
private static BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse>
1845-
alwaysErrorBidiStreamingCallable(Status status) {
1846-
return adaptOnlySend(respond -> request -> respond.onError(status.asRuntimeException()));
1847-
}
1848-
1849-
private static <ReqT extends Message, ResT> BidiStreamingCallable<ReqT, ResT> adaptOnlySend(
1850-
Function<ResponseObserver<ResT>, OnlySendClientStream<ReqT>> func) {
1851-
return adapt(func::apply);
1852-
}
1853-
1854-
private static <ReqT extends Message, ResT> BidiStreamingCallable<ReqT, ResT> adapt(
1855-
Function<ResponseObserver<ResT>, ClientStream<ReqT>> func) {
1856-
return adapt(
1857-
(respond, onReady, context) -> {
1858-
ClientStream<ReqT> clientStream = func.apply(respond);
1859-
StreamController controller = TestUtils.nullStreamController();
1860-
respond.onStart(controller);
1861-
return clientStream;
1862-
});
1863-
}
1864-
1865-
/**
1866-
* BidiStreamingCallable isn't functional even though it's a single abstract method.
1867-
*
1868-
* <p>Define a method that can adapt a TriFunc as the required implementation of {@link
1869-
* BidiStreamingCallable#internalCall(ResponseObserver, ClientStreamReadyObserver,
1870-
* ApiCallContext)}.
1871-
*
1872-
* <p>Saves several lines of boilerplate in each test.
1873-
*/
1874-
private static <ReqT, ResT> BidiStreamingCallable<ReqT, ResT> adapt(
1875-
StreamingStreamTest.TriFunc<
1876-
ResponseObserver<ResT>,
1877-
ClientStreamReadyObserver<ReqT>,
1878-
ApiCallContext,
1879-
ClientStream<ReqT>>
1880-
func) {
1881-
return new BidiStreamingCallable<ReqT, ResT>() {
1882-
@Override
1883-
public ClientStream<ReqT> internalCall(
1884-
ResponseObserver<ResT> respond,
1885-
ClientStreamReadyObserver<ReqT> onReady,
1886-
ApiCallContext context) {
1887-
return func.apply(respond, onReady, context);
1888-
}
1889-
};
1890-
}
1891-
1892-
@FunctionalInterface
1893-
interface TriFunc<A, B, C, R> {
1894-
R apply(A a, B b, C c);
1895-
}
18961840
}
18971841

18981842
public static final class BidiUploadStreamingStreamResponseObserverTest {
@@ -2289,18 +2233,4 @@ static BidiWriteObjectRequest flushOffset(long offset) {
22892233
}
22902234
return BidiWriteObjectResponse.newBuilder().setResource(f.apply(b)).build();
22912235
}
2292-
2293-
@FunctionalInterface
2294-
private interface OnlySendClientStream<ReqT> extends ClientStream<ReqT> {
2295-
@Override
2296-
default void closeSendWithError(Throwable t) {}
2297-
2298-
@Override
2299-
default void closeSend() {}
2300-
2301-
@Override
2302-
default boolean isSendReady() {
2303-
return true;
2304-
}
2305-
}
23062236
}

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

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,12 +21,19 @@
2121

2222
import com.google.api.gax.grpc.GrpcStatusCode;
2323
import com.google.api.gax.rpc.AbortedException;
24+
import com.google.api.gax.rpc.ApiCallContext;
25+
import com.google.api.gax.rpc.BidiStreamingCallable;
26+
import com.google.api.gax.rpc.ClientStream;
27+
import com.google.api.gax.rpc.ClientStreamReadyObserver;
2428
import com.google.api.gax.rpc.ErrorDetails;
29+
import com.google.api.gax.rpc.ResponseObserver;
30+
import com.google.api.gax.rpc.StreamController;
2531
import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
2632
import com.google.cloud.storage.it.ChecksummedTestContent;
2733
import com.google.common.collect.ImmutableList;
2834
import com.google.protobuf.Any;
2935
import com.google.protobuf.ByteString;
36+
import com.google.protobuf.Message;
3037
import com.google.protobuf.Timestamp;
3138
import com.google.rpc.Code;
3239
import com.google.storage.v2.BidiWriteObjectRedirectedError;
@@ -39,6 +46,7 @@
3946
import java.nio.ByteBuffer;
4047
import java.time.OffsetDateTime;
4148
import java.util.List;
49+
import java.util.function.Function;
4250
import org.checkerframework.checker.nullness.qual.NonNull;
4351

4452
final class BidiUploadTestUtils {
@@ -122,4 +130,71 @@ static BidiWriteObjectRequest withFlushAndStateLookup(BidiWriteObjectRequest ori
122130
static Timestamp timestampNow() {
123131
return Conversions.grpc().timestampCodec.encode(OffsetDateTime.now());
124132
}
133+
134+
static BidiStreamingCallable<BidiWriteObjectRequest, BidiWriteObjectResponse>
135+
alwaysErrorBidiStreamingCallable(Status status) {
136+
return adaptOnlySend(respond -> request -> respond.onError(status.asRuntimeException()));
137+
}
138+
139+
static <ReqT extends Message, ResT> BidiStreamingCallable<ReqT, ResT> adaptOnlySend(
140+
Function<ResponseObserver<ResT>, OnlySendClientStream<ReqT>> func) {
141+
return adapt(func::apply);
142+
}
143+
144+
static <ReqT extends Message, ResT> BidiStreamingCallable<ReqT, ResT> adapt(
145+
Function<ResponseObserver<ResT>, ClientStream<ReqT>> func) {
146+
return adapt(
147+
(respond, onReady, context) -> {
148+
ClientStream<ReqT> clientStream = func.apply(respond);
149+
StreamController controller = TestUtils.nullStreamController();
150+
respond.onStart(controller);
151+
return clientStream;
152+
});
153+
}
154+
155+
/**
156+
* BidiStreamingCallable isn't functional even though it's a single abstract method.
157+
*
158+
* <p>Define a method that can adapt a TriFunc as the required implementation of {@link
159+
* BidiStreamingCallable#internalCall(ResponseObserver, ClientStreamReadyObserver,
160+
* ApiCallContext)}.
161+
*
162+
* <p>Saves several lines of boilerplate in each test.
163+
*/
164+
static <ReqT, ResT> BidiStreamingCallable<ReqT, ResT> adapt(
165+
TriFunc<
166+
ResponseObserver<ResT>,
167+
ClientStreamReadyObserver<ReqT>,
168+
ApiCallContext,
169+
ClientStream<ReqT>>
170+
func) {
171+
return new BidiStreamingCallable<ReqT, ResT>() {
172+
@Override
173+
public ClientStream<ReqT> internalCall(
174+
ResponseObserver<ResT> respond,
175+
ClientStreamReadyObserver<ReqT> onReady,
176+
ApiCallContext context) {
177+
return func.apply(respond, onReady, context);
178+
}
179+
};
180+
}
181+
182+
@FunctionalInterface
183+
interface TriFunc<A, B, C, R> {
184+
R apply(A a, B b, C c);
185+
}
186+
187+
@FunctionalInterface
188+
interface OnlySendClientStream<ReqT> extends ClientStream<ReqT> {
189+
@Override
190+
default void closeSendWithError(Throwable t) {}
191+
192+
@Override
193+
default void closeSend() {}
194+
195+
@Override
196+
default boolean isSendReady() {
197+
return true;
198+
}
199+
}
125200
}

0 commit comments

Comments
 (0)