diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go index dcdfd5292..f6c4cc278 100644 --- a/cmd/serve/s3/multipart.go +++ b/cmd/serve/s3/multipart.go @@ -385,6 +385,9 @@ func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName strin etag := fmt.Sprintf("%q", hex.EncodeToString(md5Sum)) if err := up.bufferPart(ctx, partNumber, n, md5Sum, rw); err != nil { + if up.isPoisoned() { + b.failUpload(uploadID, up) + } return "", err } return etag, nil @@ -832,6 +835,10 @@ func (up *multipartUpload) pump() error { if err != nil { up.mu.Lock() up.unbuffer(psize) + // The client was told this part succeeded so won't resend it, + // and the sink may hold some of it, so the upload can never + // complete. + up.poisoned = true up.mu.Unlock() return up.stopStreaming(err) } diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go index 2c0de72d2..8173b2393 100644 --- a/cmd/serve/s3/multipart_test.go +++ b/cmd/serve/s3/multipart_test.go @@ -1599,6 +1599,51 @@ func TestMultipartPartFailsAfterWholeBody(t *testing.T) { require.ErrorIs(t, err, gofakes3.ErrNoSuchUpload) } +// flakySink is a sink which fails the write which would take it past +// failAt bytes, taking part of it, and accepts everything after that. +type flakySink struct { + bufferSink + failAt int + failed bool +} + +func (s *flakySink) Write(p []byte) (int, error) { + if s.failed || s.buf.Len()+len(p) <= s.failAt { + return s.bufferSink.Write(p) + } + s.failed = true + n, _ := s.bufferSink.Write(p[:s.failAt-s.buf.Len()]) + return n, errBoom +} + +// TestMultipartPumpFailure checks that an upload is failed when a part +// which was buffered, and so already acknowledged to the client, can't be +// written to the sink, rather than a resend of it being appended to what +// the failed write left there. +func TestMultipartPumpFailure(t *testing.T) { + b, _, bucket := newPutTestBackend(t, "", nil) + ctx := context.Background() + const object = "pump-failure.bin" + + uploadID, _ := newBufferSinkUpload(t, b, bucket, object) + up, err := b.loadUpload(ctx, uploadID) + require.NoError(t, err) + part1 := []byte(random.String(1024)) + part2 := []byte(random.String(1024)) + up.fh = &flakySink{failAt: len(part1) + len(part2)/2} + + // Part 2 arrives first so is buffered, then fails part way into the + // sink when part 1 arrives and it is pumped. + _, err = b.UploadPart(ctx, bucket, object, uploadID, 2, int64(len(part2)), bytes.NewReader(part2)) + require.NoError(t, err) + _, err = b.UploadPart(ctx, bucket, object, uploadID, 1, int64(len(part1)), bytes.NewReader(part1)) + require.ErrorIs(t, err, errBoom) + + // The upload is gone rather than accepting part 2 again. + _, err = b.UploadPart(ctx, bucket, object, uploadID, 2, int64(len(part2)), bytes.NewReader(part2)) + require.ErrorIs(t, err, gofakes3.ErrNoSuchUpload) +} + // TestMultipartPartBadMD5 checks that a part whose body doesn't match the // Content-MD5 the client declared is rejected, and that the upload it // corrupted is failed rather than left for the client to retry.