Skip to content

Commit 52c0218

Browse files
cjc25cjc25
authored
feat(storage): Unfinalized appendable objects. (#11647)
Allow callers to close an upload to an appendable object without finalizing it.
1 parent 14e8e13 commit 52c0218

File tree

7 files changed

+185
-4
lines changed

7 files changed

+185
-4
lines changed

‎storage/client.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -266,6 +266,10 @@ type openWriterParams struct {
266266
// append - Write with appendable object semantics.
267267
// Optional.
268268
append bool
269+
// finalizeOnClose - Finalize the object when the storage.Writer is closed
270+
// successfully.
271+
// Optional.
272+
finalizeOnClose bool
269273

270274
// Writer callbacks
271275

‎storage/client_test.go‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -971,6 +971,7 @@ func TestOpenAppendableWriterEmulated(t *testing.T) {
971971
vc := &Client{tc: client}
972972
w := vc.Bucket(bucket).Object(objName).NewWriter(ctx)
973973
w.Append = true
974+
w.FinalizeOnClose = true
974975
_, err = w.Write(randomBytesToWrite)
975976
if err != nil {
976977
t.Fatalf("writing test data: got %v; want ok", err)
@@ -996,6 +997,14 @@ func TestOpenAppendableWriterEmulated(t *testing.T) {
996997
if diff := cmp.Diff(got, randomBytesToWrite); diff != "" {
997998
t.Fatalf("checking written content: got(-), want(+):\n%s", diff)
998999
}
1000+
1001+
o, err := veneerClient.Bucket(bucket).Object(objName).Attrs(ctx)
1002+
if err != nil {
1003+
t.Fatalf("getting object attrs: got %v; want ok", err)
1004+
}
1005+
if o.Finalized.IsZero() {
1006+
t.Errorf("expected valid finalize time: got %v; want non-zero", o.Finalized)
1007+
}
9991008
})
10001009
}
10011010

@@ -1014,6 +1023,7 @@ func TestOpenAppendableWriterMultipleChunksEmulated(t *testing.T) {
10141023
vc := &Client{tc: client}
10151024
w := vc.Bucket(bucket).Object(objName).NewWriter(ctx)
10161025
w.Append = true
1026+
w.FinalizeOnClose = true
10171027
// This should chunk the request into three separate flushes to storage.
10181028
w.ChunkSize = MiB
10191029
var lastReportedOffset int64
@@ -1051,6 +1061,59 @@ func TestOpenAppendableWriterMultipleChunksEmulated(t *testing.T) {
10511061
if diff := cmp.Diff(got, randomBytes3MiB); diff != "" {
10521062
t.Fatalf("checking written content: got(-), want(+):\n%s", diff)
10531063
}
1064+
1065+
o, err := veneerClient.Bucket(bucket).Object(objName).Attrs(ctx)
1066+
if err != nil {
1067+
t.Fatalf("getting object attrs: got %v; want ok", err)
1068+
}
1069+
if o.Finalized.IsZero() {
1070+
t.Errorf("expected valid finalize time: got %v; want non-zero", o.Finalized)
1071+
}
1072+
})
1073+
}
1074+
1075+
func TestOpenAppendableWriterLeaveUnfinalizedEmulated(t *testing.T) {
1076+
transportClientTest(skipHTTP("appends only supported via gRPC"), t, func(t *testing.T, ctx context.Context, project, bucket string, client storageClient) {
1077+
// Populate test data.
1078+
_, err := client.CreateBucket(ctx, project, bucket, &BucketAttrs{
1079+
Name: bucket,
1080+
}, nil)
1081+
if err != nil {
1082+
t.Fatalf("client.CreateBucket: %v", err)
1083+
}
1084+
prefix := time.Now().Nanosecond()
1085+
objName := fmt.Sprintf("%d-object-%d", prefix, time.Now().Nanosecond())
1086+
1087+
vc := &Client{tc: client}
1088+
w := vc.Bucket(bucket).Object(objName).NewWriter(ctx)
1089+
w.Append = true
1090+
w.FinalizeOnClose = false
1091+
var lastReportedOffset int64
1092+
w.ProgressFunc = func(offset int64) {
1093+
lastReportedOffset = offset
1094+
}
1095+
_, err = w.Write(randomBytesToWrite)
1096+
wantLen := int64(len(randomBytesToWrite))
1097+
if err != nil {
1098+
t.Fatalf("writing test data: got %v; want ok", err)
1099+
}
1100+
if err := w.Close(); err != nil {
1101+
t.Fatalf("closing test data writer: got %v; want ok", err)
1102+
}
1103+
if lastReportedOffset != wantLen {
1104+
t.Errorf("incorrect final progress report: got %d; want %d", lastReportedOffset, wantLen)
1105+
}
1106+
// No finalize time on the object.
1107+
o, err := vc.Bucket(bucket).Object(objName).Attrs(ctx)
1108+
if err != nil {
1109+
t.Fatalf("getting object attrs: got %v; want ok", err)
1110+
}
1111+
if o.Created.IsZero() {
1112+
t.Errorf("expected valid create time: got %v; want non-zero", o.Created)
1113+
}
1114+
if !o.Finalized.IsZero() {
1115+
t.Errorf("unexpected valid finalize time: got %v; want zero", o.Finalized)
1116+
}
10541117
})
10551118
}
10561119

@@ -1069,6 +1132,7 @@ func TestWriterFlushEmulated(t *testing.T) {
10691132
vc := &Client{tc: client}
10701133
w := vc.Bucket(bucket).Object(objName).NewWriter(ctx)
10711134
w.Append = true
1135+
w.FinalizeOnClose = true
10721136
w.ChunkSize = 3 * MiB
10731137
var gotOffsets []int64
10741138
w.ProgressFunc = func(offset int64) {
@@ -1188,6 +1252,7 @@ func TestWriterFlushAtCloseEmulated(t *testing.T) {
11881252
vc := &Client{tc: client}
11891253
w := vc.Bucket(bucket).Object(objName).NewWriter(ctx)
11901254
w.Append = true
1255+
w.FinalizeOnClose = true
11911256
w.ChunkSize = MiB
11921257
var gotOffsets []int64
11931258
w.ProgressFunc = func(offset int64) {
@@ -1268,6 +1333,7 @@ func TestWriterSmallFlushEmulated(t *testing.T) {
12681333
vc := &Client{tc: client}
12691334
w := vc.Bucket(bucket).Object(objName).NewWriter(ctx)
12701335
w.Append = true
1336+
w.FinalizeOnClose = true
12711337
w.ChunkSize = MiB
12721338
var gotOffsets []int64
12731339
w.ProgressFunc = func(offset int64) {

‎storage/grpc_client.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2638,6 +2638,7 @@ func newGRPCWriter(c *grpcStorageClient, s *settings, params *openWriterParams,
26382638
forceOneShot: params.chunkSize <= 0,
26392639
forceEmptyContentType: params.forceEmptyContentType,
26402640
append: params.append,
2641+
finalizeOnClose: params.finalizeOnClose,
26412642
setPipeWriter: setPipeWriter,
26422643
flushComplete: make(chan int64),
26432644
}, nil
@@ -2666,6 +2667,7 @@ type gRPCWriter struct {
26662667
forceOneShot bool
26672668
forceEmptyContentType bool
26682669
append bool
2670+
finalizeOnClose bool
26692671

26702672
streamSender gRPCBidiWriteBufferSender
26712673
flushInProgress bool // true when the pipe is being recreated for a flush.

‎storage/grpc_writer.go‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ type gRPCAppendBidiWriteBufferSender struct {
4040
firstMessage *storagepb.BidiWriteObjectRequest
4141
objectChecksums *storagepb.ObjectChecksums
4242

43+
finalizeOnClose bool
44+
4345
forceFirstMessage bool
4446
progress func(int64)
4547
flushOffset int64
@@ -63,6 +65,7 @@ func (w *gRPCWriter) newGRPCAppendBidiWriteBufferSender() (*gRPCAppendBidiWriteB
6365
CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey),
6466
},
6567
objectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs),
68+
finalizeOnClose: w.finalizeOnClose,
6669
forceFirstMessage: true,
6770
progress: w.progress,
6871
}
@@ -226,9 +229,16 @@ func (s *gRPCAppendBidiWriteBufferSender) receiveMessages(resps chan<- *storagep
226229
}
227230

228231
func (s *gRPCAppendBidiWriteBufferSender) sendOnConnectedStream(buf []byte, offset int64, flush, finishWrite, sendFirstMessage bool) (obj *storagepb.Object, err error) {
229-
req := bidiWriteObjectRequest(buf, offset, flush, finishWrite)
232+
var req *storagepb.BidiWriteObjectRequest
233+
finalizeObject := finishWrite && s.finalizeOnClose
230234
if finishWrite {
231-
// appendable objects pass checksums on the last message only
235+
// Always flush when finishing the Write, even if not finalizing.
236+
req = bidiWriteObjectRequest(buf, offset, true, finalizeObject)
237+
} else {
238+
req = bidiWriteObjectRequest(buf, offset, flush, false)
239+
}
240+
if finalizeObject {
241+
// appendable objects pass checksums on the finalize message only
232242
req.ObjectChecksums = s.objectChecksums
233243
}
234244
if sendFirstMessage {

‎storage/integration_test.go‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3170,6 +3170,80 @@ func TestIntegration_WriterChunksize(t *testing.T) {
31703170
})
31713171
}
31723172

3173+
// Basic Writer test for appendable uploads with and without finalization.
3174+
func TestIntegration_WriterAppend(t *testing.T) {
3175+
ctx := skipAllButBidi(context.Background(), "ZB test")
3176+
multiTransportTest(ctx, t, func(t *testing.T, ctx context.Context, _, prefix string, client *Client) {
3177+
h := testHelper{t}
3178+
bucketName := prefix + uidSpace.New()
3179+
bkt := client.Bucket(bucketName)
3180+
h.mustCreate(bkt, testutil.ProjID(), &BucketAttrs{
3181+
Location: "us-central1",
3182+
CustomPlacementConfig: &CustomPlacementConfig{
3183+
DataLocations: []string{"us-central1-a"},
3184+
},
3185+
StorageClass: "RAPID",
3186+
HierarchicalNamespace: &HierarchicalNamespace{
3187+
Enabled: true,
3188+
},
3189+
UniformBucketLevelAccess: UniformBucketLevelAccess{
3190+
Enabled: true,
3191+
},
3192+
})
3193+
defer h.mustDeleteBucket(bkt)
3194+
3195+
testCases := []struct {
3196+
name string
3197+
finalize bool
3198+
content []byte
3199+
chunkSize int
3200+
}{
3201+
{
3202+
name: "finalized_object",
3203+
finalize: true,
3204+
content: randomBytes9MiB,
3205+
chunkSize: 4 * MiB,
3206+
},
3207+
{
3208+
name: "unfinalized_object",
3209+
finalize: false,
3210+
content: randomBytes9MiB,
3211+
chunkSize: 4 * MiB,
3212+
},
3213+
}
3214+
for _, tc := range testCases {
3215+
t.Run(tc.name, func(t *testing.T) {
3216+
// Create writer and upload content.
3217+
obj := bkt.Object(tc.name + uidSpace.New())
3218+
defer h.mustDeleteObject(obj)
3219+
w := obj.Retryer(WithPolicy(RetryAlways)).If(Conditions{DoesNotExist: true}).NewWriter(ctx)
3220+
w.Append = true
3221+
w.FinalizeOnClose = tc.finalize
3222+
w.ChunkSize = tc.chunkSize
3223+
3224+
h.mustWrite(w, tc.content)
3225+
3226+
// Download content again and validate.
3227+
// Disabled due to b/408373388; unskip after this is resolved.
3228+
// gotBytes := h.mustRead(obj)
3229+
// if !bytes.Equal(gotBytes, tc.content) {
3230+
// t.Errorf("content mismatch: got %v bytes, want %v bytes", len(gotBytes), len(tc.content))
3231+
// }
3232+
3233+
// Check object exists and Finalized attribute set as expected.
3234+
attrs := h.mustObjectAttrs(obj)
3235+
if tc.finalize && attrs.Finalized.IsZero() {
3236+
t.Errorf("got unfinalized object, want finalized")
3237+
}
3238+
if !tc.finalize && !attrs.Finalized.IsZero() {
3239+
t.Errorf("got object finalized at %v, want unfinalized", attrs.Finalized)
3240+
}
3241+
3242+
})
3243+
}
3244+
})
3245+
}
3246+
31733247
func TestIntegration_ZeroSizedObject(t *testing.T) {
31743248
t.Parallel()
31753249
multiTransportTest(context.Background(), t, func(t *testing.T, ctx context.Context, bucket, _ string, client *Client) {
@@ -6999,6 +7073,13 @@ func skipExtraReadAPIs(ctx context.Context, reason string) context.Context {
69997073
return context.WithValue(ctx, skipTransportTestKey("jsonReads"), reason)
70007074
}
70017075

7076+
// Skip all APIs except Bidi reads. Use for ZB tests.
7077+
func skipAllButBidi(ctx context.Context, reason string) context.Context {
7078+
ctx = context.WithValue(ctx, skipTransportTestKey("http"), reason)
7079+
ctx = context.WithValue(ctx, skipTransportTestKey("grpc"), reason)
7080+
return context.WithValue(ctx, skipTransportTestKey("jsonReads"), reason)
7081+
}
7082+
70027083
func skipXMLReads(ctx context.Context, reason string) context.Context {
70037084
return context.WithValue(ctx, skipTransportTestKey("http"), reason)
70047085
}

‎storage/retry_conformance_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -643,6 +643,7 @@ var methods = map[string][]retryFunc{
643643
objW := obj.NewWriter(ctx)
644644
objW.ChunkSize = MiB
645645
objW.Append = true
646+
objW.FinalizeOnClose = true
646647

647648
if _, err := objW.Write(randomBytes3MiB); err != nil {
648649
return fmt.Errorf("Writer.Write: %v", err)
@@ -663,7 +664,7 @@ var methods = map[string][]retryFunc{
663664
}
664665

665666
if d := cmp.Diff(content, randomBytes3MiB); d != "" {
666-
return fmt.Errorf("content got(-),want(+):\n%v", d)
667+
return fmt.Errorf("content mismatch, got %v bytes, want %v bytes", len(content), len(randomBytes3MiB))
667668
}
668669
return nil
669670
},

‎storage/writer.go‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,12 +107,25 @@ type Writer struct {
107107
// Append is a parameter to indicate whether the writer should use appendable
108108
// object semantics for the new object generation. Appendable objects are
109109
// visible on the first Write() call, and can be appended to until they are
110-
// finalized. The object is finalized on a call to Close().
110+
// finalized. If Writer.FinalizeOnClose is set to true, the object is finalized
111+
// when Writer.Close() is called; otherwise, the object is left unfinalized
112+
// and can be appended to later.
111113
//
112114
// Append is only supported for gRPC. This feature is in preview and is not
113115
// yet available for general use.
114116
Append bool
115117

118+
// FinalizeOnClose indicates whether the writer should finalize an object when
119+
// closing the write stream. This only applies to Writers where Append is
120+
// true, since append semantics allow a prefix of the object to be durable and
121+
// readable. By default, objects written with Append semantics will not be
122+
// finalized, which means they can be appended to later.
123+
//
124+
// FinalizeOnClose is supported only on gRPC clients where [Writer.Append] is
125+
// set to true. This feature is in preview and is not yet available for
126+
// general use.
127+
FinalizeOnClose bool
128+
116129
// ProgressFunc can be used to monitor the progress of a large write
117130
// operation. If ProgressFunc is not nil and writing requires multiple
118131
// calls to the underlying service (see
@@ -261,6 +274,7 @@ func (w *Writer) openWriter() (err error) {
261274
encryptionKey: w.o.encryptionKey,
262275
sendCRC32C: w.SendCRC32C,
263276
append: w.Append,
277+
finalizeOnClose: w.FinalizeOnClose,
264278
donec: w.donec,
265279
setError: w.error,
266280
progress: w.progress,
@@ -332,6 +346,9 @@ func (w *Writer) validateWriteAttrs() error {
332346
if w.ChunkSize < 0 {
333347
return errors.New("storage: Writer.ChunkSize must be non-negative")
334348
}
349+
if w.FinalizeOnClose && !w.Append {
350+
return errors.New("storage: Writer.FinalizeOnClose may only be true if Writer.Append is true")
351+
}
335352
return nil
336353
}
337354

0 commit comments

Comments
 (0)