Skip to content

Commit 9ad2100

Browse files
tritonetritone
authored
fix(storage): persist resource in append retry (#13128)
In one retry case for appendable writes, the client receives both an object resource w/ write handle as well as an error from the server. After reconnect the new stream would not keep the original object resource. This change ensures that the object is not dropped. Also explicitly guarantee the contract that either w.Attrs() will be non-nil, or w.Close() will return a non-nil error. I need to make a change to the emulator as well so that the testbench behavior more closely mocks the service.
1 parent bda509d commit 9ad2100

4 files changed

Lines changed: 107 additions & 12 deletions

File tree

‎storage/client_test.go‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1327,6 +1327,63 @@ func TestWriterFlushAtCloseEmulated(t *testing.T) {
13271327
})
13281328
}
13291329

1330+
func TestWriterRetryAttrsEmulated(t *testing.T) {
1331+
transportClientTest(skipHTTP("appends only supported via gRPC"), t, func(t *testing.T, ctx context.Context, project, bucket string, client storageClient) {
1332+
// Populate test data.
1333+
_, err := client.CreateBucket(ctx, project, bucket, &BucketAttrs{
1334+
Name: bucket,
1335+
}, nil)
1336+
if err != nil {
1337+
t.Fatalf("client.CreateBucket: %v", err)
1338+
}
1339+
prefix := time.Now().Nanosecond()
1340+
objName := fmt.Sprintf("%d-object-%d", prefix, time.Now().Nanosecond())
1341+
data := generateRandomBytes(20 * MiB)
1342+
1343+
vc := &Client{tc: client}
1344+
1345+
// Setup retry test.
1346+
instructions := map[string][]string{"storage.objects.insert": {"return-503-after-4097K"}}
1347+
testID := createRetryTest(t, client, instructions)
1348+
ctx = callctx.SetHeaders(ctx, "x-retry-test-id", testID)
1349+
1350+
w := vc.Bucket(bucket).Object(objName).If(Conditions{DoesNotExist: true}).NewWriter(ctx)
1351+
w.Append = true
1352+
w.ChunkSize = 8 * MiB
1353+
1354+
if _, err := w.Flush(); err != nil {
1355+
t.Fatalf("flush at 0b: %v", err)
1356+
}
1357+
1358+
if _, err := w.Write(data); err != nil {
1359+
t.Fatalf("writing data: got %v; want ok", err)
1360+
}
1361+
1362+
if err := w.Close(); err != nil {
1363+
t.Fatalf("closing writer: %v", err)
1364+
}
1365+
1366+
if gotAttrs := w.Attrs(); gotAttrs == nil || gotAttrs.Name != objName {
1367+
t.Fatalf("w.Attrs(): got %v, want attrs for object %v", gotAttrs, objName)
1368+
}
1369+
1370+
// Download object and check data
1371+
r, err := veneerClient.Bucket(bucket).Object(objName).NewReader(ctx)
1372+
defer r.Close()
1373+
if err != nil {
1374+
t.Fatalf("opening reading: %v", err)
1375+
}
1376+
wantLen := len(data)
1377+
got, err := io.ReadAll(r)
1378+
if n := len(got); n != wantLen {
1379+
t.Fatalf("expected to read %d bytes, but got %d (%v)", wantLen, n, err)
1380+
}
1381+
if !bytes.Equal(got, data) {
1382+
t.Fatalf("got data did not match uploaded data")
1383+
}
1384+
})
1385+
}
1386+
13301387
// Tests small flush (under 512 bytes) to verify that logic avoiding
13311388
// content type sniffing works as expected in this case.
13321389
func TestWriterSmallFlushEmulated(t *testing.T) {

‎storage/grpc_writer.go‎

Lines changed: 25 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -587,9 +587,11 @@ func (s *gRPCResumableBidiWriteBufferSender) sendBuffer(ctx context.Context, buf
587587
// uploadBuffer uploads the buffer at the given offset using a bi-directional
588588
// Write stream. It will open a new stream if necessary (on the first call or
589589
// after resuming from failure) and chunk the buffer per maxPerMessageWriteSize.
590-
// The final Object is returned on success if doneReading is true.
590+
// The object resource will be returned as well when it's available from the
591+
// server (on finalization or on object creation in an appendable write).
591592
//
592-
// Returns object and any error that is not retriable.
593+
// Returns object and any error that is not retriable. Both object and non-nil
594+
// error may be returned in some cases.
593595
func (w *gRPCWriter) uploadBuffer(ctx context.Context, recvd int, start int64, doneReading bool) (obj *storagepb.Object, err error) {
594596
if w.streamSender == nil {
595597
if w.append {
@@ -620,9 +622,14 @@ func (w *gRPCWriter) uploadBuffer(ctx context.Context, recvd int, start int64, d
620622
l = len(data)
621623
flush = true
622624
}
623-
obj, err = w.streamSender.sendBuffer(ctx, data[:l], offset, flush, flush && doneReading)
625+
var recvObj *storagepb.Object
626+
recvObj, err = w.streamSender.sendBuffer(ctx, data[:l], offset, flush, flush && doneReading)
627+
if recvObj != nil {
628+
obj = recvObj
629+
}
624630
if err != nil {
625-
return nil, err
631+
// Note, an object resource may also be returned in case of non-nil error.
632+
return
626633
}
627634
data = data[l:]
628635
offset += int64(l)
@@ -922,6 +929,9 @@ func (s *gRPCAppendBidiWriteBufferSender) receiveMessages(resps chan<- *storagep
922929
close(resps)
923930
}
924931

932+
// Send contents of the buffer to the stream.
933+
// Returns object resource (if available) and error. Both object and error
934+
// may be non-nil in some cases.
925935
func (s *gRPCAppendBidiWriteBufferSender) sendOnConnectedStream(buf []byte, offset int64, flush, finishWrite, sendFirstMessage bool) (obj *storagepb.Object, err error) {
926936
var req *storagepb.BidiWriteObjectRequest
927937
finalizeObject := finishWrite && s.finalizeOnClose
@@ -957,7 +967,8 @@ func (s *gRPCAppendBidiWriteBufferSender) sendOnConnectedStream(buf []byte, offs
957967
}
958968
}
959969
if s.recvErr != io.EOF {
960-
return nil, s.recvErr
970+
err = s.recvErr
971+
return
961972
}
962973
if obj.GetSize() > s.flushOffset {
963974
s.flushOffset = obj.GetSize()
@@ -967,14 +978,16 @@ func (s *gRPCAppendBidiWriteBufferSender) sendOnConnectedStream(buf []byte, offs
967978
}
968979

969980
if flush {
970-
// We don't necessarily expect multiple responses for a single flush, but
971-
// this allows the server to send multiple responses if it wants to.
981+
// We may receive multiple responses for a single flush. In particular
982+
// if only part of the data was persisted, we may get a success response
983+
// (persisted size or object resource) followed by an error.
972984
flushOffset := s.flushOffset
973985

974986
for flushOffset < offset+int64(len(buf)) {
975987
resp, ok := <-s.recvs
976988
if !ok {
977-
return nil, s.recvErr
989+
err = s.recvErr
990+
return
978991
}
979992
pSize := resp.GetPersistedSize()
980993
rSize := resp.GetResource().GetSize()
@@ -1008,8 +1021,10 @@ func (s *gRPCAppendBidiWriteBufferSender) sendBuffer(ctx context.Context, buf []
10081021
}
10091022
}
10101023

1011-
obj, err = s.sendOnConnectedStream(buf, offset, flush, finishWrite, sendFirstMessage)
1012-
if obj != nil {
1024+
var recvObj *storagepb.Object
1025+
recvObj, err = s.sendOnConnectedStream(buf, offset, flush, finishWrite, sendFirstMessage)
1026+
if recvObj != nil {
1027+
obj = recvObj
10131028
s.objResource = obj
10141029
}
10151030
if err == nil {

‎storage/retry_conformance_test.go‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -656,6 +656,10 @@ var methods = map[string][]retryFunc{
656656
return fmt.Errorf("Writer.Close: %v", err)
657657
}
658658

659+
if objW.Attrs() == nil {
660+
return fmt.Errorf("Writer.Attrs: expected attrs for written object, got nil")
661+
}
662+
659663
// Don't reuse obj, in case preconditions were set on the write request.
660664
r, err := b.Object(obj.ObjectName()).NewReader(ctx)
661665
defer r.Close()
@@ -703,10 +707,19 @@ var methods = map[string][]retryFunc{
703707

704708
}
705709

710+
// TODO: Remove this exception and enable the attrs check below once we
711+
// figure out how to handle the redirect w/ write handle case.
712+
// See b/451594633
706713
if err := objW.Close(); err != nil {
707-
return fmt.Errorf("Writer.Close: %w", err)
714+
if !strings.Contains(err.Error(), "no object attributes returned") {
715+
return fmt.Errorf("Writer.Close: %w", err)
716+
}
708717
}
709718

719+
// if objW.Attrs() == nil {
720+
// return fmt.Errorf("Writer.Attrs: expected attrs for written object, got nil")
721+
// }
722+
710723
// Don't reuse obj, in case preconditions were set on the write request.
711724
r, err := b.Object(obj.ObjectName()).NewReader(ctx)
712725
defer r.Close()

‎storage/writer.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,12 @@ func (w *Writer) Close() error {
261261
w.closed = true
262262
w.mu.Lock()
263263
defer w.mu.Unlock()
264+
// In rare cases for gRPC, the write may succeed but the server does not
265+
// return a full object resource. This should cause an error to keep the
266+
// contract that either w.Attrs or the error should be non-nil.
267+
if w.obj == nil && w.err == nil {
268+
w.err = errors.New("storage: write succeeded but no object attributes returned from the server")
269+
}
264270
trace.EndSpan(w.ctx, w.err)
265271
return w.err
266272
}
@@ -291,7 +297,11 @@ func (w *Writer) openWriter() (err error) {
291297
donec: w.donec,
292298
setError: w.error,
293299
progress: w.progress,
294-
setObj: func(o *ObjectAttrs) { w.obj = o },
300+
setObj: func(o *ObjectAttrs) {
301+
if o != nil {
302+
w.obj = o
303+
}
304+
},
295305
setSize: func(n int64) {
296306
if w.obj != nil {
297307
w.obj.Size = n

0 commit comments

Comments
 (0)