serve s3: fix OOM and InvalidPart errors with concurrent multipart uploads - fixes #9616

Streamed multipart uploads had two problems when the client uploaded parts
concurrently:

Parts arriving ahead of the next part needed by the backend stream were buffered
in memory without limit, acknowledging each part as soon as it was received. A
client uploading faster than the backend could drain would therefore balloon the
server's memory to the size of the upload. Buffering is now bounded a new
--multipart-streaming-buffer-limit flag (default 256Mi, 0 for unlimited): a part
that would take the buffer over the limit is not read until the stream drains,
applying backpressure to the client instead of using unbounded memory.

A part uploaded again with the same number - typically a client retrying after
its request timed out - left a stale copy in the reorder buffer which made
CompleteMultipartUpload fail with InvalidPart, aborting the whole upload.
Re-uploaded parts are now handled properly: a copy still in the buffer is
replaced, an identical copy of an already streamed part is accepted as a no-op,
and only replacing an already streamed part with different content (which the
in-order stream cannot honour) is rejected.

cmd/serve/s3/multipart.go

@@ -12,6 +12,7 @@
 package s3
 
 import (
+	"bytes"
 	"context"
 	"crypto/md5"
 	"encoding/hex"
@@ -49,31 +50,37 @@ type multipartUpload struct {
 	pipeW *io.PipeWriter // parts are streamed here, in part-number order
 
 	mu        sync.Mutex
+	cond      *sync.Cond     // signalled when buffered shrinks, nextPart advances or the upload closes
 	partMD5s  map[int][]byte // raw MD5 sums per part (for the final S3 multipart ETag)
 	partSizes map[int]int64  // observed part sizes
 	closed    bool
 
-	putCancel context.CancelFunc // cancels the background PutStream
+	putCancel   context.CancelFunc // cancels the background PutStream
-	putDone   chan struct{}      // closed when the background PutStream returns
+	putDone     chan struct{}      // closed when the background PutStream returns
-	putErr    error              // PutStream result (read only after putDone is closed)
+	putErr      error              // PutStream result (read only after putDone is closed)
-	nextPart  int                // next part number to stream (1-based)
+	nextPart    int                // next part number to stream (1-based)
-	streamBuf map[int]*pool.RW   // parts received ahead of nextPart, awaiting their turn
+	streamBuf   map[int]*pool.RW   // parts received ahead of nextPart, awaiting their turn
-	pumping   bool               // a goroutine is currently writing to the pipe
+	pumping     bool               // a goroutine is currently writing to the pipe
+	buffered    int64              // bytes of parts admitted but not yet streamed or released
+	bufferLimit int64              // max buffered before parts ahead of nextPart must wait (<= 0 for no limit)
 }
 
 // newMultipartUpload allocates an upload struct.
-func newMultipartUpload(bucket, key, fp, streamFp string, meta map[string]string) *multipartUpload {
+func newMultipartUpload(bucket, key, fp, streamFp string, meta map[string]string, bufferLimit int64) *multipartUpload {
-	return &multipartUpload{
+	up := &multipartUpload{
-		bucket:    bucket,
+		bucket:      bucket,
-		key:       key,
+		key:         key,
-		fp:        fp,
+		fp:          fp,
-		streamFp:  streamFp,
+		streamFp:    streamFp,
-		meta:      meta,
+		meta:        meta,
-		partMD5s:  map[int][]byte{},
+		partMD5s:    map[int][]byte{},
-		partSizes: map[int]int64{},
+		partSizes:   map[int]int64{},
-		nextPart:  1,
+		nextPart:    1,
-		streamBuf: map[int]*pool.RW{},
+		streamBuf:   map[int]*pool.RW{},
+		bufferLimit: bufferLimit,
 	}
+	up.cond = sync.NewCond(&up.mu)
+	return up
 }
 
 // loadUpload looks up an in-flight upload by ID.
@@ -137,7 +144,7 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec
 		streamFp = path.Join(objectDir, multipartUploadPrefix+string(uploadID))
 	}
 
-	up := newMultipartUpload(bucketName, objectName, fp, streamFp, meta)
+	up := newMultipartUpload(bucketName, objectName, fp, streamFp, meta, int64(b.s.opt.MultipartStreamingBufferLimit))
 
 	src := object.NewStaticObjectInfo(streamFp, time.Now(), -1, true, nil, f)
 	pr, pw := io.Pipe()
@@ -180,6 +187,12 @@ func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName strin
 		return "", err
 	}
 
+	// Wait until there is room to buffer this part, bounding the memory a
+	// client which uploads faster than the backend drains can consume.
+	if err := up.waitForTurn(partNumber, contentLength); err != nil {
+		return "", err
+	}
+
 	// Buffer the part in a pool-backed RW so we can MD5 it (for the ETag) and
 	// stream it once it is this part's turn.
 	rw := multipart.NewRW().Reserve(contentLength)
@@ -187,10 +200,12 @@ func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName strin
 	n, err := io.Copy(rw, io.TeeReader(body, hasher))
 	if err != nil {
 		_ = rw.Close()
+		up.release(contentLength)
 		return "", err
 	}
 	if n != contentLength {
 		_ = rw.Close()
+		up.release(contentLength)
 		return "", gofakes3.ErrIncompleteBody
 	}
 	md5Sum := hasher.Sum(nil)
@@ -202,15 +217,73 @@ func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName strin
 	return etag, nil
 }
 
+// waitForTurn blocks until size bytes can be admitted to the reorder buffer,
+// then reserves them, bounding the memory an upload can consume when the
+// client sends parts faster than the backend drains them.
+//
+// The next part the stream needs (and any retry of an earlier one) is always
+// admitted so the pipe can keep draining; so is a single part bigger than the
+// limit when the buffer is empty, to guarantee progress. Reserved bytes are
+// returned with release, or by the pump as the part is streamed.
+func (up *multipartUpload) waitForTurn(partNumber int, size int64) error {
+	up.mu.Lock()
+	defer up.mu.Unlock()
+	for {
+		if up.closed {
+			return gofakes3.ErrNoSuchUpload
+		}
+		if up.bufferLimit <= 0 || partNumber <= up.nextPart || up.buffered == 0 || up.buffered+size <= up.bufferLimit {
+			up.buffered += size
+			return nil
+		}
+		up.cond.Wait()
+	}
+}
+
+// release returns size bytes reserved by waitForTurn to the reorder buffer
+// budget and wakes any parts waiting for room.
+func (up *multipartUpload) release(size int64) {
+	up.mu.Lock()
+	up.buffered -= size
+	up.cond.Broadcast()
+	up.mu.Unlock()
+}
+
 // streamPart records a part and streams the parts into the pipe in order.
 //
 // Parts must be uploaded in ascending, contiguous part-number order. A part
 // that arrives ahead of the next expected one is buffered until its turn; the
 // parts are then pumped into the pipe in order. Whichever goroutine finds the
 // next part available does the pumping, so concurrent (but in-order) clients
-// are tolerated with buffering bounded by how far ahead they run.
+// are tolerated, with the buffering bounded by waitForTurn.
+//
+// A part number may be uploaded more than once - typically a client retrying
+// after its request timed out, but real S3 also allows replacing a part. If
+// the earlier copy is still buffered it is replaced (last write wins). If it
+// has already been streamed it can't be replaced: an identical re-upload is
+// accepted idempotently (the data is already in the stream) and a different
+// one is rejected.
 func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte, rw *pool.RW) error {
 	up.mu.Lock()
+	if oldMD5, exists := up.partMD5s[partNumber]; exists {
+		if old, buffered := up.streamBuf[partNumber]; buffered {
+			_ = old.Close()
+			up.buffered -= up.partSizes[partNumber]
+			up.cond.Broadcast()
+		} else {
+			// Already streamed (or streaming right now): the stream can't be
+			// rewritten, so accept an identical part and reject the rest.
+			same := bytes.Equal(md5Sum, oldMD5) && size == up.partSizes[partNumber]
+			up.buffered -= size
+			up.cond.Broadcast()
+			up.mu.Unlock()
+			_ = rw.Close()
+			if !same {
+				return gofakes3.ErrorMessagef(gofakes3.ErrNotImplemented, "part %d has already been streamed to the backend and cannot be replaced with different contents", partNumber)
+			}
+			return nil
+		}
+	}
 	up.partMD5s[partNumber] = md5Sum
 	up.partSizes[partNumber] = size
 	up.streamBuf[partNumber] = rw
@@ -230,6 +303,7 @@ func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte,
 			return nil
 		}
 		delete(up.streamBuf, up.nextPart)
+		psize := prw.Size()
 		up.mu.Unlock()
 
 		err := pipePart(up.pipeW, prw)
@@ -237,12 +311,16 @@ func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte,
 		if err != nil {
 			up.mu.Lock()
 			up.pumping = false
+			up.buffered -= psize
+			up.cond.Broadcast()
 			up.mu.Unlock()
 			return err
 		}
 
 		up.mu.Lock()
 		up.nextPart++
+		up.buffered -= psize
+		up.cond.Broadcast()
 	}
 }
 
@@ -400,6 +478,7 @@ func (up *multipartUpload) close(ctx context.Context) error {
 		return nil
 	}
 	up.closed = true
+	up.cond.Broadcast()
 	up.mu.Unlock()
 
 	err := up.pipeW.Close()
@@ -425,6 +504,7 @@ func (up *multipartUpload) abort(ctx context.Context) error {
 	up.closed = true
 	streamBuf := up.streamBuf
 	up.streamBuf = nil
+	up.cond.Broadcast()
 	up.mu.Unlock()
 
 	for _, rw := range streamBuf {

cmd/serve/s3/multipart_test.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
@@ -43,6 +44,12 @@ func newMultipartTestServer(t *testing.T, disableStreaming bool) (*minio.Core, f
 // destination path is exercised as well as the temporary-object path that
 // local (PartialUploads=true) uses.
 func newMultipartTestServerBacking(t *testing.T, backing string, disableStreaming bool) (*minio.Core, fs.Fs, string) {
+	return newMultipartTestServerOpt(t, backing, disableStreaming, nil)
+}
+
+// newMultipartTestServerOpt is like newMultipartTestServerBacking but also
+// applies tweak (if non-nil) to the server Options before starting it.
+func newMultipartTestServerOpt(t *testing.T, backing string, disableStreaming bool, tweak func(*Options)) (*minio.Core, fs.Fs, string) {
 	fstest.Initialise()
 	ctx := context.Background()
 	if backing == "" {
@@ -67,6 +74,9 @@ func newMultipartTestServerBacking(t *testing.T, backing string, disableStreamin
 	opt.DisableMultipartStreaming = disableStreaming
 	opt.AuthKey = []string{fmt.Sprintf("%s,%s", keyid, keysec)}
 	opt.HTTP.ListenAddr = []string{endpoint}
+	if tweak != nil {
+		tweak(&opt)
+	}
 	w, err := newServer(ctx, f, &opt, &vfscommon.Opt, &proxy.Opt)
 	require.NoError(t, err)
 	go func() { _ = w.Serve() }()
@@ -279,6 +289,165 @@ func TestMultipartAbortPreservesExisting(t *testing.T) {
 	}
 }
 
+// TestMultipartRetryAfterStreamed checks that a part which is uploaded again
+// after its first copy has already been streamed to the backend - as a client
+// whose HTTP request timed out will do when it retries - is accepted
+// idempotently and doesn't fail the CompleteMultipartUpload.
+func TestMultipartRetryAfterStreamed(t *testing.T) {
+	core, f, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+	const object = "retry.bin"
+
+	part1 := []byte(random.String(60 * 1024))
+	part2 := []byte(random.String(40 * 1024))
+	want := append(append([]byte(nil), part1...), part2...)
+
+	uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+	require.NoError(t, err)
+
+	// Parts uploaded in order are streamed synchronously, so by the time
+	// PutObjectPart returns the part is already in the backend stream.
+	p1, err := core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(part1), int64(len(part1)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+	p2, err := core.PutObjectPart(ctx, bucket, object, uploadID, 2, bytes.NewReader(part2), int64(len(part2)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+
+	// Retry part 1 with identical content, as a client retrying after a
+	// timeout does.
+	p1retry, err := core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(part1), int64(len(part1)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, p1.ETag, p1retry.ETag)
+
+	_, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, []minio.CompletePart{
+		{PartNumber: 1, ETag: p1.ETag},
+		{PartNumber: 2, ETag: p2.ETag},
+	}, minio.PutObjectOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, want, readObject(t, f, bucket, object))
+}
+
+// TestMultipartRetryDifferentContent checks that re-uploading a part with
+// different content after the first copy has been streamed is rejected - the
+// in-order stream can't replace data already sent to the backend.
+func TestMultipartRetryDifferentContent(t *testing.T) {
+	core, _, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+	const object = "retry-different.bin"
+
+	uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+	require.NoError(t, err)
+
+	part1 := []byte(random.String(60 * 1024))
+	_, err = core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(part1), int64(len(part1)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+
+	other := []byte(random.String(60 * 1024))
+	_, err = core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(other), int64(len(other)), minio.PutObjectPartOptions{})
+	require.Error(t, err)
+
+	require.NoError(t, core.AbortMultipartUpload(ctx, bucket, object, uploadID))
+}
+
+// TestMultipartReplaceBuffered checks that re-uploading a part which has been
+// received but not yet streamed (it is waiting for an earlier part) replaces
+// the buffered copy, matching S3's last-write-wins semantics.
+func TestMultipartReplaceBuffered(t *testing.T) {
+	core, f, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+	const object = "replace-buffered.bin"
+
+	uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+	require.NoError(t, err)
+
+	// Part 2 arrives first so it is buffered awaiting part 1.
+	old := []byte(random.String(40 * 1024))
+	_, err = core.PutObjectPart(ctx, bucket, object, uploadID, 2, bytes.NewReader(old), int64(len(old)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+
+	// Upload part 2 again with different content - the buffered copy must be
+	// replaced.
+	part2 := []byte(random.String(40 * 1024))
+	p2, err := core.PutObjectPart(ctx, bucket, object, uploadID, 2, bytes.NewReader(part2), int64(len(part2)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+
+	part1 := []byte(random.String(60 * 1024))
+	p1, err := core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(part1), int64(len(part1)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+
+	_, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, []minio.CompletePart{
+		{PartNumber: 1, ETag: p1.ETag},
+		{PartNumber: 2, ETag: p2.ETag},
+	}, minio.PutObjectOptions{})
+	require.NoError(t, err)
+	want := append(append([]byte(nil), part1...), part2...)
+	assert.Equal(t, want, readObject(t, f, bucket, object))
+}
+
+// TestMultipartBufferLimit checks that a part arriving ahead of its turn
+// blocks once the reorder buffer limit is reached, and is admitted once the
+// missing part arrives and the buffer drains.
+func TestMultipartBufferLimit(t *testing.T) {
+	core, f, bucket := newMultipartTestServerOpt(t, "", false, func(opt *Options) {
+		opt.MultipartStreamingBufferLimit = 50 * 1024
+	})
+	ctx := context.Background()
+	const object = "buffer-limit.bin"
+
+	sizes := []int{40 * 1024, 40 * 1024, 40 * 1024}
+	datas := make([][]byte, len(sizes))
+	var want []byte
+	for i, sz := range sizes {
+		datas[i] = []byte(random.String(sz))
+		want = append(want, datas[i]...)
+	}
+
+	uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+	require.NoError(t, err)
+
+	// Part 3 arrives first: the buffer is empty so it is admitted even though
+	// it must wait for its turn.
+	p3, err := core.PutObjectPart(ctx, bucket, object, uploadID, 3, bytes.NewReader(datas[2]), int64(sizes[2]), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+
+	// Part 2 would take the buffer over the limit, so it must block until
+	// part 1 arrives and the stream drains.
+	type partResult struct {
+		part minio.ObjectPart
+		err  error
+	}
+	done := make(chan partResult, 1)
+	go func() {
+		p, err := core.PutObjectPart(ctx, bucket, object, uploadID, 2, bytes.NewReader(datas[1]), int64(sizes[1]), minio.PutObjectPartOptions{})
+		done <- partResult{p, err}
+	}()
+	select {
+	case r := <-done:
+		t.Fatalf("part 2 was not blocked by the buffer limit (err=%v)", r.err)
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	// Part 1 is the part the stream needs so it is admitted regardless of the
+	// limit; streaming it frees the buffer and unblocks part 2.
+	p1, err := core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(datas[0]), int64(sizes[0]), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+	var p2 minio.ObjectPart
+	select {
+	case r := <-done:
+		require.NoError(t, r.err)
+		p2 = r.part
+	case <-time.After(10 * time.Second):
+		t.Fatal("part 2 was never unblocked")
+	}
+
+	_, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, []minio.CompletePart{
+		{PartNumber: 1, ETag: p1.ETag},
+		{PartNumber: 2, ETag: p2.ETag},
+		{PartNumber: 3, ETag: p3.ETag},
+	}, minio.PutObjectOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, want, readObject(t, f, bucket, object))
+}
+
 // TestMultipartOverwrite checks that a completed multipart upload atomically
 // replaces an existing object of the same name.
 func TestMultipartOverwrite(t *testing.T) {

cmd/serve/s3/s3.go

@@ -41,6 +41,10 @@ var OptionsInfo = fs.Options{{
 	Name:    "disable_multipart_streaming",
 	Default: false,
 	Help:    "Buffer multipart uploads in memory instead of streaming them to the backend (see the Multipart uploads docs section)",
+}, {
+	Name:    "multipart_streaming_buffer_limit",
+	Default: fs.SizeSuffix(256 * 1024 * 1024),
+	Help:    "Maximum memory buffered per streamed multipart upload for parts arriving out of order, 0 for unlimited (see the Multipart uploads docs section)",
 }}.
 	Add(httplib.ConfigInfo).
 	Add(httplib.AuthConfigInfo)
@@ -48,13 +52,14 @@ var OptionsInfo = fs.Options{{
 // Options contains options for the s3 Server
 type Options struct {
 	//TODO add more options
-	ForcePathStyle            bool     `config:"force_path_style"`
+	ForcePathStyle                bool          `config:"force_path_style"`
-	EtagHash                  string   `config:"etag_hash"`
+	EtagHash                      string        `config:"etag_hash"`
-	AuthKey                   []string `config:"auth_key"`
+	AuthKey                       []string      `config:"auth_key"`
-	NoCleanup                 bool     `config:"no_cleanup"`
+	NoCleanup                     bool          `config:"no_cleanup"`
-	DisableMultipartStreaming bool     `config:"disable_multipart_streaming"`
+	DisableMultipartStreaming     bool          `config:"disable_multipart_streaming"`
-	Auth                      httplib.AuthConfig
+	MultipartStreamingBufferLimit fs.SizeSuffix `config:"multipart_streaming_buffer_limit"`
-	HTTP                      httplib.Config
+	Auth                          httplib.AuthConfig
+	HTTP                          httplib.Config
 }
 
 // Opt is options set by command line flags

cmd/serve/s3/serve_s3.md

@@ -129,14 +129,25 @@ also need to support a server-side move or copy.
 
 - Parts must arrive in ascending, contiguous part-number order
   (1, 2, 3, ...). Parts the client uploads concurrently or out of order
-  are buffered until their turn, so higher client upload concurrency
+  are buffered until their turn. The memory used for this buffering is
-  uses more memory; non-contiguous part numbers are rejected. Configure
+  capped, per upload, by `--multipart-streaming-buffer-limit` (default
-  the client to upload in part order, ideally with low concurrency, for
+  `256M`, `0` for no limit): a part that would take the buffer over the
-  the lowest memory use.
+  limit is stalled until the stream drains, so a client that uploads
-- No per-part retry. Once a part has been streamed it is committed, so a
+  faster than the remote can accept sees backpressure rather than
-  failure partway through aborts the whole upload and the client must
+  unbounded server memory use. Since a stalled part holds its HTTP
-  start it again, rather than retrying a single part. (The remote's own
+  request open, clients whose upload concurrency times chunk size
-  upload still retries its internal chunks.)
+  exceeds the limit may need a longer read timeout when the remote is
+  slow. Non-contiguous part numbers are rejected on completion.
+  Configure the client to upload in part order, ideally with low
+  concurrency, for the lowest memory use.
+- A part uploaded again before completion - typically a client retrying
+  after a timeout - is accepted: if the earlier copy is still buffered
+  it is replaced, and if it has already been streamed an identical
+  re-upload is a no-op. What isn't possible is replacing a part that has
+  already been streamed with *different* content - that is rejected. A
+  failure in the stream to the remote itself still aborts the whole
+  upload and the client must start it again. (The remote's own upload
+  still retries its internal chunks.)
 - Parts are serialised into one stream, so ingest from the client is
   effectively single-threaded, although the remote's own upload still
   runs concurrently.