Skip to content

Commit 20b4818

Browse files
author
Dhriti Chopra
committed
feat(storage): allow explicit checksum on appendable upload finalization
1 parent 2f915ab commit 20b4818

6 files changed

Lines changed: 232 additions & 7 deletions

File tree

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

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import java.util.concurrent.ExecutionException;
2727
import java.util.concurrent.TimeUnit;
2828
import java.util.concurrent.TimeoutException;
29+
import org.checkerframework.checker.nullness.qual.Nullable;
2930

3031
final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
3132

@@ -36,6 +37,7 @@ final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWri
3637
private boolean open;
3738
private long writeOffset;
3839
private volatile boolean nextWriteShouldFinalize;
40+
private @Nullable String expectedCrc32c;
3941
private boolean writeCalledAtLeastOnce;
4042
private long lastFlushOffset;
4143

@@ -53,6 +55,7 @@ final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWri
5355
this.open = true;
5456
this.writeOffset = writeOffset;
5557
this.nextWriteShouldFinalize = false;
58+
this.expectedCrc32c = null;
5659
this.writeThrewError = false;
5760
this.lastFlushOffset = writeOffset;
5861
}
@@ -96,7 +99,7 @@ public void close() throws IOException {
9699
}
97100
if (nextWriteShouldFinalize) {
98101
//noinspection StatementWithEmptyBody
99-
while (!stream.finishWrite(writeOffset)) {}
102+
while (!stream.finishWrite(writeOffset, expectedCrc32c)) {}
100103
} else {
101104
//noinspection StatementWithEmptyBody
102105
while (!stream.closeStream(writeOffset)) {}
@@ -113,6 +116,11 @@ public void nextWriteShouldFinalize() {
113116
this.nextWriteShouldFinalize = true;
114117
}
115118

119+
public void nextWriteShouldFinalize(String expectedCrc32c) {
120+
this.nextWriteShouldFinalize = true;
121+
this.expectedCrc32c = expectedCrc32c;
122+
}
123+
116124
void flush() throws InterruptedException {
117125
stream.flush();
118126
stream.awaitAckOf(writeOffset);
@@ -155,7 +163,7 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th
155163
if (i < lastIdx && !shouldFlush) {
156164
appended = stream.append(datum);
157165
} else if (i == lastIdx && remainingAfterPacking == 0 && nextWriteShouldFinalize) {
158-
appended = stream.appendAndFinalize(datum);
166+
appended = stream.appendAndFinalize(datum, expectedCrc32c);
159167
} else {
160168
appended = stream.appendAndFlush(datum);
161169
}

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

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -126,12 +126,12 @@ public boolean appendAndFlush(ChunkSegmenter.@NonNull ChunkSegment data) {
126126
}
127127
}
128128

129-
public boolean appendAndFinalize(ChunkSegmenter.@NonNull ChunkSegment data) {
129+
public boolean appendAndFinalize(ChunkSegmenter.@NonNull ChunkSegment data, @Nullable String expectedCrc32c) {
130130
lock.lock();
131131
try {
132132
boolean offered = state.offer(data);
133133
if (offered) {
134-
finishWrite(state.getTotalSentBytes());
134+
finishWrite(state.getTotalSentBytes(), expectedCrc32c);
135135
}
136136
return offered;
137137
} finally {
@@ -163,6 +163,10 @@ public void flush() {
163163
}
164164

165165
public boolean finishWrite(long length) {
166+
return finishWrite(length, null);
167+
}
168+
169+
public boolean finishWrite(long length, @Nullable String expectedCrc32c) {
166170
lock.lock();
167171
try {
168172
// if we're already finalizing, ack rather than enqueueing again
@@ -172,10 +176,16 @@ public boolean finishWrite(long length) {
172176

173177
BidiWriteObjectRequest.Builder b =
174178
BidiWriteObjectRequest.newBuilder().setWriteOffset(length).setFinishWrite(true);
175-
Crc32cLengthKnown cumulativeCrc32c = state.getCumulativeCrc32c();
176-
if (cumulativeCrc32c != null) {
179+
if (expectedCrc32c != null) {
180+
int crc32cInt = Utils.crc32cCodec.decode(expectedCrc32c);
177181
b.setObjectChecksums(
178-
ObjectChecksums.newBuilder().setCrc32C(cumulativeCrc32c.getValue()).build());
182+
ObjectChecksums.newBuilder().setCrc32C(crc32cInt).build());
183+
} else {
184+
Crc32cLengthKnown cumulativeCrc32c = state.getCumulativeCrc32c();
185+
if (cumulativeCrc32c != null) {
186+
b.setObjectChecksums(
187+
ObjectChecksums.newBuilder().setCrc32C(cumulativeCrc32c.getValue()).build());
188+
}
179189
}
180190
BidiWriteObjectRequest msg = b.build();
181191
boolean offer = state.offer(msg);

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

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,31 @@ interface AppendableUploadWriteableByteChannel extends WritableByteChannel {
158158
@BetaApi
159159
void finalizeAndClose() throws IOException;
160160

161+
/**
162+
* <b>This method is blocking</b>
163+
*
164+
* <p>Finalize the upload and close this instance to further {@link #write(ByteBuffer)}ing. This
165+
* will close any underlying stream and release any releasable resources once out of scope.
166+
*
167+
* <p>Once this method is called, and returns no more writes to the object will be allowed by
168+
* GCS.
169+
*
170+
* <p>This method and {@link #close()} are mutually exclusive. If one of the other methods are
171+
* called before this method, this method will be a no-op.
172+
*
173+
* @param expectedCrc32c A Base64 encoded string representing the expected CRC32c value for the
174+
* entire object. If provided, the server will validate the final object's CRC32c against this
175+
* value. If there's a mismatch, the server will return an error (such as InvalidArgument),
176+
* and this method will throw a {@link StorageException} or equivalent exception, failing the upload.
177+
* @see Storage#blobAppendableUpload(BlobInfo, BlobAppendableUploadConfig, BlobWriteOption...)
178+
* @see BlobAppendableUploadConfig.CloseAction#FINALIZE_WHEN_CLOSING
179+
* @see BlobAppendableUploadConfig#getCloseAction()
180+
* @see BlobAppendableUploadConfig#withCloseAction(CloseAction)
181+
* @since 2.51.0 This new api is in preview and is subject to breaking changes.
182+
*/
183+
@BetaApi
184+
void finalizeAndClose(String expectedCrc32c) throws IOException;
185+
161186
/**
162187
* <b>This method is blocking</b>
163188
*
@@ -197,5 +222,27 @@ interface AppendableUploadWriteableByteChannel extends WritableByteChannel {
197222
*/
198223
@BetaApi
199224
void close() throws IOException;
225+
226+
/**
227+
* <b>This method is blocking</b>
228+
*
229+
* <p>Close this instance to further {@link #write(ByteBuffer)}ing.
230+
*
231+
* <p>This method behaves like {@link #close()}, but requires that the stream was configured with
232+
* {@link CloseAction#FINALIZE_WHEN_CLOSING}. If the stream is not configured to finalize on close,
233+
* this method will throw an {@link IllegalArgumentException}.
234+
*
235+
* @param expectedCrc32c A Base64 encoded string representing the expected CRC32c value for the
236+
* entire object. If provided, the server will validate the final object's CRC32c against this
237+
* value. If there's a mismatch, the server will return an error (such as InvalidArgument),
238+
* and this method will throw a {@link StorageException} or equivalent exception, failing the upload.
239+
* @throws IllegalArgumentException if the stream was not configured with {@link CloseAction#FINALIZE_WHEN_CLOSING}.
240+
* @see Storage#blobAppendableUpload(BlobInfo, BlobAppendableUploadConfig, BlobWriteOption...)
241+
* @see BlobAppendableUploadConfig#getCloseAction()
242+
* @see BlobAppendableUploadConfig#withCloseAction(CloseAction)
243+
* @since 2.51.0 This new api is in preview and is subject to breaking changes.
244+
*/
245+
@BetaApi
246+
void close(String expectedCrc32c) throws IOException;
200247
}
201248
}

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

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,19 @@ public void finalizeAndClose() throws IOException {
132132
}
133133
}
134134

135+
@Override
136+
public void finalizeAndClose(String expectedCrc32c) throws IOException {
137+
lock.lock();
138+
try {
139+
if (buffered.isOpen()) {
140+
unbuffered.nextWriteShouldFinalize(expectedCrc32c);
141+
buffered.close();
142+
}
143+
} finally {
144+
lock.unlock();
145+
}
146+
}
147+
135148
@Override
136149
public void closeWithoutFinalizing() throws IOException {
137150
lock.lock();
@@ -152,5 +165,15 @@ public void close() throws IOException {
152165
closeWithoutFinalizing();
153166
}
154167
}
168+
169+
@Override
170+
public void close(String expectedCrc32c) throws IOException {
171+
if (finalizeOnClose) {
172+
finalizeAndClose(expectedCrc32c);
173+
} else {
174+
throw new IllegalArgumentException(
175+
"expectedCrc32c can only be provided when finalizeOnClose is true.");
176+
}
177+
}
155178
}
156179
}

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

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2195,6 +2195,32 @@ public void finalizeAndClose() throws IOException {
21952195
}
21962196
}
21972197

2198+
@Override
2199+
@BetaApi
2200+
public void finalizeAndClose(String expectedCrc32c) throws IOException {
2201+
try (Scope ignore = openSpan.makeCurrent()) {
2202+
Span span = tracer.spanBuilder("finalizeAndClose").startSpan();
2203+
try (Scope ignore2 = span.makeCurrent()) {
2204+
delegate.finalizeAndClose(expectedCrc32c);
2205+
} catch (Throwable t) {
2206+
span.recordException(t);
2207+
span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
2208+
throw t;
2209+
} finally {
2210+
span.end();
2211+
}
2212+
} catch (IOException | RuntimeException e) {
2213+
openSpan.recordException(e);
2214+
openSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2215+
uploadSpan.recordException(e);
2216+
uploadSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2217+
throw e;
2218+
} finally {
2219+
openSpan.end();
2220+
uploadSpan.end();
2221+
}
2222+
}
2223+
21982224
@Override
21992225
@BetaApi
22002226
public void closeWithoutFinalizing() throws IOException {
@@ -2247,6 +2273,32 @@ public void close() throws IOException {
22472273
}
22482274
}
22492275

2276+
@Override
2277+
@BetaApi
2278+
public void close(String expectedCrc32c) throws IOException {
2279+
try (Scope ignore = openSpan.makeCurrent()) {
2280+
Span span = tracer.spanBuilder("close").startSpan();
2281+
try (Scope ignore2 = span.makeCurrent()) {
2282+
delegate.close(expectedCrc32c);
2283+
} catch (Throwable t) {
2284+
span.recordException(t);
2285+
span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
2286+
throw t;
2287+
} finally {
2288+
span.end();
2289+
}
2290+
} catch (IOException | RuntimeException e) {
2291+
openSpan.recordException(e);
2292+
openSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2293+
uploadSpan.recordException(e);
2294+
uploadSpan.setStatus(StatusCode.ERROR, e.getClass().getSimpleName());
2295+
throw e;
2296+
} finally {
2297+
openSpan.end();
2298+
uploadSpan.end();
2299+
}
2300+
}
2301+
22502302
@Override
22512303
public void flush() throws IOException {
22522304
try (Scope ignore = openSpan.makeCurrent()) {

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

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,91 @@ public void takeoverJustToFinalizeWorks() throws Exception {
261261
() -> assertThat(done2.getSize()).isEqualTo(p.content.length()),
262262
() -> assertThat(done2.getCrc32c()).isNotNull());
263263
}
264+
@Test
265+
public void explicitFinalizeWithCorrectChecksum() throws Exception {
266+
BlobId bid = BlobId.of(bucket.getName(), UUID.randomUUID().toString());
267+
assumeTrue(
268+
"manually finalizing",
269+
p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING);
270+
271+
BlobAppendableUpload upload =
272+
storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig);
273+
274+
try (AppendableUploadWriteableByteChannel channel = upload.open()) {
275+
int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel);
276+
assertThat(written).isEqualTo(p.content.length());
277+
278+
String expectedCrc = Utils.crc32cCodec.encode(p.content.getCrc32c());
279+
channel.finalizeAndClose(expectedCrc);
280+
}
281+
282+
BlobInfo gen1 = upload.getResult().get(5, TimeUnit.SECONDS);
283+
assertThat(gen1.getSize()).isEqualTo(p.content.length());
284+
assertThat(gen1.getCrc32c()).isEqualTo(Utils.crc32cCodec.encode(p.content.getCrc32c()));
285+
}
286+
287+
@Test
288+
public void explicitFinalizeWithIncorrectChecksumFails() throws Exception {
289+
BlobId bid = BlobId.of(bucket.getName(), UUID.randomUUID().toString());
290+
assumeTrue(
291+
"manually finalizing",
292+
p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING);
293+
294+
BlobAppendableUpload upload =
295+
storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig);
296+
297+
try (AppendableUploadWriteableByteChannel channel = upload.open()) {
298+
int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel);
299+
assertThat(written).isEqualTo(p.content.length());
300+
301+
String badCrc = Utils.crc32cCodec.encode(Crc32cValue.zero().getValue());
302+
channel.finalizeAndClose(badCrc);
303+
}
304+
305+
try {
306+
upload.getResult().get(5, TimeUnit.SECONDS);
307+
org.junit.Assert.fail("Expected exception due to checksum mismatch");
308+
} catch (ExecutionException e) {
309+
// The server rejects it
310+
assertThat(e.getCause().getMessage().toLowerCase()).contains("mismatch");
311+
}
312+
}
313+
314+
@Test
315+
@CrossRun.Ignore(backends = {Backend.TEST_BENCH})
316+
public void takeoverJustToFinalizeWithIncorrectChecksumFails() throws Exception {
317+
BlobId bid = BlobId.of(bucket.getName(), UUID.randomUUID().toString());
318+
assumeTrue(
319+
"manually finalizing",
320+
p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING);
321+
322+
BlobAppendableUpload upload =
323+
storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig);
324+
try (AppendableUploadWriteableByteChannel channel = upload.open()) {
325+
int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel);
326+
assertThat(written).isEqualTo(p.content.length());
327+
}
328+
BlobInfo done1 = upload.getResult().get(5, TimeUnit.SECONDS);
329+
assertThat(done1.getSize()).isEqualTo(p.content.length());
330+
assertThat(done1.getCrc32c()).isEqualTo(Utils.crc32cCodec.encode(p.content.getCrc32c()));
331+
332+
BlobAppendableUpload takeOver =
333+
storage.blobAppendableUpload(
334+
BlobInfo.newBuilder(done1.getBlobId()).build(), p.uploadConfig);
335+
336+
String badCrc = Utils.crc32cCodec.encode(Crc32cValue.zero().getValue());
337+
try (AppendableUploadWriteableByteChannel channel = takeOver.open()) {
338+
channel.finalizeAndClose(badCrc);
339+
}
340+
341+
try {
342+
takeOver.getResult().get(5, TimeUnit.SECONDS);
343+
org.junit.Assert.fail("Expected exception due to checksum mismatch");
344+
} catch (ExecutionException e) {
345+
// The server rejects it
346+
assertThat(e.getCause().getMessage().toLowerCase()).contains("mismatch");
347+
}
348+
}
264349

265350
private void checkTestbenchIssue733() {
266351
if (backend == Backend.TEST_BENCH

0 commit comments

Comments
 (0)