serve s3: fix multipart upload corruption if the backend write of an early part fails

A multipart part which arrives before the parts ahead of it is buffered
and acknowledged to the client, then written to the backend when its
turn comes. If that write failed the part was dropped, possibly with
some of it already written, but the upload carried on as if nothing
had happened, so it could be completed with the part missing, or with
a resend of the part appended to the partial data.

Such an upload is now failed, so the client starts it again.

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)
 		}

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.